1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
# Copyright (C) 2006-2007 Jelmer Vernooij <jelmer@samba.org>
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Subversion BzrDir formats."""
from bzrlib import urlutils
from bzrlib.bzrdir import BzrDirFormat, BzrDir, format_registry
from bzrlib.errors import (NotBranchError, NotLocalUrl, NoRepositoryPresent,
NoWorkingTree, AlreadyBranchError)
from bzrlib.transport.local import LocalTransport
from svn.core import SubversionException
import svn.core, svn.repos
from errors import NoSvnRepositoryPresent
from format import get_rich_root_format, SvnRemoteFormat
from repository import SvnRepository
from transport import SvnRaTransport, bzr_to_svn_url, get_svn_ra_transport
class SvnRemoteAccess(BzrDir):
"""BzrDir implementation for Subversion connections.
This is used for all non-checkout connections
to Subversion repositories.
"""
def __init__(self, _transport, _format=None):
"""See BzrDir.__init__()."""
_transport = get_svn_ra_transport(_transport)
if _format is None:
_format = SvnRemoteFormat()
self._format = _format
self.transport = None
self.root_transport = _transport
svn_url = bzr_to_svn_url(self.root_transport.base)
self.svn_root_url = _transport.get_svn_repos_root()
self.root_url = _transport.get_repos_root()
assert svn_url.startswith(self.svn_root_url)
self.branch_path = svn_url[len(self.svn_root_url):]
def clone(self, url, revision_id=None, force_new_repo=False):
"""See BzrDir.clone().
Not supported on Subversion connections.
"""
raise NotImplementedError(SvnRemoteAccess.clone)
def sprout(self, url, revision_id=None, force_new_repo=False,
recurse='down', possible_transports=None, accelerator_tree=None,
hardlink=False):
"""See BzrDir.sprout()."""
# FIXME: Use possible_transports
# FIXME: Use recurse
format = get_rich_root_format()
result = format.initialize(url)
repo = self.find_repository()
if force_new_repo:
result_repo = repo.clone(result, revision_id)
else:
try:
result_repo = result.find_repository()
result_repo.fetch(repo, revision_id=revision_id)
except NoRepositoryPresent:
result_repo = repo.clone(result, revision_id)
branch = self.open_branch()
result_branch = branch.sprout(result, revision_id)
if result_branch.repository.make_working_trees():
result.create_workingtree(hardlink=hardlink)
return result
def open_repository(self, _unsupported=False):
"""Open the repository associated with this BzrDir.
:return: instance of SvnRepository.
"""
if self.branch_path == "":
return SvnRepository(self, self.root_transport)
raise NoSvnRepositoryPresent(self.root_transport.base)
def find_repository(self):
"""Open the repository associated with this BzrDir.
:return: instance of SvnRepository.
"""
transport = self.root_transport
if self.root_url != transport.base:
transport = transport.clone_root()
return SvnRepository(self, transport, self.branch_path)
def open_workingtree(self, _unsupported=False,
recommend_upgrade=True):
"""See BzrDir.open_workingtree().
Will always raise NotLocalUrl as this
BzrDir can not be associated with working trees.
"""
# Working trees never exist on remote Subversion repositories
raise NoWorkingTree(self.root_transport.base)
def create_workingtree(self, revision_id=None, hardlink=None):
"""See BzrDir.create_workingtree().
Will always raise NotLocalUrl as this
BzrDir can not be associated with working trees.
"""
raise NotLocalUrl(self.root_transport.base)
def needs_format_conversion(self, format=None):
"""See BzrDir.needs_format_conversion()."""
# if the format is not the same as the system default,
# an upgrade is needed.
if format is None:
format = BzrDirFormat.get_default_format()
return not isinstance(self._format, format.__class__)
def import_branch(self, source, stop_revision=None):
"""Create a new branch in this repository, possibly
with the specified history, optionally importing revisions.
:param source: Source branch
:param stop_revision: Tip of new branch
:return: Branch object
"""
from commit import push_new
if stop_revision is None:
stop_revision = source.last_revision()
target_branch_path = self.branch_path.strip("/")
repos = self.find_repository()
repos.lock_write()
try:
full_branch_url = urlutils.join(repos.transport.base,
target_branch_path)
if repos.transport.check_path(target_branch_path,
repos.get_latest_revnum()) != svn.core.svn_node_none:
raise AlreadyBranchError(full_branch_url)
push_new(repos, target_branch_path, source, stop_revision)
finally:
repos.unlock()
branch = self.open_branch()
branch.lock_write()
try:
branch.pull(source, stop_revision=stop_revision)
finally:
branch.unlock()
return branch
def create_branch(self):
"""See BzrDir.create_branch()."""
from branch import SvnBranch
repos = self.find_repository()
if self.branch_path != "":
# TODO: Set NULL_REVISION in SVN_PROP_BZR_BRANCHING_SCHEME
repos.transport.mkdir(self.branch_path.strip("/"))
elif repos.get_latest_revnum() > 0:
# Bail out if there are already revisions in this repository
raise AlreadyBranchError(self.root_transport.base)
branch = SvnBranch(self.root_transport.base, repos, self.branch_path)
branch.bzrdir = self
return branch
def open_branch(self, unsupported=True):
"""See BzrDir.open_branch()."""
from branch import SvnBranch
repos = self.find_repository()
branch = SvnBranch(self.root_transport.base, repos, self.branch_path)
branch.bzrdir = self
return branch
def create_repository(self, shared=False, format=None):
"""See BzrDir.create_repository."""
return self.open_repository()
|