mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
130 lines
3.6 KiB
Python
130 lines
3.6 KiB
Python
from datetime import date
|
|
from typing import Any, Dict
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from applications.abri.abri_trigger_request import AbriTriggerRequest
|
|
|
|
|
|
def _full_message() -> Dict[str, Any]:
|
|
return {
|
|
"hubspot_deal_id": "9876543210",
|
|
"flows": ["amend_job", "log_job", "sync_tenant_data"],
|
|
"place_ref": "1007165",
|
|
"deal_name": "49 Admers Crescent",
|
|
"confirmed_survey_date": "2026-06-24",
|
|
"confirmed_survey_time": "14:30",
|
|
}
|
|
|
|
|
|
def test_a_full_message_parses_into_a_typed_request() -> None:
|
|
# Act
|
|
request = AbriTriggerRequest.model_validate(_full_message())
|
|
|
|
# Assert
|
|
assert request.hubspot_deal_id == "9876543210"
|
|
assert request.flows == ["amend_job", "log_job", "sync_tenant_data"]
|
|
assert request.place_ref == "1007165"
|
|
assert request.deal_name == "49 Admers Crescent"
|
|
assert request.confirmed_survey_date == date(2026, 6, 24)
|
|
assert request.confirmed_survey_time == "14:30"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("flow", "missing_field"),
|
|
[
|
|
("log_job", "place_ref"),
|
|
("log_job", "deal_name"),
|
|
("log_job", "confirmed_survey_date"),
|
|
("amend_job", "confirmed_survey_date"),
|
|
("sync_tenant_data", "place_ref"),
|
|
],
|
|
)
|
|
def test_a_message_missing_a_field_its_flow_needs_fails_validation(
|
|
flow: str, missing_field: str
|
|
) -> None:
|
|
# Arrange
|
|
message = _full_message()
|
|
message["flows"] = [flow]
|
|
del message[missing_field]
|
|
|
|
# Act / Assert
|
|
with pytest.raises(ValidationError, match=missing_field):
|
|
AbriTriggerRequest.model_validate(message)
|
|
|
|
|
|
def test_a_message_missing_only_fields_of_unfired_flows_parses() -> None:
|
|
# Arrange: tenant sync needs only the place_ref (and the deal id).
|
|
message = {
|
|
"hubspot_deal_id": "9876543210",
|
|
"flows": ["sync_tenant_data"],
|
|
"place_ref": "1007165",
|
|
}
|
|
|
|
# Act
|
|
request = AbriTriggerRequest.model_validate(message)
|
|
|
|
# Assert
|
|
assert request.flows == ["sync_tenant_data"]
|
|
assert request.confirmed_survey_date is None
|
|
|
|
|
|
def test_a_message_without_a_confirmed_time_parses_for_logging() -> None:
|
|
# Arrange: the time slot is optional — no time means an all-day slot.
|
|
message = _full_message()
|
|
message["flows"] = ["log_job"]
|
|
del message["confirmed_survey_time"]
|
|
|
|
# Act
|
|
request = AbriTriggerRequest.model_validate(message)
|
|
|
|
# Assert
|
|
assert request.confirmed_survey_time is None
|
|
|
|
|
|
def test_a_message_naming_abandon_job_validates_with_only_the_deal_id() -> None:
|
|
# Arrange: abandonment needs no deal field beyond the deal id.
|
|
message = {
|
|
"hubspot_deal_id": "9876543210",
|
|
"flows": ["abandon_job"],
|
|
}
|
|
|
|
# Act
|
|
request = AbriTriggerRequest.model_validate(message)
|
|
|
|
# Assert
|
|
assert request.flows == ["abandon_job"]
|
|
|
|
|
|
def test_a_message_carries_the_outcome_for_the_abandonment_reason_mapping() -> None:
|
|
# Arrange
|
|
message = _full_message()
|
|
message["flows"] = ["abandon_job"]
|
|
message["outcome"] = "no answer"
|
|
|
|
# Act
|
|
request = AbriTriggerRequest.model_validate(message)
|
|
|
|
# Assert
|
|
assert request.outcome == "no answer"
|
|
|
|
|
|
def test_a_message_naming_an_unknown_flow_fails_validation() -> None:
|
|
# Arrange
|
|
message = _full_message()
|
|
message["flows"] = ["cancel_job"]
|
|
|
|
# Act / Assert
|
|
with pytest.raises(ValidationError):
|
|
AbriTriggerRequest.model_validate(message)
|
|
|
|
|
|
def test_a_message_naming_no_flows_fails_validation() -> None:
|
|
# Arrange
|
|
message = _full_message()
|
|
message["flows"] = []
|
|
|
|
# Act / Assert
|
|
with pytest.raises(ValidationError):
|
|
AbriTriggerRequest.model_validate(message)
|