from tempfile import NamedTemporaryFile import pytest from typing import Any, Dict from io import BytesIO from openpyxl import Workbook from datetime import datetime from backend.condition.lookups.uprn_lookup_csv import UprnLookupLocal from backend.condition.parsing.peabody_parser import PeabodyParser from backend.condition.parsing.records.peabody.peabody_asset_condition import ( PeabodyAssetCondition, ) from backend.condition.parsing.records.peabody.peabody_property import PeabodyProperty @pytest.fixture def peabody_assets_xlsx_bytes() -> BytesIO: wb = Workbook() survey_records_d_and_lower = wb.active survey_records_d_and_lower.title = "Survey Records - D & Lower" survey_records_d_and_lower.append( [ "Lo_Reference", "full_address", "location_type_code", "Parent_Lo_Reference", "Element_Code", "Element", "Sub_Element_Code", "Sub_Element", "Material_Code", "material_or_answer", "Renewal_Quantity", "Renewal_Year", "Renewal_Cost", "cloned", "lo_type_code", "condition_survey_date", ] ) survey_records_d_and_lower.append( [ "B000RAND", "1 RANDOM HOUSE LONDON", 3, "RAND2EST", 110, "ROOFS", 1, "Primary Roof", 9, "Other", 3, 2054, 330, "N", 3, datetime(2025, 12, 4, 9, 17, 0), ] ) survey_records_d_and_lower.append( [ "B000BLOCK", "1100 BLOCK", 3, "RAND2EST", 110, "ROOFS", 1, "Primary Roof", 9, "Other", 3, 2054, 330, "N", 3, datetime(2025, 12, 4, 9, 17, 0), ] ) survey_records_d_and_lower.append( [ "B000FAKE", "3 FAKE CLOSE LONDON", 3, "FAKEEST", 100, "GENERAL", 15, "External Decoration", 2, "Normal", 1, 2035, 1500.7, "N", 3, datetime(2025, 7, 5, 0, 0, 0), ] ) survey_records_d_and_lower.append( [ "B000MIS", "99 MISC ROAD LONDON", 3, "300828", 54, "HHSRS", 29, "HHSRS Structural Collapse & Falling Elements", 4, "HHSRS Moderate", 2, 2027, None, "N", 3, None, ] ) survey_records_d_and_lower.append( [ "B000MIS", "99 MISC ROAD LONDON", 3, "300828", 53, "External", 2, "Chimney", 2, "Present", 33, 2053, 3531, "N", 3, None, ] ) stream = BytesIO() wb.save(stream) stream.seek(0) return stream @pytest.fixture def prop_ref_uprn_csv_file() -> str: csv_content = """reference,out_uprn B000RAND,1 B000BLOCK,2 B000FAKE,3 B000MIS,4 """ with NamedTemporaryFile(mode="w+", delete=False, suffix=".csv") as tmp: tmp.write(csv_content) tmp.flush() return tmp.name def test_peabody_parser_parses_conditions( peabody_assets_xlsx_bytes, prop_ref_uprn_csv_file ): # arrange uprn_lookup = UprnLookupLocal(csv_path=prop_ref_uprn_csv_file) parser = PeabodyParser(uprn_lookup=uprn_lookup) # act result: Any = parser.parse(peabody_assets_xlsx_bytes) # assert assert len(result) == 3 assert all(isinstance(item, PeabodyProperty) for item in result) @pytest.fixture def asset_condition_factory(): def _factory(full_address: str) -> PeabodyAssetCondition: return PeabodyAssetCondition( lo_reference="", full_address=full_address, location_type_code=0, parent_lo_reference="", element_code=0, element="", sub_element_code=0, sub_element="", material_code=0, material_or_answer="", renewal_quantity=0, renewal_year=2026, cloned="", lo_type_code=0, renewal_cost=None, condition_survey_date=None, ) return _factory @pytest.mark.parametrize( "full_address, expected_block_level", [ ("1-80 PRINCESS ALICE HOUSE LONDON", True), ("FLATS A-D 7 ST CHARLES SQUARE LONDON", True), ("9A-9H HEDGEGATE COURT LONDON", True), ("BLOCK MILNE HOUSE LONDON", True), ("81A-B GORE ROAD LONDON", True), ("73 & 74 HARVEST COURT ST. ALBANS", True), ("25 HAVERSHAM COURT GREENFORD", False), ("FLAT 10 SPARROW COURT SOUTHMERE DRIVE LONDON SE2 9ES", False), ], ) def test_peabody_asset_is_block_level( asset_condition_factory, full_address, expected_block_level, ): # arrange asset_condition = asset_condition_factory(full_address) # act + assert assert asset_condition.is_block_level == expected_block_level