1# Copyright (C) 2005-2010 Canonical Ltd 2# 3# This program is free software; you can redistribute it and/or modify 4# it under the terms of the GNU General Public License as published by 5# the Free Software Foundation; either version 2 of the License, or 6# (at your option) any later version. 7# 8# This program is distributed in the hope that it will be useful, 9# but WITHOUT ANY WARRANTY; without even the implied warranty of 10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 11# GNU General Public License for more details. 12# 13# You should have received a copy of the GNU General Public License 14# along with this program; if not, write to the Free Software 15# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA 16 17from io import BytesIO 18 19from .lazy_import import lazy_import 20lazy_import(globals(), """ 21from breezy import ( 22 transport as _mod_transport, 23 urlutils, 24 ) 25from breezy.bzr.bundle import serializer as _serializer 26from breezy.merge_directive import MergeDirective 27from breezy.i18n import gettext 28""") 29 30from . import ( 31 errors, 32 ) 33from .trace import note 34 35 36class Mergeable(object): 37 """A mergeable object.""" 38 39 def install_revisions(self, repository): 40 """Install the data from this mergeable into the specified repository. 41 42 :param repository: Repository 43 """ 44 raise NotImplementedError(self.install_revisions) 45 46 def get_merge_request(self, repository): 47 """Extract merge request data. 48 49 :return: tuple with (base_revision_id, target_revision_id, verified) 50 """ 51 raise NotImplementedError(self.get_merge_request) 52 53 54def read_mergeable_from_url(url, _do_directive=True, possible_transports=None): 55 """Read mergable object from a given URL. 56 57 :return: An object supporting get_target_revision. Raises NotABundle if 58 the target is not a mergeable type. 59 """ 60 child_transport = _mod_transport.get_transport( 61 url, possible_transports=possible_transports) 62 transport = child_transport.clone('..') 63 filename = transport.relpath(child_transport.base) 64 mergeable, transport = read_mergeable_from_transport(transport, filename, 65 _do_directive) 66 return mergeable 67 68 69def read_mergeable_from_transport(transport, filename, _do_directive=True): 70 def get_bundle(transport): 71 return BytesIO(transport.get_bytes(filename)), transport 72 73 def redirected_transport(transport, exception, redirection_notice): 74 note(redirection_notice) 75 url, filename = urlutils.split(exception.target, 76 exclude_trailing_slash=False) 77 if not filename: 78 raise errors.NotABundle(gettext('A directory cannot be a bundle')) 79 return _mod_transport.get_transport_from_url(url) 80 81 try: 82 bytef, transport = _mod_transport.do_catching_redirections( 83 get_bundle, transport, redirected_transport) 84 except errors.TooManyRedirections: 85 raise errors.NotABundle(transport.clone(filename).base) 86 except (errors.ConnectionReset, errors.ConnectionError) as e: 87 raise 88 except (errors.TransportError, errors.PathError) as e: 89 raise errors.NotABundle(str(e)) 90 except (IOError,) as e: 91 # jam 20060707 92 # Abstraction leakage, SFTPTransport.get('directory') 93 # doesn't always fail at get() time. Sometimes it fails 94 # during read. And that raises a generic IOError with 95 # just the string 'Failure' 96 # StubSFTPServer does fail during get() (because of prefetch) 97 # so it has an opportunity to translate the error. 98 raise errors.NotABundle(str(e)) 99 100 if _do_directive: 101 try: 102 return MergeDirective.from_lines(bytef), transport 103 except errors.NotAMergeDirective: 104 bytef.seek(0) 105 106 return _serializer.read_bundle(bytef), transport 107