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
189
190
191
192
193
194
195
|
# Copyright (C) 2019 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; version 3 of the License or
# (at your option) a 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 foreign branch support.
Currently only tells the user that Subversion is not supported.
"""
from ... import version_info # noqa: F401
from ... import (
controldir,
errors,
transport as _mod_transport,
)
class SubversionUnsupportedError(errors.UnsupportedFormatError):
_fmt = ('Subversion branches are not yet supported. '
'To convert Subversion branches to Bazaar branches or vice versa, '
'use the fastimport format.')
class SvnWorkingTreeDirFormat(controldir.ControlDirFormat):
"""Subversion directory format."""
def get_converter(self):
raise NotImplementedError(self.get_converter)
def get_format_description(self):
return "Subversion working directory"
def initialize_on_transport(self, transport):
raise errors.UninitializableFormat(self)
def is_supported(self):
return False
def supports_transport(self, transport):
return False
def check_support_status(self, allow_unsupported, recommend_upgrade=True,
basedir=None):
raise SubversionUnsupportedError()
def open(self, transport):
# Raise NotBranchError if there is nothing there
SvnWorkingTreeProber().probe_transport(transport)
raise NotImplementedError(self.open)
class SvnWorkingTreeProber(controldir.Prober):
@classmethod
def priority(klass, transport):
return 100
def probe_transport(self, transport):
try:
transport.local_abspath('.')
except errors.NotLocalUrl:
raise errors.NotBranchError(path=transport.base)
else:
if transport.has(".svn"):
return SvnWorkingTreeDirFormat()
raise errors.NotBranchError(path=transport.base)
@classmethod
def known_formats(cls):
return [SvnWorkingTreeDirFormat()]
class SvnRepositoryFormat(controldir.ControlDirFormat):
"""Subversion directory format."""
def get_converter(self):
raise NotImplementedError(self.get_converter)
def get_format_description(self):
return "Subversion repository"
def initialize_on_transport(self, transport):
raise errors.UninitializableFormat(self)
def is_supported(self):
return False
def supports_transport(self, transport):
return False
def check_support_status(self, allow_unsupported, recommend_upgrade=True,
basedir=None):
raise SubversionUnsupportedError()
def open(self, transport):
# Raise NotBranchError if there is nothing there
SvnRepositoryProber().probe_transport(transport)
raise NotImplementedError(self.open)
class SvnRepositoryProber(controldir.Prober):
_supported_schemes = ["http", "https", "file", "svn"]
@classmethod
def priority(klass, transport):
if 'svn' in transport.base:
return 90
return 100
def probe_transport(self, transport):
try:
url = transport.external_url()
except errors.InProcessTransport:
raise errors.NotBranchError(path=transport.base)
scheme = url.split(":")[0]
if scheme.startswith("svn+") or scheme == "svn":
raise SubversionUnsupportedError()
if scheme not in self._supported_schemes:
raise errors.NotBranchError(path=transport.base)
if scheme == 'file':
# Cheaper way to figure out if there is a svn repo
maybe = False
subtransport = transport
while subtransport:
try:
if all([subtransport.has(name)
for name in ["format", "db", "conf"]]):
maybe = True
break
except UnicodeEncodeError:
pass
prevsubtransport = subtransport
subtransport = prevsubtransport.clone("..")
if subtransport.base == prevsubtransport.base:
break
if not maybe:
raise errors.NotBranchError(path=transport.base)
# If this is a HTTP transport, use the existing connection to check
# that the remote end supports version control.
if scheme in ("http", "https"):
priv_transport = getattr(transport, "_decorated", transport)
try:
headers = priv_transport._options('.')
except (errors.InProcessTransport, errors.NoSuchFile,
errors.InvalidHttpResponse):
raise errors.NotBranchError(path=transport.base)
else:
dav_entries = set()
for key, value in headers:
if key.upper() == 'DAV':
dav_entries.update(
[x.strip() for x in value.split(',')])
if "version-control" not in dav_entries:
raise errors.NotBranchError(path=transport.base)
return SvnRepositoryFormat()
@classmethod
def known_formats(cls):
return [SvnRepositoryFormat()]
controldir.ControlDirFormat.register_prober(SvnWorkingTreeProber)
controldir.ControlDirFormat.register_prober(SvnRepositoryProber)
_mod_transport.register_transport_proto(
'svn+ssh://',
help="Access using the Subversion smart server tunneled over SSH.")
_mod_transport.register_transport_proto(
'svn+http://')
_mod_transport.register_transport_proto(
'svn+https://')
_mod_transport.register_transport_proto(
'svn://',
help="Access using the Subversion smart server.")
|