diff --git a/backend/app/config.py b/backend/app/config.py index 8939e6ff8..d1c5f3ace 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -44,6 +44,7 @@ class Settings(BaseSettings): COMBINER_SQS_URL: str = "changeme" LANDLORD_OVERRIDES_SQS_URL: str = "changeme" FINALISER_SQS_URL: str = "changeme" + MODELLING_E2E_SQS_URL: str = "changeme" # Third parties EPC_AUTH_TOKEN: str = "changeme" diff --git a/backend/app/modelling/router.py b/backend/app/modelling/router.py new file mode 100644 index 000000000..7eff93340 --- /dev/null +++ b/backend/app/modelling/router.py @@ -0,0 +1,73 @@ +"""The Modelling Run Distributor: POST /v1/modelling/trigger-run (ADR-0055). + +Accepts a portfolio-scoped modelling request expressed as filters, resolves +them to a concrete property set (ADR-0056), pre-creates one batch sub_task per +SQS message under the app-owned task, and fans the batches out to the +modelling_e2e workers. Never models synchronously; owns nothing after the +fan-out — progress and terminal state roll up from the workers. +""" + +import json +from collections.abc import Callable, Iterator +from typing import Any, cast + +import boto3 +from fastapi import APIRouter, Depends +from sqlmodel import Session + +from backend.app.config import get_settings +from backend.app.db.connection import db_engine +from backend.app.dependencies import validate_token +from backend.app.modelling.schemas import TriggerRunRequest + +# Sends pre-serialised message bodies to the modelling_e2e queue. A seam so +# tests record bodies instead of calling AWS. +MessageSender = Callable[[list[str]], None] + + +def get_session() -> Iterator[Session]: + with Session(db_engine) as session: + yield session + + +def get_message_sender() -> MessageSender: + settings = get_settings() + client: Any = cast( + Any, boto3.client("sqs", settings.AWS_DEFAULT_REGION) + ) # pyright: ignore[reportUnknownMemberType] + queue_url = settings.MODELLING_E2E_SQS_URL + + def send(bodies: list[str]) -> None: + # send_message_batch caps at 10 entries per call — chunk accordingly. + for start in range(0, len(bodies), 10): + chunk = bodies[start : start + 10] + client.send_message_batch( + QueueUrl=queue_url, + Entries=[ + {"Id": str(index), "MessageBody": body} + for index, body in enumerate(chunk) + ], + ) + + return send + + +router = APIRouter( + prefix="/modelling", + tags=["modelling"], + dependencies=[Depends(validate_token)], +) + + +@router.post("/trigger-run", status_code=202) +async def trigger_run( + body: TriggerRunRequest, + session: Session = Depends(get_session), + send_messages: MessageSender = Depends(get_message_sender), +) -> dict[str, str]: + raise NotImplementedError + + +# json is used by the implementation; referenced here so the stub imports are +# stable for the first RED. +_ = json diff --git a/backend/app/modelling/schemas.py b/backend/app/modelling/schemas.py new file mode 100644 index 000000000..272f5ab4d --- /dev/null +++ b/backend/app/modelling/schemas.py @@ -0,0 +1,18 @@ +"""Request contract for the Modelling Run Distributor (ADR-0055/0056).""" + +from uuid import UUID + +from pydantic import BaseModel + +from backend.app.modelling.property_filters import PropertyGroupFilters + + +class TriggerRunRequest(BaseModel): + """Exactly what the app sends: the app-created task, the portfolio scope, + the scenarios to model, and the property-group filters ({} = everything). + Anything else the distributor might want is readable from the task row.""" + + task_id: UUID + portfolio_id: int + scenario_ids: list[int] + filters: PropertyGroupFilters = PropertyGroupFilters() diff --git a/tests/backend/app/modelling/test_trigger_run_router.py b/tests/backend/app/modelling/test_trigger_run_router.py new file mode 100644 index 000000000..ae9e020be --- /dev/null +++ b/tests/backend/app/modelling/test_trigger_run_router.py @@ -0,0 +1,133 @@ +"""Tests for the Modelling Run Distributor endpoint (ADR-0055). + +The router's session and SQS seams are dependency-injected; tests run against +the ephemeral Postgres and record message bodies instead of calling AWS. +""" + +import json +from collections.abc import Iterator +from dataclasses import dataclass, field +from typing import Any +from uuid import uuid4 + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import Engine +from sqlmodel import Session + +from backend.app.dependencies import validate_token +from backend.app.modelling.router import get_message_sender, get_session, router +from domain.modelling.portfolio_goal import PortfolioGoal +from domain.tasks.tasks import Task +from infrastructure.postgres.modelling import ScenarioModel +from infrastructure.postgres.property_table import PropertyRow +from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository +from repositories.tasks.task_postgres_repository import TaskPostgresRepository + +PORTFOLIO_ID = 814 + + +@dataclass +class Api: + client: TestClient + engine: Engine + sent_bodies: list[str] = field(default_factory=list) + + def seed_task(self) -> Task: + with Session(self.engine) as session: + return TaskPostgresRepository(session).create( + Task.create(task_source="app:modelling_run", service="modelling_run") + ) + + def seed_scenario(self, *, portfolio_id: int = PORTFOLIO_ID) -> int: + with Session(self.engine) as session: + row = ScenarioModel( + portfolio_id=portfolio_id, + goal=PortfolioGoal.INCREASING_EPC, + goal_value="C", + ) + session.add(row) + session.commit() + session.refresh(row) + assert row.id is not None + return row.id + + def seed_property(self, *, postcode: str = "B93 8SU") -> int: + with Session(self.engine) as session: + row = PropertyRow(portfolio_id=PORTFOLIO_ID, postcode=postcode) + session.add(row) + session.commit() + session.refresh(row) + assert row.id is not None + return row.id + + def subtasks_for(self, task: Task) -> list[Any]: + with Session(self.engine) as session: + return list(SubTaskPostgresRepository(session).list_by_task(task.id)) + + +@pytest.fixture +def api(db_engine: Engine) -> Api: + app = FastAPI() + app.include_router(router, prefix="/v1") + harness = Api(client=TestClient(app), engine=db_engine) + + def _session() -> Iterator[Session]: + with Session(db_engine) as session: + yield session + + app.dependency_overrides[get_session] = _session + app.dependency_overrides[get_message_sender] = lambda: harness.sent_bodies.extend + app.dependency_overrides[validate_token] = lambda: "test-token" + return harness + + +def test_trigger_run_fans_out_one_subtask_and_message_per_scenario_batch( + api: Api, +) -> None: + # arrange — 3 properties (one batch) × 2 scenarios = 2 messages + task = api.seed_task() + scenario_a = api.seed_scenario() + scenario_b = api.seed_scenario() + property_ids = [api.seed_property() for _ in range(3)] + + # act + response = api.client.post( + "/v1/modelling/trigger-run", + json={ + "task_id": str(task.id), + "portfolio_id": PORTFOLIO_ID, + "scenario_ids": [scenario_a, scenario_b], + "filters": {}, + }, + ) + + # assert + assert response.status_code == 202 + subtasks = api.subtasks_for(task) + assert len(subtasks) == 2 + assert {s.status.value for s in subtasks} == {"waiting"} + + messages = [json.loads(body) for body in api.sent_bodies] + assert len(messages) == 2 + assert {m["scenario_id"] for m in messages} == {scenario_a, scenario_b} + for message in messages: + assert message["task_id"] == str(task.id) + assert message["portfolio_id"] == PORTFOLIO_ID + assert message["property_ids"] == property_ids + # ADR-0055 pinned flags: live fetch, live predict, solar only-if-missing + assert message["refetch_epc"] is True + assert message["repredict_epc"] is True + assert message["refetch_solar"] is True + assert message["dry_run"] is False + + # each message's subtask_id is one of the pre-created sub_tasks, and the + # sub_task's inputs are the message payload (its re-run recipe) + subtask_ids = {str(s.id) for s in subtasks} + assert {m["subtask_id"] for m in messages} == subtask_ids + by_id = {str(s.id): s for s in subtasks} + for message in messages: + inputs = by_id[message["subtask_id"]].inputs + assert inputs["property_ids"] == message["property_ids"] + assert inputs["scenario_id"] == message["scenario_id"]