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
|
# Copyright (C) 2019 Jelmer Vernooij <jelmer@jelmer.uk>
#
# 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
"""Mercurial foreign branch support.
Currently only tells the user that Mercurial is not supported.
"""
from ... import (
controldir,
errors,
)
from ... import version_info # noqa: F401
class MercurialUnsupportedError(errors.UnsupportedFormatError):
_fmt = ('Mercurial branches are not yet supported. '
'To convert Mercurial branches to Bazaar branches or vice versa, '
'use the fastimport format. ')
class LocalHgDirFormat(controldir.ControlDirFormat):
"""Mercurial directory format."""
def get_converter(self):
raise NotImplementedError(self.get_converter)
def get_format_description(self):
return "Local Mercurial control 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 MercurialUnsupportedError()
def open(self, transport):
# Raise NotBranchError if there is nothing there
LocalHgProber().probe_transport(transport)
raise NotImplementedError(self.open)
class LocalHgProber(controldir.Prober):
@classmethod
def priority(klass, transport):
return 100
@staticmethod
def _has_hg_dumb_repository(transport):
try:
return transport.has_any([".hg/requires", ".hg/00changelog.i"])
except (errors.NoSuchFile, errors.PermissionDenied,
errors.InvalidHttpResponse):
return False
@classmethod
def probe_transport(klass, transport):
"""Our format is present if the transport has a '.hg/' subdir."""
if klass._has_hg_dumb_repository(transport):
return HgDirFormat()
raise errors.NotBranchError(path=transport.base)
@classmethod
def known_formats(cls):
return [LocalHgDirFormat()]
class SmartHgDirFormat(controldir.ControlDirFormat):
"""Mercurial directory format."""
def get_converter(self):
raise NotImplementedError(self.get_converter)
def get_format_description(self):
return "Smart Mercurial control 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 MercurialUnsupportedError()
def open(self, transport):
# Raise NotBranchError if there is nothing there
SmartHgProber().probe_transport(transport)
raise NotImplementedError(self.open)
class SmartHgProber(controldir.Prober):
# Perhaps retrieve list from mercurial.hg.schemes ?
_supported_schemes = ["http", "https"]
@classmethod
def priority(klass, transport):
return 90
@staticmethod
def _has_hg_http_smart_server(transport, external_url):
"""Check if there is a Mercurial smart server at the remote location.
:param transport: Transport to check
:param externa_url: External URL for transport
:return: Boolean indicating whether transport is backed onto hg
"""
from breezy.urlutils import urlparse
parsed_url = urlparse.urlparse(external_url)
parsed_url = parsed_url._replace(
query='cmd=capabilities', path=parsed_url.path.rstrip('/') + '/hg')
url = urlparse.urlunparse(parsed_url)
resp = transport.request(
'GET', url, headers={'Accept': 'application/mercurial-0.1'})
if resp.status == 404:
return False
ct = resp.getheader("Content-Type")
if ct is None:
return False
return ct.startswith("application/mercurial")
@classmethod
def probe_transport(klass, transport):
try:
external_url = transport.external_url()
except errors.InProcessTransport:
raise errors.NotBranchError(path=transport.base)
scheme = external_url.split(":")[0]
if scheme not in klass._supported_schemes:
raise errors.NotBranchError(path=transport.base)
from breezy import urlutils
external_url = urlutils.strip_segment_parameters(external_url)
# Explicitly check for .hg directories here, so we avoid
# loading foreign branches through Mercurial.
if (external_url.startswith("http:") or
external_url.startswith("https:")):
if klass._has_hg_http_smart_server(transport, external_url):
return SmartHgDirFormat()
raise errors.NotBranchError(path=transport.base)
@classmethod
def known_formats(cls):
return [SmartHgDirFormat()]
controldir.ControlDirFormat.register_prober(LocalHgProber)
controldir.ControlDirFormat.register_prober(SmartHgProber)
|