From ace3150363c113fcb07349c01068509fb630d09e Mon Sep 17 00:00:00 2001 From: Khalim Conn-Kowlessar Date: Tue, 14 Jul 2026 09:26:59 +0000 Subject: [PATCH] =?UTF-8?q?Build=20and=20upload=20the=20sheet-per-scenario?= =?UTF-8?q?=20export=20workbook=20=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestration/scenario_export_orchestrator.py | 81 ++++++++++++++++ .../test_scenario_export_orchestrator.py | 95 +++++++++++++++++++ 2 files changed, 176 insertions(+) create mode 100644 orchestration/scenario_export_orchestrator.py create mode 100644 tests/orchestration/test_scenario_export_orchestrator.py diff --git a/orchestration/scenario_export_orchestrator.py b/orchestration/scenario_export_orchestrator.py new file mode 100644 index 000000000..5accfc951 --- /dev/null +++ b/orchestration/scenario_export_orchestrator.py @@ -0,0 +1,81 @@ +"""The Scenario Export orchestrator (ADR-0065). + +Ties one export together: for each requested Scenario, read its Properties' +persisted rows, shape them into a sheet, render the sheet-per-scenario branded +workbook, upload it to the exports bucket, mint a presigned URL, and email the +requester. Best-effort email (the link is also recorded on ``sub_task.outputs``); +a selection that yields no rows for any Scenario is a recorded failure — never an +empty workbook emailed to the user. +""" + +from __future__ import annotations + +import html +import logging +from dataclasses import dataclass +from typing import Protocol, Sequence + +from domain.scenario_export.scenario_sheet import ( + PropertyScenarioData, + shape_scenario_sheet, +) +from domain.tasks.subtasks import SubTaskFailure +from infrastructure.s3.s3_client import S3Client +from infrastructure.xlsx.scenario_export_workbook import render_workbook +from repositories.email.email_sender import EmailSender + +logger = logging.getLogger(__name__) + + +class ExportRowReader(Protocol): + def rows_for( + self, + *, + portfolio_id: int, + scenario_id: int, + property_ids: Sequence[int], + ) -> list[PropertyScenarioData]: ... + + +class ScenarioNames(Protocol): + def name_for(self, scenario_id: int) -> str: ... + + +@dataclass(frozen=True) +class ScenarioExportResult: + """What a completed export reports (lands on ``sub_task.outputs``).""" + + presigned_url: str + export_s3_key: str + sheet_count: int + row_count: int + + +class ScenarioExportOrchestrator: + def __init__( + self, + *, + rows: ExportRowReader, + scenarios: ScenarioNames, + exports: S3Client, + email: EmailSender, + url_ttl_seconds: int, + key_prefix: str = "scenario-exports", + ) -> None: + self._rows = rows + self._scenarios = scenarios + self._exports = exports + self._email = email + self._url_ttl_seconds = url_ttl_seconds + self._key_prefix = key_prefix + + def run( + self, + *, + portfolio_id: int, + scenario_ids: Sequence[int], + property_ids: Sequence[int], + recipient_email: str, + export_name: str, + ) -> ScenarioExportResult: + raise NotImplementedError diff --git a/tests/orchestration/test_scenario_export_orchestrator.py b/tests/orchestration/test_scenario_export_orchestrator.py new file mode 100644 index 000000000..fc918d0e8 --- /dev/null +++ b/tests/orchestration/test_scenario_export_orchestrator.py @@ -0,0 +1,95 @@ +"""The Scenario Export orchestrator (ADR-0065): read → shape → render → upload → +presign → email, best-effort, verified with in-memory fakes and a moto S3.""" + +from __future__ import annotations + +from io import BytesIO +from typing import Iterator, Optional, Sequence + +import pytest +from moto import mock_aws +from openpyxl import load_workbook + +from domain.scenario_export.scenario_sheet import ExportMeasure, PropertyScenarioData +from infrastructure.s3.s3_client import S3Client +from orchestration.scenario_export_orchestrator import ScenarioExportOrchestrator +from tests.infrastructure import make_boto_client + +EXPORTS_BUCKET = "domna-exports-dev" + + +class _FakeRows: + """Returns preset export rows per scenario id.""" + + def __init__(self, by_scenario: dict[int, list[PropertyScenarioData]]) -> None: + self._by_scenario = by_scenario + + def rows_for( + self, + *, + portfolio_id: int, + scenario_id: int, + property_ids: Sequence[int], + ) -> list[PropertyScenarioData]: + return self._by_scenario.get(scenario_id, []) + + +class _FakeScenarioNames: + def __init__(self, names: dict[int, str]) -> None: + self._names = names + + def name_for(self, scenario_id: int) -> str: + return self._names[scenario_id] + + +class _RecordingEmail: + def __init__(self) -> None: + self.sent: list[tuple[str, str, str, Optional[str]]] = [] + + def send( + self, *, to: str, subject: str, body: str, html_body: Optional[str] = None + ) -> None: + self.sent.append((to, subject, body, html_body)) + + +def _row(property_id: int) -> PropertyScenarioData: + return PropertyScenarioData( + property_id=property_id, + landlord_property_id=f"LP{property_id}", + measures=[ExportMeasure(measure_type="loft_insulation", estimated_cost=1200.0)], + ) + + +@pytest.fixture +def exports() -> Iterator[S3Client]: + with mock_aws(): + boto_client = make_boto_client("s3") + boto_client.create_bucket(Bucket=EXPORTS_BUCKET) + yield S3Client(boto_client, EXPORTS_BUCKET) + + +def test_builds_and_uploads_a_sheet_per_scenario_workbook(exports: S3Client) -> None: + # arrange — two scenarios, each with a modelled property. + orchestrator = ScenarioExportOrchestrator( + rows=_FakeRows({5: [_row(1)], 9: [_row(1), _row(2)]}), + scenarios=_FakeScenarioNames({5: "Fabric first", 9: "Low carbon"}), + exports=exports, + email=_RecordingEmail(), + url_ttl_seconds=3600, + ) + + # act + result = orchestrator.run( + portfolio_id=100, + scenario_ids=[5, 9], + property_ids=[1, 2], + recipient_email="user@example.com", + export_name="portfolio-100", + ) + + # assert — the uploaded object is a real workbook with one sheet per scenario + # and the presigned URL points at it. + assert result.sheet_count == 2 + assert result.export_s3_key in result.presigned_url + workbook = load_workbook(BytesIO(exports.get_object(result.export_s3_key))) + assert workbook.sheetnames == ["Fabric first", "Low carbon"]