1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 """Support archive locations."""
20
21 import urllib
22
23 from pybaz import errors
24
25 __all__ = [
26 'ArchiveLocation',
27 'ArchiveLocationParams',
28 ]
29
30
34
38
40 """Is `url` a valid url for an archive location?
41
42 :type url: str
43 :rtype: bool
44 """
45
46 if url.startswith('/'):
47 return True
48 if '://' not in url:
49 return False
50 scheme = url.split('://')[0]
51 if scheme not in ('ftp', 'http', 'sftp', 'file'):
52 return False
53 if scheme == 'file' and not url.startswith('file:///'):
54 return False
55 return True
56
57
59 """A location identified by an url and containing a Bazaar archive."""
60
62 error_message = ("ArchiveLocation parameter should be a str containing"
63 " an http, ftp, sftp url or an absolute file path,"
64 " but was: %r")
65 if not isinstance(url, str):
66 raise TypeError(error_message % (url,))
67 if not is_valid_url(url):
68 raise ValueError(error_message % (url,))
69 self._url = str(url)
70 self._archive = None
71
74
75 url = property(_get_url, doc="""
76 Url of this location.
77
78 :type: str
79 """)
80
82 """Compare equal to instances of ArchiveLocation with the same url."""
83 if not _factory().isArchiveLocation(other):
84 return False
85 return self.url == other.url
86
88 """Logical complement of __eq__."""
89 return not self == other
90
92
93 return '%s.%s(%r)' % (
94 self.__class__.__module__, self.__class__.__name__, self.url)
95
97 """Create a new master archive at this location.
98
99 :precondition: not self.is_registered()
100 and not archive.is_registered()
101 and <url does not exist and is writable>
102 :postcondition: archive.is_registered() and archive.location == self
103 and <url exists>
104
105 :type archive: Archive
106 :type params: ArchiveLocationParams
107 """
108 self._check_make_archive_params(archive, params)
109 args = ['make-archive']
110 if params.signed:
111 args.append('--signed')
112 if params.listing:
113 args.append('--listing')
114 if params._tla:
115 args.append('--tla')
116 args.extend((archive.name, self.url))
117 _backend().null_cmd(args)
118
120 """Create a new archive mirror at this location.
121
122 :precondition: not self.is_registered()
123 and <url does not exist and is writable>
124 :postcondition: self.is_registered()
125 and <url exists>
126
127 :type archive: Archive
128 :type params: ArchiveLocationParams
129 """
130 self._check_make_archive_params(archive, params)
131 args = ['make-archive', '--mirror', archive.name]
132 if params.signed:
133 args.append('--signed')
134 if params.listing:
135 args.append('--listing')
136 if params._tla:
137 args.append('--tla')
138 args.append(self.url)
139 _backend().null_cmd(args)
140
142 """Check sanity of create_master and create_mirror params."""
143 if not _factory().isArchive(archive):
144 raise TypeError("ArchiveLocation.create_master archive argument"
145 " must be an archive, but was: %r" % archive)
146 if not isinstance(params, ArchiveLocationParams):
147 raise TypeError(
148 "ArchiveLocation.create_master params argument"
149 " must be an ArchiveLocationParams, but was %r" % archive)
150
151
152
153
155 """Is this location registered?
156
157 :rtype: bool
158 """
159 for line in _backend().sequence_cmd(['archives', '--all-locations']):
160 if not line.startswith(' '):
161 continue
162 url = urllib.unquote(line[4:])
163 if self.url == url:
164 return True
165 return False
166
168 """Unregister this location:
169
170 :precondition: self.is_registered()
171 :poscondition: not self.is_registered()
172 :raises errors.LocationNotRegistered: this location was not registered.
173 """
174 if not self.is_registered():
175
176
177 raise errors.LocationNotRegistered(self.url)
178 _backend().null_cmd(('register-archive', '--delete', self.url))
179
181 """Register this location.
182
183 :precondition: not self.is_registered()
184 :postcondition: self.is_registered()
185 :raises errors.LocationAlreadyRegistered: this location was already
186 registered.
187 """
188 if self.is_registered():
189
190
191 raise errors.LocationAlreadyRegistered(self.url)
192 _backend().null_cmd(['register-archive', self.url])
193
225
243
245 """Archive that is associated to this location.
246
247 That's a convenience method based on meta_info() that memoises its
248 result.
249
250 :rtype: `Archive`
251 """
252 if self._archive is None:
253 name = self.meta_info('name')
254 self._archive = _factory().Archive(name)
255 return self._archive
256
258 """Version string of the archive.
259
260 Contents of the ``.archive-version`` file at the root of the archive.
261
262 :rtype: str
263 :bug: only works with schemes supported by ``urllib.urlopen``.
264 """
265 url = self.url
266 version_url = self.url + '/.archive-version'
267 version_file = urllib.urlopen(version_url)
268 try:
269 version = version_file.read()
270 finally:
271 version_file.close()
272 return version.rstrip()
273
275 """Create a mirrorer to mirror from this location to the target.
276
277 :param target: specific location the `MirrorMethod` will mirror to.
278 :type target: `ArchiveLocation`
279 :rtype: `MirrorMethod`
280
281 :raises error.LocationNotRegistered: at least one of self and target is
282 not a registered location.
283 :raises errors.MirrorLocationMismatch: self and target are registered
284 locations for different archives.
285 """
286 if not _factory().isArchiveLocation(target):
287 raise TypeError("ArchiveLocation.make_mirrorer argument should be"
288 " and ArchiveLocation, but was: %r" % target)
289 if self.url == target.url:
290 raise ValueError("ArchiveLocation.make_mirrorer argument was an"
291 " ArchiveLocation with the same url as self.")
292 if not self.is_registered():
293 raise errors.LocationNotRegistered(self)
294 if not target.is_registered():
295 raise errors.LocationNotRegistered(target)
296 if self.archive() != target.archive():
297 raise errors.MirrorLocationMismatch(self, target)
298 mirrorer = MirrorMethod(self, target)
299 return mirrorer
300
301
303 """Parameter Object used for creating archives masters and mirrors.
304
305 :ivar signed: create signed location?
306 :type signed: bool
307 :ivar listing: create location with listings for http access.
308 :type listing: bool
309 """
310
312 self.signed = False
313 self.listing = False
314 self._tla = False
315
319
320
322 """Method Object used for mirroring operations.
323
324 This class should never be instantiated directly. Instead, it is created by
325 factory methods like `ArchiveLocation.make_mirrorer`.
326 """
327
329 """Do not instanciate this class directly."""
330 self._source = source
331 self._target = target
332
334 """Internal helper for sanity checking of limit argument."""
335 if _factory().isVersion(limit):
336 return False
337 if _factory().isRevision(limit):
338 return False
339 return True
340
342 """Internal helper for sanity checking of limit argument."""
343 return limit.archive != self._source.archive()
344
345 - def mirror(self, limit=None):
346 """Perform mirroring.
347
348 :param limit: if provided, limit the mirroring to the specified
349 version or revision.
350 :type limit: `pybaz.Version`, `pybaz.Revision`, None
351 """
352 assert self._source is not None
353 assert self._target is not None
354 args = ['archive-mirror', self._source.url, self._target.url]
355 if limit is not None:
356 message = ("MirrorMethod.mirror parameter should be a Version or"
357 " Revision in %s, but was: %r")
358 archive = self._source.archive()
359 if self._limit_type_is_bad(limit):
360 raise TypeError(message % (archive.name, limit))
361 if self._limit_value_is_bad(limit):
362 raise ValueError(message % (archive.name, limit))
363 args.append(limit.nonarch)
364 _backend().null_cmd(args)
365