Abri log_job returns the OpenHousing job number from the relay success response 🟥

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Daniel Roth 2026-07-03 14:22:09 +00:00
parent 1a5368571f
commit b22afcde9f
10 changed files with 168 additions and 0 deletions

0
domain/abri/__init__.py Normal file
View file

32
domain/abri/models.py Normal file
View file

@ -0,0 +1,32 @@
from dataclasses import dataclass
from datetime import date
from typing import Literal, NewType, Union
PlaceRef = NewType("PlaceRef", str)
SlotCode = Literal["AM", "PM", "AD"]
@dataclass(frozen=True)
class LogJobRequest:
client_ref: str
place_ref: PlaceRef
appointment_date: date
appointment_time: SlotCode
short_description: str
long_description: str
@dataclass(frozen=True)
class JobLogged:
job_no: str
logged_info: str
@dataclass(frozen=True)
class AbriRequestRejected:
code: str
message: str
LogJobResult = Union[JobLogged, AbriRequestRejected]

View file

View file

@ -0,0 +1,13 @@
import requests
from domain.abri.models import LogJobRequest, LogJobResult
from infrastructure.abri.config import AbriConfig
class AbriClient:
def __init__(self, config: AbriConfig) -> None:
self._config = config
self._session = requests.Session()
def log_job(self, request: LogJobRequest) -> LogJobResult:
raise NotImplementedError

View file

@ -0,0 +1,14 @@
from dataclasses import dataclass
from typing import Mapping
@dataclass(frozen=True)
class AbriConfig:
endpoint_url: str
username: str
password: str
default_resource: str
@classmethod
def from_env(cls, env: Mapping[str, str]) -> "AbriConfig":
raise NotImplementedError

View file

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="utf-8"?>
<response>
<success>false</success>
<code>RelayError</code>
<message>No property was found with UPRN/place_ref &apos;1007176aa&apos;, street name &apos;&apos;, house number &apos;0&apos;, house name &apos;&apos;, post code &apos;&apos;</message>
</response>

View file

@ -0,0 +1,20 @@
<?xml version="1.0" ?>
<message>
<Header>
<Security username="DomnaWeb" password="" />
</Header>
<Body>
<Request request_type="logjob">
<Parameters attribute="place_ref" attribute_value="1007165" />
<Parameters attribute="std_job_code" attribute_value="SCSEXT" />
<Parameters attribute="client" attribute_value="HSG" />
<Parameters attribute="short_description" attribute_value="Christian's SCS External Test." />
<Parameters attribute="long_description" attribute_value="Christian's SCS External Test." />
<Parameters attribute="client_ref" attribute_value="DOM51111" />
<Parameters attribute="appointment_date" attribute_value="18/06/2026" />
<Parameters attribute="appointment_time" attribute_value="PM" />
<Parameters attribute="resource" attribute_value="NAULKH" />
<Parameters attribute="resource_group" attribute_value="Surveyors" />
</Request>
</Body>
</message>

View file

@ -0,0 +1,6 @@
<?xml version="1.0" ?>
<Root>
<Jobs>
<Job_logged job_no="AD0226519" logged_info="Job number AD0226519 has been successfully logged for 49 Admers Crescent, Liphook, Midsomer, XX99 IOP - Short Description example job.">AD0226519</Job_logged>
</Jobs>
</Root>

View file

View file

@ -0,0 +1,77 @@
from datetime import date
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from domain.abri.models import JobLogged, LogJobRequest, PlaceRef
from infrastructure.abri.abri_client import AbriClient
from infrastructure.abri.config import AbriConfig
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."
),
)