mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
193 lines
5.3 KiB
Python
193 lines
5.3 KiB
Python
import xml.etree.ElementTree as ET
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from domain.abri.models import (
|
|
AbriRequestRejected,
|
|
JobLogged,
|
|
LogJobRequest,
|
|
PlaceRef,
|
|
)
|
|
from infrastructure.abri.abri_client import AbriClient
|
|
from infrastructure.abri.config import AbriConfig
|
|
from infrastructure.abri.errors import AbriResponseParseError, AbriTransportError
|
|
|
|
FIXTURE_DIR = Path(__file__).parents[2] / "abri"
|
|
ENDPOINT_URL = "https://relay.example.test/api/DomnaRelay?code=test-function-key"
|
|
|
|
CONFIG = AbriConfig(
|
|
endpoint_url=ENDPOINT_URL,
|
|
username="DomnaWeb",
|
|
password="",
|
|
default_resource="NAULKH",
|
|
)
|
|
|
|
|
|
def _load_fixture(name: str) -> bytes:
|
|
return (FIXTURE_DIR / name).read_bytes()
|
|
|
|
|
|
def _make_client(mock_session: MagicMock) -> AbriClient:
|
|
with patch(
|
|
"infrastructure.abri.abri_client.requests.Session",
|
|
return_value=mock_session,
|
|
):
|
|
return AbriClient(config=CONFIG)
|
|
|
|
|
|
def _spec_example_request() -> LogJobRequest:
|
|
return LogJobRequest(
|
|
client_ref="DOM51111",
|
|
place_ref=PlaceRef("1007165"),
|
|
appointment_date=date(2026, 6, 18),
|
|
appointment_time="PM",
|
|
short_description="Christian's SCS External Test.",
|
|
long_description="Christian's SCS External Test.",
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_session() -> MagicMock:
|
|
return MagicMock()
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(mock_session: MagicMock) -> AbriClient:
|
|
return _make_client(mock_session)
|
|
|
|
|
|
# --- log_job: success ---
|
|
|
|
|
|
def test_log_job_returns_job_logged_with_openhousing_job_number(
|
|
client: AbriClient, mock_session: MagicMock
|
|
) -> None:
|
|
# Arrange
|
|
mock_session.post.return_value.content = _load_fixture(
|
|
"abri_relay_logjob_success_response.xml"
|
|
)
|
|
|
|
# Act
|
|
result = client.log_job(_spec_example_request())
|
|
|
|
# Assert
|
|
assert result == JobLogged(
|
|
job_no="AD0226519",
|
|
logged_info=(
|
|
"Job number AD0226519 has been successfully logged for "
|
|
"49 Admers Crescent, Liphook, Midsomer, XX99 IOP - "
|
|
"Short Description example job."
|
|
),
|
|
)
|
|
|
|
|
|
def test_log_job_sends_the_recorded_logjob_envelope_to_the_relay_endpoint(
|
|
client: AbriClient, mock_session: MagicMock
|
|
) -> None:
|
|
# Arrange
|
|
mock_session.post.return_value.content = _load_fixture(
|
|
"abri_relay_logjob_success_response.xml"
|
|
)
|
|
|
|
# Act
|
|
client.log_job(_spec_example_request())
|
|
|
|
# Assert
|
|
# Fixture is the spec's recorded request example, except resource_group is
|
|
# "Surveyors" (the documented default) where the example used "surveyors".
|
|
(url,) = mock_session.post.call_args.args
|
|
sent_body: bytes = mock_session.post.call_args.kwargs["data"]
|
|
expected_body = _load_fixture("abri_relay_logjob_request_example.xml")
|
|
assert url == ENDPOINT_URL
|
|
assert ET.canonicalize(xml_data=sent_body, strip_text=True) == ET.canonicalize(
|
|
xml_data=expected_body, strip_text=True
|
|
)
|
|
|
|
|
|
# --- log_job: rejection ---
|
|
|
|
|
|
def test_log_job_returns_rejection_with_openhousing_error_code_and_message_verbatim(
|
|
client: AbriClient, mock_session: MagicMock
|
|
) -> None:
|
|
# Arrange
|
|
mock_session.post.return_value.content = _load_fixture(
|
|
"abri_relay_logjob_relayerror_response.xml"
|
|
)
|
|
|
|
# Act
|
|
result = client.log_job(_spec_example_request())
|
|
|
|
# Assert
|
|
assert result == AbriRequestRejected(
|
|
code="RelayError",
|
|
message=(
|
|
"No property was found with UPRN/place_ref '1007176aa', "
|
|
"street name '', house number '0', house name '', post code ''"
|
|
),
|
|
)
|
|
|
|
|
|
# --- log_job: transport failures ---
|
|
|
|
|
|
def test_log_job_raises_transport_error_on_http_error_status(
|
|
client: AbriClient, mock_session: MagicMock
|
|
) -> None:
|
|
# Arrange
|
|
mock_session.post.return_value.content = b"<html>Server Error</html>"
|
|
mock_session.post.return_value.raise_for_status.side_effect = requests.HTTPError(
|
|
"500 Server Error"
|
|
)
|
|
|
|
# Act / Assert
|
|
with pytest.raises(AbriTransportError):
|
|
client.log_job(_spec_example_request())
|
|
|
|
|
|
def test_log_job_raises_transport_error_when_the_relay_is_unreachable(
|
|
client: AbriClient, mock_session: MagicMock
|
|
) -> None:
|
|
# Arrange
|
|
mock_session.post.side_effect = requests.ConnectionError("connection refused")
|
|
|
|
# Act / Assert
|
|
with pytest.raises(AbriTransportError):
|
|
client.log_job(_spec_example_request())
|
|
|
|
|
|
# --- log_job: ambiguous responses are retriable, never success or rejection ---
|
|
|
|
|
|
def test_log_job_raises_retriable_parse_error_on_a_malformed_response_body(
|
|
client: AbriClient, mock_session: MagicMock
|
|
) -> None:
|
|
# Arrange
|
|
mock_session.post.return_value.content = b"not xml at all <<<"
|
|
|
|
# Act / Assert
|
|
with pytest.raises(AbriResponseParseError):
|
|
client.log_job(_spec_example_request())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body",
|
|
[
|
|
b"<Root><Jobs></Jobs></Root>",
|
|
b"<response><success>true</success></response>",
|
|
b"<unexpected />",
|
|
],
|
|
)
|
|
def test_log_job_raises_retriable_parse_error_on_an_unexpectedly_shaped_response(
|
|
client: AbriClient, mock_session: MagicMock, body: bytes
|
|
) -> None:
|
|
# Arrange
|
|
mock_session.post.return_value.content = body
|
|
|
|
# Act / Assert
|
|
with pytest.raises(AbriResponseParseError):
|
|
client.log_job(_spec_example_request())
|