passerelle/tests/test_utils_soap.py

# Copyright (C) 2021  Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import logging
from unittest import mock

import pytest
import requests
from django.utils.encoding import force_bytes
from zeep import Settings
from zeep.exceptions import Fault, TransportError, XMLParseError
from zeep.plugins import Plugin

from passerelle.utils import Request
from passerelle.utils.jsonresponse import APIError
from passerelle.utils.soap import SOAPClient

WSDL = 'tests/data/soap.wsdl'


class FooPlugin(Plugin):
    pass


class BarPlugin(Plugin):
    pass


class SpecialSession(requests.Session):
    pass


class SOAPResource:
    def __init__(self):
        self.wsdl_url = WSDL

    def make_requests(self, **kwargs):
        return SpecialSession()


def test_soap_client():
    soap_resource = SOAPResource()
    plugins = [FooPlugin, BarPlugin]
    client = SOAPClient(soap_resource, plugins=plugins)
    assert client.wsdl.location.endswith(WSDL)
    assert isinstance(client.transport.session, SpecialSession)
    assert client.transport.cache
    assert client.plugins == plugins


@mock.patch('requests.sessions.Session.post')
def test_disable_strict_mode(mocked_post):
    response = requests.Response()
    response.status_code = 200
    response._content = force_bytes(
        '''<?xml version='1.0' encoding='utf-8'?>
<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/">
  <soap-env:Body>
    <ns0:TradePrice xmlns:ns0="http://example.com/stockquote.xsd">
      <price>4.20</price>
    </ns0:TradePrice>
  </soap-env:Body>
</soap-env:Envelope>'''
    )
    mocked_post.return_value = response

    soap_resource = SOAPResource()
    client = SOAPClient(soap_resource)
    match = "Unexpected element %s, expected %s" % (repr('price'), repr('skipMe'))
    with pytest.raises(XMLParseError, match=match):
        client.service.GetLastTradePrice(tickerSymbol='banana')

    client = SOAPClient(soap_resource, settings=Settings(strict=False))
    result = client.service.GetLastTradePrice(tickerSymbol='banana')
    assert len(result) == 2
    assert result['skipMe'] is None
    assert result['price'] == 4.2


@mock.patch('requests.sessions.Session.send')
def test_remove_first_bytes_for_xml(mocked_send, caplog):
    response = requests.Response()
    response.status_code = 200
    response.headers = {'Content-Type': 'application/xml'}
    response._content = b'\x8b' + force_bytes(
        '''blabla \n<?xml version='1.0' encoding='utf-8'?>
<soap-env:Envelope xmlns:soap-env="http://schemas.xmlsoap.org/soap/envelope/">
  <soap-env:Body>
    <ns0:TradePrice xmlns:ns0="http://example.com/stockquote.xsd">
      <skipMe>1.2</skipMe>
      <price>4.20</price>
    </ns0:TradePrice>
  </soap-env:Body>
</soap-env:Envelope>\n bloublou'''
    )
    mocked_send.return_value = response

    soap_resource = SOAPResource()
    logger = logging.getLogger('soap_resource')
    logger.setLevel(logging.INFO)
    soap_resource.make_requests = lambda **kwargs: Request(logger=logger, **kwargs)

    client = SOAPClient(soap_resource)
    with pytest.raises(TransportError):
        client.service.GetLastTradePrice(tickerSymbol='banana')

    client = SOAPClient(soap_resource, transport_kwargs={'remove_first_bytes_for_xml': True})
    result = client.service.GetLastTradePrice(tickerSymbol='banana')
    assert len(result) == 2
    assert result['skipMe'] == 1.2
    assert result['price'] == 4.2

    assert len(caplog.records) == 2
    assert 'response_content' not in caplog.records[-1].__dict__
    logger.setLevel(logging.DEBUG)
    result = client.service.GetLastTradePrice(tickerSymbol='banana')
    assert len(caplog.records) == 3
    assert 'response_content' in caplog.records[-1].__dict__


@mock.patch('requests.sessions.Session.send')
def test_api_error(mocked_send, caplog):
    response = requests.Response()
    response.status_code = 502
    response.headers = {'Content-Type': 'application/xml'}
    response._content = b'x'
    mocked_send.return_value = response

    soap_resource = SOAPResource()
    client = SOAPClient(soap_resource)
    with pytest.raises(TransportError):
        client.service.GetLastTradePrice(tickerSymbol='banana')

    client = SOAPClient(soap_resource, api_error=True)
    with pytest.raises(APIError, match=r'SOAP service at.*is unreachable'):
        client.service.GetLastTradePrice(tickerSymbol='banana')
    with mock.patch('zeep.proxy.OperationProxy.__call__') as operation_proxy_call:

        operation_proxy_call.side_effect = Fault('boom!')
        with pytest.raises(APIError, match=r'returned an error.*"boom!"'):
            client.service.GetLastTradePrice(tickerSymbol='banana')

        operation_proxy_call.side_effect = XMLParseError('Unexpected element')
        with pytest.raises(APIError, match=r'Unexpected element'):
            client.service.GetLastTradePrice(tickerSymbol='banana')