import pandas as pd import numpy as np from pathlib import Path import time from backend.export.property_scenarios.main import process_export from backend.export.property_scenarios.input_schema import ExportRequest from backend.app.db.models.portfolio import ( PropertyModel, Portfolio, PortfolioStatus, PortfolioGoal, PropertyCreationStatus, PropertyDetailsEpcModel, ) from backend.app.db.models.recommendations import ( PlanModel, Recommendation, PlanRecommendations, RecommendationMaterials, ) from backend.app.db.models.materials import Material from domain.epc.epc import Epc from utils.logger import setup_logger FIXTURE_PATH = Path("backend/export/tests/fixtures") logger = setup_logger() def load_csv(name: str) -> pd.DataFrame: df = pd.read_csv(FIXTURE_PATH / name) df = df.replace({np.nan: None}) return df def test_default_export_integration(db_session): # ---------------------------------------- # 1) Load csvs # ---------------------------------------- t0 = time.perf_counter() portfolio_df = load_csv("portfolio_569.csv") properties_df = load_csv("properties_569.csv") property_details_epc_df = load_csv("property_details_epc_569.csv") plans_df = load_csv("plans_569.csv") plan_recs_df = load_csv("plan_recs_569.csv") recommendations_df = load_csv("recommendations_569.csv") logger.info( "Loaded CSVs in %.2f seconds | properties=%s plans=%s recs=%s", time.perf_counter() - t0, len(properties_df), len(plans_df), len(recommendations_df), ) logger.info("Starting database load") db_load_t0 = time.perf_counter() # ---------------------------------------- # 2) Insert test portfolio # ---------------------------------------- portfolios = [] for row in portfolio_df.itertuples(index=False): portfolios.append( Portfolio( id=row.id, name=row.name, status=PortfolioStatus[row.status.split(".")[-1]], goal=PortfolioGoal[row.goal.split(".")[-1]] if row.goal else None, ) ) db_session.bulk_save_objects(portfolios) db_session.flush() # ---------------------------------------- # 3) Insert test property # ---------------------------------------- properties = [] for row in properties_df.itertuples(index=False): row_dict = {field: getattr(row, field) for field in row._fields} row_dict["uprn"] = int(row_dict["uprn"]) if row_dict.get("uprn") else None row_dict["building_reference_number"] = ( int(row_dict["building_reference_number"]) if row_dict.get("building_reference_number") else None ) prop = PropertyModel( **{ col: row_dict[col] for col in PropertyModel.__table__.columns.keys() if col in row_dict } ) prop.creation_status = PropertyCreationStatus[ row_dict["creation_status"].split(".")[-1] ] prop.status = PortfolioStatus[row_dict["status"].split(".")[-1]] if row_dict.get("current_epc_rating"): prop.current_epc_rating = Epc[row_dict["current_epc_rating"].split(".")[-1]] properties.append(prop) db_session.bulk_save_objects(properties) db_session.flush() # ---------------------------------------- # 4) Insert property details - EPC # ---------------------------------------- epc_rows = [] for row in property_details_epc_df.itertuples(index=False): row_dict = {field: getattr(row, field) for field in row._fields} # Build only fields that exist on the model epc_data = { col.name: row_dict[col.name] for col in PropertyDetailsEpcModel.__table__.columns.values() if col.name in row_dict and col.name not in ["id", "property_id", "portfolio_id"] } epc = PropertyDetailsEpcModel( property_id=row.property_id, portfolio_id=row.portfolio_id, **epc_data, ) epc_rows.append(epc) db_session.bulk_save_objects(epc_rows) db_session.flush() # ---------------------------------------- # 4) Insert default plan # ---------------------------------------- plans = [] for row in plans_df.itertuples(index=False): row_dict = {field: getattr(row, field) for field in row._fields} if row_dict.get("post_epc_rating"): row_dict["post_epc_rating"] = Epc[ row_dict["post_epc_rating"].split(".")[-1] ] row_dict["scenario_id"] = None plan = PlanModel( **{ col: row_dict[col] for col in PlanModel.__table__.columns.keys() if col in row_dict } ) plans.append(plan) db_session.bulk_save_objects(plans) db_session.flush() # ---------------------------------------- # 5) Insert recommendation # ---------------------------------------- recs = [ Recommendation( **{ col: row[col] for col in Recommendation.__table__.columns.keys() if col in row } ) for _, row in recommendations_df.iterrows() ] db_session.bulk_save_objects(recs) db_session.flush() # ---------------------------------------- # 6) Insert PlanRecommendations # ---------------------------------------- links = [ PlanRecommendations( plan_id=row.plan_id, recommendation_id=row.recommendation_id, ) for row in plan_recs_df.itertuples(index=False) ] db_session.bulk_save_objects(links) db_session.commit() logger.info("Inserted all data in %.2f seconds", time.perf_counter() - db_load_t0) # ---------------------------------------- # 6) Build payload # ---------------------------------------- body_dict = { "task_id": "test", "subtask_id": "test", "portfolio_id": 569, "scenario_ids": [], "default_plans_only": True, } payload = ExportRequest.model_validate(body_dict) # ---------------------------------------- # 7) Call process_export # ---------------------------------------- logger.info( "Recommendation count in DB: %s", db_session.query(Recommendation).count() ) logger.info("Property count in DB: %s", db_session.query(PropertyModel).count()) logger.info( "Property EPC in DB: %s", db_session.query(PropertyDetailsEpcModel).count() ) logger.info("Plan count in DB: %s", db_session.query(PlanModel).count()) logger.info( "PlanRecommendatons count in DB: %s", db_session.query(PlanModel).count() ) logger.info("Starting process_export") process_t0 = time.perf_counter() result = process_export(payload, session=db_session) logger.info( "process_export finished in %.2f seconds", time.perf_counter() - process_t0 ) # ---------------------------------------- # 8) Assertions # ---------------------------------------- assert ( "default_plans" in result ), "Expected 'default_plans' in export result, got {}".format(result.keys()) df = result["default_plans"] assert df.shape[0] == 10, "Expected 10 properties in the export, got {}".format( df.shape[0] ) failed = df[df["predicted_post_works_sap"] < 69] failed_property_types = failed["property_type"].value_counts().to_dict() assert failed_property_types["Flat"] == 2 # Check the houses assert failed.shape[0] assert ( df["total_retrofit_cost"].sum() == 41706.585999999996 ), "Expected total retrofit cost to be 10000, got {}".format( df["total_retrofit_cost"].sum() ) assert ( df["predicted_post_works_sap"].sum() == 698.1 ), "Expected total predicted post works SAP to be 698.1, got {}".format( df["predicted_post_works_sap"].sum() ) assert ( df["sap_points"].sum() == 100.10000000000001 ), "Expected total SAP points increase to be 100.10000000000001, got {}".format( df["sap_points"].sum() ) def test_solar_with_battery_example(db_session): test_portfolio_id = 1 test_property_id = 1 portfolio_df = pd.DataFrame( [ { "id": test_portfolio_id, "name": "Example", "budget": None, "status": "PortfolioStatus.SCOPING", "goal": "PortfolioGoal.NONE", "cost": None, "number_of_properties": None, "co2_equivalent_savings": None, "energy_savings": None, "energy_cost_savings": None, "property_valuation_increase": None, "rental_yield_increase": None, "total_work_hours": None, "labour_days": None, "created_at": "2026-02-12 21:23:37.862000+00:00", "updated_at": "2026-02-12 21:23:37.862000+00:00", "epc_breakdown_pre_retrofit": None, "epc_breakdown_post_retrofit": None, "n_units_to_retrofit": None, "co2_per_unit_pre_retrofit": None, "co2_per_unit_post_retrofit": None, "energy_bill_per_unit_pre_retrofit": None, "energy_bill_per_unit_post_retrofit": None, "energy_consumption_per_unit_pre_retrofit": None, "energy_consumption_per_unit_post_retrofit": None, "valuation_improvement_per_unit": None, "cost_per_unit": None, "cost_per_co2_saved": None, "cost_per_sap_point": None, "valuation_return_on_investment": None, } ] ) properties_df = pd.DataFrame( [ { "id": test_property_id, "portfolio_id": test_portfolio_id, "creation_status": "PropertyCreationStatus.READY", "uprn": 100090438731, "landlord_property_id": "BARR052", "building_reference_number": 3460742868, "status": "PortfolioStatus.ASSESSMENT", "address": "52, Barrack Street", "postcode": "CO1 2LR", "has_pre_condition_report": True, "has_recommendations": True, "created_at": "2026-02-12 21:59:02.744427", "updated_at": "2026-02-19 16:18:57.941443", "property_type": "House", "built_form": "End-Terrace", "local_authority": "Colchester", "constituency": "Colchester", "number_of_rooms": 4.0, "year_built": 1900.0, "tenure": "rental (private)", "current_epc_rating": "Epc.E", "current_sap_points": 53.0, "current_valuation": 0.0, "installed_measures_sap_point_adjustment": 0.0, "is_sap_points_adjusted_for_installed_measures": False, "original_sap_points": 53.0, } ] ) property_details_epc_df = pd.DataFrame( [ { "id": 1534934, "property_id": test_property_id, "portfolio_id": test_portfolio_id, "full_address": "48, Medcalf Road", "lodgement_date": "2018-09-05", "is_expired": False, "total_floor_area": 68.0, "walls": "Solid brick, as built, no insulation", "walls_rating": 1, "roof": "Pitched, no insulation", "roof_rating": 1.0, "floor": "Solid, no insulation", "floor_rating": None, "windows": "Fully double glazed", "windows_rating": 4, "heating": "Boiler and radiators, mains gas", "heating_rating": 4, "heating_controls": "Programmer, room thermostat and trvs", "heating_controls_rating": 4, "hot_water": "From main system", "hot_water_rating": 4, "lighting": "Low energy lighting in all fixed outlets", "lighting_rating": 5, "mainfuel": "Mains gas not community", "ventilation": "natural", "solar_pv": 0.0, "solar_hot_water": False, "wind_turbine": 0.0, "floor_height": 2.55, "number_heated_rooms": None, "heat_loss_corridor": False, "unheated_corridor_length": None, "number_of_open_fireplaces": 0, "number_of_extensions": 0, "number_of_storeys": None, "mains_gas": True, "energy_tariff": "Single", "primary_energy_consumption": 278.0, "co2_emissions": 3.81, "current_energy_demand": 14643.366, "current_energy_demand_heating_hotwater": 12185.6, "estimated": False, "sap_05_overwritten": False, "sap_05_score": None, "sap_05_epc_rating": None, "heating_cost_current": 711.0628, "hot_water_cost_current": 139.06198, "lighting_cost_current": 70.770935, "appliances_cost_current": 609.7844, "gas_standing_charge": 128.0785, "electricity_standing_charge": 199.8375, "original_co2_emissions": 3.81, "original_primary_energy_consumption": 278.0, "original_current_energy_demand": 14643.366, "original_current_energy_demand_heating_hotwater": 12185.6, "installed_measures_co2_adjustment": 0.0, "installed_measures_energy_demand_adjustment": 0.0, "installed_measures_total_energy_bill_adjustment": 0.0, "installed_measures_heat_demand_adjustment": 0.0, "is_epc_adjusted_for_installed_measures": False, } ] ) plans_df = pd.DataFrame( [ { "id": 0, "name": None, "portfolio_id": test_portfolio_id, "property_id": test_property_id, "scenario_id": 1060, "created_at": "2026-02-19 16:14:45.560816", "is_default": True, "valuation_increase_lower_bound": 0.0302, "valuation_increase_upper_bound": 0.07, "valuation_increase_average": 0.048226666, "plan_type": None, "post_sap_points": 71.5, "post_epc_rating": "Epc.C", "post_co2_emissions": 4.1813498, "co2_savings": 0.71865046, "post_energy_bill": 1447.5204, "energy_bill_savings": 691.6662, "post_energy_consumption": 15303.688, "energy_consumption_savings": 3276.7622, "valuation_post_retrofit": None, "valuation_increase": None, "cost_of_works": 6984.568, "contingency_cost": 1003.9568, } ] ) plan_recs_df = pd.DataFrame([{"id": 0, "plan_id": 0, "recommendation_id": 0}]) recommendations_df = pd.DataFrame( [ { "id": 0, "property_id": test_property_id, "created_at": "2026-02-19 16:14:45.560816", "type": "solar_pv", "measure_type": "solar_pv", "description": "Fit solar", "estimated_cost": 10000, "default": True, "starting_u_value": None, "new_u_value": None, "sap_points": 1.5, "heat_demand": 14.9, "kwh_savings": 1041.2, "co2_equivalent_savings": 0.2, "energy_savings": 14.9, "energy_cost_savings": 72.639015, "property_valuation_increase": None, "rental_yield_increase": None, "total_work_hours": 4.16, "labour_days": 1.0, "already_installed": False, "plan_name": "whatever", } ] ) recommendations_materials_df = pd.DataFrame( [ { "id": 0, "recommendation_id": 0, "material_id": 0, "depth": None, "quantity": 1.0, "quantity_unit": "part", "estimated_cost": 10000, "created_at": "2026-02-19 16:14:45.560816", "updated_at": "2026-02-19 16:14:45.560816", } ] ) materials_df = pd.DataFrame( [ { "id": 0, "type": "solar_pv", "description": "Some solar product", "depth": 75.0, "depth_unit": "mm", "cost": None, "cost_unit": "gbp_per_m2", "r_value_per_mm": 0.030303031, "r_value_unit": "square_meter_kelvin_per_watt", "thermal_conductivity": 0.033, "thermal_conductivity_unit": "watt_per_meter_kelvin", "link": "Test", "created_at": "'2026-02-19 16:14:45.560816", "is_active": True, "prime_material_cost": None, "material_cost": 0.0, "labour_cost": 0.0, "labour_hours_per_unit": 0.0, "plant_cost": 0.0, "total_cost": 10000, "notes": None, "is_installer_quote": True, "innovation_rate": 0.25, "size": None, "size_unit": None, "includes_scaffolding": True, "includes_battery": True, "battery_size": 5.8, } ] ) # Load into db # ------------------------------------------------- # Insert Portfolio # ------------------------------------------------- for row in portfolio_df.itertuples(index=False): db_session.add( Portfolio( id=row.id, name=row.name, status=PortfolioStatus[row.status.split(".")[-1]], goal=PortfolioGoal[row.goal.split(".")[-1]], ) ) db_session.flush() # ------------------------------------------------- # Insert Property # ------------------------------------------------- for row in properties_df.itertuples(index=False): prop = PropertyModel( id=row.id, portfolio_id=row.portfolio_id, creation_status=PropertyCreationStatus[row.creation_status.split(".")[-1]], status=PortfolioStatus[row.status.split(".")[-1]], uprn=row.uprn, address=row.address, postcode=row.postcode, property_type=row.property_type, current_sap_points=row.current_sap_points, current_epc_rating=Epc[row.current_epc_rating.split(".")[-1]], ) db_session.add(prop) db_session.flush() # ------------------------------------------------- # Insert EPC Details # ------------------------------------------------- for row in property_details_epc_df.itertuples(index=False): epc = PropertyDetailsEpcModel( property_id=row.property_id, portfolio_id=row.portfolio_id, full_address=row.full_address, total_floor_area=row.total_floor_area, walls=row.walls, roof=row.roof, windows=row.windows, heating=row.heating, solar_pv=row.solar_pv, ) db_session.add(epc) db_session.flush() # ------------------------------------------------- # Insert Plan (default) # ------------------------------------------------- for row in plans_df.itertuples(index=False): plan = PlanModel( id=row.id, portfolio_id=row.portfolio_id, property_id=row.property_id, scenario_id=None, # default mode is_default=row.is_default, ) db_session.add(plan) db_session.flush() # ------------------------------------------------- # IMPORTANT: Force recommendation to be solar_pv # ------------------------------------------------- recommendations_df.loc[0, "measure_type"] = "solar_pv" for row in recommendations_df.itertuples(index=False): rec = Recommendation( id=row.id, property_id=row.property_id, measure_type=row.measure_type, estimated_cost=row.estimated_cost, default=row.default, already_installed=row.already_installed, sap_points=row.sap_points, type=row.type, description=row.description, ) db_session.add(rec) db_session.flush() # ------------------------------------------------- # Link Plan -> Recommendation # ------------------------------------------------- for row in plan_recs_df.itertuples(index=False): db_session.add( PlanRecommendations( plan_id=row.plan_id, recommendation_id=row.recommendation_id, ) ) db_session.flush() # ------------------------------------------------- # Insert Material (includes_battery=True) # ------------------------------------------------- for row in materials_df.itertuples(index=False): material = Material( id=row.id, type=row.type, description=row.description, depth_unit=row.depth_unit, cost_unit=row.cost_unit, r_value_unit=row.r_value_unit, thermal_conductivity_unit=row.thermal_conductivity_unit, includes_battery=row.includes_battery, is_active=row.is_active, ) db_session.add(material) db_session.flush() # ------------------------------------------------- # Link Recommendation -> Material # ------------------------------------------------- for row in recommendations_materials_df.itertuples(index=False): db_session.add( RecommendationMaterials( recommendation_id=row.recommendation_id, material_id=row.material_id, depth=row.depth or 0.0, quantity=row.quantity, quantity_unit=row.quantity_unit, estimated_cost=row.estimated_cost, ) ) db_session.commit() payload = ExportRequest.model_validate( { "task_id": "test", "subtask_id": "test", "portfolio_id": test_portfolio_id, "scenario_ids": [], "default_plans_only": True, } ) result = process_export(payload, session=db_session) assert "default_plans" in result df = result["default_plans"] assert "solar_pv_with_battery" in df.columns # solar_pv should NOT exist assert "solar_pv" not in df.columns assert df.shape[0] == 1, "Expected 1 property in the export, got {}".format( df.shape[0] ) # Cost should land in correct column assert df["solar_pv_with_battery"].iloc[0] == 10000