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"""
19Package containing different utilities used for suds project testing.
20
21"""
22
23import suds.client
24import suds.store
25
26import six
27
28import os
29import subprocess
30import sys
31from .compare_sax import CompareSAX
32
33def _assert_request_content(request, expected_xml):
34    CompareSAX.data2data(request.envelope, expected_xml)
35
36
37def client_from_wsdl(wsdl_content, *args, **kwargs):
38    """
39    Constructs a non-caching suds Client based on the given WSDL content.
40
41      The wsdl_content is expected to be a raw byte string and not a unicode
42    string. This simple structure suits us fine here because XML content holds
43    its own embedded encoding identification ('utf-8' if not specified
44    explicitly).
45
46      Stores the content directly inside the suds library internal document
47    store under a hard-coded id to avoid having to load the data from a
48    temporary file.
49
50      Uses a locally created empty document store unless one is provided
51    externally using the 'documentStore' keyword argument.
52
53      Explicitly disables caching or otherwise, because we use the same
54    hardcoded id for our main WSDL document, suds would always reuse the first
55    such local document from its cache instead of fetching it from our document
56    store.
57
58    """
59    assert wsdl_content.__class__ is suds.byte_str_class, "bad test data"
60    store = kwargs.get("documentStore")
61    if store is None:
62        store = suds.store.DocumentStore()
63        kwargs.update(documentStore=store)
64    test_file_id = "whatchamacallit"
65    store.update({test_file_id: wsdl_content})
66    kwargs.update(cache=None)
67    return suds.client.Client("suds://" + test_file_id, *args, **kwargs)
68
69
70def run_test_process(script):
71    """
72    Runs the given Python test script as a separate process.
73
74    Expects the script to return an exit code 0 and output nothing on either
75    stdout or stderr output streams.
76
77    """
78    popen = subprocess.Popen([sys.executable], stdin=subprocess.PIPE,
79        stderr=subprocess.PIPE, stdout=subprocess.PIPE, cwd=script.dirname,
80        universal_newlines=True)
81    sys_path = sys.path
82    for i in range(len(sys_path)):
83        if not sys_path[i]:
84            sys_path[i] = os.getcwd()
85    out, err = popen.communicate("""\
86import sys
87sys.path = %(sys.path)s
88import suds
89if suds.__version__ != %(suds.__version__)r:
90    print("Unexpected suds version imported - '%%s'." %% (suds.__version__))
91    sys.exit(-2)
92
93if sys.version_info >= (3, 0):
94    def exec_file(x):
95        e = getattr(__builtins__, "exec")
96        return e(open(x).read(), globals(), globals())
97else:
98    exec_file = execfile
99exec_file(%(script)r)
100""" % {"suds.__version__": suds.__version__,
101    "script": script.basename,
102    "sys.path": sys_path})
103    if popen.returncode != 0 or err or out:
104        if popen.returncode != 0:
105            print("Test process exit code: %d" % (popen.returncode,))
106        if out:
107            print("Test process stdout:")
108            print(out)
109        if err:
110            print("Test process stderr:")
111            print(err)
112        import pytest
113        pytest.fail("Test subprocess failed.")
114
115
116def run_using_pytest(caller_globals):
117    """Run the caller test script using the pytest testing framework."""
118    import sys
119    # Trick setuptools into not recognizing we are referencing __file__ here.
120    # If setuptools detects __file__ usage in a module, any package containing
121    # this module will be installed as an actual folder instead of a zipped
122    # archive. This __file__ usage is safe since it is used only when a script
123    # has been run directly, and that can not be done from a zipped package
124    # archive.
125    filename = caller_globals.get("file".join(["__"] * 2))
126    if not filename:
127        sys.exit("Internal error: can not determine test script name.")
128    try:
129        import pytest
130    except ImportError:
131        filename = filename or "<unknown-script>"
132        sys.exit("'py.test' unit testing framework not available. Can not run "
133            "'%s' directly as a script." % (filename,))
134    exit_code = pytest.main(["--pyargs", filename] + sys.argv[1:])
135    sys.exit(exit_code)
136
137
138def wsdl(schema_content, input=None, output=None, operation_name="f",
139        wsdl_target_namespace="my-wsdl-namespace",
140        xsd_target_namespace="my-xsd-namespace",
141        web_service_URL="protocol://unga-bunga-location"):
142    """
143    Returns WSDL schema content used in different suds library tests.
144
145    Defines a single operation taking an externally specified input structure
146    and returning an externally defined output structure.
147
148    Constructed WSDL schema's XML namespace prefixes:
149      * my_wsdl - the WSDL schema's target namespace.
150      * my_xsd - the embedded XSD schema's target namespace.
151
152    input/output parameters accept the following values:
153      * None - operation has no input/output message.
154      * list/tuple - operation has an input/output message consisting of
155        message parts referencing top-level XSD schema elements with the given
156        names.
157      * Otherwise operation has an input/output message consisting of a single
158        message part referencing a top-level XSD schema element with the given
159        name.
160
161    """
162    assert isinstance(schema_content, six.string_types)
163
164    has_input = input is not None
165    has_output = output is not None
166
167    wsdl = ["""\
168<?xml version='1.0' encoding='UTF-8'?>
169<wsdl:definitions targetNamespace="%(wsdl_target_namespace)s"
170    xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
171    xmlns:my_wsdl="%(wsdl_target_namespace)s"
172    xmlns:my_xsd="%(xsd_target_namespace)s"
173    xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/">
174  <wsdl:types>
175    <xsd:schema targetNamespace="%(xsd_target_namespace)s"
176        elementFormDefault="qualified"
177        xmlns:xsd="http://www.w3.org/2001/XMLSchema">
178%(schema_content)s
179    </xsd:schema>
180  </wsdl:types>""" % dict(schema_content=schema_content,
181        wsdl_target_namespace=wsdl_target_namespace,
182        xsd_target_namespace=xsd_target_namespace)]
183
184    if has_input:
185        if input.__class__ not in (list, tuple):
186            input = [input]
187        wsdl.append("""\
188  <wsdl:message name="fRequestMessage">""")
189        for element in input:
190            wsdl.append("""\
191    <wsdl:part name="parameters" element="my_xsd:%s" />""" % (element,))
192        wsdl.append("""\
193  </wsdl:message>""")
194
195    if has_output:
196        if output.__class__ not in (list, tuple):
197            output = [output]
198        wsdl.append("""\
199  <wsdl:message name="fResponseMessage">""")
200        for element in output:
201            wsdl.append("""\
202    <wsdl:part name="parameters" element="my_xsd:%s" />""" % (element,))
203        wsdl.append("""\
204  </wsdl:message>""")
205
206    wsdl.append("""\
207  <wsdl:portType name="dummyPortType">
208    <wsdl:operation name="%s">""" % (operation_name,))
209
210    if has_input:
211        wsdl.append("""\
212      <wsdl:input message="my_wsdl:fRequestMessage" />""")
213    if has_output:
214        wsdl.append("""\
215      <wsdl:output message="my_wsdl:fResponseMessage" />""")
216
217    wsdl.append("""\
218    </wsdl:operation>
219  </wsdl:portType>
220  <wsdl:binding name="dummy" type="my_wsdl:dummyPortType">
221    <soap:binding style="document"
222        transport="http://schemas.xmlsoap.org/soap/http" />
223    <wsdl:operation name="%s">
224      <soap:operation soapAction="my-soap-action" style="document" />""" %
225        (operation_name,))
226
227    if has_input:
228        wsdl.append("""\
229      <wsdl:input><soap:body use="literal" /></wsdl:input>""")
230    if has_output:
231        wsdl.append("""\
232      <wsdl:output><soap:body use="literal" /></wsdl:output>""")
233
234    wsdl.append("""\
235    </wsdl:operation>
236  </wsdl:binding>
237  <wsdl:service name="dummy">
238    <wsdl:port name="dummy" binding="my_wsdl:dummy">
239      <soap:address location="%s" />
240    </wsdl:port>
241  </wsdl:service>
242</wsdl:definitions>
243""" % (web_service_URL,))
244
245    return suds.byte_str("\n".join(wsdl))
246