diff --git a/applications/modelling_e2e/Dockerfile b/applications/modelling_e2e/Dockerfile new file mode 100644 index 00000000..fa13f8ef --- /dev/null +++ b/applications/modelling_e2e/Dockerfile @@ -0,0 +1,24 @@ +FROM public.ecr.aws/lambda/python:3.11 + +WORKDIR /var/task + +COPY applications/modelling_e2e/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY datatypes/ datatypes/ +COPY domain/ domain/ +COPY infrastructure/ infrastructure/ +COPY orchestration/ orchestration/ +COPY repositories/ repositories/ +COPY utilities/ utilities/ +COPY harness/ harness/ + +# harness/console.py imports in-memory fakes from tests/orchestration/ at module +# load time; the fakes have no pytest dependency and are safe to ship. +COPY tests/__init__.py tests/__init__.py +COPY tests/orchestration/__init__.py tests/orchestration/__init__.py +COPY tests/orchestration/fakes.py tests/orchestration/fakes.py + +COPY applications/ applications/ + +CMD ["applications.modelling_e2e.handler.handler"] diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py new file mode 100644 index 00000000..d45951e3 --- /dev/null +++ b/applications/modelling_e2e/handler.py @@ -0,0 +1,358 @@ +"""SQS-triggered Lambda: fetch EPC (or predict) → run modelling → persist plan. + +One SQS message = one batch of properties sharing a portfolio, scenario, and +(by caller convention) postcode. The handler reads ``property_ids``, +``portfolio_id``, ``scenario_id``, ``no_solar``, and ``dry_run`` from the +message body, fetches or predicts each property's EPC, runs the full modelling +pipeline (SAP10 → optimiser) via ``harness.console.run_modelling``, and +persists the resulting Plan via ``PostgresUnitOfWork`` in one atomic transaction +per property. + +When no lodged EPC is found, EPC Prediction (Path 3, ADR-0031) synthesises one +from the postcode cohort. ``_cohort_cache`` is module-level so warm Lambda +containers re-processing the same postcode avoid redundant fetches. + +``secondary_heating_removal`` is excluded unconditionally: the live ``material`` +catalogue does not yet carry this measure type, causing a crash during catalogue +reads for properties with a lodged secondary heater. + +DB engine is module-scoped so the connection pool is reused across warm +invocations (ADR-0012). +""" + +from __future__ import annotations + +import io +import os +from collections.abc import Callable +from typing import Any, Optional, cast + +import boto3 +import pandas as pd # pyright: ignore[reportMissingTypeStubs] +from sqlalchemy import Engine, text +from sqlmodel import Session + +from datatypes.epc.domain.epc_property_data import ( + BuildingPartIdentifier, + EpcPropertyData, +) +from domain.epc_prediction.comparable_properties import ( + ComparableProperty, + select_comparables, +) +from domain.epc_prediction.epc_prediction import EpcPrediction +from domain.epc_prediction.prediction_target import build_prediction_target +from domain.geospatial.coordinates import Coordinates +from domain.geospatial.planning_restrictions import PlanningRestrictions +from domain.geospatial.spatial_reference import SpatialReference +from domain.modelling.measure_type import MeasureType +from domain.property.property import Property, PropertyIdentity +from domain.tasks.tasks import Source +from harness.console import run_modelling +from infrastructure.epc_client.epc_client_service import EpcClientService +from infrastructure.postgres.config import PostgresConfig +from infrastructure.postgres.engine import make_engine +from infrastructure.solar.google_solar_api_client import ( + BuildingInsightsNotFoundError, + GoogleSolarApiClient, +) +from applications.modelling_e2e.modelling_e2e_trigger_body import ( + ModellingE2ETriggerBody, +) +from repositories.comparable_properties.epc_comparable_properties_repository import ( + EpcComparablePropertiesRepository, +) +from repositories.geospatial.geospatial_s3_repository import ( + GeospatialS3Repository, + ParquetReader, +) +from repositories.postgres_unit_of_work import PostgresUnitOfWork +from repositories.product.product_postgres_repository import ProductPostgresRepository +from repositories.property.landlord_override_overlays import overlays_from +from repositories.property.override_backed_prediction_attributes_reader import ( + OverrideBackedPredictionAttributesReader, +) +from repositories.property.property_overrides_postgres_reader import ( + PropertyOverridesPostgresReader, +) +from repositories.scenario.scenario_postgres_repository import ( + ScenarioPostgresRepository, +) +from utilities.aws_lambda.task_handler import task_handler +from utilities.logger import setup_logger + +_engine: Optional[Engine] = None +_cohort_cache: dict[str, list[ComparableProperty]] = {} + +logger = setup_logger() + + +def _get_engine() -> Engine: + global _engine + if _engine is None: + _engine = make_engine(PostgresConfig.from_env(dict(os.environ))) + return _engine + + +def _s3_parquet_reader() -> ParquetReader: + bucket = os.environ["DATA_BUCKET"] + + def read(key: str) -> pd.DataFrame: + s3: Any = cast( + Any, boto3.client("s3") + ) # pyright: ignore[reportUnknownMemberType] + raw = cast(bytes, s3.get_object(Bucket=bucket, Key=key)["Body"].read()) + return pd.read_parquet(io.BytesIO(raw)) # type: ignore[return-value] + + return read + + +def _spatial_for( + geospatial: GeospatialS3Repository, uprn: int +) -> Optional[SpatialReference]: + try: + return geospatial.spatial_for(uprn) + except Exception: # noqa: BLE001 + return None + + +def _solar_insights_for( + solar_client: GoogleSolarApiClient, spatial: Optional[SpatialReference] +) -> Optional[dict[str, Any]]: + if spatial is None or spatial.coordinates is None: + return None + try: + return solar_client.get_building_insights( + spatial.coordinates.longitude, spatial.coordinates.latitude + ) + except BuildingInsightsNotFoundError: + return None + + +def _predict_epc( + *, + property_id: int, + uprn: int, + postcode: str, + portfolio_id: int, + attributes_reader: OverrideBackedPredictionAttributesReader, + coordinates: Optional[Coordinates], + cohort_for: Callable[[str], list[ComparableProperty]], + predictor: EpcPrediction, +) -> Optional[EpcPropertyData]: + """Synthesise an EpcPropertyData for an EPC-less property from its postcode + cohort (EPC Prediction Path 3, ADR-0031), or None when ineligible. + + Returns None when property_type is unresolvable (hard cohort filter cannot + fire) or when the postcode cohort is empty after filtering. + """ + attributes = attributes_reader.attributes_for(property_id) + identity = PropertyIdentity( + portfolio_id=portfolio_id, postcode=postcode, address="", uprn=uprn + ) + target = build_prediction_target(identity, coordinates, attributes) + if target is None: + return None + comparables = select_comparables(target, cohort_for(target.postcode)) + if not comparables.members: + return None + predicted = predictor.predict(target, comparables) + if not any( + part.identifier is BuildingPartIdentifier.MAIN + for part in predicted.sap_building_parts + ): + return None + return predicted + + +@task_handler(task_source="modelling_e2e", source=Source.PROPERTY) +def handler(body: dict[str, Any], context: Any) -> None: + trigger = ModellingE2ETriggerBody.model_validate(body) + property_ids = trigger.property_ids + portfolio_id = trigger.portfolio_id + scenario_id = trigger.scenario_id + no_solar = trigger.no_solar + dry_run = trigger.dry_run + + logger.info( + f"start property_ids={property_ids} portfolio={portfolio_id} " + f"scenario={scenario_id} no_solar={no_solar} dry_run={dry_run}" + ) + + engine = _get_engine() + epc_client = EpcClientService(os.environ["OPEN_EPC_API_TOKEN"]) + geospatial = GeospatialS3Repository(_s3_parquet_reader()) + solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"]) + + with engine.connect() as conn: + uprn_rows = conn.execute( + text("SELECT id, uprn FROM property WHERE id = ANY(:ids)"), + {"ids": property_ids}, + ).fetchall() + postcode_rows = conn.execute( + text("SELECT id, postcode FROM property WHERE id = ANY(:ids)"), + {"ids": property_ids}, + ).fetchall() + + uprns: dict[int, int] = {int(row[0]): int(row[1]) for row in uprn_rows} + postcodes: dict[int, str] = {int(row[0]): (row[1] or "") for row in postcode_rows} + + overrides_reader = PropertyOverridesPostgresReader(lambda: Session(engine)) + prediction_attrs_reader = OverrideBackedPredictionAttributesReader(overrides_reader) + comparables_repo = EpcComparablePropertiesRepository(epc_client, geospatial) + predictor = EpcPrediction() + + def _get_cohort(postcode: str) -> list[ComparableProperty]: + if postcode not in _cohort_cache: + _cohort_cache[postcode] = ( + comparables_repo.candidates_for(postcode) if postcode else [] + ) + return _cohort_cache[postcode] + + read_session = Session(engine) + try: + scenario = ScenarioPostgresRepository(read_session).get_many([scenario_id])[0] + products = ProductPostgresRepository(read_session) + + errors: list[int] = [] + + for property_id in property_ids: + try: + uprn = uprns[property_id] + postcode = postcodes.get(property_id, "") + logger.info(f"property={property_id} uprn={uprn} postcode={postcode!r}") + + spatial = _spatial_for(geospatial, uprn) + restrictions = ( + spatial.restrictions + if spatial is not None + else PlanningRestrictions() + ) + coordinates: Optional[Coordinates] = ( + spatial.coordinates if spatial is not None else None + ) + + epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn) + overrides = overlays_from(overrides_reader.overrides_for(property_id)) + + if epc is not None: + logger.info(f"property={property_id} lodged EPC found") + effective_epc = Property( + identity=PropertyIdentity( + portfolio_id=portfolio_id, + postcode=postcode, + address="", + uprn=uprn, + ), + epc=epc, + landlord_overrides=overrides, + ).effective_epc + else: + logger.info( + f"property={property_id} no lodged EPC — attempting prediction" + ) + predicted_epc = _predict_epc( + property_id=property_id, + uprn=uprn, + postcode=postcode, + portfolio_id=portfolio_id, + attributes_reader=prediction_attrs_reader, + coordinates=coordinates, + cohort_for=_get_cohort, + predictor=predictor, + ) + if predicted_epc is None: + raise ValueError( + f"no EPC for UPRN {uprn} and not predictable " + f"(unresolved property_type or empty '{postcode}' cohort)" + ) + effective_epc = Property( + identity=PropertyIdentity( + portfolio_id=portfolio_id, + postcode=postcode, + address="", + uprn=uprn, + ), + epc=None, + predicted_epc=predicted_epc, + landlord_overrides=overrides, + ).effective_epc + + solar_insights: Optional[dict[str, Any]] = ( + None if no_solar else _solar_insights_for(solar_client, spatial) + ) + + # secondary_heating_removal is absent from the live material.type + # enum; exclude unconditionally until the catalogue gap is resolved. + considered: Optional[frozenset[MeasureType]] = ( + frozenset(MeasureType) + - {MeasureType.SECONDARY_HEATING_REMOVAL} + - {MeasureType.SYSTEM_TUNE_UP_ZONED} + ) + + plan = run_modelling( + effective_epc, + planning_restrictions=restrictions, + solar_insights=solar_insights, + considered_measures=considered, + products=products, + scenario=scenario, + print_table=False, + ) + logger.info( + f"property={property_id} modelling complete " + f"measures={len(plan.measures)}" + ) + + if dry_run: + measure_types = ( + ", ".join(m.measure_type for m in plan.measures) or "none" + ) + logger.info( + f"[dry_run] property={property_id} " + f"measures=[{measure_types}] — skipping DB write" + ) + continue + + with PostgresUnitOfWork(lambda: Session(engine)) as uow: + if epc is not None: + uow.epc.save( + epc, property_id=property_id, portfolio_id=portfolio_id + ) + if spatial is not None: + uow.spatial.save(uprn, spatial) + if ( + solar_insights is not None + and spatial is not None + and spatial.coordinates is not None + ): + uow.solar.save( + uprn, + longitude=spatial.coordinates.longitude, + latitude=spatial.coordinates.latitude, + insights=solar_insights, + ) + uow.plan.save( + plan, + property_id=property_id, + scenario_id=scenario_id, + portfolio_id=portfolio_id, + is_default=scenario.is_default, + ) + uow.property.mark_modelled( + property_id, has_recommendations=bool(plan.measures) + ) + uow.commit() + logger.info(f"property={property_id} plan saved") + + except Exception as error: # noqa: BLE001 + logger.error( + f"property={property_id}: {type(error).__name__}: {error}", + exc_info=True, + ) + errors.append(property_id) + + if errors: + raise RuntimeError(f"failed property_ids: {errors}") + finally: + read_session.close() diff --git a/applications/modelling_e2e/local_handler/.env.local.example b/applications/modelling_e2e/local_handler/.env.local.example new file mode 100644 index 00000000..8054ecd8 --- /dev/null +++ b/applications/modelling_e2e/local_handler/.env.local.example @@ -0,0 +1,36 @@ +# Local-test environment for the modelling_e2e Lambda. +# +# cp .env.local.example .env.local then fill in the values below. +# +# .env.local is gitignored. The container hits REAL AWS and a REAL Postgres, +# so every value here points at infrastructure that exists. +# +# Set dry_run=true in invoke_local_lambda.py to run the full pipeline without +# writing anything to the DB — safe for local testing. +# +# Keep comments on their own lines — docker-compose's env_file parser folds a +# trailing "# ..." into the value. + +# --- Postgres (infrastructure/postgres/config.py -> PostgresConfig.from_env) --- +# POSTGRES_HOST <- DB_HOST, PORT <- DB_PORT, USERNAME <- DB_USERNAME, +# PASSWORD <- DB_PASSWORD, DATABASE <- DB_NAME. +POSTGRES_HOST= +POSTGRES_PORT=5432 +POSTGRES_USERNAME= +POSTGRES_PASSWORD= +POSTGRES_DATABASE= +# POSTGRES_DRIVER=psycopg2 (optional; defaults to psycopg2) + +# --- Handler config (applications/modelling_e2e/handler.py) --- +# OPEN_EPC_API_TOKEN: gov.uk EPC API token (root .env: OPEN_EPC_API_TOKEN). +# GOOGLE_SOLAR_API_KEY: Google Solar API key (root .env: GOOGLE_SOLAR_API_KEY). +# DATA_BUCKET: S3 bucket holding geospatial parquet files (root .env: DATA_BUCKET). +OPEN_EPC_API_TOKEN= +GOOGLE_SOLAR_API_KEY= +DATA_BUCKET= + +# --- AWS credentials for boto3 (S3 + EPC client) --- +AWS_ACCESS_KEY_ID= +AWS_SECRET_ACCESS_KEY= +AWS_DEFAULT_REGION=eu-west-2 +# AWS_SESSION_TOKEN= (only if using temporary/SSO credentials) diff --git a/applications/modelling_e2e/local_handler/docker-compose.yml b/applications/modelling_e2e/local_handler/docker-compose.yml new file mode 100644 index 00000000..09e41d65 --- /dev/null +++ b/applications/modelling_e2e/local_handler/docker-compose.yml @@ -0,0 +1,9 @@ +services: + modelling-e2e: + build: + context: ../../../ + dockerfile: applications/modelling_e2e/Dockerfile + ports: + - "9004:8080" + env_file: + - ../../../.env diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py new file mode 100644 index 00000000..b9a0d584 --- /dev/null +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +import json +import requests + +HOST = "localhost" +PORT = "9004" + +LAMBDA_URL = f"http://{HOST}:{PORT}/2015-03-31/functions/function/invocations" + +payload = { + "Records": [ + { + "body": json.dumps( + { + "property_ids": [722473], + "portfolio_id": 796, + "scenario_id": 1268, + "no_solar": False, + "dry_run": False, + } + ) + } + ] +} + +response = requests.post(LAMBDA_URL, json=payload) + +print("Status code:", response.status_code) +print("Response:") +print(response.text) diff --git a/applications/modelling_e2e/modelling_e2e_trigger_body.py b/applications/modelling_e2e/modelling_e2e_trigger_body.py new file mode 100644 index 00000000..cb83118a --- /dev/null +++ b/applications/modelling_e2e/modelling_e2e_trigger_body.py @@ -0,0 +1,11 @@ +from pydantic import BaseModel, ConfigDict + + +class ModellingE2ETriggerBody(BaseModel): + model_config = ConfigDict(extra="allow") + + property_ids: list[int] + portfolio_id: int + scenario_id: int + no_solar: bool = False + dry_run: bool = False diff --git a/applications/modelling_e2e/requirements.txt b/applications/modelling_e2e/requirements.txt new file mode 100644 index 00000000..bac38184 --- /dev/null +++ b/applications/modelling_e2e/requirements.txt @@ -0,0 +1,11 @@ +awslambdaric +boto3 +numpy==2.1.2 +pandas==2.2.2 +pyarrow==17.0.0 +httpx +requests +pydantic +sqlalchemy==2.0.36 +sqlmodel +psycopg2-binary==2.9.10 diff --git a/deployment/terraform/lambda/modelling_e2e/main.tf b/deployment/terraform/lambda/modelling_e2e/main.tf new file mode 100644 index 00000000..826e83aa --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/main.tf @@ -0,0 +1,51 @@ +data "terraform_remote_state" "shared" { + backend = "s3" + config = { + bucket = "assessment-model-terraform-state" + key = "env:/${var.stage}/terraform.tfstate" + region = "eu-west-2" + } +} + +data "aws_secretsmanager_secret_version" "db_credentials" { + secret_id = "${var.stage}/assessment_model/db_credentials" +} + +locals { + db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string) +} + +module "lambda" { + source = "../../modules/lambda_with_sqs" + + name = var.lambda_name + stage = var.stage + + image_uri = local.image_uri + + reserved_concurrent_executions = var.reserved_concurrent_executions + + batch_size = var.batch_size + maximum_concurrency = var.maximum_concurrency + + timeout = 900 + memory_size = 3008 + + environment = { + STAGE = var.stage + LOG_LEVEL = "info" + POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username + POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password + POSTGRES_HOST = var.db_host + POSTGRES_DATABASE = var.db_name + POSTGRES_PORT = var.db_port + OPEN_EPC_API_TOKEN = var.open_epc_api_token + GOOGLE_SOLAR_API_KEY = var.google_solar_api_key + DATA_BUCKET = "retrofit-data-${var.stage}" + } +} + +resource "aws_iam_role_policy_attachment" "modelling_e2e_s3_read" { + role = module.lambda.role_name + policy_arn = data.terraform_remote_state.shared.outputs.modelling_e2e_s3_read_arn +} diff --git a/deployment/terraform/lambda/modelling_e2e/outputs.tf b/deployment/terraform/lambda/modelling_e2e/outputs.tf new file mode 100644 index 00000000..34cf4cef --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/outputs.tf @@ -0,0 +1,9 @@ +output "modelling_e2e_queue_url" { + value = module.lambda.queue_url + description = "URL of the modelling-e2e SQS queue (pass to trigger_modelling_e2e_sqs.py --sqs-url)" +} + +output "modelling_e2e_queue_arn" { + value = module.lambda.queue_arn + description = "ARN of the modelling-e2e SQS queue" +} diff --git a/deployment/terraform/lambda/modelling_e2e/provider.tf b/deployment/terraform/lambda/modelling_e2e/provider.tf new file mode 100644 index 00000000..7100be24 --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/provider.tf @@ -0,0 +1,20 @@ +terraform { + required_providers { + aws = { + source = "hashicorp/aws" + version = ">= 5.0" + } + } + + backend "s3" { + bucket = "modelling-e2e-terraform-state" + key = "terraform.tfstate" + region = "eu-west-2" + } + + required_version = ">= 1.2.0" +} + +provider "aws" { + region = "eu-west-2" +} diff --git a/deployment/terraform/lambda/modelling_e2e/variables.tf b/deployment/terraform/lambda/modelling_e2e/variables.tf new file mode 100644 index 00000000..417f2780 --- /dev/null +++ b/deployment/terraform/lambda/modelling_e2e/variables.tf @@ -0,0 +1,65 @@ +variable "lambda_name" { + type = string + description = "Logical name of the lambda" +} + +variable "stage" { + description = "Deployment stage (e.g. dev, prod)" + type = string +} + +variable "ecr_repo_url" { + type = string + description = "ECR repository URL (no tag, no digest)" +} + +variable "image_digest" { + type = string + description = "Image digest (sha256:...)" +} + +variable "reserved_concurrent_executions" { + type = number + default = -1 + description = "Reserved concurrency for the Lambda. -1 = unreserved." +} + +variable "maximum_concurrency" { + type = number + default = 2 + description = "Maximum concurrent Lambda invocations from the SQS trigger." +} + +variable "batch_size" { + type = number + default = 1 +} + +variable "db_host" { + type = string + sensitive = true +} + +variable "db_name" { + type = string + sensitive = true +} + +variable "db_port" { + type = string + sensitive = true +} + +variable "open_epc_api_token" { + type = string + sensitive = true +} + +variable "google_solar_api_key" { + type = string + sensitive = true +} + +locals { + image_uri = "${var.ecr_repo_url}@${var.image_digest}" +} diff --git a/deployment/terraform/shared/main.tf b/deployment/terraform/shared/main.tf index 3d6bbd39..bca65bb3 100644 --- a/deployment/terraform/shared/main.tf +++ b/deployment/terraform/shared/main.tf @@ -858,3 +858,35 @@ module "sharepoint_renamer_registry" { stage = var.stage } +################################################ +# Modelling E2E – Lambda +################################################ +module "modelling_e2e_state_bucket" { + source = "../modules/tf_state_bucket" + bucket_name = "modelling-e2e-terraform-state" +} + +module "modelling_e2e_registry" { + source = "../modules/container_registry" + name = "modelling-e2e" + stage = var.stage +} + +module "modelling_e2e_s3_read" { + source = "../modules/s3_iam_policy" + + policy_name = "ModellingE2EReadS3" + policy_description = "Allow modelling-e2e Lambda to read spatial parquet from the data bucket" + bucket_arns = ["arn:aws:s3:::retrofit-data-${var.stage}"] + actions = ["s3:GetObject", "s3:ListBucket"] + resource_paths = ["/*"] +} + +output "modelling_e2e_s3_read_arn" { + value = module.modelling_e2e_s3_read.policy_arn +} + +output "modelling_e2e_ecr_url" { + value = module.modelling_e2e_registry.repository_url +} + diff --git a/domain/modelling/generators/heating_recommendation.py b/domain/modelling/generators/heating_recommendation.py index 475039df..9eae0353 100644 --- a/domain/modelling/generators/heating_recommendation.py +++ b/domain/modelling/generators/heating_recommendation.py @@ -284,6 +284,7 @@ def recommend_heating( epc: EpcPropertyData, products: ProductRepository, restrictions: PlanningRestrictions = PlanningRestrictions(), + considered_measures: Optional[frozenset[MeasureType]] = None, ) -> Optional[Recommendation]: """Return a "Heating & Hot Water" Recommendation of competing whole-system bundles for the dwelling, else None when no bundle is eligible. ASHP is @@ -302,7 +303,7 @@ def recommend_heating( if boiler_option is not None: options.append(boiler_option) - options.extend(_system_tune_up_options(epc, products)) + options.extend(_system_tune_up_options(epc, products, considered_measures)) if not options: return None @@ -310,7 +311,9 @@ def recommend_heating( def _system_tune_up_options( - epc: EpcPropertyData, products: ProductRepository + epc: EpcPropertyData, + products: ProductRepository, + considered_measures: Optional[frozenset[MeasureType]] = None, ) -> list[MeasureOption]: """The system tune-up options: keep the existing wet boiler but install better heating controls (standard 2106 and/or zone 2110, as competing @@ -338,7 +341,10 @@ def _system_tune_up_options( ), ) ) - if control_code not in _ZONE_CONTROL_CODES: + if control_code not in _ZONE_CONTROL_CODES and ( + considered_measures is None + or _SYSTEM_TUNE_UP_ZONED_MEASURE_TYPE in considered_measures + ): options.append( _tune_up_option( epc, diff --git a/domain/tasks/tasks.py b/domain/tasks/tasks.py index 177258d6..ead253ab 100644 --- a/domain/tasks/tasks.py +++ b/domain/tasks/tasks.py @@ -17,6 +17,7 @@ class TaskStatus(str, Enum): class Source(str, Enum): PORTFOLIO = "portfolio_id" HUBSPOT_DEAL = "hubspot_deal_id" + PROPERTY = "property_id" @dataclass diff --git a/orchestration/modelling_orchestrator.py b/orchestration/modelling_orchestrator.py index 55ae531d..e3180a4d 100644 --- a/orchestration/modelling_orchestrator.py +++ b/orchestration/modelling_orchestrator.py @@ -331,7 +331,7 @@ def _candidate_recommendations( MeasureType.SYSTEM_TUNE_UP, MeasureType.SYSTEM_TUNE_UP_ZONED, ), - lambda: recommend_heating(effective_epc, products, planning_restrictions), + lambda: recommend_heating(effective_epc, products, planning_restrictions, considered_measures), ), ( admitted(MeasureType.SECONDARY_HEATING_REMOVAL), diff --git a/scripts/trigger_modelling_e2e_sqs.py b/scripts/trigger_modelling_e2e_sqs.py new file mode 100644 index 00000000..3bcfeaf9 --- /dev/null +++ b/scripts/trigger_modelling_e2e_sqs.py @@ -0,0 +1,105 @@ +"""Enqueue one SQS message per property for the modelling_e2e Lambda. + +Reads all property IDs for the given portfolio from the DB and sends a batch of +SQS messages, one per property. The Lambda then processes each message +independently, enabling concurrent modelling at scale. + +Edit the CONFIG block below, then run via VSCode Run button or Jupyter. +AWS creds come from the ambient ~/.aws profile; DB creds from backend/.env. +""" + +from __future__ import annotations + +# -------------------------------------------------------------------------- +# CONFIG — edit these before running +# --------------------------------------------------------------------------- +PORTFOLIO_ID: int = 785 +SCENARIO_ID: int = 1266 +SQS_URL: str = "https://sqs.eu-west-2.amazonaws.com/ACCOUNT_ID/modelling-e2e-STAGE" + +# Set to a positive integer to enqueue only the first N properties (trial run). +LIMIT: int | None = 10 + +# True → Lambda runs the full pipeline but skips all DB writes (safe for testing). +DRY_RUN: bool = True + +# True → Lambda skips the Google Solar fetch. +NO_SOLAR: bool = False +# --------------------------------------------------------------------------- + +import json +import sys +from pathlib import Path +from typing import Any, cast +from uuid import uuid4 + +_REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(_REPO_ROOT)) + +import boto3 # noqa: E402 +from sqlalchemy import text # noqa: E402 + +from scripts.e2e_common import ENV_PATH, build_engine, load_env # noqa: E402 + +_BATCH_SIZE = 10 + + +def _property_ids(portfolio_id: int, limit: int | None, engine: object) -> list[int]: + from sqlalchemy.engine import Engine + + assert isinstance(engine, Engine) + query = "SELECT id FROM property WHERE portfolio_id = :pid ORDER BY id" + if limit is not None: + query += f" LIMIT {int(limit)}" + with engine.connect() as conn: + rows = conn.execute(text(query), {"pid": portfolio_id}).fetchall() + return [int(r[0]) for r in rows] + + +def _batches(items: list[int], size: int) -> list[list[int]]: + return [items[i : i + size] for i in range(0, len(items), size)] + + +def main() -> None: + load_env(ENV_PATH) + engine = build_engine() + + ids = _property_ids(PORTFOLIO_ID, LIMIT, engine) + if not ids: + print(f"no properties found for portfolio {PORTFOLIO_ID}") + return + + print( + f"enqueuing {len(ids)} properties " + f"(portfolio={PORTFOLIO_ID}, scenario={SCENARIO_ID}, " + f"no_solar={NO_SOLAR}, dry_run={DRY_RUN}) → {SQS_URL}" + ) + + sqs: Any = cast( + Any, boto3.client("sqs") + ) # pyright: ignore[reportUnknownMemberType] + sent = 0 + for batch in _batches(ids, _BATCH_SIZE): + entries = [ + { + "Id": str(uuid4()).replace("-", "")[:8] + str(i), + "MessageBody": json.dumps( + { + "property_id": [pid], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": NO_SOLAR, + "dry_run": DRY_RUN, + } + ), + } + for i, pid in enumerate(batch) + ] + sqs.send_message_batch(QueueUrl=SQS_URL, Entries=entries) + sent += len(batch) + print(f" sent {sent}/{len(ids)}", end="\r") + + print(f"\ndone — {sent} messages enqueued") + + +main() diff --git a/tests/applications/modelling_e2e/__init__.py b/tests/applications/modelling_e2e/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py new file mode 100644 index 00000000..5979fe1b --- /dev/null +++ b/tests/applications/modelling_e2e/test_handler.py @@ -0,0 +1,679 @@ +"""Tests for the modelling_e2e Lambda handler. + +Tests exercise the handler's external behaviour through handler.__wrapped__, +patching I/O boundaries (EPC client, DB reads, UoW) so no real DB or network +is needed. One test per distinct behaviour path. +""" + +from __future__ import annotations + +from contextlib import ExitStack +from typing import Any +from unittest.mock import MagicMock, call, patch + +import pytest +from pydantic import ValidationError + +from applications.modelling_e2e.modelling_e2e_trigger_body import ( + ModellingE2ETriggerBody, +) + +PROPERTY_ID = 12345 +UPRN = 987654321 +POSTCODE = "SW1A 1AA" +PORTFOLIO_ID = 100 +SCENARIO_ID = 200 + +_ENV = { + "OPEN_EPC_API_TOKEN": "test-token", + "DATA_BUCKET": "test-bucket", + "GOOGLE_SOLAR_API_KEY": "test-solar-key", +} + +_BODY = { + "property_ids": [PROPERTY_ID], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, +} + + +def _call_handler(body: dict[str, Any]) -> Any: + from applications.modelling_e2e.handler import handler + + return handler.__wrapped__(body, None) # type: ignore[attr-defined] + + +def _engine_mock( + property_ids: list[int], + uprns: list[int], + postcodes: list[str], +) -> MagicMock: + """Mock engine whose connect() returns UPRN then postcode rows.""" + mock_engine = MagicMock() + mock_conn = mock_engine.connect.return_value.__enter__.return_value + + uprn_result = MagicMock() + uprn_result.fetchall.return_value = list(zip(property_ids, uprns)) + + postcode_result = MagicMock() + postcode_result.fetchall.return_value = list(zip(property_ids, postcodes)) + + mock_conn.execute.side_effect = [uprn_result, postcode_result] + return mock_engine + + +def _plan_mock() -> MagicMock: + plan = MagicMock() + plan.measures = [] + plan.cost_of_works = 0.0 + return plan + + +# --------------------------------------------------------------------------- +# Fixture: clear module-level cohort cache between tests +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _clear_cohort_cache() -> None: + import applications.modelling_e2e.handler as h + + h._cohort_cache.clear() + + +# --------------------------------------------------------------------------- +# Trigger body validation +# --------------------------------------------------------------------------- + + +def test_trigger_body_requires_property_ids() -> None: + """property_ids (plural) must be provided and must be a list of ints.""" + # Arrange + body = { + "property_ids": [1, 2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + } + + # Act + result = ModellingE2ETriggerBody.model_validate(body) + + # Assert + assert result.property_ids == [1, 2] + + +def test_trigger_body_rejects_missing_property_ids() -> None: + with pytest.raises(ValidationError): + ModellingE2ETriggerBody.model_validate( + {"portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID} + ) + + +# --------------------------------------------------------------------------- +# Lodged EPC path +# --------------------------------------------------------------------------- + + +def test_lodged_epc_path_saves_epc_plan_and_marks_modelled() -> None: + """When get_by_uprn returns an EPC the handler saves it, saves the plan, + and marks the property as modelled — all inside one UoW per property.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + mock_epc = MagicMock() + mock_plan = _plan_mock() + mock_uow = MagicMock() + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = mock_epc + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.Session") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + _call_handler(_BODY) + + # Assert — EPC saved (lodged path), plan saved, property marked modelled + mock_uow.epc.save.assert_called_once_with( + mock_epc, property_id=PROPERTY_ID, portfolio_id=PORTFOLIO_ID + ) + mock_uow.plan.save.assert_called_once() + mock_uow.property.mark_modelled.assert_called_once_with( + PROPERTY_ID, has_recommendations=False + ) + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# EPC Prediction path +# --------------------------------------------------------------------------- + + +def test_prediction_path_saves_plan_without_epc_save() -> None: + """When get_by_uprn returns None the handler synthesises an EPC via + prediction and saves the plan — but never calls epc.save.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + mock_predicted_epc = MagicMock() + # _predict_epc checks for a MAIN building part + from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier + + mock_part = MagicMock() + mock_part.identifier = BuildingPartIdentifier.MAIN + mock_predicted_epc.sap_building_parts = [mock_part] + + mock_comparables = MagicMock() + mock_comparables.members = [MagicMock()] # non-empty cohort + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = None # no lodged EPC + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + # Prediction infrastructure + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = mock_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ).return_value.predict.return_value = mock_predicted_epc + stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ).return_value.candidates_for.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.Session") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + _call_handler(_BODY) + + # Assert — epc.save NOT called (no lodged cert), plan IS saved + mock_uow.epc.save.assert_not_called() + mock_uow.plan.save.assert_called_once() + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# Prediction gate-out (empty cohort) +# --------------------------------------------------------------------------- + + +def test_empty_cohort_gates_property_out_and_raises() -> None: + """When candidates_for returns an empty list the property cannot be + predicted; the handler records it as an error and raises RuntimeError.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + + empty_comparables = MagicMock() + empty_comparables.members = [] # empty cohort → gate-out + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = None + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = empty_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ).return_value.candidates_for.return_value = [] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + + # Act + with pytest.raises(RuntimeError, match=str(PROPERTY_ID)): + _call_handler(_BODY) + + # UoW never entered — the property errored before the write block + MockUoW.return_value.__enter__.assert_not_called() + + +# --------------------------------------------------------------------------- +# Partial batch failure +# --------------------------------------------------------------------------- + + +def test_partial_batch_failure_raises_runtime_error_listing_failed_ids() -> None: + """Two properties: property 1 succeeds, property 2 raises during modelling. + Handler raises RuntimeError naming only the failed property; property 1's + UoW was committed.""" + # Arrange + pid1, pid2 = 111, 222 + mock_engine = _engine_mock([pid1, pid2], [1001, 1002], [POSTCODE, POSTCODE]) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + def _run_modelling_side_effect(*args: Any, **kwargs: Any) -> Any: + # Fail on second call (pid2) + if not hasattr(_run_modelling_side_effect, "_calls"): + _run_modelling_side_effect._calls = 0 # type: ignore[attr-defined] + _run_modelling_side_effect._calls += 1 # type: ignore[attr-defined] + if _run_modelling_side_effect._calls == 2: # type: ignore[attr-defined] + raise ValueError("modelling exploded") + return mock_plan + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = MagicMock() # lodged EPC + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + side_effect=_run_modelling_side_effect, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + with pytest.raises(RuntimeError, match=str(pid2)): + _call_handler( + { + "property_ids": [pid1, pid2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, + } + ) + + # Property 1 succeeded — its UoW was committed + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# Cohort cache hit +# --------------------------------------------------------------------------- + + +def test_cohort_cache_prevents_duplicate_candidates_for_calls() -> None: + """Two properties sharing a postcode: candidates_for is fetched once and + cached — EpcComparablePropertiesRepository.candidates_for called once.""" + # Arrange + pid1, pid2 = 301, 302 + mock_engine = _engine_mock( + [pid1, pid2], [3001, 3002], [POSTCODE, POSTCODE] + ) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + mock_predicted_epc = MagicMock() + from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier + + mock_part = MagicMock() + mock_part.identifier = BuildingPartIdentifier.MAIN + mock_predicted_epc.sap_building_parts = [mock_part] + + mock_comparables = MagicMock() + mock_comparables.members = [MagicMock()] + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = None # force prediction path + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = mock_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ).return_value.predict.return_value = mock_predicted_epc + MockCandidates = stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ) + MockCandidates.return_value.candidates_for.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + MockUoW.return_value.__enter__.return_value = mock_uow + MockUoW.return_value.__exit__.return_value = False + + # Act + _call_handler( + { + "property_ids": [pid1, pid2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, + } + ) + + # Assert — cohort fetched exactly once despite two properties + MockCandidates.return_value.candidates_for.assert_called_once_with(POSTCODE) + + +# --------------------------------------------------------------------------- +# Dry-run +# --------------------------------------------------------------------------- + + +def test_dry_run_skips_all_db_writes() -> None: + """dry_run=True: run_modelling executes but PostgresUnitOfWork is never + entered — no DB writes occur for any property in the batch.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + + with ExitStack() as stack: + stack.enter_context( + patch("applications.modelling_e2e.handler.os.environ", _ENV) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._get_engine", + return_value=mock_engine, + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcClientService") + ).return_value.get_by_uprn.return_value = MagicMock() + stack.enter_context( + patch("applications.modelling_e2e.handler.GeospatialS3Repository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.GoogleSolarApiClient") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._spatial_for", return_value=None + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler._solar_insights_for", + return_value=None, + ) + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=_plan_mock(), + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + + # Act + _call_handler({**_BODY, "dry_run": True}) + + # Assert — UoW never entered + MockUoW.return_value.__enter__.assert_not_called() diff --git a/tests/lambdas/__init__.py b/tests/lambdas/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/lambdas/modelling_e2e/__init__.py b/tests/lambdas/modelling_e2e/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index 82c7198e..093d6bd6 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/utilities/aws_lambda/task_handler.py @@ -5,6 +5,7 @@ TaskOrchestrator.create_task_with_subtask(...) + run_subtask(...). """ import json +import logging from contextlib import AbstractContextManager from functools import wraps from typing import Any, Callable, Optional, cast @@ -13,6 +14,8 @@ from utilities.aws_lambda.default_orchestrator import default_orchestrator from domain.tasks.tasks import Source from orchestration.task_orchestrator import TaskOrchestrator +logger = logging.getLogger(__name__) + OrchestratorCM = Callable[[], AbstractContextManager[TaskOrchestrator]] @@ -63,6 +66,11 @@ def task_handler( ) results.append(result) except Exception: + logger.exception( + "subtask failed (task_source=%s source_id=%s)", + task_source, + source_id, + ) if "Records" in event: message_id = record.get("messageId", "") failures.append({"itemIdentifier": message_id})