import uuid
from io import StringIO
from pretend import stub
from tests.utils import DummyTransport, assert_nodes_equal
from zeep import Client, wsa, wsdl
from zeep.settings import Settings
def test_require_wsa(recwarn, monkeypatch):
monkeypatch.setattr(
uuid, "uuid4", lambda: uuid.UUID("ada686f9-5995-4088-bea4-239f694b2eaf")
)
wsdl_main = StringIO(
"""
Test service
""".strip()
)
client = stub(plugins=[], wsse=None)
transport = DummyTransport()
client = Client(wsdl_main, transport=transport)
binding = client.wsdl.services.get("TestService").ports.get("TestPortType").binding
envelope, headers = binding._create(
"TestOperation1",
args=["foo"],
kwargs={},
client=client,
options={"address": "http://tests.python-zeep.org/test"},
)
expected = """
urn:dummyRequest
urn:uuid:ada686f9-5995-4088-bea4-239f694b2eaf
http://tests.python-zeep.org/test
foo
"""
assert_nodes_equal(expected, envelope)
def test_duplicate_message_wsa(recwarn, monkeypatch):
monkeypatch.setattr(
uuid, "uuid4", lambda: uuid.UUID("ada686f9-5995-4088-bea4-239f694b2eaf")
)
wsdl_main = StringIO(
"""
Test service
""".strip()
)
client = stub(plugins=[], wsse=None)
transport = DummyTransport()
client = Client(wsdl_main, transport=transport)
binding = client.wsdl.services.get("TestService").ports.get("TestPortType").binding
envelope, headers = binding._create(
"TestOperation1",
args=["foo"],
kwargs={},
client=client,
options={"address": "http://tests.python-zeep.org/test"},
)
expected = """
urn:action1
urn:uuid:ada686f9-5995-4088-bea4-239f694b2eaf
http://tests.python-zeep.org/test
foo
"""
assert_nodes_equal(expected, envelope)
def test_force_wsa(recwarn, monkeypatch):
monkeypatch.setattr(
uuid, "uuid4", lambda: uuid.UUID("ada686f9-5995-4088-bea4-239f694b2eaf")
)
wsdl_main = StringIO(
"""
Test service
""".strip()
)
transport = DummyTransport()
client = Client(wsdl_main, transport=transport, plugins=[wsa.WsAddressingPlugin()])
binding = client.wsdl.services.get("TestService").ports.get("TestPortType").binding
envelope, headers = binding._create(
"TestOperation1",
args=["foo"],
kwargs={},
client=client,
options={"address": "http://tests.python-zeep.org/test"},
)
expected = """
urn:dummyRequest
urn:uuid:ada686f9-5995-4088-bea4-239f694b2eaf
http://tests.python-zeep.org/test
foo
"""
assert_nodes_equal(expected, envelope)
def test_force_wsa_soap12(recwarn, monkeypatch):
monkeypatch.setattr(
uuid, "uuid4", lambda: uuid.UUID("ada686f9-5995-4088-bea4-239f694b2eaf")
)
wsdl_main = StringIO(
"""
Test service
""".strip()
)
client = stub(plugins=[wsa.WsAddressingPlugin()], wsse=None, settings=Settings())
transport = DummyTransport()
doc = wsdl.Document(wsdl_main, transport, settings=client.settings)
binding = doc.services.get("TestService").ports.get("TestPortType").binding
envelope, headers = binding._create(
"TestOperation1",
args=["foo"],
kwargs={},
client=client,
options={"address": "http://tests.python-zeep.org/test"},
)
expected = """
urn:dummyRequest
urn:uuid:ada686f9-5995-4088-bea4-239f694b2eaf
http://tests.python-zeep.org/test
foo
"""
assert_nodes_equal(expected, envelope)
assert headers["Content-Type"] == (
'application/soap+xml; charset=utf-8; action="urn:dummyRequest"'
)
def test_require_wsa_new(recwarn, monkeypatch):
monkeypatch.setattr(
uuid, "uuid4", lambda: uuid.UUID("ada686f9-5995-4088-bea4-239f694b2eaf")
)
wsdl_main = StringIO(
"""
Test service
""".strip()
)
client = stub(plugins=[], wsse=None)
transport = DummyTransport()
client = Client(wsdl_main, transport=transport)
binding = client.wsdl.services.get("TestService").ports.get("TestPortType").binding
envelope, headers = binding._create(
"TestOperation1",
args=["foo"],
kwargs={},
client=client,
options={"address": "http://tests.python-zeep.org/test"},
)
expected = """
urn:dummyRequest
urn:uuid:ada686f9-5995-4088-bea4-239f694b2eaf
http://tests.python-zeep.org/test
foo
"""
assert_nodes_equal(expected, envelope)