1# -*- coding: utf-8 -*-
2
3# This program is free software; you can redistribute it and/or modify it under
4# the terms of the (LGPL) GNU Lesser General Public License as published by the
5# Free Software Foundation; either version 3 of the License, or (at your
6# option) any later version.
7#
8# This program is distributed in the hope that it will be useful, but WITHOUT
9# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
10# FOR A PARTICULAR PURPOSE. See the GNU Library Lesser General Public License
11# for more details at ( http://www.gnu.org/licenses/lgpl.html ).
12#
13# You should have received a copy of the GNU Lesser General Public License
14# along with this program; if not, write to the Free Software Foundation, Inc.,
15# 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
16# written by: Jurko Gospodnetić ( jurko.gospodnetic@pke.hr )
17
18"""
19Suds library HTTP transport related unit tests.
20
21Implemented using the 'pytest' testing framework.
22
23"""
24
25if __name__ == "__main__":
26    import __init__
27    __init__.runUsingPyTest(globals())
28
29
30import suds
31import suds.client
32import suds.store
33import suds.transport.http
34
35import pytest
36
37import base64
38import sys
39import urllib2
40
41
42class MyException(Exception):
43    """Local exception used in this test module."""
44    pass
45
46
47def test_authenticated_http():
48    t = suds.transport.http.HttpAuthenticated(username="Habul AfuFa",
49        password="preCious")
50    assert t.credentials() == ("Habul AfuFa", "preCious")
51
52    t = suds.transport.http.HttpAuthenticated(username="macro")
53    assert t.credentials() == ("macro", None)
54
55
56def test_authenticated_http_add_credentials_to_request():
57    class MockRequest:
58        def __init__(self):
59            self.headers = {}
60
61    t = suds.transport.http.HttpAuthenticated(username="Humpty")
62    r = MockRequest()
63    t.addcredentials(r)
64    assert len(r.headers) == 0
65
66    t = suds.transport.http.HttpAuthenticated(password="Dumpty")
67    r = MockRequest()
68    t.addcredentials(r)
69    assert len(r.headers) == 0
70
71    username = "Habul Afufa"
72    password = "preCious"
73    t = suds.transport.http.HttpAuthenticated(username=username,
74        password=password)
75    r = MockRequest()
76    t.addcredentials(r)
77    _check_Authorization_header(r, username, password)
78
79    #   Regression test: Extremely long username & password combinations must
80    # not cause suds to add additional newlines in the constructed
81    # 'Authorization' HTTP header.
82    username = ("An Extremely Long Username that could be usable only to "
83        "Extremely Important People whilst on Extremely Important Missions.")
84    password = ("An Extremely Long Password that could be usable only to "
85        "Extremely Important People whilst on Extremely Important Missions. "
86        "And some extra 'funny' characters to improve security: "
87        "!@#$%^&*():|}|{{.\nEven\nSome\nNewLines\n"
88        "  and spaces at the start of a new line.   ")
89    t = suds.transport.http.HttpAuthenticated(username=username,
90        password=password)
91    r = MockRequest()
92    t.addcredentials(r)
93    _check_Authorization_header(r, username, password)
94
95
96@pytest.mark.parametrize("url", (
97    "http://my little URL",
98    "https://my little URL",
99    "xxx://my little URL",
100    "xxx:my little URL",
101    "xxx:"))
102def test_http_request_URL(url):
103    """Make sure suds makes a HTTP request targeted at an expected URL."""
104    class MockURLOpener:
105        def open(self, request, timeout=None):
106            assert request.get_full_url() == url
107            raise MyException
108    transport = suds.transport.http.HttpTransport()
109    transport.urlopener = MockURLOpener()
110    store = suds.store.DocumentStore(wsdl=_wsdl_with_no_input_data(url))
111    client = suds.client.Client("suds://wsdl", cache=None, documentStore=store,
112        transport=transport)
113    pytest.raises(MyException, client.service.f)
114
115
116@pytest.mark.parametrize("url", (
117    "my no-protocol URL",
118    ":my no-protocol URL"))
119def test_http_request_URL_with_a_missing_protocol_identifier(url):
120    """
121    Test suds reporting URLs with a missing protocol identifier.
122
123    Python urllib library makes this check under Python 3.x, but does not under
124    earlier Python versions.
125
126    """
127    class MockURLOpener:
128        def open(self, request, timeout=None):
129            raise MyException
130    transport = suds.transport.http.HttpTransport()
131    transport.urlopener = MockURLOpener()
132    store = suds.store.DocumentStore(wsdl=_wsdl_with_no_input_data(url))
133    client = suds.client.Client("suds://wsdl", cache=None, documentStore=store,
134        transport=transport)
135    exceptionClass = ValueError
136    if sys.version_info < (3, 0):
137        exceptionClass = MyException
138    pytest.raises(exceptionClass, client.service.f)
139
140
141def test_sending_unicode_data(monkeypatch):
142    """
143    Original suds implementation passed its request location URL to the
144    underlying HTTP request object as a unicode string.
145
146    Under Python 2.4 this causes no problems as that implementation simply
147    sends all the request data over the network as-is (and treats all unicode
148    data as bytes anyway).
149
150    Under Python 2.7 this causes the httplib HTTP request implementation to
151    convert all of its data to unicode, and do so by simply assuming that data
152    contains only ASCII characters. If any other characters are encountered, it
153    fails with an exception like "UnicodeDecodeError: 'ascii' codec can't
154    decode byte 0xd0 in position 290: ordinal not in range(128)".
155
156    Under Python 3.x the httplib HTTP request implementation automatically
157    converts its received URL to a bytes object (assuming it contains only
158    ASCII characters), thus avoiding the need to convert all the other request
159    data.
160
161    In order to trigger the problematic httplib behaviour we need to make suds
162    attempt to send a HTTP request over the network. On the other hand, we want
163    this test to work even on computers not connected to a network so we
164    monkey-patch the underlying network socket APIs, log all the data suds
165    attempt to send over the network and consider the test run successful once
166    suds attempt to read back data from the network.
167
168    """
169    def callOnce(f):
170        """Method decorator making sure its function only gets called once."""
171        def wrapper(self, *args, **kwargs):
172            fTag = "_%s__%s_called" % (self.__class__.__name__, f.__name__)
173            assert not hasattr(self, fTag)
174            setattr(self, fTag, True)
175            return f(self, *args, **kwargs)
176        return wrapper
177
178    class Mocker:
179        def __init__(self, expectedHost, expectedPort):
180            self.expectedHost = expectedHost
181            self.expectedPort = expectedPort
182            self.sentData = suds.byte_str()
183            self.hostAddress = object()
184        @callOnce
185        def getaddrinfo(self, host, port, *args, **kwargs):
186            assert host == self.expectedHost
187            assert port == self.expectedPort
188            return [(None, None, None, None, self.hostAddress)]
189        @callOnce
190        def socket(self, *args, **kwargs):
191            self.socket = MockSocket(self)
192            return self.socket
193
194    class MockSocketReader:
195        @callOnce
196        def readline(self, *args, **kwargs):
197            raise MyException
198
199    class MockSocket:
200        def __init__(self, mocker):
201            self.__mocker = mocker
202        @callOnce
203        def connect(self, address):
204            assert address is self.__mocker.hostAddress
205        @callOnce
206        def makefile(self, *args, **kwargs):
207            return MockSocketReader()
208        def sendall(self, data):
209            # Python 2.4 urllib implementation calls this function twice - once
210            # for sending the HTTP request headers and once for its body.
211            self.__mocker.sentData += data
212        @callOnce
213        def settimeout(self, *args, **kwargs):
214            assert not hasattr(self, "settimeout_called")
215            self.settimeout_called = True
216
217    host = "an-easily-recognizable-host-name-214894932"
218    port = 9999
219    mocker = Mocker(host, port)
220    monkeypatch.setattr("socket.getaddrinfo", mocker.getaddrinfo)
221    monkeypatch.setattr("socket.socket", mocker.socket)
222    url = "http://%s:%s/svc" % (host, port)
223    store = suds.store.DocumentStore(wsdl=_wsdl_with_input_data(url))
224    client = suds.client.Client("suds://wsdl", cache=None, documentStore=store)
225    data = u"Дмитровский район"
226    pytest.raises(MyException, client.service.f, data)
227    assert data.encode("utf-8") in mocker.sentData
228
229
230def test_sending_non_ascii_location():
231    """
232    Suds should refuse to send HTTP requests with a target location string
233    containing non-ASCII characters. URLs are supposed to consist of
234    characters only.
235
236    """
237    class MockURLOpener:
238        def open(self, request, timeout=None):
239            raise MyException
240    url = u"http://Дмитровский-район-152312306:9999/svc"
241    transport = suds.transport.http.HttpTransport()
242    transport.urlopener = MockURLOpener()
243    store = suds.store.DocumentStore(wsdl=_wsdl_with_no_input_data(url))
244    client = suds.client.Client("suds://wsdl", cache=None, documentStore=store,
245        transport=transport)
246    pytest.raises(UnicodeEncodeError, client.service.f)
247
248
249@pytest.mark.skipif(sys.version_info >= (3, 0),
250    reason="Python 2 specific functionality")
251@pytest.mark.parametrize(("urlString", "expectedException"), (
252    ("http://jorgula", MyException),
253    ("http://jorgula_\xe7", UnicodeDecodeError)))
254def test_sending_py2_bytes_location(urlString, expectedException):
255    """
256    Suds should accept single-byte string URL values under Python 2, but should
257    still report an error if those strings contain any non-ASCII characters.
258
259    """
260    class MockURLOpener:
261        def open(self, request, timeout=None):
262            raise MyException
263    transport = suds.transport.http.HttpTransport()
264    transport.urlopener = MockURLOpener()
265    store = suds.store.DocumentStore(wsdl=_wsdl_with_no_input_data("http://x"))
266    client = suds.client.Client("suds://wsdl", cache=None, documentStore=store,
267        transport=transport)
268    client.options.location = suds.byte_str(urlString)
269    pytest.raises(expectedException, client.service.f)
270
271
272@pytest.mark.skipif(sys.version_info < (3, 0),
273    reason="requires at least Python 3")
274@pytest.mark.parametrize("urlString", (
275    "http://jorgula",
276    "http://jorgula_\xe7"))
277def test_sending_py3_bytes_location(urlString):
278    """
279    Suds should refuse to send HTTP requests with a target location specified
280    as either a Python 3 bytes or bytearray object.
281
282    """
283    class MockURLOpener:
284        def open(self, request, timeout=None):
285            raise MyException
286    transport = suds.transport.http.HttpTransport()
287    transport.urlopener = MockURLOpener()
288    store = suds.store.DocumentStore(wsdl=_wsdl_with_no_input_data("http://x"))
289    client = suds.client.Client("suds://wsdl", cache=None, documentStore=store,
290        transport=transport)
291
292    expectedException = AssertionError
293    if sys.flags.optimize:
294        expectedException = AttributeError
295
296    for url in (bytes(urlString, encoding="utf-8"),
297        bytearray(urlString, encoding="utf-8")):
298        # Under Python 3.x we can not use the client's 'location' option to set
299        # a bytes URL as it accepts only strings and in Python 3.x all strings
300        # are unicode strings. Therefore, we use an ugly hack, modifying suds's
301        # internal web service description structure to force it to think it
302        # has a bytes object specified as a location for its 'f' web service
303        # operation.
304        client.sd[0].ports[0][0].methods['f'].location = url
305        pytest.raises(expectedException, client.service.f)
306
307
308def _check_Authorization_header(request, username, password):
309    assert len(request.headers) == 1
310    header = request.headers["Authorization"]
311    assert header == _encode_basic_credentials(username, password)
312
313
314def _encode_basic_credentials(username, password):
315    """
316      Encodes user credentials as used in basic HTTP authentication.
317
318      This is the value expected to be added to the 'Authorization' HTTP
319    header.
320
321    """
322    data = suds.byte_str("%s:%s" % (username, password))
323    return "Basic %s" % base64.b64encode(data).decode("utf-8")
324
325
326def _wsdl_with_input_data(url):
327    """
328    Return a WSDL schema with a single operation f taking a single parameter.
329
330    Included operation takes a single string parameter and returns no values.
331    Externally specified URL is used as the web service location.
332
333    """
334    return suds.byte_str(u"""\
335<?xml version="1.0" encoding="utf-8"?>
336<wsdl:definitions targetNamespace="myNamespace"
337  xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
338  xmlns:tns="myNamespace"
339  xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
340  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
341  <wsdl:types>
342    <xsd:schema targetNamespace="myNamespace">
343      <xsd:element name="fRequest" type="xsd:string"/>
344    </xsd:schema>
345  </wsdl:types>
346  <wsdl:message name="fInputMessage">
347    <wsdl:part name="parameters" element="tns:fRequest"/>
348  </wsdl:message>
349  <wsdl:portType name="Port">
350    <wsdl:operation name="f">
351      <wsdl:input message="tns:fInputMessage"/>
352    </wsdl:operation>
353  </wsdl:portType>
354  <wsdl:binding name="Binding" type="tns:Port">
355    <soap:binding style="document"
356      transport="http://schemas.xmlsoap.org/soap/http"/>
357    <wsdl:operation name="f">
358      <soap:operation/>
359      <wsdl:input><soap:body use="literal"/></wsdl:input>
360    </wsdl:operation>
361  </wsdl:binding>
362  <wsdl:service name="Service">
363    <wsdl:port name="Port" binding="tns:Binding">
364      <soap:address location="%s"/>
365    </wsdl:port>
366  </wsdl:service>
367</wsdl:definitions>""" % (url,))
368
369
370def _wsdl_with_no_input_data(url):
371    """
372    Return a WSDL schema with a single operation f taking no parameters.
373
374    Included operation returns no values. Externally specified URL is used as
375    the web service location.
376
377    """
378    return suds.byte_str(u"""\
379<?xml version="1.0" encoding="utf-8"?>
380<wsdl:definitions targetNamespace="myNamespace"
381  xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
382  xmlns:tns="myNamespace"
383  xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
384  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
385  <wsdl:portType name="Port">
386    <wsdl:operation name="f"/>
387  </wsdl:portType>
388  <wsdl:binding name="Binding" type="tns:Port">
389    <soap:binding style="document"
390      transport="http://schemas.xmlsoap.org/soap/http"/>
391    <wsdl:operation name="f"/>
392  </wsdl:binding>
393  <wsdl:service name="Service">
394    <wsdl:port name="Port" binding="tns:Binding">
395      <soap:address location="%s"/>
396    </wsdl:port>
397  </wsdl:service>
398</wsdl:definitions>""" % (url,))
399