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