1# Copyright (C) 2006-2012, 2016-2017 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 xmlrpc.client import (
20    loads as xmlrpc_loads,
21    Transport,
22    )
23
24from ...tests import TestCaseWithTransport
25
26# local import
27from .lp_registration import (
28    BaseRequest,
29    ResolveLaunchpadPathRequest,
30    LaunchpadService,
31    )
32
33
34# TODO: Test that the command-line client, making sure that it'll pass the
35# request through to a dummy transport, and that the transport will validate
36# the results passed in.  Not sure how to get the transport object back out to
37# validate that its OK - may not be necessary.
38
39class InstrumentedXMLRPCConnection(object):
40    """Stands in place of an http connection for the purposes of testing"""
41
42    def __init__(self, testcase):
43        self.testcase = testcase
44
45    def getreply(self):
46        """Fake the http reply.
47
48        :returns: (errcode, errmsg, headers)
49        """
50        return (200, 'OK', [])
51
52    def getresponse(self, buffering=True):
53        """Fake the http reply.
54
55        This is used when running on Python 2.7, where xmlrpclib uses
56        httplib.HTTPConnection in a different way than before.
57        """
58        class FakeHttpResponse(object):
59
60            def __init__(self, status, reason, body):
61                self.status = status
62                self.reason = reason
63                self.body = body
64
65            def read(self, size=-1):
66                return self.body.read(size)
67
68            def getheader(self, name, default):
69                # We don't have headers
70                return default
71
72        return FakeHttpResponse(200, 'OK', self.getfile())
73
74    def getfile(self):
75        """Return a fake file containing the response content."""
76        return BytesIO(b'''\
77<?xml version="1.0" ?>
78<methodResponse>
79    <params>
80        <param>
81            <value>
82                <string>victoria dock</string>
83            </value>
84        </param>
85    </params>
86</methodResponse>''')
87
88
89class InstrumentedXMLRPCTransport(Transport):
90
91    # Python 2.5's xmlrpclib looks for this.
92    _use_datetime = False
93    _use_builtin_types = False
94
95    def __init__(self, testcase):
96        self.testcase = testcase
97        self._connection = (None, None)
98
99    def make_connection(self, host):
100        host, http_headers, x509 = self.get_host_info(host)
101        test = self.testcase
102        self.connected_host = host
103        if http_headers:
104            raise AssertionError()
105        return InstrumentedXMLRPCConnection(test)
106
107    def send_request(self, host, handler_path, request_body,
108                     verbose=None):
109        self.connected_host = host
110        test = self.testcase
111        self.got_request = True
112        unpacked, method = xmlrpc_loads(request_body)
113        if None in unpacked:
114            raise AssertionError(
115                "xmlrpc result %r shouldn't contain None" % (unpacked,))
116        self.sent_params = unpacked
117        return InstrumentedXMLRPCConnection(test)
118
119    def send_host(self, conn, host):
120        pass
121
122    def send_user_agent(self, conn):
123        # TODO: send special user agent string, including breezy version
124        # number
125        pass
126
127    def send_content(self, conn, request_body):
128        unpacked, method = xmlrpc_loads(request_body)
129        if None in unpacked:
130            raise AssertionError(
131                "xmlrpc result %r shouldn't contain None" % (unpacked,))
132        self.sent_params = unpacked
133
134
135class MockLaunchpadService(LaunchpadService):
136
137    def send_request(self, method_name, method_params, verbose=None):
138        """Stash away the method details rather than sending them to a real server"""
139        self.called_method_name = method_name
140        self.called_method_params = method_params
141
142
143class TestResolveLaunchpadPathRequest(TestCaseWithTransport):
144
145    def setUp(self):
146        super(TestResolveLaunchpadPathRequest, self).setUp()
147        # make sure we have a reproducible standard environment
148        self.overrideEnv('BRZ_LP_XMLRPC_URL', None)
149
150    def test_onto_transport(self):
151        """A request is transmitted across a mock Transport"""
152        transport = InstrumentedXMLRPCTransport(self)
153        service = LaunchpadService(transport)
154        resolve = ResolveLaunchpadPathRequest('bzr')
155        resolve.submit(service)
156        self.assertEqual(transport.connected_host, 'xmlrpc.launchpad.net')
157        self.assertEqual(len(transport.sent_params), 1)
158        self.assertEqual(transport.sent_params, ('bzr', ))
159        self.assertTrue(transport.got_request)
160
161    def test_subclass_request(self):
162        """Define a new type of xmlrpc request"""
163        class DummyRequest(BaseRequest):
164            _methodname = 'dummy_request'
165
166            def _request_params(self):
167                return (42,)
168
169        service = MockLaunchpadService()
170        service.registrant_email = 'test@launchpad.net'
171        service.registrant_password = ''
172        request = DummyRequest()
173        request.submit(service)
174        self.assertEqual(service.called_method_name, 'dummy_request')
175        self.assertEqual(service.called_method_params, (42,))
176
177    def test_mock_resolve_lp_url(self):
178        test_case = self
179
180        class MockService(MockLaunchpadService):
181            def send_request(self, method_name, method_params,
182                             verbose=None):
183                test_case.assertEqual(method_name, "resolve_lp_path")
184                test_case.assertEqual(list(method_params), ['bzr'])
185                return dict(urls=[
186                    'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk',
187                    'sftp://bazaar.launchpad.net~bzr/bzr/trunk',
188                    'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk',
189                    'http://bazaar.launchpad.net~bzr/bzr/trunk'])
190        service = MockService()
191        resolve = ResolveLaunchpadPathRequest('bzr')
192        result = resolve.submit(service)
193        self.assertTrue('urls' in result)
194        self.assertEqual(result['urls'], [
195            'bzr+ssh://bazaar.launchpad.net~bzr/bzr/trunk',
196            'sftp://bazaar.launchpad.net~bzr/bzr/trunk',
197            'bzr+http://bazaar.launchpad.net~bzr/bzr/trunk',
198            'http://bazaar.launchpad.net~bzr/bzr/trunk'])
199