← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Vincent Ladeuil
  • Date: 2010-01-25 15:55:48 UTC
  • mto: (4985.1.4 add-attr-cleanup)
  • mto: This revision was merged to the branch mainline in revision 4988.
  • Revision ID: v.ladeuil+lp@free.fr-20100125155548-0l352pujvt5bzl5e
Deploy addAttrCleanup on the whole test suite.

Several use case worth mentioning:

- setting a module or any other object attribute is the majority
by far. In some cases the setting itself is deferred but most of
the time we want to set at the same time we add the cleanup.

- there multiple occurrences of protecting hooks or ui factory
which are now useless (the test framework takes care of that now),

- there was some lambda uses that can now be avoided.

That first cleanup already simplifies things a lot.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/plugins/launchpad/lp_registration.py
21
21
import urllib
22
22
import xmlrpclib
23
23
 
24
 
from bzrlib.lazy_import import lazy_import
25
 
lazy_import(globals(), """
26
 
from bzrlib import urlutils
27
 
""")
28
 
 
29
24
from bzrlib import (
30
25
    config,
31
26
    errors,
 
27
    urlutils,
32
28
    __version__ as _bzrlib_version,
33
29
    )
 
30
from bzrlib.transport.http import _urllib2_wrappers
 
31
 
34
32
 
35
33
# for testing, do
36
34
'''
53
51
        errors.BzrError.__init__(self, url=url)
54
52
 
55
53
 
 
54
class XMLRPCTransport(xmlrpclib.Transport):
 
55
 
 
56
    def __init__(self, scheme):
 
57
        # In python2.4 xmlrpclib.Transport is a old-style class, and does not
 
58
        # define __init__, so we check first
 
59
        init = getattr(xmlrpclib.Transport, '__init__', None)
 
60
        if init is not None:
 
61
            init(self)
 
62
        self._scheme = scheme
 
63
        self._opener = _urllib2_wrappers.Opener()
 
64
        self.verbose = 0
 
65
 
 
66
    def request(self, host, handler, request_body, verbose=0):
 
67
        self.verbose = verbose
 
68
        url = self._scheme + "://" + host + handler
 
69
        request = _urllib2_wrappers.Request("POST", url, request_body)
 
70
        # FIXME: _urllib2_wrappers will override user-agent with its own
 
71
        # request.add_header("User-Agent", self.user_agent)
 
72
        request.add_header("Content-Type", "text/xml")
 
73
 
 
74
        response = self._opener.open(request)
 
75
        if response.code != 200:
 
76
            raise xmlrpclib.ProtocolError(host + handler, response.code,
 
77
                                          response.msg, response.info())
 
78
        return self.parse_response(response)
 
79
 
 
80
 
56
81
class LaunchpadService(object):
57
82
    """A service to talk to Launchpad via XMLRPC.
58
83
 
90
115
        self._lp_instance = lp_instance
91
116
        if transport is None:
92
117
            uri_type = urllib.splittype(self.service_url)[0]
93
 
            if uri_type == 'https':
94
 
                transport = xmlrpclib.SafeTransport()
95
 
            else:
96
 
                transport = xmlrpclib.Transport()
 
118
            transport = XMLRPCTransport(uri_type)
97
119
            transport.user_agent = 'bzr/%s (xmlrpclib/%s)' \
98
120
                    % (_bzrlib_version, xmlrpclib.__version__)
99
121
        self.transport = transport
100
122
 
101
 
 
102
123
    @property
103
124
    def service_url(self):
104
125
        """Return the http or https url for the xmlrpc server.
116
137
        else:
117
138
            return self.DEFAULT_SERVICE_URL
118
139
 
 
140
    @classmethod
 
141
    def for_url(cls, url, **kwargs):
 
142
        """Return the Launchpad service corresponding to the given URL."""
 
143
        result = urlsplit(url)
 
144
        lp_instance = result[1]
 
145
        if lp_instance == '':
 
146
            lp_instance = None
 
147
        elif lp_instance not in cls.LAUNCHPAD_INSTANCE:
 
148
            raise errors.InvalidURL(path=url)
 
149
        return cls(lp_instance=lp_instance, **kwargs)
 
150
 
119
151
    def get_proxy(self, authenticated):
120
152
        """Return the proxy for XMLRPC requests."""
121
153
        if authenticated:
191
223
            instance = self._lp_instance
192
224
        return self.LAUNCHPAD_DOMAINS[instance]
193
225
 
194
 
    def get_web_url_from_branch_url(self, branch_url, _request_factory=None):
195
 
        """Get the Launchpad web URL for the given branch URL.
196
 
 
197
 
        :raise errors.InvalidURL: if 'branch_url' cannot be identified as a
198
 
            Launchpad branch URL.
199
 
        :return: The URL of the branch on Launchpad.
200
 
        """
 
226
    def _guess_branch_path(self, branch_url, _request_factory=None):
201
227
        scheme, hostinfo, path = urlsplit(branch_url)[:3]
202
228
        if _request_factory is None:
203
229
            _request_factory = ResolveLaunchpadPathRequest
215
241
                for domain in self.LAUNCHPAD_DOMAINS.itervalues())
216
242
            if hostinfo not in domains:
217
243
                raise NotLaunchpadBranch(branch_url)
 
244
        return path.lstrip('/')
 
245
 
 
246
    def get_web_url_from_branch_url(self, branch_url, _request_factory=None):
 
247
        """Get the Launchpad web URL for the given branch URL.
 
248
 
 
249
        :raise errors.InvalidURL: if 'branch_url' cannot be identified as a
 
250
            Launchpad branch URL.
 
251
        :return: The URL of the branch on Launchpad.
 
252
        """
 
253
        path = self._guess_branch_path(branch_url, _request_factory)
218
254
        return urlutils.join('https://code.%s' % self.domain, path)
219
255
 
220
256