diff --git a/.gitignore b/.gitignore index ff4a83d..ce47739 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,12 @@ *.pyc *~ *.swp +.pytest_cache/ +.vscode/ +.tox/ /.idea/ /Results/ /Tests/Results/ +/build/ +/dist/ +/robotframework_zeeplibrary.egg-info diff --git a/ZeepLibrary/zeeplibrary.py b/ZeepLibrary/zeeplibrary.py index b1d170b..d23c6aa 100644 --- a/ZeepLibrary/zeeplibrary.py +++ b/ZeepLibrary/zeeplibrary.py @@ -1,57 +1,74 @@ -import os -import zeep -import requests -import requests.auth -from requests import Session -from robot.api import logger -from robot.api.deco import keyword -from lxml import etree -from zeep import Client -from zeep.transports import Transport +"""Library to provide Robot Framework keywords for testing SOAP services. +Wraps the python zeep module's functionalities to accomplish this. +""" +import base64 import mimetypes +import os + from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase from email.mime.text import MIMEText from email.mime.application import MIMEApplication from email.mime.image import MIMEImage -from email.encoders import encode_7or8bit, encode_base64, encode_noop -import base64 +# from email.encoders import encode_7or8bit, encode_base64, encode_noop +from email.encoders import encode_7or8bit, encode_noop + +import requests +import requests.auth +import zeep + +from lxml import etree +# from requests import Session +from robot.api.deco import keyword +from robot.api import logger +# from zeep import Client +# from zeep.transports import Transport class ZeepLibraryException(Exception): + """Custom base class for ZeepLibrary exceptions. + New exceptions should base this and overwrite the err_msg attribute. + """ + def __init__(self): + self.err_msg = "Default error message" + def __str__(self): - return self.message + return self.err_msg class AliasAlreadyInUseException(ZeepLibraryException): + """Raise when an alias is already in use.""" def __init__(self, alias): - self.message = "The alias `{}' is already in use.".format(alias) + self.err_msg = "The alias '{}' is already in use.".format(alias) class ClientNotFoundException(ZeepLibraryException): + """Raise when a client could not be found with given alias.""" def __init__(self, alias): - self.message = "Could not find a client with alias `{}'."\ - .format(alias) + self.err_msg = "Could not find a client with alias '{}'.".format(alias) class AliasNotFoundException(ZeepLibraryException): + """Raise when an alias could not be found for a client.""" def __init__(self): - self.message = "Could not find alias for the provided client." + self.err_msg = "Could not find alias for the provided client." class AliasRequiredException(ZeepLibraryException): + """Raise when there is more than one client but no alias was specified.""" def __init__(self): - self.message = ("When using more than one client, providing an alias " + self.err_msg = ("When using more than one client, providing an alias " "is required.") -class ZeepLibrary: +class ZeepLibrary(object): """This library is built on top of the library Zeep in order to bring its functionality to Robot Framework. Following in the footsteps of the (now unmaintained) SudsLibrary, it allows testing SOAP communication. Zeep offers a more intuitive and modern approach than Suds does, and especially since the latter is unmaintained now, it seemed time to write a library to enable Robot Framework to use Zeep. + Note: inheriting from 'object' makes ZeepLibrary a new style class in py2. """ __version__ = '0.9.2' @@ -161,8 +178,8 @@ def add_attachment(self, else: file_mode = 'rt' - with open(filepath, file_mode) as f: - contents = f.read() + with open(filepath, file_mode) as file_object: + contents = file_object.read() attachment = { 'filename': filename, @@ -183,25 +200,22 @@ def call_operation(self, operation, xop=False, **kwargs): # xop=xop) # return original_post_method(address, body, headers) def post_with_attachments(address, message, headers): - message = self.create_message(operation, **kwargs) - logger.debug("HTTP Post to {}:\n".format(address)) - - headers, body = self._build_multipart_request(message, - xop=xop) - - response = self.active_client.transport.session.post(address, - data=body, - headers=headers, - timeout=self.active_client.transport.operation_timeout) - - logger.debug(_prettify_request(response.request)) - logger.debug("HTTP Response from {0} (status: {1}):\n".format(address, response.status_code)) - logger.debug(_prettify_response(response)) - - return response + message = self.create_message(operation, **kwargs) + logger.debug("HTTP Post to {}:\n".format(address)) + headers, body = self._build_multipart_request(message, xop=xop) + response = self.active_client.transport.session.post( + address, + data=body, + headers=headers, + timeout=self.active_client.transport.operation_timeout + ) + logger.debug(_prettify_request(response.request)) + logger.debug("HTTP Response from {0} (status: {1}):\n" + .format(address, response.status_code)) + logger.debug(_prettify_response(response)) + return response self.active_client.transport.post = post_with_attachments - operation_method = getattr(self.active_client.service, operation) return operation_method(**kwargs) @@ -217,7 +231,12 @@ def close_client(self, alias=None): @keyword('Close all clients') def close_all_clients(self): - for alias in self.clients.keys(): + # FIX + # old code produced error: dictionary changed size during iteration + # for alias in self.clients.keys(): + # self.close_client(alias) + aliases = list(self.clients.keys()) + for alias in aliases: self.close_client(alias) def _add_client(self, client, alias=None): @@ -255,13 +274,13 @@ def create_message(self, operation, to_string=True, **kwargs): operation, **kwargs) if to_string: - return etree.tostring(message) - else: - return message + # etree.tostring returns byte object without encoding kw arg. + message = etree.tostring(message, encoding='unicode') + return message @keyword('Create object') - def create_object(self, type, *args, **kwargs): - type_ = self.active_client.get_type(type) + def create_object(self, type_to_get, *args, **kwargs): + type_ = self.active_client.get_type(type_to_get) return type_(*args, **kwargs) @keyword('Get alias') @@ -269,7 +288,7 @@ def get_alias(self, client=None): if not client: return self.active_client_alias else: - for alias, client_ in self.clients.iteritems(): + for alias, client_ in self.clients.items(): if client_ == client: return alias raise AliasNotFoundException() @@ -281,9 +300,10 @@ def get_client(self, alias=None): If no ``alias`` is provided, the active client will be assumed. """ if alias: - return self.clients[alias] + client = self.clients[alias] else: - return self.active_client + client = self.active_client + return client @keyword('Get clients') def get_clients(self): @@ -291,9 +311,10 @@ def get_clients(self): @keyword('Get namespace prefix') def get_namespace_prefix_for_uri(self, uri): - for prefix, uri_ in self.active_client.namespaces.iteritems(): + for prefix, uri_ in self.active_client.namespaces.items(): if uri == uri_: return prefix + return None @keyword('Get namespace URI') def get_namespace_uri_by_prefix(self, prefix): @@ -338,13 +359,14 @@ def _log(item, to_log=True, to_console=False): elif to_console: logger.console(item) + def _perform_xop_magic(message): doc = etree.fromstring(message) for element in doc.iter(): if (element.text and - len(element.text) > 0 and - len(element.text) % 4 == 0): + len(element.text) > 0 and + len(element.text) % 4 == 0): try: decoded_val = base64.b64decode(element.text) if decoded_val.startswith('cid:'): @@ -356,43 +378,44 @@ def _perform_xop_magic(message): except TypeError: continue - message = etree.tostring(doc) + message = etree.tostring(doc, encoding='unicode') return message def _prettify_request(request, hide_auth=True): - """Pretty prints the request for the supplied `requests.Request` - object. Especially useful after having performed the request, in - order to inspect what was truly sent. To access the used request - on the `requests.Response` object use the `request` attribute. - """ - if hide_auth: - logger.warn(("Hiding the `Authorization' header for security " - "reasons. If you wish to display it anyways, pass " - "`hide_auth=False`.")) - result = ('{}\n{}\n{}\n\n{}{}'.format( - '----------- REQUEST BEGIN -----------', - request.method + ' ' + request.url, - '\n'.join('{}: {}'.format(key, value) - for key, value in request.headers.items() - if not(key == 'Authorization' and hide_auth)), - request.body, - "\n" - '------------ REQUEST END ------------' - )) - return result + """Pretty prints the request for the supplied `requests.Request` + object. Especially useful after having performed the request, in + order to inspect what was truly sent. To access the used request + on the `requests.Response` object use the `request` attribute. + """ + if hide_auth: + logger.warn(("Hiding the `Authorization' header for security " + "reasons. If you wish to display it anyways, pass " + "`hide_auth=False`.")) + result = ('{}\n{}\n{}\n\n{}{}'.format( + '----------- REQUEST BEGIN -----------', + request.method + ' ' + request.url, + '\n'.join('{}: {}'.format(key, value) + for key, value in request.headers.items() + if not(key == 'Authorization' and hide_auth)), + request.body, + "\n" + '------------ REQUEST END ------------' + )) + return result + def _prettify_response(reponse): - result = ('{}\n{}\n{}\n\n{}{}'.format( - '----------- RESPONSE BEGIN -----------', - reponse.url, - '\n'.join('{}: {}'.format(key, value) - for key, value in reponse.headers.items()), - reponse.text, - "\n" - '------------ RESPONSE END ------------' - )) - return result + result = ('{}\n{}\n{}\n\n{}{}'.format( + '----------- RESPONSE BEGIN -----------', + reponse.url, + '\n'.join('{}: {}'.format(key, value) + for key, value in reponse.headers.items()), + reponse.text, + "\n" + '------------ RESPONSE END ------------' + )) + return result def _add_or_replace_http_header_if_passed(mime_object, headers, key): diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e784d19 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +lxml +mock +mockito +pytest +pytest-mockito +robotframework>=3.2 +zeep>=2.5.0 \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/regression_set.robot b/tests/regression_set.robot index c514ca8..9799dac 100644 --- a/tests/regression_set.robot +++ b/tests/regression_set.robot @@ -69,12 +69,12 @@ Namespace trickery Log namespace prefix map ${client}= Get client - ${some prefix}= Set variable ${client.namespaces.keys()[0]} + ${some prefix}= Evaluate list(${client.namespaces})[0] ${uri}= Get namespace URI ${some prefix} Should start with ${uri} http - ${some uri}= Set variable ${client.namespaces.items()[0][1]} + ${some uri}= Set Variable ${{ list(${client.namespaces}.values()) }}[0] ${prefix}= Get namespace prefix ${some uri} Close client diff --git a/tests/sample.xml b/tests/sample.xml new file mode 100644 index 0000000..b5f26b0 --- /dev/null +++ b/tests/sample.xml @@ -0,0 +1,10 @@ + + + + This is as far as it gets. + + + Or maybe this. + + + diff --git a/tests/test_client_operations.py b/tests/test_client_operations.py new file mode 100644 index 0000000..1954787 --- /dev/null +++ b/tests/test_client_operations.py @@ -0,0 +1,55 @@ +"""Test ZeepLibrary's methods and properties related to client objects. + Note: 'monkeypatch' is a fixture from pytest.""" +import sys +import pytest +try: + from ZeepLibrary.zeeplibrary import ClientNotFoundException, ZeepLibrary +except ImportError: + from .ZeepLibrary.zeeplibrary import ClientNotFoundException, ZeepLibrary + + +def test_library_init(): + """Testing library initialization sets certain attributes.""" + zl_instance = ZeepLibrary() + assert ZeepLibrary.ROBOT_LIBRARY_SCOPE == 'GLOBAL' + assert isinstance(zl_instance._clients, dict) + assert len(zl_instance._clients) == 0 + assert zl_instance._active_client_alias is None + assert hasattr(ZeepLibrary, '__version__') + +def test_active_client(monkeypatch): + """Testing active client.""" + zl_instance = ZeepLibrary() + with monkeypatch.context() as mc: + mc.setitem(zl_instance._clients, "first", "value") + mc.setattr(zl_instance, "_active_client_alias", "first") + assert zl_instance.active_client == "value" + +def test_active_client_alias_getting(monkeypatch): + """Testing active client property getter.""" + zl_instance = ZeepLibrary() + with monkeypatch.context() as mc: + mc.setattr(zl_instance, "_active_client_alias", "first") + assert zl_instance.active_client_alias == "first" + +def test_active_client_alias_setting_succeeds(monkeypatch): + """Test setting the active client.""" + zl_instance = ZeepLibrary() + with monkeypatch.context() as mc: + mc.setitem(zl_instance._clients, "first", "value") + zl_instance.active_client_alias = "first" + assert zl_instance._active_client_alias == "first" + +def test_active_client_alias_setting_raises(): + """Test setting active client fails with non-existing alias.""" + zl_instance = ZeepLibrary() + with pytest.raises(ClientNotFoundException, match= + "Could not find a client with alias 'Alien'."): + zl_instance.active_client_alias = "Alien" + +def test_clients(monkeypatch): + """Test clients property getting.""" + zl_instance = ZeepLibrary() + with monkeypatch.context() as mc: + mc.setitem(zl_instance._clients, "first", "value") + assert zl_instance.clients == {"first": "value"} diff --git a/tests/test_custom_exceptions.py b/tests/test_custom_exceptions.py new file mode 100644 index 0000000..73cec71 --- /dev/null +++ b/tests/test_custom_exceptions.py @@ -0,0 +1,52 @@ +"""Test ZeepLibrary custom exceptions using pytest. +capsys is a pytest fixture used to capture stdout and stderr content.""" +try: + from ZeepLibrary.zeeplibrary import (AliasAlreadyInUseException, + AliasNotFoundException, + AliasRequiredException, + ClientNotFoundException, + ZeepLibraryException) +except ImportError: + from .ZeepLibrary.zeeplibrary import (AliasAlreadyInUseException, + AliasNotFoundException, + AliasRequiredException, + ClientNotFoundException, + ZeepLibraryException) + + +def test_alias_already_in_use(capsys): + """Test that custom error message is placed in proper attribute and + printing the exception produces the same error message.""" + excep = AliasAlreadyInUseException('Smith') + assert excep.err_msg == "The alias '{}' is already in use.".format( + 'Smith' + ) + print(excep) # to test the __str__ in base class ZeepLibraryException + captured = capsys.readouterr() + assert captured.out.strip('\n') == excep.err_msg + +def test_client_not_found(): + """Test that custom error message is placed in proper attribute.""" + excep = ClientNotFoundException('Smith') + assert excep.err_msg ==\ + "Could not find a client with alias '{}'.".format('Smith') + +def test_alias_not_found(capsys): + """Test that custom error message is placed in proper attribute and + printing the exception produces the same error message.""" + excep = AliasNotFoundException() + assert excep.err_msg == "Could not find alias for the provided client." + print(excep) # to test the __str__ in base class ZeepLibraryException + captured = capsys.readouterr() + assert captured.out.strip('\n') == excep.err_msg + +def test_alias_required(): + """Test that custom error message is placed in proper attribute.""" + excep = AliasRequiredException() + assert excep.err_msg == ("When using more than one client, " + "providing an alias is required.") + +def test_base_exception(): + """Test that custom error message is placed in proper attribute.""" + excep = ZeepLibraryException() + assert excep.err_msg == ("Default error message") diff --git a/tests/test_keywords.py b/tests/test_keywords.py new file mode 100644 index 0000000..ba03181 --- /dev/null +++ b/tests/test_keywords.py @@ -0,0 +1,524 @@ +"""Unit test ZeepLibrary keyword methods.""" +from copy import deepcopy +from lxml import etree +import pytest +import requests +import zeep +from mockito import mock, verify, when +try: + from ZeepLibrary.zeeplibrary import ZeepLibrary, AliasNotFoundException + from ZeepLibrary import zeeplibrary as zl_module +except ImportError: + from .ZeepLibrary.zeeplibrary import ZeepLibrary, AliasNotFoundException + from .ZeepLibrary import zeeplibrary as zl_module +try: + from mock import Mock +except (ImportError, ModuleNotFoundError): + from unittest.mock import Mock + +# Fixture definitions +@pytest.fixture(scope='function') +def attachment_dir(tmp_path): + """Provide a temporary directory with one text and one binary file.""" + text_file = tmp_path / 'text_file.txt' + text_file.write_text(u"This content is meant to be text.") + bin_file = tmp_path / 'bin_file.bmp' + bin_file.write_bytes(b'1010101010101010') + return tmp_path + +@pytest.fixture(scope='function') +def client_fixture(): + """Create an object to mock a client in ZeepLibrary instance.""" + class Clnt(object): + """Create an object to mock a client in ZeepLibrary instance.""" + def __init__(self): + self.attachments = [] + + mock_client = Clnt() + return mock_client + +@pytest.fixture(scope="function") +def zl_with_clients(client_fixture): + """Create a ZeepLibrary instance with four 'clients'.""" + zl_instance = ZeepLibrary() + aliases = ["first", "second", "third", "fourth"] + for alias in aliases: + client = deepcopy(client_fixture) + client.alias_name = alias + zl_instance._clients[alias] = client + zl_instance._active_client_alias = alias + return zl_instance + + +# Tests begin +@pytest.mark.parametrize( + ["file_name", "mime_type", "binary", "http_headers"], + [ + ("text_file.txt", "text/plain", False, "header1"), + ("bin_file.bmp", "image/bmp", True, "header2, header-bmp"), + ("", "text_plain", False, "header2, header-txt"), + ("bin_file.bmp", "", True, "header2, header-bmp"), + (None, None, None, None) + ], + ids=[ + "text file", + "binary file", + "no file name", + "no mime-type", + "only file_path" + ] +) +def test_add_attachment(file_name, mime_type, binary, http_headers, + attachment_dir, client_fixture, monkeypatch): + """Test add attachment -method.""" + # Arrange + zl_instance = ZeepLibrary() + zl_instance._clients = {'first': client_fixture} + zl_instance._active_client_alias = 'first' + file_path = attachment_dir / (file_name or 'text_file.txt') + file_path_str = str(file_path) + file_content = file_path.read_bytes() if binary else \ + file_path.read_text() + exp_mime_type = mime_type or 'image/bmp' + monkeypatch.setattr("ZeepLibrary.zeeplibrary._guess_mimetype", + lambda x: exp_mime_type) + expected = { + "filename": (file_name or 'text_file.txt'), + "contents": file_content, + "mimetype": exp_mime_type, + "http_headers": http_headers + } + # Act + zl_instance.add_attachment(file_path_str, + file_name, + mime_type, + binary, + http_headers) + # Assert + assert zl_instance.active_client.attachments[0] == expected + +@pytest.mark.parametrize( + "alias", + [ + ("first"), + (None), + ("cannot_be_found") + ], + ids=[ + "correct alias", + "no alias", + "non existent alias" + ] +) +def test_close_client(alias, client_fixture): + """Test closing a client, i.e. popping it out of client list. + There won't be a KeyError for non-existent aliases because close_client + uses a default (None) when popping clients out of dict.""" + # Arrange + zl_instance = ZeepLibrary() + zl_instance._clients = {"first": client_fixture, "second": client_fixture} + zl_instance._active_client_alias = 'first' + # Act + zl_instance.close_client(alias) + # Assert + assert alias not in zl_instance._clients.keys() + +@pytest.mark.parametrize( + "aliases", + [ + ([]), + (["first"]), + (["first", "second", "third"]) + ], + ids=[ + "zero aliases", + "one alias", + "three aliases" + ] +) +def test_close_all_clients(aliases, client_fixture): + """Test closing all opened clients. + """ + # Arrange + zl_instance = ZeepLibrary() + for alias in aliases: + zl_instance._clients[alias] = client_fixture + zl_instance._active_client_alias = alias + # Act + zl_instance.close_all_clients() + # Assert + assert zl_instance._clients == {} + +def test_create_client_no_auth(client_fixture): + """Test client creation. + Mocking out requests.Session, zeep.Client and + zeep.transport.Transport with mockito mocks.""" + # Arrange + zl_instance = ZeepLibrary() + wsdl = 'calculator.wsdl' + alias = 'first' + auth = None + proxies = '127.0.0.1' + cert = 'have_certificate_will_access' + verify = False + mock_session = mock(spec=requests.Session) + mock_session.cert = cert + mock_session.proxies = proxies + mock_session.verify = verify + when(requests).Session().thenReturn(mock_session) + mock_transport = mock(spec=zeep.transports.Transport) + when(zeep.transports).Transport(session=mock_session).thenReturn(mock_transport) + when(zeep).Client(wsdl, transport=mock_transport).thenReturn(client_fixture) + # Act + new_client = zl_instance.create_client(wsdl, alias, auth, proxies, + cert, verify) + # Assert + assert zl_instance._clients == {alias: client_fixture} + assert new_client == client_fixture + assert new_client.attachments == [] + +def test_create_client_with_auth(client_fixture): + """Test client creation. + Mocking out requests.Session, requests.auth.HTTPBasicAuth, + zeep.Client and zeep.transport.Transport with mockito mocks.""" + # Arrange + zl_instance = ZeepLibrary() + wsdl = 'calculator.wsdl' + alias = 'first' + auth = ('user', 'pwd') + proxies = '127.0.0.1' + cert = 'have_certificate_will_access' + verify = False + mock_session = mock(spec=requests.Session) + mock_session.cert = cert + mock_session.proxies = proxies + mock_session.verify = verify + mock_session.auth = auth + when(requests).Session().thenReturn(mock_session) + when(requests.auth).HTTPBasicAuth(auth[0], auth[1]).thenReturn(auth) + mock_transport = mock(spec=zeep.transports.Transport) + when(zeep.transports).Transport(session=mock_session).thenReturn(mock_transport) + when(zeep).Client(wsdl, transport=mock_transport).thenReturn(client_fixture) + # Act + new_client = zl_instance.create_client(wsdl, alias, auth, proxies, + cert, verify) + # Assert + assert zl_instance._clients == {alias: client_fixture} + assert new_client == client_fixture + assert new_client.attachments == [] + +def test_create_message_xml(client_fixture): + """Test message creation. Do not convert to string.""" + # Arrange + zl_instance = ZeepLibrary() + with open('tests/sample.xml', 'r') as f_o: + content = f_o.read() + xml_msg = etree.fromstring(content) + operation = 'operation' + kws = {'x': 1, 'y': '2'} + client_fixture.service = 'soap.service.com' + client_fixture.create_message = Mock(return_value=xml_msg) + zl_instance._clients['first'] = client_fixture + zl_instance._active_client_alias = 'first' + to_string = False + # Act + msg = zl_instance.create_message(operation, to_string, **kws) + # Assert + assert msg == xml_msg + client_fixture.create_message.assert_called_once_with( + zl_instance.active_client.service, operation, **kws + ) + +def test_create_message_to_string(client_fixture): + """Test message creation. Convert message to unicode string.""" + # Arrange + zl_instance = ZeepLibrary() + with open('tests/sample.xml', 'r') as f_o: + content = f_o.read() + xml_msg = etree.fromstring(content) + operation = 'operation' + kws = {'x': 1, 'y': '2'} + client_fixture.service = 'soap.service.com' + client_fixture.create_message = Mock(return_value=xml_msg) + zl_instance._clients['first'] = client_fixture + zl_instance._active_client_alias = 'first' + to_string = True + # Act + msg = zl_instance.create_message(operation, to_string, **kws) + # Assert + assert msg == etree.tostring(xml_msg, encoding='unicode') + client_fixture.create_message.assert_called_once_with( + zl_instance.active_client.service, operation, **kws + ) + +@pytest.mark.parametrize( + "type_to_get, args, kwargs", + [ + ("Add", [], {'x': '1'}), + ("Update", ["just_one"], {}), + ("Get", ["just_one"], {'x': '1'}), + ("You're my Type", ["just_one", "or_maybe_two", "one_more"], + {'x': '1', 'y':'2', 'z':'three'}) + ], + ids=[ + "no arg, one kwarg", + "one arg, no kwarg", + "one arg, one kwarg", + "many args, many kwargs" + ] +) +def test_create_object(type_to_get, args, kwargs, client_fixture): + """Test creating an object by mocking get_type() call.""" + # Arrange + zl_instance = ZeepLibrary() + zl_instance._clients['first'] = client_fixture + zl_instance._active_client_alias = 'first' + mock_type_ = mock() + mock_type_.args = args + mock_type_.kws = kwargs + mock_type_.type_ = type_to_get + when(mock_type_).__call__(*args, **kwargs).thenReturn(mock_type_) + client_fixture.get_type = Mock(return_value=mock_type_) + expected = mock_type_ + # Act + actual = zl_instance.create_object(type_to_get, *args, **kwargs) + # Assert + assert actual == expected + assert actual.type_ == type_to_get + assert actual.args == args + assert actual.kws == kwargs + client_fixture.get_type.assert_called_once_with(type_to_get) + +def test_get_alias_no_client_argument_given(zl_with_clients): + """Test getting of an alias""" + # Arrange + expected = zl_with_clients._active_client_alias + # Act + actual = zl_with_clients.get_alias() + # Assert + assert actual == expected + +def test_get_alias_with_client_argument(zl_with_clients): + """Test getting of an alias""" + # Arrange + expected = "second" + # Act + actual = zl_with_clients.get_alias(zl_with_clients._clients[expected]) + # Assert + assert actual == expected + +def test_get_alias_with_client_raises(zl_with_clients): + """Test getting of an alias""" + # Arrange + non_existing_client = zl_with_clients._clients.pop("second") + # Act + with pytest.raises(AliasNotFoundException, + match="Could not find alias for the provided client."): + zl_with_clients.get_alias(non_existing_client) + +def test_get_client_no_alias_argument_given(zl_with_clients): + """Test getting of an alias""" + # Arrange + expected = zl_with_clients.active_client + # Act + actual = zl_with_clients.get_client() + # Assert + assert actual == expected + +def test_get_client_with_alias_argument(zl_with_clients): + """Test getting of an alias""" + # Arrange + expected = zl_with_clients._clients["second"] + # Act + actual = zl_with_clients.get_client("second") + # Assert + assert actual == expected + +def test_get_client_with_alias_raises(zl_with_clients): + """Test getting of an alias""" + # Arrange + zl_with_clients._clients.pop("second") + # Act + with pytest.raises(KeyError, match="second"): + zl_with_clients.get_client("second") + +def test_get_clients(zl_with_clients): + """Test getting clients list property.""" + # Arrange + expected = zl_with_clients._clients + # Act + actual = zl_with_clients.get_clients() + # Assert + assert actual == expected + +def test_get_namespace_prefix_for_uri_is_found(zl_with_clients): + """Test getting the namespace prefix using its uri.""" + # Arrange + namespace_dict = {'ns1': 'http://www.ns.fi', 'ns2': 'http://yyy.xx.com'} + zl_with_clients.active_client.namespaces = namespace_dict + expected = 'ns1' + # Act + actual = zl_with_clients.get_namespace_prefix_for_uri('http://www.ns.fi') + # Assert + assert actual == expected + +def test_get_namespace_prefix_for_uri_is_not_found(zl_with_clients): + """Test getting the namespace prefix using its uri + but argument uri is not found.""" + # Arrange + namespace_dict = {'ns1': 'http://www.ns.fi', 'ns2': 'http://yyy.xx.com'} + zl_with_clients.active_client.namespaces = namespace_dict + expected = None + # Act + actual = zl_with_clients.get_namespace_prefix_for_uri('http://www.ns.se') + # Assert + assert actual == expected + +def test_get_namespace_prefix_for_uri_no_namespaces(zl_with_clients): + """Test getting the namespace prefix using its uri + but no namespaces exist.""" + # Arrange + namespace_dict = {} + zl_with_clients.active_client.namespaces = namespace_dict + expected = None + # Act + actual = zl_with_clients.get_namespace_prefix_for_uri('http://www.ns.se') + # Assert + assert actual == expected + +def test_get_namespace_uri_by_prefix_is_found(zl_with_clients): + """Test getting the namespace uri using its prefix.""" + # Arrange + namespace_dict = {'ns1': 'http://www.ns.fi', 'ns2': 'http://yyy.xx.com'} + zl_with_clients.active_client.namespaces = namespace_dict + expected = 'http://yyy.xx.com' + # Act + actual = zl_with_clients.get_namespace_uri_by_prefix('ns2') + # Assert + assert actual == expected + +def test_get_namespace_uri_by_prefix_is_not_found(zl_with_clients): + """Test getting the namespace uri using its prefix, + but namespace prefix is not found.""" + # Arrange + namespace_dict = {'ns1': 'http://www.ns.fi', 'ns2': 'http://yyy.xx.com'} + zl_with_clients.active_client.namespaces = namespace_dict + # Act & Assert + with pytest.raises(KeyError, match='ns3'): + zl_with_clients.get_namespace_uri_by_prefix('ns3') + +def test_get_namespace_uri_by_prefix_no_namespaces(zl_with_clients): + """Test getting the namespace uri using its prefix, + no namespaces are present.""" + # Arrange + namespace_dict = {} + zl_with_clients.active_client.namespaces = namespace_dict + # Act & Assert + with pytest.raises(KeyError, match='ns1'): + zl_with_clients.get_namespace_uri_by_prefix('ns1') + +@pytest.mark.parametrize( + "to_log, to_console", + [ + (True, True), + (True, False), + (False, True), + (False, False) + ], + ids=[ + "log to both", + "log to log only", + "log to console only", + "no logging" + ] +) +def test_log_namespace_prefix_map(to_log, to_console, zl_with_clients): + """Test logging of namespace prefixes using different arguments.""" + # Arrange + namespace_dict = {'ns1': 'http://www.ns.fi', 'ns2': 'http://yyy.xx.com'} + zl_with_clients.active_client.namespaces = namespace_dict + when(zl_module)._log(namespace_dict, to_log, to_console) + # Act + zl_with_clients.log_namespace_prefix_map(to_log, to_console) + # Assert + verify(zl_module, times=1)._log(namespace_dict, to_log, to_console) + +def test_log_namespace_prefix_map_using_defaults(zl_with_clients): + """Test logging of namespace prefixes using default arguments.""" + # Arrange + namespace_dict = {'ns1': 'http://www.ns.fi', 'ns2': 'http://yyy.xx.com'} + zl_with_clients.active_client.namespaces = namespace_dict + when(zl_module)._log(namespace_dict, True, False) + # Act + zl_with_clients.log_namespace_prefix_map() + # Assert + # mockito seems to count a function call twice when the function's + # arguments include properties that need to be resolved before + # actual call, hence times=2. + verify(zl_module, times=2)._log(namespace_dict, True, False) + +@pytest.mark.parametrize( + "to_log, to_console", + [ + (True, True), + (True, False), + (False, True), + (False, False) + ], + ids=[ + "log to both", + "log to log only", + "log to console only", + "no logging" + ] +) +def test_log_opened_clients(to_log, to_console, zl_with_clients): + """Test logging of opened clients using different arguments.""" + # Arrange + when(zl_module)._log(zl_with_clients._clients, to_log, to_console) + # Act + zl_with_clients.log_opened_clients(to_log, to_console) + # Assert + verify(zl_module, times=1)._log(zl_with_clients._clients, to_log, to_console) + +def test_log_opened_clients_using_defaults(zl_with_clients): + """Test logging of opened clients using default arguments.""" + # Arrange + when(zl_module)._log(zl_with_clients._clients, True, False) + # Act + zl_with_clients.log_opened_clients() + # Assert + verify(zl_module, times=1)._log(zl_with_clients._clients, True, False) + +def test_dump_wsdl(zl_with_clients): + """Test that the wsdl.dump() method is called. + It is mocked so you have to verify __call__() instead of dump()""" + # Arrange + zl_with_clients.active_client.wsdl = mock() + zl_with_clients.active_client.wsdl.dump = mock() + # Act + zl_with_clients.dump_wsdl() + # Assert + verify(zl_with_clients._clients["fourth"].wsdl.dump, times=1).__call__() + +@pytest.mark.parametrize( + "alias_arg", + [ + "first", "fourth", + pytest.param("cannot_find_me", marks=pytest.mark.xfail) + ], + ids=[ + "Switch to another alias", + "Switch to the same alias", + "Switch to non-existing alias" + ] +) +def test_switch_client(alias_arg, zl_with_clients): + """Test client switching by alias. + switch_client() returns previously active alias.""" + # Arrange + expected = zl_with_clients._active_client_alias + # Act + actual = zl_with_clients.switch_client(alias_arg) + # Assert + assert actual == expected diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..19d2271 --- /dev/null +++ b/tox.ini @@ -0,0 +1,15 @@ +# tox (https://tox.readthedocs.io/) is a tool for running tests +# in multiple virtualenvs. This configuration file will run the +# test suite on all supported python versions. To use it, "pip install tox" +# and then run "tox" from this directory. + +[tox] +envlist = py27, py39 +skip_missing_interpreters = true + +[testenv] +deps = + pip + -rrequirements.txt +commands = + pytest tests