1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
from unittest import TestCase, mock
from tests.fixtures.calculator import (
Add,
CalculatorSoapAdd,
CalculatorSoapAddInput,
CalculatorSoapAddOutput,
)
from tests.fixtures.hello import HelloGetHelloAsString
from xsdata.exceptions import ClientValueError
from xsdata.formats.dataclass.client import Client, Config, TransportTypes
from xsdata.formats.dataclass.transports import DefaultTransport
response = """
<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope
xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<soap:Body>
<AddResponse xmlns="http://tempuri.org/">
<AddResult>7</AddResult>
</AddResponse>
</soap:Body>
</soap:Envelope>"""
class ClientTests(TestCase):
def test_from_service(self):
client = Client.from_service(CalculatorSoapAdd, location="http://testurl.com")
actual = client.config._asdict()
expected = {
"style": "document",
"input": CalculatorSoapAddInput,
"location": "http://testurl.com",
"soap_action": "http://tempuri.org/Add",
"output": CalculatorSoapAddOutput,
"transport": "http://schemas.xmlsoap.org/soap/http",
"encoding": None,
}
self.assertEqual(expected, actual)
@mock.patch.object(DefaultTransport, "post")
def test_send_with_dict_params(self, mock_post):
mock_post.return_value = response.encode()
client = Client.from_service(CalculatorSoapAdd)
params = {"Body": {"Add": {"intA": 3, "intB": 4}}}
result = client.send(params, headers={"User-Agent": "xsdata"})
self.assertIsInstance(result, CalculatorSoapAddOutput)
self.assertEqual(7, result.body.add_response.add_result)
obj = CalculatorSoapAddInput(body=CalculatorSoapAddInput.Body(add=Add(3, 4)))
request = client.serializer.render(obj)
mock_post.assert_called_once_with(
"http://www.dneonline.com/calculator.asmx",
data=request,
headers={
"User-Agent": "xsdata",
"content-type": "text/xml",
"SOAPAction": "http://tempuri.org/Add",
},
)
@mock.patch.object(DefaultTransport, "post")
def test_send_with_instance_object(self, mock_post):
mock_post.return_value = response.encode()
client = Client.from_service(CalculatorSoapAdd)
obj = CalculatorSoapAddInput(body=CalculatorSoapAddInput.Body(add=Add(3, 4)))
result = client.send(obj)
self.assertIsInstance(result, CalculatorSoapAddOutput)
self.assertEqual(7, result.body.add_response.add_result)
request = client.serializer.render(obj)
mock_post.assert_called_once_with(
"http://www.dneonline.com/calculator.asmx",
data=request,
headers={
"content-type": "text/xml",
"SOAPAction": "http://tempuri.org/Add",
},
)
def test_prepare_payload_with_encoding(self):
client = Client.from_service(HelloGetHelloAsString, encoding="utf-8")
result = client.prepare_payload(
{"Body": {"getHelloAsString": {"arg0": "Χριστόδουλος"}}}
)
self.assertIn("Χριστόδουλος".encode(), result)
def test_prepare_payload_raises_error_with_type_mismatch(self):
client = Client.from_service(CalculatorSoapAdd)
with self.assertRaises(ClientValueError) as cm:
client.prepare_payload(CalculatorSoapAddOutput())
self.assertEqual(
"Invalid input service type, expected "
"`CalculatorSoapAddInput` got `CalculatorSoapAddOutput`",
str(cm.exception),
)
def test_prepare_headers(self):
config = Config(
style="document",
location="",
transport=TransportTypes.SOAP,
soap_action="",
input=None,
output=None,
)
client = Client(config=config)
headers = {"foo": "bar"}
result = client.prepare_headers(headers)
self.assertEqual({"content-type": "text/xml", "foo": "bar"}, result)
self.assertEqual(1, len(headers))
config = config._replace(soap_action="add")
client = Client(config=config)
result = client.prepare_headers({})
self.assertEqual({"SOAPAction": "add", "content-type": "text/xml"}, result)
def test_prepare_headers_raises_error_with_unsupported_binding_transport(self):
config = Config.from_service(CalculatorSoapAdd, transport="foobar")
client = Client(config=config)
with self.assertRaises(ClientValueError) as cm:
client.prepare_headers({})
self.assertEqual("Unsupported binding transport: `foobar`", str(cm.exception))
|
