junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Abandon a job through the relay canceljob route 🟥

Browse source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Daniel Roth 2026-07-07 15:30:24 +00:00
parent 596f5ff785
commit c206a76b22
5 changed files with 101 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
domain/abri/models.py
View file

@ -1,5 +1,6 @@
from dataclasses import dataclass
from datetime import date
from enum import Enum
from typing import Literal, NewType, Optional, Tuple, Union
PlaceRef = NewType("PlaceRef", str)
@ -7,6 +8,16 @@ PlaceRef = NewType("PlaceRef", str)
SlotCode = Literal["AM", "PM", "AD"]
class AbandonReason(str, Enum):
"""OpenHousing's coded reason a job was abandoned.
Only CARDED is used for now; the full HubSpot-outcome-to-code table is a
documented follow-up (needs Abri's code list).
"""
CARDED = "CARDED"
@dataclass(frozen=True)
class LogJobRequest:
client_ref: str
@ -78,3 +89,18 @@ class TenancyData:
GetTenantDataResult = Union[TenancyData, AbriRequestRejected]
@dataclass(frozen=True)
class AbandonJobRequest:
job_no: str
reason: AbandonReason
date_abandoned: date
@dataclass(frozen=True)
class JobAbandoned:
job_no: str
AbandonJobResult = Union[JobAbandoned, AbriRequestRejected]

6
infrastructure/abri/abri_client.py
View file

@ -5,11 +5,14 @@ from typing import Dict, List, Tuple, Union
import requests
from domain.abri.models import (
AbandonJobRequest,
AbandonJobResult,
AbriRequestRejected,
AmendJobRequest,
AmendJobResult,
AppointmentAmended,
GetTenantDataResult,
JobAbandoned,
JobLogged,
LogJobRequest,
LogJobResult,
@ -94,6 +97,9 @@ class AbriClient:
return self._parse_appointment_amended(outcome)
def abandon_job(self, request: AbandonJobRequest) -> AbandonJobResult:
raise NotImplementedError
def get_tenant_data(self, place_ref: PlaceRef) -> GetTenantDataResult:
outcome = self._exchange(
request_type="getSCSTenantData",

13
tests/abri/abri_relay_canceljob_request_example.xml Normal file
View file

@ -0,0 +1,13 @@
<?xml version="1.0" ?>
<message>
<Header>
<Security username="DomnaWeb" password="" />
</Header>
<Body>
<Request request_type="canceljob">
<Parameters attribute="job_no" attribute_value="AC0439951" />
<Parameters attribute="abandon_reason" attribute_value="CARDED" />
<Parameters attribute="date_abandoned" attribute_value="07/07/2026" />
</Request>
</Body>
</message>

6
tests/abri/abri_relay_canceljob_success_response.xml Normal file
View file

@ -0,0 +1,6 @@
<?xml version="1.0" ?>
<Root>
<CancelJobs>
<JobCancelled job_no="AC0439951" />
</CancelJobs>
</Root>

50
tests/infrastructure/abri/test_abri_client.py
View file

@ -7,9 +7,12 @@ import pytest
import requests
from domain.abri.models import (
AbandonJobRequest,
AbandonReason,
AbriRequestRejected,
AmendJobRequest,
AppointmentAmended,
JobAbandoned,
JobLogged,
LogJobRequest,
PlaceRef,
@ -60,6 +63,14 @@ def _spec_example_amend_request() -> AmendJobRequest:
)
def _spec_example_abandon_request() -> AbandonJobRequest:
return AbandonJobRequest(
job_no="AC0439951",
reason=AbandonReason.CARDED,
date_abandoned=date(2026, 7, 7),
)
@pytest.fixture()
def mock_session() -> MagicMock:
return MagicMock()
@ -348,3 +359,42 @@ def test_amend_job_raises_transport_error_when_the_relay_is_unreachable(
# Act / Assert
with pytest.raises(AbriTransportError):
client.amend_job(_spec_example_amend_request())
# --- abandon_job: success ---
def test_abandon_job_returns_job_abandoned_with_the_cancelled_job_number(
client: AbriClient, mock_session: MagicMock
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_canceljob_success_response.xml"
)
# Act
result = client.abandon_job(_spec_example_abandon_request())
# Assert
assert result == JobAbandoned(job_no="AC0439951")
def test_abandon_job_sends_the_recorded_canceljob_envelope_to_the_relay_endpoint(
client: AbriClient, mock_session: MagicMock
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_canceljob_success_response.xml"
)
# Act
client.abandon_job(_spec_example_abandon_request())
# Assert
(url,) = mock_session.post.call_args.args
sent_body: bytes = mock_session.post.call_args.kwargs["data"]
expected_body = _load_fixture("abri_relay_canceljob_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
)