diff --git a/backend-services/tests/test_soap_gateway_content_types.py b/backend-services/tests/test_soap_gateway_content_types.py new file mode 100644 index 0000000..0b256c5 --- /dev/null +++ b/backend-services/tests/test_soap_gateway_content_types.py @@ -0,0 +1,142 @@ +import pytest + + +class _FakeXMLResponse: + def __init__(self, status_code=200, text='', headers=None): + self.status_code = status_code + self.text = text + base = {'Content-Type': 'text/xml'} + if headers: + base.update(headers) + self.headers = base + self.content = self.text.encode('utf-8') + + +def _mk_xml_client(captured): + class _FakeXMLClient: + def __init__(self, *args, **kwargs): + pass + async def __aenter__(self): + return self + async def __aexit__(self, exc_type, exc, tb): + return False + async def post(self, url, content=None, params=None, headers=None): + captured.append({'url': url, 'headers': dict(headers or {}), 'content': content}) + return _FakeXMLResponse(200, '', {'X-Upstream': 'yes', 'Content-Type': 'text/xml'}) + return _FakeXMLClient + + +async def _setup_api(client, name, ver): + r = await client.post('/platform/api', json={ + 'api_name': name, + 'api_version': ver, + 'api_description': f'{name} {ver}', + 'api_allowed_roles': ['admin'], + 'api_allowed_groups': ['ALL'], + 'api_servers': ['http://soap.up'], + 'api_type': 'REST', + 'api_allowed_retry_count': 0, + }) + assert r.status_code in (200, 201) + r2 = await client.post('/platform/endpoint', json={ + 'api_name': name, + 'api_version': ver, + 'endpoint_method': 'POST', + 'endpoint_uri': '/call', + 'endpoint_description': 'soap call', + }) + assert r2.status_code in (200, 201) + rme = await client.get('/platform/user/me') + username = (rme.json().get('username') if rme.status_code == 200 else 'admin') + rs = await client.post('/platform/subscription/subscribe', json={'username': username, 'api_name': name, 'api_version': ver}) + assert rs.status_code in (200, 201) + + +@pytest.mark.asyncio +async def test_soap_incoming_application_xml_sets_text_xml_outgoing(monkeypatch, authed_client): + import services.gateway_service as gs + name, ver = 'soapct1', 'v1' + await _setup_api(authed_client, name, ver) + captured = [] + monkeypatch.setattr(gs.httpx, 'AsyncClient', _mk_xml_client(captured)) + envelope = '' + r = await authed_client.post( + f'/api/soap/{name}/{ver}/call', + headers={'Content-Type': 'application/xml'}, + content=envelope, + ) + assert r.status_code == 200 + assert len(captured) == 1 + h = captured[0]['headers'] + assert h.get('Content-Type') == 'text/xml; charset=utf-8' + + +@pytest.mark.asyncio +async def test_soap_incoming_text_xml_passes_through(monkeypatch, authed_client): + import services.gateway_service as gs + name, ver = 'soapct2', 'v1' + await _setup_api(authed_client, name, ver) + captured = [] + monkeypatch.setattr(gs.httpx, 'AsyncClient', _mk_xml_client(captured)) + envelope = '' + r = await authed_client.post( + f'/api/soap/{name}/{ver}/call', + headers={'Content-Type': 'text/xml'}, + content=envelope, + ) + assert r.status_code == 200 + h = captured[0]['headers'] + assert h.get('Content-Type') == 'text/xml' + + +@pytest.mark.asyncio +async def test_soap_incoming_application_soap_xml_passes_through(monkeypatch, authed_client): + import services.gateway_service as gs + name, ver = 'soapct3', 'v1' + await _setup_api(authed_client, name, ver) + captured = [] + monkeypatch.setattr(gs.httpx, 'AsyncClient', _mk_xml_client(captured)) + envelope = '' + r = await authed_client.post( + f'/api/soap/{name}/{ver}/call', + headers={'Content-Type': 'application/soap+xml'}, + content=envelope, + ) + assert r.status_code == 200 + h = captured[0]['headers'] + assert h.get('Content-Type') == 'application/soap+xml' + + +@pytest.mark.asyncio +async def test_soap_adds_default_soapaction_when_missing(monkeypatch, authed_client): + import services.gateway_service as gs + name, ver = 'soapct4', 'v1' + await _setup_api(authed_client, name, ver) + captured = [] + monkeypatch.setattr(gs.httpx, 'AsyncClient', _mk_xml_client(captured)) + envelope = '' + r = await authed_client.post( + f'/api/soap/{name}/{ver}/call', + headers={'Content-Type': 'application/xml'}, + content=envelope, + ) + assert r.status_code == 200 + h = captured[0]['headers'] + assert 'SOAPAction' in h and h['SOAPAction'] == '""' + + +@pytest.mark.asyncio +async def test_soap_parses_xml_response_success(monkeypatch, authed_client): + import services.gateway_service as gs + name, ver = 'soapct5', 'v1' + await _setup_api(authed_client, name, ver) + captured = [] + monkeypatch.setattr(gs.httpx, 'AsyncClient', _mk_xml_client(captured)) + envelope = '' + r = await authed_client.post( + f'/api/soap/{name}/{ver}/call', + headers={'Content-Type': 'application/xml'}, + content=envelope, + ) + assert r.status_code == 200 + assert '' in (r.text or '')