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