Model/tests/condition/parsing/test_lbwf_parser.py
Khalim Conn-Kowlessar 735e83cef2 Relocate condition-data ingestion into the DDD layout 🟪
Behaviour-preserving lift-and-shift of the condition module out of legacy
backend/condition/ into domain/condition, infrastructure/condition,
infrastructure/postgres/condition_tables.py, repositories/condition,
applications/condition, and tests/condition. Imports rewritten to the DDD
paths; ConditionPostgres and the ORM models keep the legacy backend.app.db
Base/db_session bridges so the existing suite proves behaviour is unchanged
(SQLModel + session-DI conversion tracked as a follow-up in ADR-0064).

16 condition tests pass at the new location.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-13 14:51:25 +00:00

134 lines
No EOL
3.3 KiB
Python

from typing import Any
import pytest
from io import BytesIO
from openpyxl import Workbook
from datetime import datetime
from infrastructure.condition.parsing.lbwf_parser import LbwfParser
from domain.condition.records.lbwf.lbwf_asset_condition import LbwfAssetCondition
from domain.condition.records.lbwf.lbwf_house import LbwfHouse
@pytest.fixture
def lbwf_homes_xlsx_bytes() -> BytesIO:
wb = Workbook()
houses_asset_data = wb.active
houses_asset_data.title = "Houses Asset Data"
houses_asset_data.append([
"PROP REF",
"Domna",
"ADDRESS",
"OWNERSHIP",
"PROP STATUS",
"PROP TYPE",
"PROP SUB TYPE",
"ELEMENT GROUP",
"ELEMENT CODE",
"ELEMENT CODE DESCRIPTION",
"ATTRIBUTE CODE",
"ATTRIBUTE CODE DESCRIPTION",
"ELEMENT DATE VALUE",
"ELEMENT NUMERIC VALUE",
"ELEMENT TEXT VALUE",
"QUANTITY",
"INSTALL DATE",
"REMAINING LIFE",
"ELEMENT COMMENTS"
]
)
houses_asset_data.append([
12345,
12345,
"123 Fake Street, London, A10 1AB",
"LBWF_OWNED",
"OCCP",
"HOU",
"TERRACED",
"ASSETS",
"AHR_CAT",
"Accessible Housing Register Category",
"F",
"General Needs",
None,
None,
None,
1,
None,
None,
None,
])
houses_asset_data.append([
54321,
54321,
"100 Random Road, London, A10 1AB",
"LBWF_OWNED",
"OCCP",
"HOU",
"EOT",
"ASSETS",
"INTSMKDET",
"Smoke Detectors in Property",
"HARDWRDMNS",
"Hard Wired Mains Smoke Alarm in Property",
None,
None,
None,
2,
datetime(2019,4,1),
4,
"Source of Data = Joe Bloggs",
])
houses = wb.create_sheet("Houses")
houses.append(["Reference", "Address", "EPC ", "SHDF", "HOSUE", "Fail Decency"])
houses.append([12345, "123 Fake Street, London, A10 1AB", "E", "NO", "HOUSE", 2025])
houses.append([54321, "100 Random Road, London, A10 1AB", "F", "NO", "HOUSE", 2025])
all_energy_breakdown = wb.create_sheet("All Energy Breakdown ") # Trailing space is intentional; matches source
all_energy_breakdown.append([
"UPRN",
"Organisation Reference",
"Alternate Organisation Reference",
"Address",
"Postcode"
])
all_energy_breakdown.append([
1,
200,
None,
"123 FAKE STREET",
"A10 1AB"
])
all_energy_breakdown.append([
2,
100,
101,
"100 RANDOM ROAD",
"A10 1AB"
])
stream = BytesIO()
wb.save(stream)
stream.seek(0)
return stream
def test_lbwf_parser_parses_houses(lbwf_homes_xlsx_bytes):
# arrange
parser = LbwfParser()
# act
result: Any = parser.parse(lbwf_homes_xlsx_bytes)
# assert
# TODO: Improve these asserts
assert len(result) == 2
assert isinstance(result[0], LbwfHouse)
assert result[0].uprn == 1
assert len(result[0].assets) == 1
assert isinstance(result[0].assets[0], LbwfAssetCondition)
assert isinstance(result[1], LbwfHouse)
assert result[1].uprn == 2
assert len(result[1].assets) == 1
assert isinstance(result[1].assets[0], LbwfAssetCondition)