Model/tests/condition/test_calico_load.py
Khalim Conn-Kowlessar a6a48dd337 Run a Calico load end to end and return its reconciliation 🟥
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-13 15:13:06 +00:00

88 lines
2.7 KiB
Python

from datetime import date
from io import BytesIO
from typing import List
from openpyxl import Workbook
from applications.condition.calico_load import run_calico_load
from applications.condition.load_report import LoadReport
from domain.condition.property_condition_survey import PropertyConditionSurvey
from repositories.condition.property_reference_reader import (
PropertyReference,
PropertyReferenceReader,
)
from repositories.condition.property_uprn_lookup import PropertyUprnLookup
HEADERS = [
"Asset Reference",
"Asset Type",
"Address",
"Path",
"Type",
"Fitted / Renewed Date",
"Quantity",
"Remaining Life",
]
class FakePropertyReferenceReader(PropertyReferenceReader):
def __init__(self, references: List[PropertyReference]) -> None:
self._references = references
def references_for_portfolio(self, portfolio_id: int) -> List[PropertyReference]:
return self._references
class FakeConditionWriter:
def __init__(self) -> None:
self.surveys: List[PropertyConditionSurvey] = []
def bulk_insert_surveys(self, surveys: List[PropertyConditionSurvey]) -> None:
self.surveys.extend(surveys)
def _workbook(rows: List[List[object]]) -> BytesIO:
wb = Workbook()
ws = wb.active
assert ws is not None
ws.append(HEADERS)
for row in rows:
ws.append(row)
buffer = BytesIO()
wb.save(buffer)
buffer.seek(0)
return buffer
def test_run_calico_load_persists_surveys_and_reports_reconciliation():
# Arrange -- one loaded (443), one blank placeholder (438), one unmatched (777)
workbook = _workbook(
[
[443, "HOUSE", "31 Venice Avenue", "Roof Covering", "Concrete Tiles",
date(1998, 4, 1), 55, 36],
[438, "FLAT", "137 Brownhill Avenue", "Roof Covering", "N/A",
date(2000, 4, 1), 0, 973],
[777, "HOUSE", "8 Marine Avenue", "Roof Covering", "Natural Slate",
date(2010, 4, 1), 60, 50],
]
)
reader = FakePropertyReferenceReader(
[
PropertyReference(landlord_property_id="443", uprn=100093305101),
PropertyReference(landlord_property_id="999", uprn=None),
]
)
lookup = PropertyUprnLookup(reader, portfolio_id=824)
writer = FakeConditionWriter()
# Act
report = run_calico_load(workbook, lookup, export_year=2026, writer=writer)
# Assert
assert report == LoadReport(
loaded=1, unmatched_to_portfolio=1, null_uprn=1, blank_placeholder=1
)
assert len(writer.surveys) == 1
persisted = writer.surveys[0]
assert persisted.uprn == 100093305101
assert persisted.elements[0].aspect_conditions[0].renewal_year == 2026 + 36