Merging upstream version 2.5.0.

This commit is contained in:
Mathias Behrle 2018-03-30 18:26:53 +02:00
parent 039b5b30b2
commit af618ce5e3
51 changed files with 402 additions and 256 deletions

12
CHANGES
View File

@ -1,3 +1,15 @@
2.5.0 (2018-01-06)
------------------
- Fix AnyType value rendering by guessing the xsd type for the value (#552)
- Fix AnySimpleType.xmlvalue() not implemented exception (#644)
- Add __dir__ method to value objects returned by Zeep
- Don't require content for 201 and 202 status codes (#613)
- Fix wheel package by cleaning the build directory correctly (#634)
- Handle Nil values on complexType with SimpleContent elements (#604)
- Add Client.namespaces method to list all namespaces available
- Improve support for auto-completion (#537)
2.4.0 (2017-08-26)
------------------
- Add support for tornado async transport via gen.coroutine (#530, Kateryna Burda)

View File

@ -1,6 +1,6 @@
Metadata-Version: 1.1
Name: zeep
Version: 2.4.0
Version: 2.5.0
Summary: A modern/fast Python SOAP client based on lxml / requests
Home-page: http://docs.python-zeep.org
Author: Michael van Tellingen

View File

@ -4,4 +4,4 @@ import zeep
client = zeep.Client(
wsdl='http://ec.europa.eu/taxation_customs/vies/checkVatService.wsdl')
print(client.service.checkVat('NL', '170944128B01'))
print(client.service.checkVat('NL', '123456789B01'))

View File

@ -1,5 +1,5 @@
[bumpversion]
current_version = 2.4.0
current_version = 2.5.0
commit = true
tag = true
tag_name = {new_version}

View File

@ -39,7 +39,7 @@ tests_require = [
'pytest-tornado==0.4.5',
# Linting
'isort==4.2.5',
'isort==4.2.15',
'flake8==3.3.0',
'flake8-blind-except==0.1.1',
'flake8-debugger==1.4.0',
@ -58,7 +58,7 @@ with open('README.rst') as fh:
setup(
name='zeep',
version='2.4.0',
version='2.5.0',
description='A modern/fast Python SOAP client based on lxml / requests',
long_description=long_description,
author="Michael van Tellingen",
@ -78,7 +78,6 @@ setup(
package_dir={'': 'src'},
packages=['zeep'],
include_package_data=True,
license='MIT',
classifiers=[
'Development Status :: 5 - Production/Stable',

View File

@ -1,6 +1,6 @@
Metadata-Version: 1.1
Name: zeep
Version: 2.4.0
Version: 2.5.0
Summary: A modern/fast Python SOAP client based on lxml / requests
Home-page: http://docs.python-zeep.org
Author: Michael van Tellingen

View File

@ -28,7 +28,6 @@ src/zeep.egg-info/PKG-INFO
src/zeep.egg-info/SOURCES.txt
src/zeep.egg-info/dependency_links.txt
src/zeep.egg-info/not-zip-safe
src/zeep.egg-info/pbr.json
src/zeep.egg-info/requires.txt
src/zeep.egg-info/top_level.txt
src/zeep/asyncio/__init__.py

View File

@ -1 +0,0 @@
{"is_release": false, "git_version": "70e9199"}

View File

@ -22,7 +22,7 @@ pytest-cov==2.5.1
pytest==3.1.3
requests_mock>=0.7.0
pytest-tornado==0.4.5
isort==4.2.5
isort==4.2.15
flake8==3.3.0
flake8-blind-except==0.1.1
flake8-debugger==1.4.0

View File

@ -3,4 +3,4 @@ from zeep.transports import Transport # noqa
from zeep.plugins import Plugin # noqa
from zeep.xsd.valueobjects import AnyObject # noqa
__version__ = '2.4.0'
__version__ = '2.5.0'

View File

@ -4,11 +4,11 @@ Adds asyncio support to Zeep. Contains Python 3.5+ only syntax!
"""
import asyncio
import logging
from . import bindings
import aiohttp
from requests import Response
from zeep.asyncio import bindings
from zeep.exceptions import TransportError
from zeep.transports import Transport
from zeep.utils import get_version

View File

@ -14,6 +14,10 @@ class OperationProxy(object):
self._proxy = service_proxy
self._op_name = operation_name
@property
def __doc__(self):
return str(self._proxy._binding._operations[self._op_name])
def __call__(self, *args, **kwargs):
"""Call the operation with the given args and kwargs.
@ -67,6 +71,13 @@ class ServiceProxy(object):
raise AttributeError('Service has no operation %r' % key)
return OperationProxy(self, key)
def __dir__(self):
""" Return the names of the operations. """
return list(dir(super(ServiceProxy, self))
+ list(self.__dict__)
+ list(self._binding.port_type.operations))
# using list() on the dicts for Python 3 compatibility
class Factory(object):
def __init__(self, types, kind, namespace):
@ -133,6 +144,10 @@ class Client(object):
self._default_port_name = port_name
self._default_soapheaders = None
@property
def namespaces(self):
return self.wsdl.types.prefix_map
@property
def service(self):
"""The default ServiceProxy instance

View File

@ -1,6 +1,7 @@
from zeep.wsdl import bindings
from tornado import gen
from zeep.wsdl import bindings
__all__ = ['AsyncSoap11Binding', 'AsyncSoap12Binding']

View File

@ -4,12 +4,12 @@ Adds async tornado.gen support to Zeep.
"""
import logging
import urllib
from . import bindings
from tornado import gen, httpclient
from requests import Response, Session
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
from tornado import gen, httpclient
from zeep.tornado import bindings
from zeep.transports import Transport
from zeep.utils import get_version
from zeep.wsdl.utils import etree_to_string
@ -89,7 +89,7 @@ class TornadoAsyncTransport(Transport):
auth_password = self.session.password
auth_mode = 'digest'
else:
raise StandardError('Not supported authentication.')
raise Exception('Not supported authentication.')
# extracting client cert
client_cert = None
@ -111,7 +111,8 @@ class TornadoAsyncTransport(Transport):
'auth_username': auth_username,
'auth_password': auth_password,
'auth_mode': auth_mode,
'validate_cert': self.session.verify,
'validate_cert': self.session.verify is not None,
'ca_certs': self.session.verify,
'client_key': client_key,
'client_cert': client_cert
}
@ -130,4 +131,4 @@ class TornadoAsyncTransport(Transport):
new._content = response.body
new.status_code = response.code
new.headers = dict(response.headers.get_all())
return new
return new

View File

@ -131,7 +131,10 @@ class SoapBinding(Binding):
:type response: requests.Response
"""
if response.status_code != 200 and not response.content:
if response.status_code in (201, 202) and not response.content:
return None
elif response.status_code != 200 and not response.content:
raise TransportError(
u'Server returned HTTP status %d (no content available)'
% response.status_code,

View File

@ -255,8 +255,14 @@ class Definition(object):
for import_node in doc.findall("wsdl:import", namespaces=NSMAP):
namespace = import_node.get('namespace')
location = import_node.get('location')
location = absolute_location(location, self.location)
if not location:
logger.debug(
"Skipping import for namespace %s (empty location)",
namespace)
continue
location = absolute_location(location, self.location)
key = (namespace, location)
if key in self.wsdl._definitions:
self.imports[key] = self.wsdl._definitions[key]

View File

@ -55,7 +55,7 @@ class Any(Base):
qname = etree.QName(xmlelement.tag)
if context and context.schemas:
for context_schema in context.schemas:
if context_schema._has_schema_document(qname.namespace):
if context_schema.documents.has_schema_document_for_ns(qname.namespace):
schema = context_schema
break

View File

@ -225,7 +225,7 @@ class Element(Base):
path=render_path)
elif self.max_occurs != 'unbounded' and len(value) > self.max_occurs:
raise exceptions.ValidationError(
"Expected at most %d items (maxOccurs check)" % self.min_occurs,
"Expected at most %d items (maxOccurs check)" % self.max_occurs,
path=render_path)
for val in value:

View File

@ -28,7 +28,7 @@ class Schema(object):
self._transport = transport
self._documents = OrderedDict()
self.documents = _SchemaContainer()
self._prefix_map_auto = {}
self._prefix_map_custom = {}
@ -40,11 +40,12 @@ class Schema(object):
nodes = node
self.add_documents(nodes, location)
@property
def documents(self):
for documents in self._documents.values():
for document in documents:
yield document
def __repr__(self):
main_doc = self.root_document
if main_doc:
return '<Schema(location=%r, tns=%r)>' % (
main_doc._location, main_doc._target_namespace)
return '<Schema()>'
@property
def prefix_map(self):
@ -69,7 +70,7 @@ class Schema(object):
@property
def namespaces(self):
return set(self._documents.keys())
return self.documents.get_all_namespaces()
@property
def elements(self):
@ -99,20 +100,13 @@ class Schema(object):
yield type_
seen.add(type_.qname)
def __repr__(self):
main_doc = self.root_document
if main_doc:
return '<Schema(location=%r, tns=%r)>' % (
main_doc._location, main_doc._target_namespace)
return '<Schema()>'
def add_documents(self, schema_nodes, location):
documents = []
resolve_queue = []
for node in schema_nodes:
document = self.create_new_document(node, location)
documents.append(document)
resolve_queue.append(document)
for document in documents:
for document in resolve_queue:
document.resolve()
self._prefix_map_auto = self._create_prefix_map()
@ -199,19 +193,24 @@ class Schema(object):
return namespace
def create_new_document(self, node, url, base_url=None):
"""
:rtype: zeep.xsd.schema.SchemaDocument
"""
namespace = node.get('targetNamespace') if node is not None else None
if base_url is None:
base_url = url
schema = SchemaDocument(namespace, url, base_url)
self._add_schema_document(schema)
self.documents.add(schema)
schema.load(self, node)
return schema
def merge(self, schema):
"""Merge an other XSD schema in this one"""
for document in schema.documents:
self._add_schema_document(document)
self.documents.add(document)
self._prefix_map_auto = self._create_prefix_map()
def _load_default_documents(self):
@ -226,7 +225,7 @@ class Schema(object):
schema.register_element(cls.qname, instance)
schema._is_internal = True
self._add_schema_document(schema)
self.documents.add(schema)
return schema
def _get_instance(self, qname, method_name, name):
@ -277,7 +276,7 @@ class Schema(object):
'xsd': 'http://www.w3.org/2001/XMLSchema',
}
i = 0
for namespace in self._documents.keys():
for namespace in self.documents.get_all_namespaces():
if namespace is None or namespace in prefix_map.values():
continue
@ -285,30 +284,6 @@ class Schema(object):
i += 1
return prefix_map
def _has_schema_document(self, namespace):
"""Return a boolean if there is a SchemaDocumnet for the namespace.
:rtype: boolean
"""
return namespace in self._documents
def _add_schema_document(self, document):
logger.debug("Add document with tns %s to schema %s", document.namespace, id(self))
documents = self._documents.setdefault(document.namespace, [])
documents.append(document)
def _get_schema_document(self, namespace, location):
"""Return a list of SchemaDocument's for the given namespace AND
location.
:rtype: SchemaDocument
"""
for document in self._documents.get(namespace, []):
if document._location == location:
return document
def _get_schema_documents(self, namespace, fail_silently=False):
"""Return a list of SchemaDocument's for the given namespace.
@ -316,21 +291,81 @@ class Schema(object):
"""
if (
namespace not in self._documents
not self.documents.has_schema_document_for_ns(namespace)
and namespace in const.AUTO_IMPORT_NAMESPACES
):
logger.debug("Auto importing missing known schema: %s", namespace)
self.add_document_by_url(namespace)
if namespace not in self._documents:
return self.documents.get_by_namespace(namespace, fail_silently)
class _SchemaContainer(object):
"""Container instances to store multiple SchemaDocument objects per
namespace.
"""
def __init__(self):
self._instances = OrderedDict()
def __iter__(self):
for document in self.values():
yield document
def add(self, document):
"""Append a schema document
:param document: zeep.xsd.schema.SchemaDocument
"""
logger.debug("Add document with tns %s to schema %s", document.namespace, id(self))
documents = self._instances.setdefault(document.namespace, [])
documents.append(document)
def get_all_namespaces(self):
return self._instances.keys()
def get_by_namespace(self, namespace, fail_silently):
if namespace not in self._instances:
if fail_silently:
return []
raise exceptions.NamespaceError(
"No schema available for the namespace %r" % namespace)
return self._documents[namespace]
return self._instances[namespace]
def get_by_namespace_and_location(self, namespace, location):
"""Return list of SchemaDocument's for the given namespace AND
location.
:rtype: zeep.xsd.schema.SchemaDocument
"""
documents = self.get_by_namespace(namespace, fail_silently=True)
for document in documents:
if document._location == location:
return document
def has_schema_document_for_ns(self, namespace):
"""Return a boolean if there is a SchemaDocument for the namespace.
:rtype: boolean
"""
return namespace in self._instances
def values(self):
for documents in self._instances.values():
for document in documents:
yield document
class SchemaDocument(object):
"""A Schema Document consists of a set of schema components for a
specific target namespace.
"""
def __init__(self, namespace, location, base_url):
logger.debug("Init schema document for %r", location)
@ -340,6 +375,7 @@ class SchemaDocument(object):
self._target_namespace = namespace
self._is_internal = False
# Containers for specific types
self._attribute_groups = {}
self._attributes = {}
self._elements = {}
@ -365,10 +401,16 @@ class SchemaDocument(object):
return not bool(self._imports or self._types or self._elements)
def load(self, schema, node):
"""Load the XML Schema passed in via the node attribute.
:type schema: zeep.xsd.schema.Schema
:type node: etree._Element
"""
if node is None:
return
if not schema._has_schema_document(self._target_namespace):
if not schema.documents.has_schema_document_for_ns(self._target_namespace):
raise RuntimeError(
"The document needs to be registered in the schema before " +
"it can be loaded")
@ -415,44 +457,62 @@ class SchemaDocument(object):
_resolve_dict(self._types)
def register_import(self, namespace, schema):
"""Register an import for an other schema document.
:type namespace: str
:type schema: zeep.xsd.schema.SchemaDocument
"""
schemas = self._imports.setdefault(namespace, [])
schemas.append(schema)
def is_imported(self, namespace):
return namespace in self._imports
def register_type(self, name, value):
assert not isinstance(value, type)
assert value is not None
def register_type(self, qname, value):
"""Register a xsd.Type in this schema
if isinstance(name, etree.QName):
name = name.text
logger.debug("register_type(%r, %r)", name, value)
self._types[name] = value
:type qname: str or etree.QName
:type value: zeep.xsd.Type
def register_element(self, name, value):
if isinstance(name, etree.QName):
name = name.text
logger.debug("register_element(%r, %r)", name, value)
self._elements[name] = value
"""
self._add_component(qname, value, self._types, 'type')
def register_group(self, name, value):
if isinstance(name, etree.QName):
name = name.text
logger.debug("register_group(%r, %r)", name, value)
self._groups[name] = value
def register_element(self, qname, value):
"""Register a xsd.Element in this schema
def register_attribute(self, name, value):
if isinstance(name, etree.QName):
name = name.text
logger.debug("register_attribute(%r, %r)", name, value)
self._attributes[name] = value
:type qname: str or etree.QName
:type value: zeep.xsd.Element
def register_attribute_group(self, name, value):
if isinstance(name, etree.QName):
name = name.text
logger.debug("register_attribute_group(%r, %r)", name, value)
self._attribute_groups[name] = value
"""
self._add_component(qname, value, self._elements, 'element')
def register_group(self, qname, value):
"""Register a xsd.Element in this schema
:type qname: str or etree.QName
:type value: zeep.xsd.Element
"""
self._add_component(qname, value, self._groups, 'group')
def register_attribute(self, qname, value):
"""Register a xsd.Element in this schema
:type qname: str or etree.QName
:type value: zeep.xsd.Attribute
"""
self._add_component(qname, value, self._attributes, 'attribute')
def register_attribute_group(self, qname, value):
"""Register a xsd.Element in this schema
:type qname: str or etree.QName
:type value: zeep.xsd.Group
"""
self._add_component(qname, value, self._attribute_groups, 'attribute_group')
def get_type(self, qname):
"""Return a xsd.Type object from this schema
@ -460,7 +520,7 @@ class SchemaDocument(object):
:rtype: zeep.xsd.ComplexType or zeep.xsd.AnySimpleType
"""
return self._get_instance(qname, self._types, 'type')
return self._get_component(qname, self._types, 'type')
def get_element(self, qname):
"""Return a xsd.Element object from this schema
@ -468,7 +528,7 @@ class SchemaDocument(object):
:rtype: zeep.xsd.Element
"""
return self._get_instance(qname, self._elements, 'element')
return self._get_component(qname, self._elements, 'element')
def get_group(self, qname):
"""Return a xsd.Group object from this schema.
@ -476,7 +536,7 @@ class SchemaDocument(object):
:rtype: zeep.xsd.Group
"""
return self._get_instance(qname, self._groups, 'group')
return self._get_component(qname, self._groups, 'group')
def get_attribute(self, qname):
"""Return a xsd.Attribute object from this schema
@ -484,7 +544,7 @@ class SchemaDocument(object):
:rtype: zeep.xsd.Attribute
"""
return self._get_instance(qname, self._attributes, 'attribute')
return self._get_component(qname, self._attributes, 'attribute')
def get_attribute_group(self, qname):
"""Return a xsd.AttributeGroup object from this schema
@ -492,9 +552,15 @@ class SchemaDocument(object):
:rtype: zeep.xsd.AttributeGroup
"""
return self._get_instance(qname, self._attribute_groups, 'attributeGroup')
return self._get_component(qname, self._attribute_groups, 'attributeGroup')
def _get_instance(self, qname, items, item_name):
def _add_component(self, name, value, items, item_name):
if isinstance(name, etree.QName):
name = name.text
logger.debug("register_%s(%r, %r)", item_name, name, value)
items[name] = value
def _get_component(self, qname, items, item_name):
try:
return items[qname]
except KeyError:

View File

@ -94,7 +94,22 @@ class AnyType(Type):
return self
def xmlvalue(self, value):
return value
"""Guess the xsd:type for the value and use corresponding serializer"""
from zeep.xsd.types import builtins
available_types = [
builtins.String,
builtins.Boolean,
builtins.Decimal,
builtins.Float,
builtins.DateTime,
builtins.Date,
builtins.Time,
]
for xsd_type in available_types:
if isinstance(value, xsd_type.accepted_types):
return xsd_type().xmlvalue(value)
return str(value)
def pythonvalue(self, value, schema=None):
return value

View File

@ -4,7 +4,7 @@ import six
from lxml import etree
from zeep.exceptions import ValidationError
from zeep.xsd.const import xsd_ns
from zeep.xsd.const import Nil, xsd_ns, xsi_ns
from zeep.xsd.types.any import AnyType
logger = logging.getLogger(__name__)
@ -68,6 +68,9 @@ class AnySimpleType(AnyType):
'%s.pytonvalue() not implemented' % self.__class__.__name__)
def render(self, parent, value, xsd_type=None, render_path=None):
if value is Nil:
parent.set(xsi_ns('nil'), 'true')
return
parent.text = self.xmlvalue(value)
def signature(self, schema=None, standalone=True):
@ -76,7 +79,3 @@ class AnySimpleType(AnyType):
def validate(self, value, required=False):
if required and value is None:
raise ValidationError("Value is required")
def xmlvalue(self, value):
raise NotImplementedError(
'%s.xmlvalue() not implemented' % self.__class__.__name__)

View File

@ -13,6 +13,7 @@ class AnyObject(object):
:param value: The value
"""
def __init__(self, xsd_object, value):
self.xsd_obj = xsd_object
self.value = value
@ -110,6 +111,9 @@ class CompoundValue(object):
def __iter__(self):
return self.__values__.__iter__()
def __dir__(self):
return list(self.__values__.keys())
def __repr__(self):
return PrettyPrinter().pformat(self.__values__)
@ -194,7 +198,8 @@ def _process_signature(xsd_type, args, kwargs):
available_kwargs = set(kwargs.keys())
for element_name, element in xsd_type.elements_nested:
if element.accepts_multiple:
values = element.parse_kwargs(kwargs, element_name, available_kwargs)
values = element.parse_kwargs(
kwargs, element_name, available_kwargs)
else:
values = element.parse_kwargs(kwargs, None, available_kwargs)

View File

@ -170,7 +170,7 @@ class SchemaVisitor(object):
# Check if the schema is already imported before based on the
# namespace. Schema's without namespace are registered as 'None'
document = self.schema._get_schema_document(namespace, location)
document = self.schema.documents.get_by_namespace_and_location(namespace, location)
if document:
logger.debug("Returning existing schema: %r", location)
self.register_import(namespace, document)

View File

@ -1,10 +1,10 @@
import pytest
from pretend import stub
from lxml import etree
import aiohttp
import pytest
from aioresponses import aioresponses
from lxml import etree
from pretend import stub
from zeep import cache, asyncio, exceptions
from zeep import asyncio, exceptions
@pytest.mark.requests

View File

@ -43,6 +43,20 @@ def test_service_proxy_non_existing():
assert client_obj.service.NonExisting
def test_service_proxy_dir_operations():
client_obj = client.Client('tests/wsdl_files/soap.wsdl')
operations = [op for op in dir(client_obj.service) if not op.startswith('_')]
assert set(operations) == set(['GetLastTradePrice', 'GetLastTradePriceNoOutput'])
def test_operation_proxy_doc():
client_obj = client.Client('tests/wsdl_files/soap.wsdl')
assert (client_obj.service.GetLastTradePrice.__doc__
== 'GetLastTradePrice(tickerSymbol: xsd:string, '
'account: ns0:account, '
'country: ns0:country) -> price: xsd:float')
def test_open_from_file_object():
with open('tests/wsdl_files/soap_transport_err.wsdl', 'rb') as fh:
client_obj = client.Client(fh)

View File

@ -13,7 +13,7 @@ def test_factory_namespace():
def test_factory_no_reference():
client = Client('tests/wsdl_files/soap.wsdl')
factory = client.type_factory('http://example.com/stockquote.xsd')
obj_1 = client.get_type('ns0:ArrayOfAddress')()
obj_1.Address.append({
'NameFirst': 'J',

View File

@ -4,9 +4,7 @@ from collections import OrderedDict
from lxml import etree
from tests.utils import assert_nodes_equal, load_xml, render_node
from zeep import xsd
from six import binary_type
from zeep import helpers
from zeep import helpers, xsd
from zeep.helpers import serialize_object
@ -197,6 +195,6 @@ def test_create_xml_soap_map():
<value xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="xs:date">2016-01-14</value>
</item>
</document>
""" # noqa
""" # noqa
node = render_node(value._xsd_type, value)
assert_nodes_equal(expected, node)

View File

@ -1,5 +1,6 @@
from zeep.loader import parse_xml
from tests.utils import DummyTransport
from zeep.loader import parse_xml
def test_huge_text():
# libxml2>=2.7.3 has XML_MAX_TEXT_LENGTH 10000000 without XML_PARSE_HUGE

View File

@ -11,7 +11,7 @@ def test_dict():
'foo': '1',
'bar': {
'bala': 'qwe',
},
},
'x': [1, 2, 3, 4],
'y': [],
}

View File

@ -1,13 +1,9 @@
import io
import pytest
import requests_mock
from lxml import etree
from pretend import stub
from six import StringIO
from tests.utils import DummyTransport, assert_nodes_equal
from zeep import Client, wsdl
from zeep import Client
from zeep.transports import Transport
@ -74,10 +70,12 @@ def test_parse_multiref_soap_response():
<wsdl:operation name="TestOperation">
<soap:operation soapAction=""/>
<wsdl:input name="TestOperationRequest">
<soap:body use="encoded" encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" />
<soap:body use="encoded"
encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" />
</wsdl:input>
<wsdl:output name="TestOperationResponse">
<soap:body use="encoded" encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" />
<soap:body use="encoded"
encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" />
</wsdl:output>
</wsdl:operation>
</wsdl:binding>
@ -88,7 +86,7 @@ def test_parse_multiref_soap_response():
</wsdl:port>
</wsdl:service>
</wsdl:definitions>
""".strip())
""".strip()) # noqa
content = """
<?xml version="1.0"?>
@ -135,7 +133,6 @@ def test_parse_multiref_soap_response():
assert result.item_2.subitem_2 == 'bar'
@pytest.mark.requests
def test_parse_multiref_soap_response_child():
wsdl_file = io.StringIO(u"""
@ -218,7 +215,7 @@ def test_parse_multiref_soap_response_child():
</wsdl:port>
</wsdl:service>
</wsdl:definitions>
""".strip())
""".strip()) # noqa
content = """
<?xml version="1.0"?>
@ -247,7 +244,7 @@ def test_parse_multiref_soap_response_child():
</multiRef>
</soapenv:Body>
</soapenv:Envelope>
""".strip()
""".strip() # noqa
client = Client(wsdl_file, transport=Transport(),)
response = stub(
@ -264,4 +261,3 @@ def test_parse_multiref_soap_response_child():
assert result.item_2.subitem_1.subitem_1 == 'foo'
assert result.item_2.subitem_1.subitem_2 == 'bar'
assert result.item_2.subitem_2 == 'bar'

View File

@ -1,11 +1,13 @@
import io
from requests_toolbelt.multipart.decoder import MultipartDecoder
from pretend import stub
import requests_mock
from lxml import etree
from tests.utils import load_xml, assert_nodes_equal
from pretend import stub
from requests_toolbelt.multipart.decoder import MultipartDecoder
from six import StringIO
from tests.utils import assert_nodes_equal
from zeep import Client
from zeep.transports import Transport
from zeep.wsdl.attachments import MessagePack
from zeep.wsdl.messages import xop
@ -47,11 +49,6 @@ def test_rebuild_xml():
'Content-Type': 'multipart/related; boundary=MIME_boundary; type="application/soap+xml"; start="<claim@insurance.com>" 1'
}
)
client = stub(
transport=None,
wsdl=stub(strict=True),
xml_huge_tree=False)
decoder = MultipartDecoder(
response.content, response.headers['Content-Type'], 'utf-8')
@ -74,16 +71,6 @@ def test_rebuild_xml():
assert_nodes_equal(etree.tostring(document), expected)
import pytest
import requests_mock
from six import StringIO
from zeep import Client
from zeep.transports import Transport
def test_xop():
wsdl_main = StringIO("""
<?xml version="1.0"?>
@ -238,8 +225,8 @@ def test_xop():
print(response1)
with requests_mock.mock() as m:
m.post('http://tests.python-zeep.org/test',
content=response2.encode("utf-8"),
headers={"Content-Type": content_type})
content=response2.encode("utf-8"),
headers={"Content-Type": content_type})
result = service.TestOperation2("")
assert result["_value_1"] == "BINARYDATA".encode()
@ -249,5 +236,3 @@ def test_xop():
headers={"Content-Type": content_type})
result = service.TestOperation1("")
assert result == "BINARYDATA".encode()

View File

@ -1,11 +1,11 @@
import pytest
from pretend import stub
from lxml import etree
from tornado.httpclient import HTTPResponse, HTTPRequest
from tornado.testing import gen_test, AsyncTestCase
from tornado.concurrent import Future
from mock import patch
from pretend import stub
from tornado.concurrent import Future
from tornado.httpclient import HTTPRequest, HTTPResponse
from tornado.testing import AsyncTestCase, gen_test
from zeep.tornado import TornadoAsyncTransport

View File

@ -5,7 +5,6 @@ from pretend import stub
from zeep import cache, transports
@pytest.mark.requests
def test_no_cache():
transport = transports.Transport(cache=None)

View File

@ -3,7 +3,8 @@ import io
import pytest
from lxml import etree
from tests.utils import DummyTransport, assert_nodes_equal, load_xml, render_node
from tests.utils import (
DummyTransport, assert_nodes_equal, load_xml, render_node)
from zeep import xsd

View File

@ -65,58 +65,6 @@ def test_parse():
assert operation.output.signature(as_output=True) == 'xsd:string'
def test_empty_input_parse():
wsdl_content = StringIO("""
<definitions xmlns="http://schemas.xmlsoap.org/wsdl/"
xmlns:tns="http://tests.python-zeep.org/tns"
xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
targetNamespace="http://tests.python-zeep.org/tns">
<types>
<xsd:schema targetNamespace="http://tests.python-zeep.org/tns">
<xsd:element name="Request" type="xsd:string"/>
<xsd:element name="Response" type="xsd:string"/>
</xsd:schema>
</types>
<message name="Input"/>
<message name="Output">
<part element="tns:Response"/>
</message>
<portType name="TestPortType">
<operation name="TestOperation">
<input message="Input"/>
<output message="Output"/>
</operation>
</portType>
<binding name="TestBinding" type="tns:TestPortType">
<soap:binding style="document" transport="http://schemas.xmlsoap.org/soap/http"/>
<operation name="TestOperation">
<soap:operation soapAction=""/>
<input>
<soap:body use="literal"/>
</input>
<output>
<soap:body use="literal"/>
</output>
</operation>
</binding>
</definitions>
""".strip())
root = wsdl.Document(wsdl_content, None)
binding = root.bindings['{http://tests.python-zeep.org/tns}TestBinding']
operation = binding.get('TestOperation')
assert operation.input.body.signature(schema=root.types) == 'soap-env:Body()'
assert operation.input.header.signature(schema=root.types) == 'soap-env:Header()'
assert operation.input.envelope.signature(schema=root.types) == 'soap-env:envelope(body: {})'
assert operation.input.signature(as_output=False) == ''
def test_parse_with_header():
wsdl_content = StringIO("""
<definitions xmlns="http://schemas.xmlsoap.org/wsdl/"

View File

@ -1,14 +1,13 @@
# -*- coding: utf-8 -*-
import platform
import pytest
from lxml import etree
from pretend import stub
from tests.utils import load_xml
from zeep import Client
from zeep.exceptions import Fault
from zeep.exceptions import TransportError
from zeep.exceptions import Fault, TransportError
from zeep.wsdl import bindings
@ -167,6 +166,8 @@ def test_no_content_type():
assert result == 120.123
@pytest.mark.skipif(platform.python_implementation() == 'PyPy',
reason="Fails on PyPy")
def test_wrong_content():
data = """
The request is answered something unexpected,
@ -190,6 +191,8 @@ def test_wrong_content():
assert data == exc.value.content
@pytest.mark.skipif(platform.python_implementation() == 'PyPy',
reason="Fails on PyPy")
def test_wrong_no_unicode_content():
data = """
The request is answered something unexpected,
@ -214,6 +217,8 @@ def test_wrong_no_unicode_content():
assert data == exc.value.content
@pytest.mark.skipif(platform.python_implementation() == 'PyPy',
reason="Fails on PyPy")
def test_http_error():
data = """
Unauthorized!
@ -384,3 +389,35 @@ def test_unexpected_headers():
assert result.body.price == 120.123
assert result.header.body is None
assert len(result.header._raw_elements) == 1
def test_response_201():
client = Client('tests/wsdl_files/soap_header.wsdl')
binding = client.service._binding
response = stub(
status_code=201,
content='',
encoding='utf-8',
headers={}
)
result = binding.process_reply(
client, binding.get('GetLastTradePrice'), response)
assert result is None
def test_response_202():
client = Client('tests/wsdl_files/soap_header.wsdl')
binding = client.service._binding
response = stub(
status_code=202,
content='',
encoding='utf-8',
headers={}
)
result = binding.process_reply(
client, binding.get('GetLastTradePrice'), response)
assert result is None

View File

@ -2,7 +2,6 @@ import os
import sys
import pytest
from lxml import etree
from tests.utils import load_xml
from zeep import wsse

View File

@ -72,6 +72,24 @@ def test_element_simple_type():
assert_nodes_equal(expected, node)
def test_complex_type():
custom_type = xsd.ComplexType(
xsd.Sequence([
xsd.Element(
etree.QName('http://tests.python-zeep.org/', 'username'),
xsd.String()),
xsd.Element(
etree.QName('http://tests.python-zeep.org/', 'password'),
xsd.String()),
])
)
obj = custom_type('user', 'pass')
assert {key: obj[key] for key in obj} == {
'username': 'user',
'password': 'pass'
}
def test_nil_elements():
custom_type = xsd.Element(
'{http://tests.python-zeep.org/}container',
@ -149,8 +167,6 @@ def test_invalid_kwarg_simple_type():
elm(something='is-wrong')
def test_any():
some_type = xsd.Element(
etree.QName('http://tests.python-zeep.org/', 'doei'),

View File

@ -26,7 +26,6 @@ def get_any_schema():
"""))
def test_default_xsd_type():
schema = xsd.Schema(load_xml("""
<?xml version="1.0"?>
@ -257,20 +256,21 @@ def test_element_any_type():
schema = xsd.Schema(node)
container_elm = schema.get_element('{http://tests.python-zeep.org/}container')
obj = container_elm(something='bar')
obj = container_elm(something=datetime.time(18, 29, 59))
node = etree.Element('document')
container_elm.render(node, obj)
expected = """
<document>
<ns0:container xmlns:ns0="http://tests.python-zeep.org/">
<ns0:something>bar</ns0:something>
<ns0:something>18:29:59</ns0:something>
</ns0:container>
</document>
"""
assert_nodes_equal(expected, node)
item = container_elm.parse(node.getchildren()[0], schema)
assert item.something == 'bar'
assert item.something == '18:29:59'
def test_element_any_type_unknown_type():
node = etree.fromstring("""

View File

@ -1,5 +1,5 @@
from lxml import etree
import pytest
from lxml import etree
from tests.utils import assert_nodes_equal, load_xml, render_node
from zeep import xsd
@ -295,3 +295,38 @@ def test_xml_unparsed_elements():
obj = container_elm.parse(expected[0], schema)
assert obj.item == 'bar'
assert obj._raw_elements
def test_xml_simple_content_nil():
schema = xsd.Schema(load_xml("""
<?xml version="1.0"?>
<schema xmlns="http://www.w3.org/2001/XMLSchema"
xmlns:tns="http://tests.python-zeep.org/"
targetNamespace="http://tests.python-zeep.org/"
elementFormDefault="qualified">
<element name="container" nillable="true">
<complexType>
<simpleContent>
<restriction base="string">
<maxLength value="1000"/>
</restriction>
</simpleContent>
</complexType>
</element>
</schema>
"""))
schema.set_ns_prefix('tns', 'http://tests.python-zeep.org/')
container_elm = schema.get_element('tns:container')
obj = container_elm(xsd.Nil)
result = render_node(container_elm, obj)
expected = """
<document>
<ns0:container xmlns:ns0="http://tests.python-zeep.org/" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:nil="true" />
</document>
"""
result = render_node(container_elm, obj)
assert_nodes_equal(result, expected)
obj = container_elm.parse(result[0], schema)
assert obj._value_1 is None

View File

@ -1,6 +1,6 @@
from lxml import etree
from tests.utils import assert_nodes_equal, render_node, load_xml
from tests.utils import assert_nodes_equal, load_xml, render_node
from zeep import xsd

View File

@ -1,10 +1,11 @@
from collections import deque
import pytest
from lxml import etree
from tests.utils import assert_nodes_equal, load_xml, render_node
from zeep import xsd
from zeep.exceptions import XMLParseError, ValidationError
from zeep.exceptions import ValidationError, XMLParseError
from zeep.helpers import serialize_object
@ -95,6 +96,7 @@ def test_choice_element_second_elm():
assert value.item_2 == 'foo'
assert value.item_3 is None
def test_choice_element_second_elm_positional():
node = etree.fromstring("""
<?xml version="1.0"?>
@ -768,6 +770,8 @@ def test_choice_with_sequence_change():
element.render(node, elm)
assert_nodes_equal(expected, node)
value = element.parse(node[0], schema)
assert value.item_1 == 'bla-1'
assert value.item_2 == 'bla-2'
def test_choice_with_sequence_change_named():
@ -813,6 +817,8 @@ def test_choice_with_sequence_change_named():
element.render(node, elm)
assert_nodes_equal(expected, node)
value = element.parse(node[0], schema)
assert value.item_1 == 'bla-1'
assert value.item_2 == 'bla-2'
def test_choice_with_sequence_multiple():

View File

@ -1,7 +1,7 @@
import pytest
from lxml import etree
from tests.utils import assert_nodes_equal, render_node, load_xml
from tests.utils import assert_nodes_equal, load_xml, render_node
from zeep import xsd

View File

@ -1,7 +1,7 @@
import pytest
from lxml import etree
from tests.utils import load_xml, render_node, assert_nodes_equal
from tests.utils import assert_nodes_equal, load_xml, render_node
from zeep import xsd
@ -81,7 +81,6 @@ def test_build_min_occurs_2_max_occurs_2():
assert custom_type.signature()
elm = custom_type(_value_1=[
{'item_1': 'foo-1', 'item_2': 'bar-1'},
{'item_1': 'foo-2', 'item_2': 'bar-2'},

View File

@ -198,7 +198,6 @@ def test_sequence_parse_anytype_regression_17():
assert result.getCustomFieldReturn.value.content == 'Test Solution'
def test_nested_complex_type():
custom_type = xsd.Element(
etree.QName('http://tests.python-zeep.org/', 'authentication'),
@ -324,13 +323,13 @@ def test_nested_choice_optional():
etree.QName('http://tests.python-zeep.org/', 'item_1'),
xsd.String()),
xsd.Choice([
xsd.Element(
'{http://tests.python-zeep.org/}item_2',
xsd.String()),
xsd.Element(
'{http://tests.python-zeep.org/}item_3',
xsd.String()),
],
xsd.Element(
'{http://tests.python-zeep.org/}item_2',
xsd.String()),
xsd.Element(
'{http://tests.python-zeep.org/}item_3',
xsd.String()),
],
min_occurs=0, max_occurs=1
),
])

View File

@ -1,11 +1,11 @@
import pytest
from lxml import etree
from tests.utils import DummyTransport, load_xml
from tests.utils import (
DummyTransport, assert_nodes_equal, load_xml, render_node)
from zeep import exceptions, xsd
from zeep.xsd import Schema
from zeep.xsd.types.unresolved import UnresolvedType
from tests.utils import assert_nodes_equal, load_xml, render_node
def test_default_types():
@ -776,6 +776,7 @@ def test_include_different_form_defaults():
""")
assert_nodes_equal(expected, node)
def test_merge():
node_a = etree.fromstring("""
<?xml version="1.0"?>

View File

@ -1,7 +1,7 @@
from lxml import etree
from zeep import xsd
from tests.utils import load_xml
from zeep import xsd
def test_signature_complex_type_choice():
@ -189,6 +189,7 @@ def test_signature_complex_type_sequence_with_anys():
'({item_1: xsd:string} | {item_2: {_value_1: ANY, _value_2: ANY}})' +
')')
def test_schema_recursive_ref():
schema = xsd.Schema(load_xml("""
<?xml version="1.0"?>
@ -211,4 +212,3 @@ def test_schema_recursive_ref():
elm = schema.get_element('ns0:Container')
elm.signature(schema)

View File

@ -125,6 +125,7 @@ def test_restriction_anon():
"""
assert_nodes_equal(expected, node)
def test_simple_type_list():
schema = xsd.Schema(load_xml("""
<?xml version="1.0"?>

View File

@ -40,13 +40,6 @@ def test_simpletype_parse():
assert item.parse_xmlelement(node) is None
def test_simpletype_xmlvalue():
item = types.AnySimpleType()
with pytest.raises(NotImplementedError):
item.xmlvalue(None)
def test_simpletype_pythonvalue():
item = types.AnySimpleType()

View File

@ -75,7 +75,6 @@ def test_validate_required_attribute():
result = render_node(container_elm, obj)
assert 'The attribute item is not valid: Value is required (container.item)' in str(exc)
obj.item = 'bar'
result = render_node(container_elm, obj)

View File

@ -1,4 +1,3 @@
import json
import pickle
import pytest