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
import responses
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, to_json
from passerelle.utils.soap import SOAPClient, SOAPFault

with open('tests/data/soap.wsdl') as fd:
    WSDL = fd.read()

WSDL_URL = 'http://example.com/soap.wsdl'


class FooPlugin(Plugin):
    pass


class BarPlugin(Plugin):
    pass


FOO_XSD = '''\
<schema xmlns="http://www.w3.org/2001/XMLSchema"
        targetNamespace="http://example.com/foo.xsd">
  <complexType name="Address">
    <sequence>
      <element minOccurs="0" maxOccurs="1" name="NameFirst" type="string"/>
      <element minOccurs="0" maxOccurs="1" name="NameLast" type="string"/>
      <element minOccurs="0" maxOccurs="1" name="Email" type="string"/>
    </sequence>
  </complexType>
</schema>
'''


@pytest.fixture
def http_mock():
    with responses.RequestsMock() as http_mock:
        yield http_mock


@pytest.fixture
def wsdl_response(http_mock):
    http_mock.get('http://example.com/foo.xsd', FOO_XSD)
    return http_mock.get(WSDL_URL, WSDL)


@responses.activate
@pytest.fixture
def soap_resource(http_mock, wsdl_response):
    class SOAPResource:
        def __init__(self):
            self.wsdl_url = WSDL_URL
            self.logger = mock.Mock()

        def make_requests(self, **kwargs):
            return Request(logger=self.logger, **kwargs)

    yield SOAPResource()


def test_soap_client(soap_resource):
    plugins = [FooPlugin, BarPlugin]
    client = SOAPClient(soap_resource, plugins=plugins)
    assert client.wsdl.location == WSDL_URL
    assert client.plugins == plugins


def test_disable_strict_mode(soap_resource, http_mock):
    http_mock.post(
        'http://example.com/stockquote',
        body=b'''\
<?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>''',
    )

    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


def test_remove_first_bytes_for_xml(http_mock, soap_resource, caplog):
    http_mock.add(
        responses.POST,
        'http://example.com/stockquote',
        body=b'\x8b'
        + b'''\
<?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>''',
    )

    soap_resource.logger = logging.getLogger('soap_resource')
    soap_resource.logger.setLevel(logging.INFO)
    client = SOAPClient(soap_resource)
    with pytest.raises(TransportError):
        client.service.GetLastTradePrice(tickerSymbol='banana')
    caplog.clear()

    client = SOAPClient(soap_resource, transport_kwargs={'remove_first_bytes_for_xml': True})
    result = client.service.GetLastTradePrice(tickerSymbol='banana')
    assert result['skipMe'] == 1.2
    assert result['price'] == 4.2
    assert caplog.messages == ['POST http://example.com/stockquote (=> 200)']
    assert 'response_content' not in caplog.records[-1].__dict__
    caplog.clear()

    soap_resource.logger.setLevel(logging.DEBUG)
    result = client.service.GetLastTradePrice(tickerSymbol='banana')
    assert caplog.messages == ['POST http://example.com/stockquote (=> 200)']
    assert 'response_content' in caplog.records[-1].__dict__


def test_api_error(soap_resource, http_mock, caplog):
    http_mock.add(responses.POST, 'http://example.com/stockquote', status=502, body=b'x')

    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')

    http_mock.replace(
        responses.POST, 'http://example.com/stockquote', status=502, body=requests.ConnectTimeout('too long!')
    )
    with pytest.raises(APIError, match=r'SOAP service at.*is unreachable.*too long!'):
        client.service.GetLastTradePrice(tickerSymbol='banana')


def test_fault_detail_on_500(soap_resource, http_mock):
    http_mock.add(
        responses.POST,
        'http://example.com/stockquote',
        body=b'''<?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>''',
        status=500,
    )
    client = SOAPClient(soap_resource, api_error=True)
    with pytest.raises(SOAPFault) as exc:
        client.service.GetLastTradePrice(tickerSymbol='banana')

    response = to_json().err_to_response(exc.value)
    assert response['err'] == 1
    assert 'xmlns:soap' in response['data']['soap_fault']['detail']


def test_wsdl_cache(freezer, http_mock, wsdl_response, soap_resource):
    assert wsdl_response.call_count == 0
    SOAPClient(soap_resource, api_error=True)
    assert wsdl_response.call_count == 1
    SOAPClient(soap_resource, api_error=True)
    assert wsdl_response.call_count == 1
    freezer.tick(299)
    SOAPClient(soap_resource, api_error=True)
    assert wsdl_response.call_count == 1
    freezer.tick(1)
    SOAPClient(soap_resource, api_error=True)
    assert wsdl_response.call_count == 2