← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/plugins/launchpad/lp_directory.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2008-03-12 04:19:21 UTC
  • mfrom: (3251.4.8 lp-service)
  • Revision ID: pqm@pqm.ubuntu.com-20080312041921-iqibnzp8pcqx18mk
Convert lp transport into a directory service (abentley)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/lp_directory.py
1
 
# Copyright (C) 2007 Canonical Ltd
 
1
# Copyright (C) 2007, 2008 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
 
18
 
"""Transport indirection that uses Launchpad as a directory lookup.
19
 
 
20
 
When the transport is opened, it immediately redirects to a url
21
 
on Launchpad, which can then either serve the branch itself or redirect
22
 
again.
23
 
"""
 
18
"""Directory lookup that uses Launchpad."""
24
19
 
25
20
from urlparse import urlsplit, urlunsplit
26
21
import xmlrpclib
34
29
from bzrlib.transport import (
35
30
    get_transport,
36
31
    register_urlparse_netloc_protocol,
37
 
    Transport,
38
32
    )
39
33
 
40
34
from bzrlib.plugins.launchpad.lp_registration import (
48
42
register_urlparse_netloc_protocol('lp')
49
43
 
50
44
 
51
 
class LaunchpadTransport(Transport):
52
 
    """lp:/// URL transport
53
 
 
54
 
    This transport redirects requests to the real branch location
55
 
    after resolving the URL via an XMLRPC request to Launchpad.
56
 
    """
57
 
 
58
 
    def __init__(self, base):
59
 
        super(LaunchpadTransport, self).__init__(base)
60
 
        # We only support URLs without a netloc
61
 
        self.lp_instance = urlsplit(base)[1]
62
 
        if self.lp_instance == '':
63
 
            self.lp_instance = None
64
 
        elif self.lp_instance not in LaunchpadService.LAUNCHPAD_INSTANCE:
65
 
            raise errors.InvalidURL(path=base)
 
45
class LaunchpadDirectory(object):
66
46
 
67
47
    def _requires_launchpad_login(self, scheme, netloc, path, query,
68
48
                                  fragment):
75
55
                and (netloc.endswith('launchpad.net')
76
56
                     or netloc.endswith('launchpad.dev')))
77
57
 
78
 
    def _resolve(self, abspath,
 
58
    def look_up(self, name, url):
 
59
        """See DirectoryService.look_up"""
 
60
        return self._resolve(url)
 
61
 
 
62
    def _resolve(self, url,
79
63
                 _request_factory=ResolveLaunchpadPathRequest,
80
64
                 _lp_login=None):
81
65
        """Resolve the base URL for this transport."""
82
 
        path = urlsplit(abspath)[2].lstrip('/')
 
66
        result = urlsplit(url)
83
67
        # Perform an XMLRPC request to resolve the path
84
 
        resolve = _request_factory(path)
85
 
        service = LaunchpadService(lp_instance=self.lp_instance)
 
68
        lp_instance = result[1]
 
69
        if lp_instance == '':
 
70
            lp_instance = None
 
71
        elif lp_instance not in LaunchpadService.LAUNCHPAD_INSTANCE:
 
72
            raise errors.InvalidURL(path=url)
 
73
        resolve = _request_factory(result[2].strip('/'))
 
74
        service = LaunchpadService(lp_instance=lp_instance)
86
75
        try:
87
76
            result = resolve.submit(service)
88
77
        except xmlrpclib.Fault, fault:
89
78
            raise errors.InvalidURL(
90
 
                path=abspath, extra=fault.faultString)
 
79
                path=url, extra=fault.faultString)
91
80
 
92
81
        if 'launchpad' in debug.debug_flags:
93
82
            trace.mutter("resolve_lp_path(%r) == %r", path, result)
114
103
                else:
115
104
                    break
116
105
        else:
117
 
            raise errors.InvalidURL(path=abspath,
118
 
                                    extra='no supported schemes')
 
106
            raise errors.InvalidURL(path=url, extra='no supported schemes')
119
107
        return url
120
108
 
121
 
    def _request_redirect(self, relpath):
122
 
        source = urlutils.join(self.base, relpath)
123
 
        # Split the source location into the branch location, and the
124
 
        # extra path components.
125
 
        pos = source.find('/.bzr/')
126
 
        if pos >= 0:
127
 
            branchpath = source[:pos]
128
 
            extra = source[pos:]
129
 
        else:
130
 
            branchpath = source
131
 
            extra = ''
132
 
        target = self._resolve(branchpath) + extra
133
 
        raise errors.RedirectRequested(
134
 
            source=source,
135
 
            target=target)
136
 
 
137
 
    def get(self, relpath):
138
 
        """See Transport.get()."""
139
 
        self._request_redirect(relpath)
140
 
 
141
 
    def mkdir(self, relpath, mode=None):
142
 
        """See Transport.mkdir()."""
143
 
        self._request_redirect(relpath)
144
 
 
145
109
 
146
110
def get_test_permutations():
147
111
    # Since this transport doesn't do anything once opened, it's not subjected