diff --git a/.github/workflows/v2-integration-check.yml b/.github/workflows/v2-integration-check.yml index baf3e27ae..d0f5cdf5f 100644 --- a/.github/workflows/v2-integration-check.yml +++ b/.github/workflows/v2-integration-check.yml @@ -74,7 +74,7 @@ jobs: - name: Verify the managed v2 roles hold their declared privileges run: uv run --frozen --extra dev python scripts/reconcile_v2_privileges.py verify - name: Test v2 metadata publication, resource routes, the countries list, reports and simulations - run: uv run --frozen --extra dev coverage run --branch -m pytest -q tests/integration/test_v2_catalog_publication.py tests/integration/test_v2_catalog_publication_time_periods.py tests/integration/test_v2_catalog_publication_pruning.py tests/integration/test_v2_catalog_spm_availability.py tests/integration/test_v2_metadata_routes.py tests/integration/test_v2_country_routes.py tests/integration/test_v2_report_reads.py tests/integration/test_v2_report_creation.py tests/integration/test_v2_budget_window_report_reads.py tests/integration/test_v2_budget_window_report_creation.py + run: uv run --frozen --extra dev coverage run --branch -m pytest -q tests/integration/test_v2_catalog_publication.py tests/integration/test_v2_catalog_publication_time_periods.py tests/integration/test_v2_catalog_publication_pruning.py tests/integration/test_v2_catalog_spm_availability.py tests/integration/test_v2_metadata_routes.py tests/integration/test_v2_country_routes.py tests/integration/test_v2_report_reads.py tests/integration/test_v2_report_creation.py tests/integration/test_v2_budget_window_report_reads.py tests/integration/test_v2_budget_window_report_creation.py tests/integration/test_v2_report_runs.py - name: Test v2 policy persistence and immediate v1 mirroring run: uv run --frozen --extra dev coverage run -a --branch -m pytest -q tests/integration/test_v2_policy_persistence.py tests/integration/test_v1_policy_dual_write.py tests/integration/test_mysql_policy_dual_write.py tests/integration/test_v2_user_policy_mirroring.py tests/integration/test_v1_user_policy_dual_write.py - name: Test signed-in identity provisioning, user associations, saved detailed-save records and user types diff --git a/changelog.d/v2-report-reruns.added.md b/changelog.d/v2-report-reruns.added.md new file mode 100644 index 000000000..af7519366 --- /dev/null +++ b/changelog.d/v2-report-reruns.added.md @@ -0,0 +1 @@ +`/v2/reports` runs an economy report or a budget window again and lists its runs. `POST /v2/reports/{report_id}/runs?country_id=` (body `{"force": false}`) checks the report's country and the policies its simulations name, as a new report is checked, then has the simulation service create and start the report's new run on its default worker's versions, reusing each side's successful simulation run on those versions unless `force` is true; it answers 201 with the report and its new run, 409 naming the run when one is pending or running, and keeps every earlier run. `GET /v2/reports/{report_id}/runs?country_id=` lists the report's runs, newest first (a budget window's budget-window runs, each with its years), each with its `country_package_version` and `policyengine_version`, beside the versions the Public API runs now, so a caller can mark a report whose newest run is on older versions as out of date. Household reports are not run again or listed yet. diff --git a/docs/engineering/migration-contracts.md b/docs/engineering/migration-contracts.md index d9f04b4bd..6a7e5fe3a 100644 --- a/docs/engineering/migration-contracts.md +++ b/docs/engineering/migration-contracts.md @@ -8,7 +8,7 @@ Generated from `policyengine_api/migration_registry.py` and `tests/contract/regi | --- | ---: | | route group count | 9 | | workflow count | 18 | -| request count | 97 | +| request count | 99 | | db entity count | 6 | | sim flow count | 3 | @@ -207,6 +207,8 @@ Generated from `policyengine_api/migration_registry.py` and `tests/contract/regi | `POST` | `/v2/reports?country_id=us` | 201 | `report` | none | `status`, `message`, `result.item.id`, `result.item.country_id`, `result.item.kind`, `result.item.status`, `result.item.run`, `result.item.current_versions`, `result.item.result` | | `GET` | `/v2/reports?country_id=us&report_id={report_id}` | 200 | `report` | none | `status`, `message`, `result.items` | | `GET` | `/v2/reports/{report_id}?country_id=us` | 200 | `report` | none | `status`, `message`, `result.item.id`, `result.item.status`, `result.item.run`, `result.item.current_versions`, `result.item.result` | +| `POST` | `/v2/reports/{report_id}/runs?country_id=us` | 201 | `report` | none | `status`, `message`, `result.item.id`, `result.item.status`, `result.item.run`, `result.item.current_versions` | +| `GET` | `/v2/reports/{report_id}/runs?country_id=us` | 200 | `report` | none | `status`, `message`, `result.report_id`, `result.kind`, `result.items`, `result.current_versions` | | `POST` | `/v2/simulations?country_id=us` | 201 | `simulation` | none | `status`, `message`, `result.item.id`, `result.item.kind`, `result.item.status`, `result.item.run` | | `GET` | `/v2/simulations/{simulation_id}?country_id=us` | 200 | `simulation` | none | `status`, `message`, `result.item.id`, `result.item.status`, `result.item.run`, `result.item.current_versions` | diff --git a/docs/generated/migration_contracts.json b/docs/generated/migration_contracts.json index 6eca98bb9..48869197b 100644 --- a/docs/generated/migration_contracts.json +++ b/docs/generated/migration_contracts.json @@ -1,7 +1,7 @@ { "metadata": { "db_entity_count": 6, - "request_count": 97, + "request_count": 99, "route_group_count": 9, "sim_flow_count": 3, "workflow_count": 18 @@ -1423,6 +1423,36 @@ "result.item.result" ] }, + { + "expected_status": 201, + "headers": [], + "method": "POST", + "path": "/v2/reports/{report_id}/runs?country_id=us", + "route_group": "report", + "stable_response_fields": [ + "status", + "message", + "result.item.id", + "result.item.status", + "result.item.run", + "result.item.current_versions" + ] + }, + { + "expected_status": 200, + "headers": [], + "method": "GET", + "path": "/v2/reports/{report_id}/runs?country_id=us", + "route_group": "report", + "stable_response_fields": [ + "status", + "message", + "result.report_id", + "result.kind", + "result.items", + "result.current_versions" + ] + }, { "expected_status": 201, "headers": [], diff --git a/policyengine_api/fastapi_routes/resource_service_protocols.py b/policyengine_api/fastapi_routes/resource_service_protocols.py index 32542319a..d771ea3b5 100644 --- a/policyengine_api/fastapi_routes/resource_service_protocols.py +++ b/policyengine_api/fastapi_routes/resource_service_protocols.py @@ -37,6 +37,7 @@ ) from policyengine_api.services.v2.stored_reports.types import ( StoredReportRead, + StoredReportRunsRead, StoredReportSummaryRead, ) from policyengine_api.services.v2.stored_simulations.types import SimulationRead @@ -222,6 +223,35 @@ def create_report(self, report_input: ReportCreationInput) -> StoredReportRead: """ ... + def rerun_report( + self, *, country_id: str, report_id: UUID, force: bool + ) -> StoredReportRead: + """Run the country's economy report or budget window again; return it. + + It is returned with its new run as its newest. Raise + ``StoredReportNotFoundError`` when the country has no such report, + ``ReportKindUnsupportedError`` for a kind whose runs the run routes do + not start yet (household reports), the + checks' errors as ``create_report`` does for the policies of the + report's simulations, the simulation service's errors, among them + ``SimulationServiceConflictError`` when it will not run the report + again now, and ``RerunReportUnreadableError`` when the report cannot + be read back once its new run was started. + """ + ... + + def list_report_runs( + self, *, country_id: str, report_id: UUID + ) -> StoredReportRunsRead: + """Return the runs of the country's economy report or budget window. + + They are newest first; a budget window's are its budget-window runs, + each with its years. Raise ``StoredReportNotFoundError`` when the + country has no such report, and ``ReportKindUnsupportedError`` for a + kind whose runs the run routes do not list yet (household reports). + """ + ... + class V2StoredSimulationResourceService(Protocol): """Route-independent simulation operations for one request.""" diff --git a/policyengine_api/fastapi_routes/v2/reports/routes.py b/policyengine_api/fastapi_routes/v2/reports/routes.py index 4a1349005..8b20e9e58 100644 --- a/policyengine_api/fastapi_routes/v2/reports/routes.py +++ b/policyengine_api/fastapi_routes/v2/reports/routes.py @@ -6,7 +6,9 @@ reading never calls the simulation service and never starts a run. A report is an economy report or a budget window, which its ``kind`` names. Creating a report checks it, has the simulation service create and start it, and -reads it back. +reads it back. Running a report again checks it as a new report is checked, +has the simulation service start its new run, and reads the report back; +listing its runs reads them, newest first. """ from __future__ import annotations @@ -41,6 +43,15 @@ report_item, report_summary_item, ) +from policyengine_api.fastapi_routes.v2.reports.run_models import ( + RERUN_RESPONSES, + RUN_LIST_RESPONSES, + ReportRerunConflictResponse, + ReportRerunRequest, + ReportRunListResponse, + RerunReportUnreadableResponse, + report_run_list_result, +) from policyengine_api.fastapi_routes.v2.simulation_resources_support import ( SUBMISSION_ERROR_MAPPINGS, SUBMISSION_REFUSALS, @@ -55,8 +66,10 @@ ReportCollectionQuery, ReportCreateQuery, ReportDetailQuery, + ReportRunsQuery, ) from policyengine_api.services.v2.simulation_service.types import ( + SimulationServiceConflictError, SimulationServiceUnavailableError, ) from policyengine_api.services.v2.stored_reports.budget_window_results.types import ( @@ -69,6 +82,7 @@ CreatedReportUnreadableError, ReportKindUnsupportedError, ReportStorageError, + RerunReportUnreadableError, StoredReportNotFoundError, ) @@ -93,6 +107,9 @@ ReportDetailQueryDependency = Annotated[ ReportDetailQuery, Depends(query_dependency(ReportDetailQuery)) ] +ReportRunsQueryDependency = Annotated[ + ReportRunsQuery, Depends(query_dependency(ReportRunsQuery)) +] # Refusals a caller can act on carry their own message and are not logged. REPORT_REFUSALS: tuple[type[Exception], ...] = ( @@ -176,6 +193,117 @@ def _create_report( return ReportDetailResponse(result=ReportDetailResult(item=report_item(report))) +def _rerun_report( + service: V2StoredReportResourceService, + *, + country_id: str, + report_id: UUID, + force: bool, +) -> ReportDetailResponse | JSONResponse: + """Run the report again and answer it with its new run. + + The simulation service's refusal to run it again now answers 409 with + the run it names, and a report that cannot be read back once its new run + started answers 502 with its ``report_id``. Every other failure is + raised, for the route's error mappings to answer. + """ + try: + report = service.rerun_report( + country_id=country_id, report_id=report_id, force=force + ) + except SimulationServiceConflictError as error: + conflict = ReportRerunConflictResponse(message=str(error), run_id=error.run_id) + return JSONResponse(status_code=409, content=conflict.model_dump(mode="json")) + except RerunReportUnreadableError as error: + return created_record_unreadable_answer( + "v2_rerun_report_unreadable", + error, + RerunReportUnreadableResponse( + message=( + "The simulation service started a new run of report " + f"{error.report_id}, but the report could not be read back. " + f"Read it with GET /v2/reports/{error.report_id}?country_id=" + f"{error.country_id} rather than running it again." + ), + report_id=error.report_id, + ), + ) + return ReportDetailResponse(result=ReportDetailResult(item=report_item(report))) + + +def _add_report_run_routes( + router: APIRouter, service_factory: Callable[[], V2StoredReportResourceService] +) -> None: + """Add the routes that run a report again and list its runs.""" + + @router.post( + "/reports/{report_id}/runs", + response_model=ReportDetailResponse, + status_code=201, + responses=RERUN_RESPONSES, + summary="Run an economy report or budget window again", + description=( + "Checks the report as a new report is checked (the country has " + "economy reports, and each policy its simulations name is stored and " + "sets only parameters the running catalog version has), then has the " + "simulation service create the report's new run, with the manual " + "trigger, on its default worker's PolicyEngine.py version, and start " + "it. Every earlier run is kept. Without force, each side reuses a " + "run of its own simulation that succeeded on those versions; with " + "force, both simulations are calculated again. A budget window's new " + "run is a budget-window run with a report run for each year, and " + "with force each year calculates its simulations again. Answers the " + "report as stored, whose newest run is then the new one, since no " + "other run " + "starts while it is pending or running: a report with a run pending " + "or running answers 409 naming that run. Read the report again to " + "follow the run." + ), + ) + def rerun_report( + report_id: UUID, + body: ReportRerunRequest, + query: ReportRunsQueryDependency, + _size: Annotated[None, Depends(enforce_report_request_size)], + ) -> ReportDetailResponse | JSONResponse: + def rerun() -> ReportDetailResponse | JSONResponse: + return _rerun_report( + service_factory(), + country_id=query.country_id, + report_id=report_id, + force=body.force, + ) + + return _report_operation(rerun) + + @router.get( + "/reports/{report_id}/runs", + response_model=ReportRunListResponse, + responses=RUN_LIST_RESPONSES, + summary="List an economy report's or budget window's runs", + description=( + "Lists every run of the report, newest first, whatever its status " + "and versions, each with the country package and PolicyEngine.py " + "versions it ran on, beside the versions the Public API runs now. A " + "run on other versions than those is out of date. An economy " + "report's runs are its report runs (kind economy); a budget " + "window's are its budget-window runs, each with its years (kind " + "budget_window). Reading starts no run." + ), + ) + def list_report_runs( + report_id: UUID, + query: ReportRunsQueryDependency, + ) -> ReportRunListResponse | JSONResponse: + def read() -> ReportRunListResponse: + runs = service_factory().list_report_runs( + country_id=query.country_id, report_id=report_id + ) + return ReportRunListResponse(result=report_run_list_result(runs)) + + return _report_operation(read) + + def build_v2_report_router(dependencies: NativeRouteDependencies) -> APIRouter: """Build the report routes without opening a connection.""" router = APIRouter(prefix="/v2", responses=REPORT_ERROR_RESPONSES) @@ -274,4 +402,5 @@ def read() -> ReportDetailResponse: return _report_operation(read) + _add_report_run_routes(router, service_factory) return router diff --git a/policyengine_api/fastapi_routes/v2/reports/run_models.py b/policyengine_api/fastapi_routes/v2/reports/run_models.py new file mode 100644 index 000000000..a45b8aecd --- /dev/null +++ b/policyengine_api/fastapi_routes/v2/reports/run_models.py @@ -0,0 +1,198 @@ +"""Strict HTTP models of a v2 report's runs: running it again and listing them.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Annotated, Any, Literal +from uuid import UUID + +from pydantic import Field, StrictBool + +from policyengine_api.fastapi_routes.v2.errors_support import ( + V2ErrorResponse, + V2MissingParametersErrorResponse, +) +from policyengine_api.fastapi_routes.v2.reports.response_models import ( + BudgetWindowRunItem, + PackageVersionsItem, + ReportRunItem, + ReportSuccessResponse, + StrictReportAPIModel, +) +from policyengine_api.fastapi_routes.v2.simulation_resources_support import ( + error_response, +) +from policyengine_api.services.v2.simulation_service.types import ( + StrictSubmissionModel, +) + +if TYPE_CHECKING: + from policyengine_api.services.v2.stored_reports.types import ( + BudgetWindowRunsRead, + ReportRunsRead, + StoredReportRunsRead, + ) + + +class ReportRerunRequest(StrictSubmissionModel): + """Run a stored report again on the simulation service's current versions. + + Without ``force`` (``{}``), each side reuses a run of its own simulation + that succeeded on those versions; with it, both simulations are + calculated again. The new run is kept beside every earlier one. + """ + + force: StrictBool = False + + +class ReportRunListResult(StrictReportAPIModel): + """An economy report's runs in ``items``, newest first, whatever their status. + + Each run names the country package and PolicyEngine.py versions it ran + on; ``current_versions`` are those the Public API runs now, so a run on + other versions is out of date. + """ + + report_id: UUID + kind: Literal["economy"] + items: list[ReportRunItem] + current_versions: PackageVersionsItem + + @classmethod + def from_read(cls, read: ReportRunsRead) -> ReportRunListResult: + """Build the list from the service's read of the report's runs.""" + return cls( + report_id=read.report_id, + kind=read.kind, + items=[ReportRunItem.model_validate(run) for run in read.runs], + current_versions=PackageVersionsItem.model_validate(read.current_versions), + ) + + +class BudgetWindowRunListResult(StrictReportAPIModel): + """A budget window's runs in ``items``, newest first, each with its years. + + Each budget-window run names the country package and PolicyEngine.py + versions its years ran on; ``current_versions`` are those the Public API + runs now, so a run on other versions is out of date. + """ + + report_id: UUID + kind: Literal["budget_window"] + items: list[BudgetWindowRunItem] + current_versions: PackageVersionsItem + + @classmethod + def from_read(cls, read: BudgetWindowRunsRead) -> BudgetWindowRunListResult: + """Build the list from the service's read of the window's runs.""" + return cls( + report_id=read.report_id, + kind=read.kind, + items=[BudgetWindowRunItem.model_validate(run) for run in read.runs], + current_versions=PackageVersionsItem.model_validate(read.current_versions), + ) + + +# A report's runs, as the list of its kind; ``kind`` says which. +StoredReportRunListResult = Annotated[ + ReportRunListResult | BudgetWindowRunListResult, Field(discriminator="kind") +] + + +def report_run_list_result( + read: StoredReportRunsRead, +) -> ReportRunListResult | BudgetWindowRunListResult: + """Build the run list of the report's kind from the service's read.""" + if read.kind == "budget_window": + return BudgetWindowRunListResult.from_read(read) + return ReportRunListResult.from_read(read) + + +class ReportRunListResponse(ReportSuccessResponse[StoredReportRunListResult]): + """Success response holding a report's runs; ``result.kind`` says which list.""" + + +class ReportRerunConflictResponse(V2ErrorResponse): + """The 409 answer when the simulation service will not run the report again now. + + The message gives the service's code and reason: ``run_in_progress``, + when the report has a run pending or running, which ``run_id`` names, or + ``rerun_not_supported`` for a report it does not run again, such as one + whose simulation, region or dataset is gone. ``run_id`` is null when the + service named no run. + """ + + run_id: UUID | None + + +class RerunReportUnreadableResponse(V2ErrorResponse): + """The 502 answer when the report cannot be read back after its new run started. + + The simulation service started the run, so the caller reads the report + with ``GET /v2/reports/{report_id}`` rather than running it again. + """ + + report_id: UUID + + +NOT_FOUND = "No report has this ID in this country." +OTHER_KIND = ( + "or the report is a household report, whose runs these routes do not start " + "or list yet" +) +RUN_LIST_RESPONSES: dict[int | str, dict[str, Any]] = { + 404: error_response(NOT_FOUND), + 422: error_response(f"The request does not match the schema, {OTHER_KIND}."), +} +RERUN_RESPONSES: dict[int | str, dict[str, Any]] = { + 404: error_response( + f"{NOT_FOUND} Or a policy a simulation of the report names is no longer " + "stored in the country." + ), + 409: { + "model": ReportRerunConflictResponse, + "description": ( + "The simulation service will not run the report again now, and " + "started nothing: the report has a run pending or running " + "(run_in_progress), which run_id names, or it is one the service " + "does not run again (rerun_not_supported)." + ), + }, + 413: error_response("The request body exceeds 16 KiB."), + 422: { + "model": V2MissingParametersErrorResponse | V2ErrorResponse, + "description": ( + f"The request does not match the schema, {OTHER_KIND}; the country has " + "no economy reports; a policy of the report's simulations sets " + "parameters the running catalog version does not have (code " + "missing_parameters, which names them); or the simulation service " + "cannot run the report as stored: its 422, or its 400 for a region, " + "dataset or dynamic the report names that the current catalog version " + "does not have, with its code and reason." + ), + }, + 502: { + "model": V2ErrorResponse | RerunReportUnreadableResponse, + "description": ( + "The simulation service may have started a new run, but its answer " + "does not say so: its answer was lost, a proxy answered in its place, " + "it failed (500), or it answered in a shape or status the Public API " + "does not expect. The message asks the caller to read the report to " + "see whether it has a new run before running it again; while that " + "run is pending or running, running it again answers 409. Or it " + "started the run, but the report cannot be read back: the answer " + "gives its report_id." + ), + }, + 503: error_response( + "The simulation service started nothing that runs: it cannot be reached " + "or is not configured; it answered a readable 503, and the message gives " + "its status and code, and names the new run it failed when Modal refused " + "the run's call; it answered a 400 about its own state rather than a row " + "the report names (such as a catalog version missing while a release " + "deploys), and the message gives its code and reason; it answered 400, " + "409 or 422 in a shape the Public API cannot read; or it answered 401, " + "403, 404, 405, 413 or 429, which it sends before writing anything. Or " + "the v2 database, or the catalog the policies are checked against, " + "cannot be read." + ), +} diff --git a/policyengine_api/libs/simulation_report_reruns.py b/policyengine_api/libs/simulation_report_reruns.py new file mode 100644 index 000000000..75d4204fb --- /dev/null +++ b/policyengine_api/libs/simulation_report_reruns.py @@ -0,0 +1,291 @@ +"""Run a stored report again through the simulation service. + +``start_report_rerun`` is what ``SimulationServiceReportClient.rerun_report`` +does: it posts ``{"force": …}`` to the Cloud Run simulation service's +``POST /reports/{report_id}/runs``, through the same HTTP client as the +service's other v2 routes (``policyengine_api.libs.simulation_reports``). +The service creates the stored report's new run on its default worker's +versions and starts it, keeping every earlier run. Without ``force``, each +side reuses a run of its own simulation that succeeded on those versions. + +The Public API reads the report in its country before asking, so it never +asks about a report it does not have. The simulation service answers: + +- 201 or 202 naming the report: the new run exists. One whose body cannot + be read, or that names another report, is + SimulationServiceAnswerUnreadableError; +- 409 with ``code`` and ``detail`` when it will not run the report again + now: ``run_in_progress``, naming the run pending or running in + ``run_id``, or ``rerun_not_supported`` for a report it does not run again. + SimulationServiceConflictError, with the code, reason and run. It answers + 409 before writing anything, so one that cannot be read is + SimulationServiceUnavailableError; +- 422 with its reasons when the stored policies or dynamics cannot be + calculated on the current catalog version, and 400 with a code in + ``CALLER_ROW_REFUSAL_CODES`` when that catalog version lacks the region, + dataset or a dynamic the report names: SimulationServiceInvalidRequestError, + with the code and reasons. Any other 400 is the service's own state, such + as a catalog version missing while a release deploys: + SimulationServiceUnavailableError, with the code and reason; +- 503 with its code when it cannot answer now, before writing anything, and + ``modal_call_refused`` naming the report and its new run (``run_id``) when + Modal refused the run's call and the service failed the run: either is + SimulationServiceUnavailableError, whose message gives the status and + code, and names the failed run. Nothing is running, so running the report + again later is safe; +- 401, 403, 404, 405, 413 or 429, sent before its route writes anything: + SimulationServiceUnavailableError. A 404 is the service's state too, a + route it does not have yet or a database it reads differently, since the + Public API has just read the report. + +Any other answer may follow a new run the service created, so it is never +SimulationServiceUnavailableError, whose 503 says nothing was started: an +answer lost after the request may have been sent and a proxy's 502 or 504 +(SimulationServiceAnswerUnreadableError), a 500 +(SimulationServiceInternalError), a 503 that cannot be read or names another +record, and any other status (SimulationServiceAnswerUnreadableError). Each +asks the caller to read the report to see whether it has a new run before +running it again. Running it again while that run is pending or running +answers 409, so a second request starts no second calculation. + +As for a submission, a 400's or 422's or 409's message gives the service's +code and reason, but never its status, and any other message gives the +status and code, never the reason, which may be an exception's text. +""" + +from __future__ import annotations + +from http import HTTPStatus +from typing import TYPE_CHECKING, Any, NoReturn +from uuid import UUID + +from pydantic import ValidationError + +from policyengine_api.libs.simulation_service_answers import ( + CALLER_ROW_REFUSAL_CODES, + CREATED_STATUSES, + PROXY_STAND_IN_STATUSES, + REFUSAL_STATUSES, + SENT_BEFORE_ROUTE_STATUSES, + AnswerModel, + Refusal, + SimulationAnswer, + answer_lost, + fixed_identifier_code, + parsed_body, + refusal_reason, + send_to_simulation_service, + status_and_code, +) +from policyengine_api.services.v2.simulation_service.types import ( + SimulationServiceAnswerUnreadableError, + SimulationServiceConflictError, + SimulationServiceInternalError, + SimulationServiceInvalidRequestError, + SimulationServiceUnavailableError, +) + +if TYPE_CHECKING: + from collections.abc import Callable + + import httpx + + from policyengine_api.services.v2.simulation_service.types import ( + ReportRerunInput, + ) + +READ_BEFORE_RUNNING_AGAIN = ( + "Read the report to see whether it has a new run before running it again." +) +UNREADABLE_START = ( + "The simulation service answered that it started a new run of the report, " + f"but its answer could not be read. {READ_BEFORE_RUNNING_AGAIN}" +) + + +class _RerunStarted(AnswerModel): + report_id: UUID + + +class _RerunRefusal(Refusal): + """A refusal's or failure's body: a reason, or each refused field. + + It may name its ``code``; a 409 for a run in progress names that run in + ``run_id``, and a 503 after Modal refused the new run's call names the + report and the run it failed. + """ + + code: str | None = None + report_id: UUID | None = None + simulation_id: UUID | None = None + run_id: UUID | None = None + + +def report_runs_path(report_id: UUID) -> str: + """Return the simulation service's path for the report's runs.""" + return f"/reports/{report_id}/runs" + + +def report_rerun_request(rerun: ReportRerunInput) -> dict[str, Any]: + """Return the simulation service's request body for the re-run.""" + return {"force": rerun.force} + + +def post_report_rerun( + http_client: httpx.Client, report_id: UUID, body: dict[str, Any] +) -> SimulationAnswer: + """Send one re-run to the simulation service and return its answer. + + Raise httpx.HTTPError when no answer arrives, and GatewayAuthError when + the gateway token cannot be fetched. ``http_client`` has the service's + base URL. With the generated simulation client, this body becomes the + route's operation (``rerun_report.sync_detailed`` with ``report_id`` and + ``ReportRerunRequest.from_dict(body)``), as + ``policyengine_api.libs.simulation_reports.post_report`` describes. + """ + response = http_client.post(report_runs_path(report_id), json=body) + return SimulationAnswer(status_code=response.status_code, content=response.content) + + +def _refusal(answer: SimulationAnswer) -> _RerunRefusal | None: + try: + return _RerunRefusal.model_validate_json(answer.content) + except ValidationError: + return None + + +def _status_and_code(answer: SimulationAnswer) -> str: + """Return the answer's status and fixed-identifier code, never its reason.""" + return status_and_code( + answer.status_code, fixed_identifier_code(parsed_body(answer)) + ) + + +def _code_and_reason(refusal: _RerunRefusal) -> str: + """Return `` (code): reason``, or ``: reason`` when the service gave no code.""" + code = "" if refusal.code is None else f" ({refusal.code})" + return f"{code}: {refusal_reason(refusal)}" + + +def _possibly_started(answer: SimulationAnswer) -> str: + """Say the service may have started a new run, but failed, and what to do.""" + return ( + "The simulation service may have started a new run of the report, but " + f"it failed ({_status_and_code(answer)}). {READ_BEFORE_RUNNING_AGAIN}" + ) + + +def _raise_for_refused_request( + answer: SimulationAnswer, refusal: _RerunRefusal +) -> NoReturn: + """Raise the error a readable 400 or 422 means.""" + code_and_reason = _code_and_reason(refusal) + if ( + answer.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + or refusal.code in CALLER_ROW_REFUSAL_CODES + ): + msg = f"The simulation service cannot run the report again{code_and_reason}" + raise SimulationServiceInvalidRequestError(msg) + msg = f"The simulation service refused to run the report again{code_and_reason}" + raise SimulationServiceUnavailableError(msg) + + +def _raise_for_conflict(answer: SimulationAnswer) -> NoReturn: + """Raise the error a 409 means: a conflict, or unavailable when unreadable.""" + refusal = _refusal(answer) + if refusal is None or refusal.code is None: + msg = f"The simulation service answered {answer.status_code} unreadably." + raise SimulationServiceUnavailableError(msg) + msg = ( + "The simulation service will not run the report again now" + f"{_code_and_reason(refusal)}" + ) + raise SimulationServiceConflictError(msg, run_id=refusal.run_id) + + +def _raise_for_unavailable(answer: SimulationAnswer, report_id: UUID) -> NoReturn: + """Raise the error a 503 means. + + One naming this report says the service started its new run and failed + it; one naming no record says the service wrote nothing. One that cannot + be read, such as Cloud Run's own, or naming another record, may follow a + new run the service created. + """ + refusal = _refusal(answer) + if ( + refusal is None + or refusal.simulation_id is not None + or refusal.report_id not in {None, report_id} + ): + raise SimulationServiceAnswerUnreadableError(_possibly_started(answer)) + unavailable = f"is unavailable ({_status_and_code(answer)})" + if refusal.report_id is None: + msg = f"The simulation service {unavailable}." + raise SimulationServiceUnavailableError(msg) + new_run = "a new run" if refusal.run_id is None else f"run {refusal.run_id}" + msg = ( + f"The simulation service started {new_run} of report {report_id}, then " + f"failed it, and {unavailable}. Run the report again later." + ) + raise SimulationServiceUnavailableError(msg) + + +def _raise_for_refusal(answer: SimulationAnswer, report_id: UUID) -> NoReturn: + """Raise the error any answer but a started re-run means.""" + status_code = answer.status_code + if status_code in REFUSAL_STATUSES: + refusal = _refusal(answer) + if refusal is None: + msg = f"The simulation service answered {status_code} unreadably." + raise SimulationServiceUnavailableError(msg) + _raise_for_refused_request(answer, refusal) + if status_code == HTTPStatus.CONFLICT: + _raise_for_conflict(answer) + if status_code in PROXY_STAND_IN_STATUSES: + msg = answer_lost( + f"a proxy answered {status_code} in its place", READ_BEFORE_RUNNING_AGAIN + ) + raise SimulationServiceAnswerUnreadableError(msg) + if status_code == HTTPStatus.SERVICE_UNAVAILABLE: + _raise_for_unavailable(answer, report_id) + if status_code in SENT_BEFORE_ROUTE_STATUSES: + msg = f"The simulation service is unavailable ({_status_and_code(answer)})." + raise SimulationServiceUnavailableError(msg) + if status_code == HTTPStatus.INTERNAL_SERVER_ERROR: + raise SimulationServiceInternalError(_possibly_started(answer)) + raise SimulationServiceAnswerUnreadableError(_possibly_started(answer)) + + +def _require_started(answer: SimulationAnswer, report_id: UUID) -> None: + """Raise unless the started answer can be read and names the report.""" + try: + started = _RerunStarted.model_validate_json(answer.content) + except ValidationError as error: + raise SimulationServiceAnswerUnreadableError(UNREADABLE_START) from error + if started.report_id != report_id: + raise SimulationServiceAnswerUnreadableError(UNREADABLE_START) + + +def start_report_rerun( + http_client: httpx.Client, + rerun: ReportRerunInput, + *, + send: Callable[[httpx.Client, UUID, dict[str, Any]], SimulationAnswer], +) -> None: + """Have the simulation service create and start the report's new run. + + Send through ``send`` on ``http_client``, and raise the errors the module + docstring lists for the service's answers. A connection never made, or a + gateway token that cannot be fetched, is + SimulationServiceUnavailableError. + """ + answer = send_to_simulation_service( + lambda client, body: send(client, rerun.report_id, body), + http_client, + report_rerun_request(rerun), + submit_again_sentence=READ_BEFORE_RUNNING_AGAIN, + ) + if answer.status_code in CREATED_STATUSES: + _require_started(answer, rerun.report_id) + return + _raise_for_refusal(answer, rerun.report_id) diff --git a/policyengine_api/libs/simulation_reports.py b/policyengine_api/libs/simulation_reports.py index 7c936e3f3..2c31974a5 100644 --- a/policyengine_api/libs/simulation_reports.py +++ b/policyengine_api/libs/simulation_reports.py @@ -86,11 +86,16 @@ come from ``policyengine_api.libs.simulation_service_answers``, which the household calculation client shares. -Wiring the generated simulation client: ``post_simulation`` and -``post_report`` are the only code that sends a request, and each returns the -status and body, which the rest of this module reads. Once a -``policyengine-api-simulation-client`` release has the two routes, depend on -that release and make each function call the generated operation, as its +``rerun_report`` runs a stored report again through the service's +``POST /reports/{report_id}/runs``; how it reads that route's answers is in +``policyengine_api.libs.simulation_report_reruns``. + +Wiring the generated simulation client: ``post_simulation``, +``post_report`` and ``post_report_rerun`` are the only code that sends a +request, and each returns the status and body, which the rest of these +modules read. Once a +``policyengine-api-simulation-client`` release has the three routes, depend +on that release and make each function call the generated operation, as its docstring says. """ @@ -103,8 +108,16 @@ from pydantic import ValidationError +from policyengine_api.libs.simulation_report_reruns import ( + post_report_rerun, + start_report_rerun, +) from policyengine_api.libs.simulation_service_answers import ( + CALLER_ROW_REFUSAL_CODES, + CREATED_STATUSES, PROXY_STAND_IN_STATUSES, + REFUSAL_STATUSES, + SENT_BEFORE_ROUTE_STATUSES, AnswerModel, Refusal, SimulationAnswer, @@ -139,36 +152,11 @@ EconomySimulationCreationInput, PolicyChoice, ReportCreationInput, + ReportRerunInput, ) SIMULATIONS_PATH = "/simulations" REPORTS_PATH = "/reports" -CREATED_STATUSES = frozenset({HTTPStatus.CREATED, HTTPStatus.ACCEPTED}) -REFUSAL_STATUSES = frozenset({HTTPStatus.BAD_REQUEST, HTTPStatus.UNPROCESSABLE_ENTITY}) -# The codes of the simulation service's 400 that name a row the caller chose -# and the Public API does not check: the region, the dataset, or a dynamic -# the catalog version its default worker runs does not have. Each is the -# caller's error and answers 422. Any other 400 is a state of the service, -# such as a catalog version missing while a release deploys, or a row the -# Public API already checked, and answers 503. -CALLER_ROW_REFUSAL_CODES = frozenset( - {"unknown_region", "unknown_dataset", "unknown_dynamic"} -) -# Statuses the gateway, Cloud Run or the service's framework answers before -# the service's route runs, so the service has created nothing: a call not -# authenticated or not allowed, a route or method the service does not have, -# a body over the size limit, and too many requests. The route may have -# written rows before answering any other status the client does not expect. -SENT_BEFORE_ROUTE_STATUSES = frozenset( - { - HTTPStatus.UNAUTHORIZED, - HTTPStatus.FORBIDDEN, - HTTPStatus.NOT_FOUND, - HTTPStatus.METHOD_NOT_ALLOWED, - HTTPStatus.REQUEST_ENTITY_TOO_LARGE, - HTTPStatus.TOO_MANY_REQUESTS, - } -) # What a submission creates. @@ -428,7 +416,7 @@ def _raise_for_refusal(answer: SimulationAnswer, record: RecordName) -> NoReturn class SimulationServiceReportClient: - """Create simulations and reports through the simulation service.""" + """Create simulations and reports, and run reports again, through the service.""" def __init__( self, @@ -440,14 +428,18 @@ def __init__( send_report: Callable[ [httpx.Client, dict[str, Any]], SimulationAnswer ] = post_report, + send_report_rerun: Callable[ + [httpx.Client, UUID, dict[str, Any]], SimulationAnswer + ] = post_report_rerun, ) -> None: - """Send through ``send_simulation`` and ``send_report`` on ``http_client``. + """Send through ``send_simulation``, ``send_report`` and ``send_report_rerun``. - ``http_client`` has the service's base URL. + Each sends on ``http_client``, which has the service's base URL. """ self._http_client = http_client self._send_simulation = send_simulation self._send_report = send_report + self._send_report_rerun = send_report_rerun @classmethod def from_environment( @@ -564,6 +556,14 @@ def submit_budget_window( budget_window_run_id=created.budget_window_run_id, ) + def rerun_report(self, rerun: ReportRerunInput) -> None: + """Have the service create and start the stored report's new run. + + Raise the errors ``policyengine_api.libs.simulation_report_reruns`` + lists for the service's answers. + """ + start_report_rerun(self._http_client, rerun, send=self._send_report_rerun) + @lru_cache(maxsize=1) def process_simulation_service_client() -> SimulationServiceReportClient: diff --git a/policyengine_api/libs/simulation_service_answers.py b/policyengine_api/libs/simulation_service_answers.py index b54667121..885b03a6e 100644 --- a/policyengine_api/libs/simulation_service_answers.py +++ b/policyengine_api/libs/simulation_service_answers.py @@ -22,7 +22,12 @@ message gives in place of the service's reason (``status_and_code``), with the code only when it is a fixed identifier (``FIXED_IDENTIFIER``); - the statuses a proxy in front of the service answers in its place - (``PROXY_STAND_IN_STATUSES``); + (``PROXY_STAND_IN_STATUSES``), and, for the report and simulation client + and the report re-run client, the statuses of a created record + (``CREATED_STATUSES``), of a refusal (``REFUSAL_STATUSES``) and of an + answer sent before the service's route writes anything + (``SENT_BEFORE_ROUTE_STATUSES``), and the codes of a refusal of a row the + Public API does not check (``CALLER_ROW_REFUSAL_CODES``); - the messages for an answer lost after the request may have arrived (``answer_lost``) and for a failure after the service may have created what was submitted (``created_but_failed``), each ending with @@ -64,6 +69,36 @@ PROXY_STAND_IN_STATUSES = frozenset( {HTTPStatus.BAD_GATEWAY, HTTPStatus.GATEWAY_TIMEOUT} ) +# The statuses of a created record and of a refusal of the request. +CREATED_STATUSES = frozenset({HTTPStatus.CREATED, HTTPStatus.ACCEPTED}) +REFUSAL_STATUSES = frozenset({HTTPStatus.BAD_REQUEST, HTTPStatus.UNPROCESSABLE_ENTITY}) +# The codes of the simulation service's report and simulation routes' 400 +# that name a row the caller chose, or a stored report names, and the Public +# API does not check: the region, the dataset, or a dynamic the catalog +# version its default worker runs does not have. Each is the caller's error +# and answers 422. Any other 400 is a state of the service, such as a catalog +# version missing while a release deploys, or a row the Public API already +# checked, and answers 503. +CALLER_ROW_REFUSAL_CODES = frozenset( + {"unknown_region", "unknown_dataset", "unknown_dynamic"} +) +# Statuses the gateway, Cloud Run or the service's framework answers before +# the service's report and simulation routes run, so the service has created +# nothing: a call not authenticated or not allowed, a route or method the +# service does not have, a body over the size limit, and too many requests. +# A re-run route's own 404, for a record it does not find, also comes before +# any write. The route may have written rows before answering any other +# status the client does not expect. +SENT_BEFORE_ROUTE_STATUSES = frozenset( + { + HTTPStatus.UNAUTHORIZED, + HTTPStatus.FORBIDDEN, + HTTPStatus.NOT_FOUND, + HTTPStatus.METHOD_NOT_ALLOWED, + HTTPStatus.REQUEST_ENTITY_TOO_LARGE, + HTTPStatus.TOO_MANY_REQUESTS, + } +) # Transport errors raised before the request could reach the service. After # any other, the service may have received the submission and created what # it asks for. diff --git a/policyengine_api/query_parameters.py b/policyengine_api/query_parameters.py index 82456d4c7..568e148c8 100644 --- a/policyengine_api/query_parameters.py +++ b/policyengine_api/query_parameters.py @@ -433,6 +433,10 @@ class ReportDetailQuery(CountryQuery): """Query contract for reading one report in a country.""" +class ReportRunsQuery(CountryQuery): + """Query contract for a report's runs in a country: listing or starting one.""" + + class ReportCollectionQuery(CountryQuery): """Query contract for listing chosen reports of a country. diff --git a/policyengine_api/services/v2/simulation_service/services.py b/policyengine_api/services/v2/simulation_service/services.py index 13900c3a6..21f8f9942 100644 --- a/policyengine_api/services/v2/simulation_service/services.py +++ b/policyengine_api/services/v2/simulation_service/services.py @@ -8,10 +8,12 @@ country does not), that each policy the submission names is stored in the country, through the same service as ``/v2/policies`` (current law names none), and that none of those policies sets a parameter the running catalog -version lacks (``require_runnable_policies``, which re-runs should call too). +version lacks (``require_runnable_policies``, which re-runs call too). Only then is the simulation service asked. The dynamics a submission names, its region and its dataset are checked by the simulation service, which reads -them in the catalog version its default worker runs. +them in the catalog version its default worker runs. Running a stored report +again (``POST /reports/{report_id}/runs``) goes through the same checks, of +the policies its simulations name. """ from __future__ import annotations @@ -42,6 +44,7 @@ EconomyReportCreationInput, EconomySimulationCreationInput, PolicySource, + ReportRerunInput, SimulationServiceClient, SubmittedBudgetWindow, SubmittedReport, @@ -75,7 +78,7 @@ def require_runnable_policies( names no policy, and a policy the running catalog could not check passes. Raise SubmissionStorageError when the database or its catalog cannot be read. Anything else that starts a run of stored policies, such as a re-run, - should check them here first. + checks them here first. """ with _storage_failures(): stored_policies = [ @@ -155,3 +158,13 @@ def submit_budget_window( report.country_id, (report.baseline.policy_id, report.reform.policy_id) ) return self._client_factory().submit_budget_window(report) + + def rerun_report(self, rerun: ReportRerunInput) -> None: + """Check the stored report, then have the simulation service run it again. + + The country is checked, then the policies of the report's + simulations, the baseline's first, as for a new report. Raise the + checks' errors, then the simulation service's. + """ + self._check(rerun.country_id, rerun.policy_ids) + self._client_factory().rerun_report(rerun) diff --git a/policyengine_api/services/v2/simulation_service/types.py b/policyengine_api/services/v2/simulation_service/types.py index 3603394de..744678d4b 100644 --- a/policyengine_api/services/v2/simulation_service/types.py +++ b/policyengine_api/services/v2/simulation_service/types.py @@ -2,10 +2,12 @@ The simulation service creates simulations, reports and their runs in the detailed-save tables, and starts their calculation on Modal. Its -``POST /simulations`` and ``POST /reports`` are reached through -``SimulationServiceClient``; ``policyengine_api.libs.simulation_reports`` -implements it over HTTP, and tests replace it with a stand-in. Reading what -it created needs no call: the Public API reads the tables. +``POST /simulations``, ``POST /reports`` and ``POST +/reports/{report_id}/runs``, which runs a stored report again, are reached +through ``SimulationServiceClient``; +``policyengine_api.libs.simulation_reports`` implements it over HTTP, and +tests replace it with a stand-in. Reading what it created needs no call: the +Public API reads the tables. The simulation service runs every calculation on its default worker's PolicyEngine.py version, so no request names a version. @@ -70,8 +72,9 @@ class SimulationServiceUnavailableError(RuntimeError): service's route runs; the service's own readable 503; a 400 that describes the service's state, or a row the Public API checks first, rather than the caller's request (such as a catalog version missing - while a release deploys); and a 400 or 422 the client cannot read. The - route answers 503. + while a release deploys); a 400 or 422 the client cannot read; and a + re-run's 409 the client cannot read, which the service answers only + before writing anything. The route answers 503. A message gives the service's reason only for a readable 400. For any other answer it gives the status, and the code when that is a fixed @@ -137,6 +140,22 @@ class SimulationServiceInvalidRequestError(ValueError): """ +class SimulationServiceConflictError(RuntimeError): + """The simulation service will not run the stored report again now. + + The report has a run pending or running, which ``run_id`` names, or it is + one the service does not run again, such as a kind of report it cannot + re-run yet or one whose simulation, region or dataset is gone. The + service wrote nothing. The message gives the service's code and reason; + the route answers 409, naming ``run_id`` when the service named a run. + """ + + def __init__(self, message: str, *, run_id: UUID | None = None) -> None: + """Say why, and name the run under way, if the service named one.""" + super().__init__(message) + self.run_id = run_id + + @dataclass(frozen=True) class PackageVersions: """A country package version and the PolicyEngine.py version beside it.""" @@ -295,6 +314,24 @@ class SubmittedBudgetWindow: budget_window_run_id: UUID +@dataclass(frozen=True) +class ReportRerunInput: + """A stored report of the country to calculate again as a new run. + + The simulation service runs it on its default worker's versions. Without + ``force``, each side reuses a run of its own simulation that succeeded on + those versions; with it, both simulations are calculated again. + ``policy_ids`` are the policies of the report's simulations, baseline + first, None for current law or a simulation that is gone, for the checks + before the simulation service is asked. + """ + + country_id: str + report_id: UUID + force: bool + policy_ids: tuple[UUID | None, ...] + + class SimulationServiceClient(Protocol): """The simulation service's routes that create and start calculations. @@ -318,6 +355,12 @@ class SimulationServiceClient(Protocol): record the service created when the service's 503 names the submission's own. + ``rerun_report`` raises the same errors, apart from naming a created + record, and SimulationServiceConflictError when the service will not run + the report again now; a 409 that cannot be read is + SimulationServiceUnavailableError, since the service answers 409 only + before writing anything. + ``SimulationSubmissionService``, which checks a submission before sending it through a client, also raises EconomyCalculationsUnsupportedError, PolicyNotFoundError, PolicyParametersMissingError and @@ -344,6 +387,10 @@ def submit_budget_window( """ ... + def rerun_report(self, rerun: ReportRerunInput) -> None: + """Create the stored report's new run on the current versions, and start it.""" + ... + class CountrySource(Protocol): """The countries list a submission checks first.""" diff --git a/policyengine_api/services/v2/stored_reports/database_connectors/reads.py b/policyengine_api/services/v2/stored_reports/database_connectors/reads.py index 9f369f17b..90a20910a 100644 --- a/policyengine_api/services/v2/stored_reports/database_connectors/reads.py +++ b/policyengine_api/services/v2/stored_reports/database_connectors/reads.py @@ -1,4 +1,4 @@ -"""Database reads of v2 reports and their newest runs. +"""Database reads of v2 reports, their newest runs and budget-window years. A report's own runs are report runs that are not a year of a budget window; a budget-window report's are budget-window runs, each with a report run per @@ -114,7 +114,7 @@ def read_report_rows( .where(col(Report.country) == country_id, col(Report.id).in_(report_ids)) ) rows = session.execute(statement).all() - years = _budget_window_years( + years = read_budget_window_years( session, [window_run.id for *_, window_run in rows if window_run is not None], ) @@ -143,7 +143,7 @@ def read_report_rows( ] -def _budget_window_years( +def read_budget_window_years( session: Session, budget_window_run_ids: Sequence[UUID] ) -> dict[UUID, tuple[ReportRun, ...]]: """Return each budget-window run's year report runs, in year order.""" diff --git a/policyengine_api/services/v2/stored_reports/database_connectors/reads_runs.py b/policyengine_api/services/v2/stored_reports/database_connectors/reads_runs.py new file mode 100644 index 000000000..ab6e5eca6 --- /dev/null +++ b/policyengine_api/services/v2/stored_reports/database_connectors/reads_runs.py @@ -0,0 +1,52 @@ +"""Database reads of a v2 report's runs, and of a budget window's runs.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from sqlalchemy import select +from sqlmodel import col + +from policyengine_api.data.v2.models import BudgetWindowRun, ReportRun + +if TYPE_CHECKING: + from uuid import UUID + + from sqlmodel import Session + + +def read_report_runs(session: Session, *, report_id: UUID) -> list[ReportRun]: + """Return every run of the report, newest first. + + Newest is created last, ties broken by the larger ID, as for the report's + newest run. Runs of a budget window's years are not the report's own + runs, so they are left out. + """ + statement = ( + select(ReportRun) + .where( + col(ReportRun.report_id) == report_id, + col(ReportRun.budget_window_run_id).is_(None), + ) + .order_by(col(ReportRun.created_at).desc(), col(ReportRun.id).desc()) + ) + return list(session.scalars(statement).all()) + + +def read_budget_window_runs( + session: Session, *, report_id: UUID +) -> list[BudgetWindowRun]: + """Return every budget-window run of the report, newest first. + + Newest is created last, ties broken by the larger ID, as for the + window's newest run. Each run's years are read separately + (``reads.read_budget_window_years``). + """ + statement = ( + select(BudgetWindowRun) + .where(col(BudgetWindowRun.report_id) == report_id) + .order_by( + col(BudgetWindowRun.created_at).desc(), col(BudgetWindowRun.id).desc() + ) + ) + return list(session.scalars(statement).all()) diff --git a/policyengine_api/services/v2/stored_reports/services.py b/policyengine_api/services/v2/stored_reports/services.py index 436108d75..c9e149222 100644 --- a/policyengine_api/services/v2/stored_reports/services.py +++ b/policyengine_api/services/v2/stored_reports/services.py @@ -24,7 +24,17 @@ back from the tables. Once the simulation service has answered with the report's ID, the report exists: a failure to read it back raises CreatedReportUnreadableError, which names it, so the caller reads it rather than submitting it again. -Re-runs and run lists are not here. + +Running a report again reads it in its country, checks its country and the +policies its simulations name as a new report's are checked, has the +simulation service create and start its new run on the current versions, +then reads the report back with that run as its newest. The database session +is closed while the simulation service is asked. A failure to read it back +raises RerunReportUnreadableError, since the new run exists. A report's runs +are listed newest first, whatever their status and versions, beside the +versions the Public API runs, so a caller can tell which are out of date: an +economy report's report runs, or a budget window's budget-window runs with +their years. Household reports are neither run again nor listed here yet. """ from __future__ import annotations @@ -36,6 +46,7 @@ from policyengine_api.data.v2.models import ReportRunStatus from policyengine_api.services.v2.simulation_service.types import ( + ReportRerunInput, SimulationServiceUnavailableError, ) from policyengine_api.services.v2.stored_reports.budget_window_results import ( @@ -48,25 +59,35 @@ reads_results as result_reads, ) from policyengine_api.services.v2.stored_reports.database_connectors.reads import ( + read_budget_window_years, read_report_rows, ) +from policyengine_api.services.v2.stored_reports.database_connectors.reads_runs import ( + read_budget_window_runs, + read_report_runs, +) from policyengine_api.services.v2.stored_reports.economy_results import ( transformations as result_transformations, ) from policyengine_api.services.v2.stored_reports.transformations import ( budget_window_read, + budget_window_runs_read, budget_window_summary_read, listed_reports, report_kind, + report_policy_ids, report_read, + report_runs_read, report_summary_read, ) from policyengine_api.services.v2.stored_reports.validators import ( CreatedReportUnreadableError, ReportStorageError, + RerunReportUnreadableError, require_readable_report, require_reform_run, require_report, + require_run_route_kind, ) if TYPE_CHECKING: @@ -87,6 +108,7 @@ BudgetWindowReportRead, StoredReportRead, StoredReportRows, + StoredReportRunsRead, StoredReportSummaryRead, ) @@ -105,8 +127,21 @@ def _storage_failures() -> Iterator[None]: raise ReportStorageError(msg) from error +def _stored_report( + session: Session, *, country_id: str, report_id: UUID +) -> StoredReportRows: + """Return the country's report with its newest run. + + Raise StoredReportNotFoundError when the country has no such report. + """ + rows = read_report_rows(session, country_id=country_id, report_ids=(report_id,)) + return require_report( + next(iter(rows), None), report_id=report_id, country_id=country_id + ) + + class V2StoredReportService: - """Read v2 reports and their results, and create economy reports and windows.""" + """Read v2 reports and results; create reports and windows; run reports again.""" def __init__( self, @@ -120,12 +155,19 @@ def __init__( ``current_versions`` returns the country package and PolicyEngine.py versions the Public API runs for a country. ``submissions`` is the checking ``SimulationSubmissionService``; without it, creating a - report raises SubmissionsNotConfiguredError. + report or running one again raises SubmissionsNotConfiguredError. """ self._database_session = database_session self._current_versions = current_versions self._submissions = submissions + def _configured_submissions(self) -> SimulationServiceClient: + """Return the submissions, or raise SubmissionsNotConfiguredError.""" + if self._submissions is None: + msg = "The simulation service is not configured here" + raise SubmissionsNotConfiguredError(msg) + return self._submissions + def get_report(self, *, country_id: str, report_id: UUID) -> StoredReportRead: """Return the country's report, its newest run and that run's result. @@ -138,18 +180,7 @@ def get_report(self, *, country_id: str, report_id: UUID) -> StoredReportRead: when the database cannot be read. """ with _storage_failures(), self._database_session.read() as session: - stored = require_report( - next( - iter( - read_report_rows( - session, country_id=country_id, report_ids=(report_id,) - ) - ), - None, - ), - report_id=report_id, - country_id=country_id, - ) + stored = _stored_report(session, country_id=country_id, report_id=report_id) if require_readable_report(stored) == "budget_window": return self._budget_window(session, stored) run = stored.newest_run @@ -220,13 +251,11 @@ def create_report(self, report_input: ReportCreationInput) -> StoredReportRead: created cannot be read back in the country: not stored there, the database unavailable, or any other failure of the read. """ - if self._submissions is None: - msg = "The simulation service is not configured here" - raise SubmissionsNotConfiguredError(msg) + submissions = self._configured_submissions() if report_input.kind == "budget_window": - report_id = self._submissions.submit_budget_window(report_input).report_id + report_id = submissions.submit_budget_window(report_input).report_id else: - report_id = self._submissions.submit_report(report_input).report_id + report_id = submissions.submit_report(report_input).report_id try: return self.get_report( country_id=report_input.country_id, report_id=report_id @@ -236,3 +265,74 @@ def create_report(self, report_input: ReportCreationInput) -> StoredReportRead: raise CreatedReportUnreadableError( report_id=report_id, country_id=report_input.country_id ) from error + + def rerun_report( + self, *, country_id: str, report_id: UUID, force: bool + ) -> StoredReportRead: + """Run the country's economy report or budget window again; return it. + + It is returned with its new run as its newest. Without ``force``, each + side reuses a run of its own simulation that succeeded on the current + versions; with it, both are calculated again (a budget window's + new run then recalculates each year). Raise StoredReportNotFoundError + when the country has no such report, ReportKindUnsupportedError for a + household report, ReportStorageError when it cannot be read, the + submission's + checks' and the simulation service's errors (see + ``SimulationSubmissionService.rerun_report``), and + RerunReportUnreadableError, caused by the read's own error, when the + report cannot be read back once its new run was started. + """ + submissions = self._configured_submissions() + with _storage_failures(), self._database_session.read() as session: + stored = _stored_report(session, country_id=country_id, report_id=report_id) + require_run_route_kind(stored) + submissions.rerun_report( + ReportRerunInput( + country_id=country_id, + report_id=report_id, + force=force, + policy_ids=report_policy_ids(stored), + ) + ) + try: + return self.get_report(country_id=country_id, report_id=report_id) + except Exception as error: + # The new run exists whatever failed, so the caller reads it. + raise RerunReportUnreadableError( + report_id=report_id, country_id=country_id + ) from error + + def list_report_runs( + self, *, country_id: str, report_id: UUID + ) -> StoredReportRunsRead: + """Return the runs of the country's economy report or budget window. + + They are newest first: an economy report's report runs, or a budget + window's budget-window runs, each with its years. Raise + StoredReportNotFoundError when the country has no such report, + ReportKindUnsupportedError for a household report, and + ReportStorageError when the database cannot be read. + """ + with _storage_failures(), self._database_session.read() as session: + stored = _stored_report(session, country_id=country_id, report_id=report_id) + current_versions = self._current_versions(country_id) + # A kind the run routes take later fails mypy (a missing return) + # until it has a case here. + match require_run_route_kind(stored): + case "budget_window": + window_runs = read_budget_window_runs(session, report_id=report_id) + return budget_window_runs_read( + stored, + window_runs, + read_budget_window_years( + session, [window_run.id for window_run in window_runs] + ), + current_versions=current_versions, + ) + case "economy": + return report_runs_read( + stored, + read_report_runs(session, report_id=report_id), + current_versions=current_versions, + ) diff --git a/policyengine_api/services/v2/stored_reports/transformations.py b/policyengine_api/services/v2/stored_reports/transformations.py index f58c37926..c7f9b51ae 100644 --- a/policyengine_api/services/v2/stored_reports/transformations.py +++ b/policyengine_api/services/v2/stored_reports/transformations.py @@ -8,16 +8,18 @@ BudgetWindowReportRead, BudgetWindowReportSummaryRead, BudgetWindowRunRead, + BudgetWindowRunsRead, BudgetWindowYearRead, ReportDefinitionRead, ReportRead, ReportRunRead, + ReportRunsRead, ReportSimulationRead, ReportSummaryRead, ) if TYPE_CHECKING: - from collections.abc import Iterable, Sequence + from collections.abc import Iterable, Mapping, Sequence from uuid import UUID from policyengine_api.data.v2.models import ( @@ -215,3 +217,52 @@ def listed_reports( if stored is not None: listed.append(stored) return listed + + +def report_policy_ids(rows: StoredReportRows) -> tuple[UUID | None, ...]: + """Return the policies of the report's simulations, baseline first. + + A simulation of current law, or one that is gone, names none. + """ + return tuple( + None if simulation is None else simulation.policy_id + for simulation in (rows.baseline_simulation, rows.reform_simulation) + ) + + +def report_runs_read( + rows: StoredReportRows, + runs: Iterable[ReportRun], + *, + current_versions: PackageVersions, +) -> ReportRunsRead: + """Return an economy report's runs, in the order given, and the current versions.""" + return ReportRunsRead( + report_id=rows.report.id, + kind="economy", + runs=tuple(report_run_read(run) for run in runs), + current_versions=current_versions, + ) + + +def budget_window_runs_read( + rows: StoredReportRows, + window_runs: Iterable[BudgetWindowRun], + years: Mapping[UUID, Sequence[ReportRun]], + *, + current_versions: PackageVersions, +) -> BudgetWindowRunsRead: + """Return a budget window's runs, in the order given, and the current versions. + + ``years`` holds each window run's year report runs, in year order; a run + with none has no years. + """ + return BudgetWindowRunsRead( + report_id=rows.report.id, + kind="budget_window", + runs=tuple( + budget_window_run_read(window_run, years.get(window_run.id, ())) + for window_run in window_runs + ), + current_versions=current_versions, + ) diff --git a/policyengine_api/services/v2/stored_reports/types.py b/policyengine_api/services/v2/stored_reports/types.py index c4aadd560..80a13a986 100644 --- a/policyengine_api/services/v2/stored_reports/types.py +++ b/policyengine_api/services/v2/stored_reports/types.py @@ -43,6 +43,8 @@ # What a report compares: two economy simulations, two household simulations, # or two economy simulations in each year of a budget window. ReportKind = Literal["economy", "household", "budget_window"] +# The kinds whose runs the run routes start and list. +RunRouteKind = Literal["economy", "budget_window"] ReportRunTriggerName = Literal["initial", "manual", "system"] @@ -177,6 +179,38 @@ class BudgetWindowReportRead(BudgetWindowReportSummaryRead): StoredReportSummaryRead = ReportSummaryRead | BudgetWindowReportSummaryRead +@dataclass(frozen=True) +class ReportRunsRead: + """An economy report's runs, newest first, and the versions the Public API runs. + + Each run names the country package and PolicyEngine.py versions it ran + on; one on other versions than ``current_versions`` is out of date. + """ + + report_id: UUID + kind: Literal["economy"] + runs: tuple[ReportRunRead, ...] + current_versions: PackageVersions + + +@dataclass(frozen=True) +class BudgetWindowRunsRead: + """A budget window's runs, newest first, each with its years. + + Each budget-window run names the versions its years ran on; one on other + versions than ``current_versions`` is out of date. + """ + + report_id: UUID + kind: Literal["budget_window"] + runs: tuple[BudgetWindowRunRead, ...] + current_versions: PackageVersions + + +# The runs of a report of either kind the run routes take. +StoredReportRunsRead = ReportRunsRead | BudgetWindowRunsRead + + @dataclass(frozen=True) class StoredReportRows: """A report's row, its newest runs and the catalog and simulation rows it names. diff --git a/policyengine_api/services/v2/stored_reports/validators.py b/policyengine_api/services/v2/stored_reports/validators.py index bfd888c89..21a102230 100644 --- a/policyengine_api/services/v2/stored_reports/validators.py +++ b/policyengine_api/services/v2/stored_reports/validators.py @@ -18,10 +18,17 @@ from policyengine_api.data.v2.models import ReportRun from policyengine_api.services.v2.stored_reports.types import ( ReportKind, + RunRouteKind, StoredReportRows, ) KIND_NAMES = {"household": "a household report"} +# Each kind of report, as the run routes name one they refuse. +REPORT_KIND_NAMES: dict[ReportKind, str] = { + "economy": "an economy report", + "household": "a household report", + "budget_window": "a budget window", +} class StoredReportNotFoundError(LookupError): @@ -51,11 +58,30 @@ def __init__(self, *, report_id: UUID, country_id: str) -> None: self.country_id = country_id +class RerunReportUnreadableError(RuntimeError): + """The simulation service started the report's new run, but it could not be read. + + The new run exists, so the caller reads the report rather than running + it again. ``report_id`` and ``country_id`` say where to read it; the + error's cause says why reading it back failed. + """ + + def __init__(self, *, report_id: UUID, country_id: str) -> None: + """Name the report and its country.""" + super().__init__( + f"The simulation service started a new run of report {report_id} in " + f"country {country_id!r}, but the report could not be read back" + ) + self.report_id = report_id + self.country_id = country_id + + class ReportKindUnsupportedError(ValueError): - """The report is of a kind these routes do not read yet. + """The report is of a kind a route does not read, or start and list runs of, yet. - Household reports are stored in the same tables, and are read once their - own routes exist. + Household reports are stored in the same tables, and each route takes + them once its own PR adds them; the message says which kind the report + is and what the route does not do with it. """ @@ -84,6 +110,23 @@ def require_readable_report(rows: StoredReportRows) -> ReportKind: return kind +def require_run_route_kind(rows: StoredReportRows) -> RunRouteKind: + """Return the report's kind, or raise ReportKindUnsupportedError. + + The run routes start and list runs of economy reports and budget + windows, and refuse any other kind. They take household reports once + their own PR adds them, separately from the read routes' kinds. + """ + kind = report_kind(rows.report) + if kind in ("economy", "budget_window"): + return kind + msg = ( + f"Report {rows.report.id} is {REPORT_KIND_NAMES[kind]}, whose runs " + "these routes do not start or list yet" + ) + raise ReportKindUnsupportedError(msg) + + def require_reform_run(report_run: ReportRun) -> UUID: """Return the reform simulation run an economy report run compares. diff --git a/tests/contract/registry.py b/tests/contract/registry.py index a35ea84b5..4d7fc840c 100644 --- a/tests/contract/registry.py +++ b/tests/contract/registry.py @@ -658,6 +658,34 @@ class WorkflowContract: ), route_group="report", ), + ContractRequest( + method="POST", + path="/v2/reports/{report_id}/runs?country_id=us", + expected_status=201, + stable_response_fields=( + "status", + "message", + "result.item.id", + "result.item.status", + "result.item.run", + "result.item.current_versions", + ), + route_group="report", + ), + ContractRequest( + method="GET", + path="/v2/reports/{report_id}/runs?country_id=us", + expected_status=200, + stable_response_fields=( + "status", + "message", + "result.report_id", + "result.kind", + "result.items", + "result.current_versions", + ), + route_group="report", + ), ContractRequest( method="POST", path="/v2/simulations?country_id=us", diff --git a/tests/integration/test_v2_report_runs.py b/tests/integration/test_v2_report_runs.py new file mode 100644 index 000000000..b98d02331 --- /dev/null +++ b/tests/integration/test_v2_report_runs.py @@ -0,0 +1,649 @@ +"""Run v2 reports again and list their runs, on migrated PostgreSQL. + +A stand-in for the simulation service's client stores the new run the +simulation service would create and records what it was asked; the routes +run as the Public API's runtime role, reading the report, checking its +country and policies, and reading the report and its runs back. +""" + +from __future__ import annotations + +from contextlib import contextmanager +from dataclasses import dataclass, field, replace +from datetime import timedelta +from typing import TYPE_CHECKING, Any +from uuid import UUID, uuid4 + +import pytest +from alembic import command +from alembic.config import Config +from fastapi.testclient import TestClient +from flask import Flask +from sqlalchemy import create_engine, text +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import NullPool +from sqlmodel import Session + +from policyengine_api.asgi_factory import create_asgi_app +from policyengine_api.constants import REPO +from policyengine_api.fastapi_routes.dependencies import NativeRouteDependencies +from policyengine_api.migration_flags import ( + RouteImplementation, + RouteImplementationSettings, +) +from policyengine_api.services.v2.countries.database_session import ( + CountryDatabaseSession, +) +from policyengine_api.services.v2.countries.services import V2CountryService +from policyengine_api.services.v2.policies.database_session import ( + PolicyDatabaseSession, +) +from policyengine_api.services.v2.policies.services import V2PolicyService +from policyengine_api.services.v2.simulation_service.services import ( + SimulationSubmissionService, +) +from policyengine_api.services.v2.simulation_service.types import ( + PackageVersions, + ReportRerunInput, + SimulationServiceConflictError, + SimulationServiceUnavailableError, +) +from policyengine_api.services.v2.stored_reports.database_session import ( + StoredReportDatabaseSession, +) +from policyengine_api.services.v2.stored_reports.services import V2StoredReportService +from tests.integration.simulation_rows import insert_row +from tests.integration.stored_report_rows import ( + STARTED_AT, + BudgetWindowYear, + RunSettings, + StoredReport, + delete_catalog, + store_budget_window_run, + store_catalog, + store_report, + store_report_run, + store_year_simulations, +) + +if TYPE_CHECKING: + from collections.abc import Iterator + + from sqlalchemy import Connection, Engine + + from tests.integration.stored_report_rows import CatalogRows + +CURRENT_VERSIONS = {"country_package_version": "1.5.0", "policyengine_version": "4.2.0"} +OLDER_VERSIONS = {"country_package_version": "1.4.0", "policyengine_version": "4.1.0"} + + +def current_versions(country_id: str) -> PackageVersions: + return PackageVersions(**CURRENT_VERSIONS) + + +def mark_manual(connection: Connection, run_id: UUID) -> None: + """Record the report run as a re-run, as the simulation service creates one.""" + connection.execute( + text("UPDATE report_runs SET trigger = 'manual' WHERE id = :id"), + {"id": run_id}, + ) + + +def mark_window_manual( + connection: Connection, window_run_id: UUID, *, recalculate: bool +) -> None: + """Record the window run as a re-run, as the simulation service creates one.""" + connection.execute( + text( + "UPDATE budget_window_runs SET trigger = 'manual', recalculate = " + ":recalculate WHERE id = :id" + ), + {"id": window_run_id, "recalculate": recalculate}, + ) + + +def store_window_run( + connection: Connection, + catalog: CatalogRows, + report: StoredReport, + settings: RunSettings, + years: tuple[int, ...], +) -> UUID: + """Store a budget-window run and a report run of each year, with ``settings``.""" + window_run_id = store_budget_window_run(connection, catalog, report, settings) + for year in years: + store_report_run( + connection, + catalog, + store_year_simulations(connection, catalog, report, year), + replace( + settings, + budget_window_year=BudgetWindowYear( + budget_window_run_id=window_run_id, year=year + ), + ), + ) + return window_run_id + + +@dataclass +class RerunningSimulationService: + """Store the new run the simulation service would start, and record the ask. + + With ``delete_report``, the report is deleted once its new run is stored, + so reading it back finds nothing. With ``refusal``, it stores nothing and + raises it, as the simulation service refuses. With ``database_session``, + it records how many of the report service's sessions are open when asked. + With ``window_years``, the report is a budget window, and the new run is + a budget-window run with a pending report run for each of those years. + """ + + engine: Engine + catalog: CatalogRows + delete_report: bool = False + refusal: Exception | None = None + database_session: CountingDatabaseSession | None = None + window_years: tuple[int, ...] = () + reruns: list[ReportRerunInput] = field(default_factory=list) + new_run_ids: list[UUID] = field(default_factory=list) + open_sessions_when_asked: list[int] = field(default_factory=list) + + def rerun_report(self, rerun: ReportRerunInput) -> None: + self.reruns.append(rerun) + if self.database_session is not None: + self.open_sessions_when_asked.append(self.database_session.open_sessions) + if self.refusal is not None: + raise self.refusal + with self.engine.begin() as connection: + simulations = connection.execute( + text( + "SELECT baseline_simulation_id, reform_simulation_id FROM " + "reports WHERE id = :id" + ), + {"id": rerun.report_id}, + ).one() + report = StoredReport( + report_id=rerun.report_id, + baseline_simulation_id=simulations.baseline_simulation_id, + reform_simulation_id=simulations.reform_simulation_id, + ) + settings = RunSettings( + versions=CURRENT_VERSIONS, + status="pending", + created_at=STARTED_AT + timedelta(days=30), + ) + if self.window_years: + window_run_id = store_window_run( + connection, self.catalog, report, settings, self.window_years + ) + mark_window_manual(connection, window_run_id, recalculate=rerun.force) + self.new_run_ids.append(window_run_id) + else: + run_ids = store_report_run(connection, self.catalog, report, settings) + mark_manual(connection, run_ids.report_run_id) + self.new_run_ids.append(run_ids.report_run_id) + if self.delete_report: + connection.execute( + text("DELETE FROM reports WHERE id = :id"), + {"id": rerun.report_id}, + ) + + +@pytest.fixture +def owner_engine(disposable_v2_database_url: str) -> Iterator[Engine]: + command.upgrade(Config(str(REPO / "alembic-v2.ini")), "head") + engine = create_engine(disposable_v2_database_url, poolclass=NullPool) + yield engine + engine.dispose() + + +@pytest.fixture +def catalog(owner_engine: Engine) -> Iterator[CatalogRows]: + with owner_engine.begin() as connection: + stored = store_catalog(connection, "us") + yield stored + with owner_engine.begin() as connection: + delete_catalog(connection, stored) + + +@pytest.fixture +def runtime_sessions(managed_v2_roles, v2_engine_as_role) -> Iterator[Any]: + engine = v2_engine_as_role(managed_v2_roles.runtime, poolclass=NullPool) + yield sessionmaker(engine, class_=Session) + engine.dispose() + + +class CountingDatabaseSession(StoredReportDatabaseSession): + """Count the report service's read sessions open at each moment.""" + + def __init__(self, sessions: Any) -> None: + super().__init__(sessions) + self.open_sessions = 0 + + @contextmanager + def read(self) -> Iterator[Session]: + self.open_sessions += 1 + try: + with super().read() as session: + yield session + finally: + self.open_sessions -= 1 + + +def client_for( + sessions: Any, + simulation_service: RerunningSimulationService, + database_session: StoredReportDatabaseSession | None = None, +) -> TestClient: + """Serve the routes as the runtime role, running reports through the stand-in.""" + submissions = SimulationSubmissionService( + countries=V2CountryService(CountryDatabaseSession(sessions)), + policies=V2PolicyService( + PolicyDatabaseSession(sessions), running_policyengine_version="4.2.0" + ), + client_factory=lambda: simulation_service, + ) + dependencies = NativeRouteDependencies( + readiness_probe=lambda: True, + gateway_client_factory=lambda: None, + metadata_reader_factory=lambda: None, + v2_stored_report_service_factory=lambda: V2StoredReportService( + database_session or StoredReportDatabaseSession(sessions), + current_versions=current_versions, + submissions=submissions, + ), + ) + settings = RouteImplementationSettings( + health=RouteImplementation.FLASK_FALLBACK, + specification=RouteImplementation.FLASK_FALLBACK, + metadata=RouteImplementation.FLASK_FALLBACK, + ) + return TestClient( + create_asgi_app( + Flask(__name__), dependencies=dependencies, route_settings=settings + ) + ) + + +def test_a_reports_runs_are_listed_newest_first_with_their_versions( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any +) -> None: + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + first = store_report_run( + connection, + catalog, + report, + RunSettings(versions=OLDER_VERSIONS, status="succeeded"), + ) + failed = store_report_run( + connection, + catalog, + report, + RunSettings( + versions=CURRENT_VERSIONS, + status="failed", + created_at=STARTED_AT + timedelta(days=1), + error_message="The reform simulation failed", + ), + ) + mark_manual(connection, failed.report_run_id) + newest = store_report_run( + connection, + catalog, + report, + RunSettings( + versions=CURRENT_VERSIONS, + status="running", + created_at=STARTED_AT + timedelta(days=2), + ), + ) + mark_manual(connection, newest.report_run_id) + other_report = store_report(connection, catalog) + store_report_run( + connection, + catalog, + other_report, + RunSettings(versions=CURRENT_VERSIONS), + ) + client = client_for( + runtime_sessions, RerunningSimulationService(owner_engine, catalog) + ) + + response = client.get(f"/v2/reports/{report.report_id}/runs?country_id=us") + + assert response.status_code == 200, response.text + result = response.json()["result"] + assert result["report_id"] == str(report.report_id) + assert result["kind"] == "economy" + assert result["current_versions"] == CURRENT_VERSIONS + items = result["items"] + assert [item["id"] for item in items] == [ + str(newest.report_run_id), + str(failed.report_run_id), + str(first.report_run_id), + ] + assert [(item["status"], item["trigger"]) for item in items] == [ + ("running", "manual"), + ("failed", "manual"), + ("succeeded", "initial"), + ] + assert [item["policyengine_version"] for item in items] == [ + "4.2.0", + "4.2.0", + "4.1.0", + ] + assert items[1]["error_message"] == "The reform simulation failed" + assert items[2]["baseline_simulation_run_id"] == str( + first.baseline_simulation_run_id + ) + assert items[2]["reform_simulation_run_id"] == str(first.reform_simulation_run_id) + + +def test_a_report_with_no_run_lists_none( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any +) -> None: + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + client = client_for( + runtime_sessions, RerunningSimulationService(owner_engine, catalog) + ) + + response = client.get(f"/v2/reports/{report.report_id}/runs?country_id=us") + + assert response.status_code == 200, response.text + assert response.json()["result"]["items"] == [] + + +@pytest.mark.parametrize("method", ["get", "post"]) +def test_another_countrys_report_or_an_unknown_one_is_not_found( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any, method: str +) -> None: + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + simulation_service = RerunningSimulationService(owner_engine, catalog) + client = client_for(runtime_sessions, simulation_service) + + other_country = client.request( + method, f"/v2/reports/{report.report_id}/runs?country_id=uk", json={} + ) + unknown = client.request( + method, f"/v2/reports/{uuid4()}/runs?country_id=us", json={} + ) + + assert other_country.status_code == 404 + assert unknown.status_code == 404 + assert simulation_service.reruns == [] + + +def store_household_report(connection: Connection, catalog: CatalogRows) -> UUID: + """Store a household and a report over it; return the report's ID.""" + household_id = uuid4() + insert_row( + connection, + "households", + id=household_id, + country_id=catalog.country, + household_data="{}", + canonicalization_version=1, + content_hash=uuid4().hex * 2, + ) + catalog.household_ids.append(household_id) + return store_report(connection, catalog, household_id=household_id).report_id + + +@pytest.mark.parametrize("method", ["get", "post"]) +def test_a_household_reports_runs_are_not_listed_or_started_here_yet( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any, method: str +) -> None: + with owner_engine.begin() as connection: + report_id = store_household_report(connection, catalog) + simulation_service = RerunningSimulationService(owner_engine, catalog) + client = client_for(runtime_sessions, simulation_service) + + response = client.request( + method, f"/v2/reports/{report_id}/runs?country_id=us", json={} + ) + + assert response.status_code == 422 + assert response.json()["message"] == ( + f"Report {report_id} is a household report, whose runs these routes do " + "not start or list yet" + ) + assert simulation_service.reruns == [] + + +@pytest.mark.parametrize("force", [False, True]) +def test_running_a_report_again_answers_it_with_its_new_run( + owner_engine: Engine, + catalog: CatalogRows, + runtime_sessions: Any, + force: bool, # noqa: FBT001 # pytest passes parameters positionally +) -> None: + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + first = store_report_run( + connection, catalog, report, RunSettings(versions=OLDER_VERSIONS) + ) + simulation_service = RerunningSimulationService(owner_engine, catalog) + client = client_for(runtime_sessions, simulation_service) + + response = client.post( + f"/v2/reports/{report.report_id}/runs?country_id=us", json={"force": force} + ) + + assert response.status_code == 201, response.text + assert simulation_service.reruns == [ + ReportRerunInput( + country_id="us", + report_id=report.report_id, + force=force, + policy_ids=(None, catalog.policy_id), + ) + ] + item = response.json()["result"]["item"] + (new_run_id,) = simulation_service.new_run_ids + assert item["id"] == str(report.report_id) + assert item["status"] == "pending" + assert item["run"]["id"] == str(new_run_id) + assert item["run"]["trigger"] == "manual" + assert item["run"]["policyengine_version"] == "4.2.0" + assert item["current_versions"] == CURRENT_VERSIONS + assert item["result"] is None + runs = client.get(f"/v2/reports/{report.report_id}/runs?country_id=us").json() + assert [run["id"] for run in runs["result"]["items"]] == [ + str(new_run_id), + str(first.report_run_id), + ] + + +def test_a_report_whose_reform_simulation_is_gone_is_left_to_the_service_to_refuse( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any +) -> None: + # The Public API checks the policies that remain and asks the simulation + # service, which refuses a report missing a simulation. + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + connection.execute( + text("UPDATE reports SET reform_simulation_id = NULL WHERE id = :id"), + {"id": report.report_id}, + ) + message = ( + "The simulation service will not run the report again now " + "(rerun_not_supported): The report has no reform simulation." + ) + simulation_service = RerunningSimulationService( + owner_engine, catalog, refusal=SimulationServiceConflictError(message) + ) + client = client_for(runtime_sessions, simulation_service) + + response = client.post( + f"/v2/reports/{report.report_id}/runs?country_id=us", json={} + ) + + assert response.status_code == 409 + assert response.json() == {"status": "error", "message": message, "run_id": None} + assert [rerun.policy_ids for rerun in simulation_service.reruns] == [(None, None)] + + +def test_a_new_run_whose_report_cannot_be_read_back_answers_the_reports_id( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any +) -> None: + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + simulation_service = RerunningSimulationService( + owner_engine, catalog, delete_report=True + ) + + response = client_for(runtime_sessions, simulation_service).post( + f"/v2/reports/{report.report_id}/runs?country_id=us", json={} + ) + + assert response.status_code == 502 + assert response.json() == { + "status": "error", + "message": ( + f"The simulation service started a new run of report " + f"{report.report_id}, but the report could not be read back. Read it " + f"with GET /v2/reports/{report.report_id}?country_id=us rather than " + "running it again." + ), + "report_id": str(report.report_id), + } + + +def test_no_report_session_is_open_while_the_simulation_service_is_asked( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any +) -> None: + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + database_session = CountingDatabaseSession(runtime_sessions) + simulation_service = RerunningSimulationService( + owner_engine, catalog, database_session=database_session + ) + client = client_for(runtime_sessions, simulation_service, database_session) + + response = client.post( + f"/v2/reports/{report.report_id}/runs?country_id=us", json={} + ) + + assert response.status_code == 201, response.text + assert simulation_service.open_sessions_when_asked == [0] + assert database_session.open_sessions == 0 + + +def test_the_simulation_services_refusal_is_answered_as_its_own( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any +) -> None: + # The report exists, so the service's refusal is not mistaken for a + # report that cannot be read back. + with owner_engine.begin() as connection: + report = store_report(connection, catalog) + message = "The simulation service is unavailable (503 manifest_unavailable)." + simulation_service = RerunningSimulationService( + owner_engine, catalog, refusal=SimulationServiceUnavailableError(message) + ) + + response = client_for(runtime_sessions, simulation_service).post( + f"/v2/reports/{report.report_id}/runs?country_id=us", json={} + ) + + assert response.status_code == 503 + assert response.json() == {"status": "error", "message": message} + + +def test_a_budget_windows_runs_are_listed_newest_first_with_their_years( + owner_engine: Engine, catalog: CatalogRows, runtime_sessions: Any +) -> None: + with owner_engine.begin() as connection: + window = store_report(connection, catalog, budget_window_size=2) + first_run_id = store_window_run( + connection, + catalog, + window, + RunSettings(versions=OLDER_VERSIONS, status="succeeded"), + (2026, 2027), + ) + newest_run_id = store_window_run( + connection, + catalog, + window, + RunSettings( + versions=CURRENT_VERSIONS, + status="running", + created_at=STARTED_AT + timedelta(days=1), + ), + (2026, 2027), + ) + mark_window_manual(connection, newest_run_id, recalculate=True) + client = client_for( + runtime_sessions, RerunningSimulationService(owner_engine, catalog) + ) + + response = client.get(f"/v2/reports/{window.report_id}/runs?country_id=us") + + assert response.status_code == 200, response.text + result = response.json()["result"] + assert result["report_id"] == str(window.report_id) + assert result["kind"] == "budget_window" + assert result["current_versions"] == CURRENT_VERSIONS + items = result["items"] + assert [item["id"] for item in items] == [str(newest_run_id), str(first_run_id)] + assert [ + (item["status"], item["trigger"], item["recalculate"]) for item in items + ] == [("running", "manual", True), ("succeeded", "initial", False)] + assert [item["policyengine_version"] for item in items] == ["4.2.0", "4.1.0"] + assert [[year["year"] for year in item["years"]] for item in items] == [ + [2026, 2027], + [2026, 2027], + ] + + +@pytest.mark.parametrize("force", [False, True]) +def test_running_a_budget_window_again_answers_it_with_its_new_run( + owner_engine: Engine, + catalog: CatalogRows, + runtime_sessions: Any, + force: bool, # noqa: FBT001 # pytest passes parameters positionally +) -> None: + with owner_engine.begin() as connection: + window = store_report(connection, catalog, budget_window_size=2) + first_run_id = store_window_run( + connection, + catalog, + window, + RunSettings(versions=OLDER_VERSIONS), + (2026, 2027), + ) + simulation_service = RerunningSimulationService( + owner_engine, catalog, window_years=(2026, 2027) + ) + client = client_for(runtime_sessions, simulation_service) + + response = client.post( + f"/v2/reports/{window.report_id}/runs?country_id=us", json={"force": force} + ) + + assert response.status_code == 201, response.text + assert simulation_service.reruns == [ + ReportRerunInput( + country_id="us", + report_id=window.report_id, + force=force, + policy_ids=(None, catalog.policy_id), + ) + ] + item = response.json()["result"]["item"] + (new_run_id,) = simulation_service.new_run_ids + assert item["kind"] == "budget_window" + assert item["status"] == "pending" + assert item["run"]["id"] == str(new_run_id) + assert item["run"]["trigger"] == "manual" + assert item["run"]["recalculate"] is force + assert item["run"]["policyengine_version"] == "4.2.0" + assert item["result"] is None + runs = client.get(f"/v2/reports/{window.report_id}/runs?country_id=us").json() + assert [run["id"] for run in runs["result"]["items"]] == [ + str(new_run_id), + str(first_run_id), + ] diff --git a/tests/unit/libs/test_simulation_report_reruns.py b/tests/unit/libs/test_simulation_report_reruns.py new file mode 100644 index 000000000..977de08bd --- /dev/null +++ b/tests/unit/libs/test_simulation_report_reruns.py @@ -0,0 +1,522 @@ +"""Run a stored report again through the simulation service; read its answers.""" + +from __future__ import annotations + +import json +from typing import Any +from uuid import UUID + +import httpx +import pytest + +from policyengine_api.fastapi_routes.v2.reports.routes import REPORT_ERROR_MAPPINGS +from policyengine_api.libs.simulation_report_reruns import report_runs_path +from policyengine_api.libs.simulation_reports import SimulationServiceReportClient +from policyengine_api.libs.simulation_service_answers import SimulationAnswer +from policyengine_api.services.v2.simulation_service.types import ( + ReportRerunInput, + SimulationServiceAnswerUnreadableError, + SimulationServiceConflictError, + SimulationServiceInternalError, + SimulationServiceInvalidRequestError, + SimulationServiceUnavailableError, +) + +SIMULATION_URL = "https://simulation.example.test" +ENVIRONMENT_VARIABLES = ( + "GATEWAY_AUTH_ISSUER", + "GATEWAY_AUTH_AUDIENCE", + "GATEWAY_AUTH_CLIENT_ID", + "GATEWAY_AUTH_CLIENT_SECRET", + "GATEWAY_AUTH_CLIENT_SECRET_RESOURCE", + "GATEWAY_AUTH_REQUIRED", + "SIM_ENTRYPOINT", + "OLD_SIMULATION_GATEWAY_URL", + "SIMULATION_ENTRYPOINT_URL", +) +REPORT_ID = UUID("00000000-0000-0000-0000-00000000d001") +OTHER_REPORT_ID = UUID("00000000-0000-0000-0000-00000000d002") +NEW_RUN_ID = UUID("00000000-0000-0000-0000-00000000d003") +RUNNING_RUN_ID = UUID("00000000-0000-0000-0000-00000000d004") +POLICY_ID = UUID("00000000-0000-0000-0000-00000000d005") +SIMULATION_ID = UUID("00000000-0000-0000-0000-00000000d006") +# What the simulation service answers once it has started the report's new run. +RERUN_STARTED = { + "report_id": str(REPORT_ID), + "report_run_id": str(NEW_RUN_ID), + "status": "pending", + "baseline_simulation_id": "00000000-0000-0000-0000-00000000d007", + "baseline_simulation_run_id": "00000000-0000-0000-0000-00000000d008", + "baseline_reused": True, + "reform_simulation_id": "00000000-0000-0000-0000-00000000d009", + "reform_simulation_run_id": "00000000-0000-0000-0000-00000000d00a", + "reform_reused": False, +} +READ_BEFORE_RUNNING_AGAIN = ( + "Read the report to see whether it has a new run before running it again." +) + + +@pytest.fixture(autouse=True) +def configured_simulation_service(monkeypatch): + """Name the Cloud Run simulation service; set no gateway credentials.""" + for name in ENVIRONMENT_VARIABLES: + monkeypatch.delenv(name, raising=False) + monkeypatch.setenv("SIMULATION_ENTRYPOINT_URL", f"{SIMULATION_URL}/") + monkeypatch.setenv("SIM_ENTRYPOINT", "old_gateway_direct") + monkeypatch.setenv("OLD_SIMULATION_GATEWAY_URL", "https://old.example.test") + + +def rerun_input(**changes: Any) -> ReportRerunInput: + values: dict[str, Any] = { + "country_id": "us", + "report_id": REPORT_ID, + "force": False, + "policy_ids": (None, POLICY_ID), + } + return ReportRerunInput(**{**values, **changes}) + + +def route_status(error: Exception) -> int: + """Return the status the report routes answer ``error`` with.""" + return next( + mapping.status_code + for mapping in REPORT_ERROR_MAPPINGS + if isinstance(error, mapping.error_types) + ) + + +class ScriptedService: + """The simulation service, answering every request with one scripted answer.""" + + def __init__(self, status_code: int = 201, body: object = RERUN_STARTED) -> None: + self.status_code = status_code + self.body = body + self.requests: list[httpx.Request] = [] + + def answer(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + if isinstance(self.body, bytes): + return httpx.Response(self.status_code, content=self.body) + return httpx.Response(self.status_code, json=self.body) + + def rerun(self, **changes: Any) -> None: + client = SimulationServiceReportClient.from_environment( + transport=httpx.MockTransport(self.answer) + ) + client.rerun_report(rerun_input(**changes)) + + +def possibly_started_message(status_and_code: str) -> str: + return ( + "The simulation service may have started a new run of the report, but it " + f"failed ({status_and_code}). {READ_BEFORE_RUNNING_AGAIN}" + ) + + +@pytest.mark.parametrize("force", [False, True]) +def test_a_rerun_posts_force_to_the_reports_runs_route( + force: bool, # noqa: FBT001 # pytest passes parameters positionally +) -> None: + service = ScriptedService() + + service.rerun(force=force) + + (request,) = service.requests + assert request.method == "POST" + assert str(request.url) == f"{SIMULATION_URL}/reports/{REPORT_ID}/runs" + assert report_runs_path(REPORT_ID) == f"/reports/{REPORT_ID}/runs" + assert json.loads(request.content) == {"force": force} + + +@pytest.mark.parametrize("status_code", [201, 202]) +def test_a_started_rerun_naming_the_report_is_accepted(status_code: int) -> None: + service = ScriptedService( + status_code=status_code, body={"report_id": str(REPORT_ID)} + ) + + service.rerun() + + assert len(service.requests) == 1 + + +@pytest.mark.parametrize( + "body", + [ + {**RERUN_STARTED, "report_id": str(OTHER_REPORT_ID)}, + {"report_run_id": str(NEW_RUN_ID)}, + b"not json", + ], + ids=["another-report", "no-report", "not-json"], +) +def test_a_started_answer_that_cannot_be_read_may_have_started_a_run( + body: object, +) -> None: + with pytest.raises(SimulationServiceAnswerUnreadableError) as unreadable: + ScriptedService(body=body).rerun() + + assert route_status(unreadable.value) == 502 + assert str(unreadable.value) == ( + "The simulation service answered that it started a new run of the " + f"report, but its answer could not be read. {READ_BEFORE_RUNNING_AGAIN}" + ) + + +def test_a_run_in_progress_is_a_conflict_naming_that_run() -> None: + service = ScriptedService( + status_code=409, + body={ + "code": "run_in_progress", + "detail": f"Report {REPORT_ID} has run {RUNNING_RUN_ID} running.", + "run_id": str(RUNNING_RUN_ID), + }, + ) + + with pytest.raises(SimulationServiceConflictError) as conflict: + service.rerun() + + assert conflict.value.run_id == RUNNING_RUN_ID + assert str(conflict.value) == ( + "The simulation service will not run the report again now " + f"(run_in_progress): Report {REPORT_ID} has run {RUNNING_RUN_ID} running." + ) + + +def test_a_report_the_service_does_not_run_again_is_a_conflict() -> None: + service = ScriptedService( + status_code=409, + body={"code": "rerun_not_supported", "detail": "The report has no dataset."}, + ) + + with pytest.raises(SimulationServiceConflictError) as conflict: + service.rerun() + + assert conflict.value.run_id is None + assert str(conflict.value) == ( + "The simulation service will not run the report again now " + "(rerun_not_supported): The report has no dataset." + ) + + +@pytest.mark.parametrize( + "body", + [b"conflict", {"detail": 3}, {"code": "run_in_progress"}], + ids=["not-json", "no-reason", "code-only"], +) +def test_a_conflict_that_cannot_be_read_is_unavailable(body: object) -> None: + # The service answers 409 only before writing anything. + with pytest.raises(SimulationServiceUnavailableError) as unavailable: + ScriptedService(status_code=409, body=body).rerun() + + assert route_status(unavailable.value) == 503 + assert str(unavailable.value) == "The simulation service answered 409 unreadably." + + +def test_stored_values_the_service_cannot_calculate_are_refused_with_its_reasons() -> ( + None +): + service = ScriptedService( + status_code=422, + body={ + "code": "missing_parameters", + "detail": [ + { + "loc": ["report", "reform", "policy_id", "gov.example.amount"], + "msg": "The catalog version has no parameter gov.example.amount", + } + ], + }, + ) + + with pytest.raises(SimulationServiceInvalidRequestError) as refused: + service.rerun() + + assert route_status(refused.value) == 422 + assert str(refused.value) == ( + "The simulation service cannot run the report again (missing_parameters): " + "report.reform.policy_id.gov.example.amount: The catalog version has no " + "parameter gov.example.amount" + ) + + +@pytest.mark.parametrize( + "code", ["unknown_region", "unknown_dataset", "unknown_dynamic"] +) +def test_a_stored_row_the_catalog_version_lacks_is_refused(code: str) -> None: + service = ScriptedService( + status_code=400, + body={ + "code": code, + "detail": "The catalog version lacks a row the report names.", + }, + ) + + with pytest.raises(SimulationServiceInvalidRequestError) as refused: + service.rerun() + + assert route_status(refused.value) == 422 + assert str(refused.value) == ( + f"The simulation service cannot run the report again ({code}): The " + "catalog version lacks a row the report names." + ) + + +def test_a_400_about_the_services_own_state_is_unavailable() -> None: + service = ScriptedService( + status_code=400, + body={ + "code": "unknown_catalog_version", + "detail": "No catalog version for .py 4.3.0.", + }, + ) + + with pytest.raises(SimulationServiceUnavailableError) as unavailable: + service.rerun() + + assert route_status(unavailable.value) == 503 + assert unavailable.value.created_record_id is None + assert str(unavailable.value) == ( + "The simulation service refused to run the report again " + "(unknown_catalog_version): No catalog version for .py 4.3.0." + ) + + +@pytest.mark.parametrize("status_code", [400, 422]) +def test_a_refusal_that_cannot_be_read_is_unavailable(status_code: int) -> None: + with pytest.raises(SimulationServiceUnavailableError) as unavailable: + ScriptedService(status_code=status_code, body=b"refused").rerun() + + assert str(unavailable.value) == ( + f"The simulation service answered {status_code} unreadably." + ) + + +@pytest.mark.parametrize( + ("status_code", "body", "status_and_code"), + [ + ( + 404, + {"code": "unknown_report", "detail": "No such report."}, + "404 unknown_report", + ), + (404, {"detail": "Not Found"}, "404"), + (401, b"", "401"), + (403, {"code": "forbidden", "detail": "No permission."}, "403 forbidden"), + (405, b"", "405"), + (413, b"", "413"), + (429, b"", "429"), + ], + ids=["unknown-report", "no-route", "401", "403", "405", "413", "429"], +) +def test_a_status_sent_before_the_service_writes_is_unavailable( + status_code: int, body: object, status_and_code: str +) -> None: + # The Public API has just read the report itself, so the service's 404 is + # its own state: a route it does not have yet, or a database it cannot read. + with pytest.raises(SimulationServiceUnavailableError) as unavailable: + ScriptedService(status_code=status_code, body=body).rerun() + + assert route_status(unavailable.value) == 503 + assert str(unavailable.value) == ( + f"The simulation service is unavailable ({status_and_code})." + ) + + +def test_the_services_unavailability_is_passed_on_with_its_code_not_its_reason() -> ( + None +): + service = ScriptedService( + status_code=503, + body={"code": "manifest_unavailable", "detail": "secret text"}, + ) + + with pytest.raises(SimulationServiceUnavailableError) as unavailable: + service.rerun() + + assert route_status(unavailable.value) == 503 + assert str(unavailable.value) == ( + "The simulation service is unavailable (503 manifest_unavailable)." + ) + + +def test_a_new_run_the_unavailable_service_failed_is_named() -> None: + service = ScriptedService( + status_code=503, + body={ + "code": "modal_call_refused", + "detail": "secret text", + "report_id": str(REPORT_ID), + "run_id": str(NEW_RUN_ID), + }, + ) + + with pytest.raises(SimulationServiceUnavailableError) as unavailable: + service.rerun() + + assert route_status(unavailable.value) == 503 + assert str(unavailable.value) == ( + f"The simulation service started run {NEW_RUN_ID} of report {REPORT_ID}, " + "then failed it, and is unavailable (503 modal_call_refused). Run the " + "report again later." + ) + + +def test_a_failed_new_run_the_503_does_not_name_is_still_the_reports() -> None: + service = ScriptedService( + status_code=503, + body={ + "code": "modal_call_refused", + "detail": "secret text", + "report_id": str(REPORT_ID), + }, + ) + + with pytest.raises(SimulationServiceUnavailableError) as unavailable: + service.rerun() + + assert str(unavailable.value) == ( + f"The simulation service started a new run of report {REPORT_ID}, then " + "failed it, and is unavailable (503 modal_call_refused). Run the report " + "again later." + ) + + +@pytest.mark.parametrize( + ("body", "status_and_code"), + [ + (b"Service Unavailable", "503"), + ( + { + "code": "modal_call_refused", + "detail": "x", + "report_id": str(OTHER_REPORT_ID), + }, + "503 modal_call_refused", + ), + ( + { + "code": "modal_call_refused", + "detail": "x", + "simulation_id": str(SIMULATION_ID), + }, + "503 modal_call_refused", + ), + ( + {"code": "modal_call_refused", "detail": "x", "report_id": "not-a-uuid"}, + "503 modal_call_refused", + ), + ], + ids=["not-json", "another-report", "a-simulation", "malformed-id"], +) +def test_a_503_that_cannot_be_read_or_names_another_record_may_have_started_a_run( + body: object, status_and_code: str +) -> None: + with pytest.raises(SimulationServiceAnswerUnreadableError) as unreadable: + ScriptedService(status_code=503, body=body).rerun() + + assert route_status(unreadable.value) == 502 + assert str(unreadable.value) == possibly_started_message(status_and_code) + + +def test_an_internal_error_may_have_started_a_run() -> None: + service = ScriptedService( + status_code=500, body={"code": "internal_error", "detail": "secret text"} + ) + + with pytest.raises(SimulationServiceInternalError) as failed: + service.rerun() + + assert route_status(failed.value) == 502 + assert str(failed.value) == possibly_started_message("500 internal_error") + + +@pytest.mark.parametrize( + ("status_code", "status_and_code"), + [(200, "200"), (204, "204"), (302, "302"), (418, "418")], +) +def test_an_unexpected_status_may_have_started_a_run( + status_code: int, status_and_code: str +) -> None: + with pytest.raises(SimulationServiceAnswerUnreadableError) as unreadable: + ScriptedService(status_code=status_code, body=b"").rerun() + + assert route_status(unreadable.value) == 502 + assert str(unreadable.value) == possibly_started_message(status_and_code) + + +@pytest.mark.parametrize("status_code", [502, 504]) +def test_a_proxys_answer_in_place_of_the_services_may_have_started_a_run( + status_code: int, +) -> None: + with pytest.raises(SimulationServiceAnswerUnreadableError) as lost: + ScriptedService(status_code=status_code, body=b"bad gateway").rerun() + + assert route_status(lost.value) == 502 + assert str(lost.value) == ( + "The simulation service may have received the request, but its answer " + f"was lost (a proxy answered {status_code} in its place). " + f"{READ_BEFORE_RUNNING_AGAIN}" + ) + + +def _client_raising( + error_type: type[httpx.TransportError], +) -> SimulationServiceReportClient: + def fail(request: httpx.Request) -> httpx.Response: + msg = "the transport failed" + raise error_type(msg, request=request) + + return SimulationServiceReportClient.from_environment( + transport=httpx.MockTransport(fail) + ) + + +@pytest.mark.parametrize( + "error_type", [httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout] +) +def test_a_connection_never_made_is_unavailable( + error_type: type[httpx.TransportError], +) -> None: + with pytest.raises( + SimulationServiceUnavailableError, match="could not be reached" + ) as unavailable: + _client_raising(error_type).rerun_report(rerun_input()) + + assert route_status(unavailable.value) == 503 + + +@pytest.mark.parametrize( + "error_type", [httpx.ReadTimeout, httpx.ReadError, httpx.RemoteProtocolError] +) +def test_an_answer_lost_after_sending_may_have_started_a_run( + error_type: type[httpx.TransportError], +) -> None: + with pytest.raises(SimulationServiceAnswerUnreadableError) as lost: + _client_raising(error_type).rerun_report(rerun_input()) + + assert route_status(lost.value) == 502 + assert str(lost.value) == ( + "The simulation service may have received the request, but its answer " + f"was lost ({error_type.__name__}). {READ_BEFORE_RUNNING_AGAIN}" + ) + + +def test_a_rerun_is_sent_through_its_one_replaceable_send_function() -> None: + sent: list[tuple[UUID, dict[str, Any]]] = [] + + def send_rerun( + http_client: httpx.Client, report_id: UUID, body: dict[str, Any] + ) -> SimulationAnswer: + assert str(http_client.base_url).rstrip("/") == SIMULATION_URL + sent.append((report_id, body)) + return SimulationAnswer( + status_code=201, content=json.dumps(RERUN_STARTED).encode() + ) + + client = SimulationServiceReportClient( + httpx.Client(base_url=SIMULATION_URL), send_report_rerun=send_rerun + ) + + client.rerun_report(rerun_input(force=True)) + + assert sent == [(REPORT_ID, {"force": True})] diff --git a/tests/unit/test_alembic_workflows.py b/tests/unit/test_alembic_workflows.py index 13d70219a..494f62a1c 100644 --- a/tests/unit/test_alembic_workflows.py +++ b/tests/unit/test_alembic_workflows.py @@ -247,6 +247,7 @@ def test_reusable_v2_integration_check_uses_disposable_databases_and_coverage(): assert "test_v2_report_creation.py" in workflow assert "test_v2_budget_window_report_reads.py" in workflow assert "test_v2_budget_window_report_creation.py" in workflow + assert "test_v2_report_runs.py" in workflow assert "test_v2_saved_detailed_save_records.py" in workflow assert "test_v2_saved_budget_window_reports.py" in workflow assert "test_v2_policy_persistence.py" in workflow diff --git a/tests/unit/test_migration_contract_artifacts.py b/tests/unit/test_migration_contract_artifacts.py index e8f5bcc1a..7c9a2694e 100644 --- a/tests/unit/test_migration_contract_artifacts.py +++ b/tests/unit/test_migration_contract_artifacts.py @@ -14,7 +14,7 @@ def test_migration_contract_payload_summarizes_route_contracts(): assert payload["metadata"] == { "route_group_count": 9, "workflow_count": 18, - "request_count": 97, + "request_count": 99, "db_entity_count": 6, "sim_flow_count": 3, } diff --git a/tests/unit/v2/report_run_support.py b/tests/unit/v2/report_run_support.py new file mode 100644 index 000000000..2988ef581 --- /dev/null +++ b/tests/unit/v2/report_run_support.py @@ -0,0 +1,127 @@ +"""A stand-in report service that also runs reports again and lists their runs.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import timedelta +from typing import TYPE_CHECKING +from uuid import UUID + +from policyengine_api.services.v2.stored_reports.types import ( + BudgetWindowRunsRead, + ReportRunsRead, +) +from tests.unit.v2.budget_window_report_support import ( + BUDGET_WINDOW_REPORT_ID, + budget_window_run_read, + budget_window_year_read, +) +from tests.unit.v2.stored_report_support import ( + CREATED_AT, + CURRENT_VERSIONS, + REPORT_ID, + FakeReportService, + report_run_read, +) + +if TYPE_CHECKING: + from policyengine_api.services.v2.stored_reports.types import ( + StoredReportRead, + StoredReportRunsRead, + ) + +NEW_RUN_ID = UUID("00000000-0000-0000-0000-00000000a00b") +NEW_WINDOW_RUN_ID = UUID("00000000-0000-0000-0000-00000000a10b") + + +def report_runs_read() -> ReportRunsRead: + """Return a report's two runs, newest first: a re-run on the current versions.""" + return ReportRunsRead( + report_id=REPORT_ID, + kind="economy", + runs=( + report_run_read( + id=NEW_RUN_ID, + status="pending", + trigger="manual", + country_package_version=CURRENT_VERSIONS.country_package_version, + policyengine_version=CURRENT_VERSIONS.policyengine_version, + created_at=CREATED_AT + timedelta(days=3), + started_at=None, + completed_at=None, + ), + report_run_read(), + ), + current_versions=CURRENT_VERSIONS, + ) + + +def budget_window_runs_read() -> BudgetWindowRunsRead: + """Return a budget window's two runs, newest first: a forced re-run, pending. + + The re-run's years are on the current versions, which are the window + run's, as its first year's always are. + """ + rerun_at = CREATED_AT + timedelta(days=3) + current = { + "country_package_version": CURRENT_VERSIONS.country_package_version, + "policyengine_version": CURRENT_VERSIONS.policyengine_version, + } + return BudgetWindowRunsRead( + report_id=BUDGET_WINDOW_REPORT_ID, + kind="budget_window", + runs=( + budget_window_run_read( + id=NEW_WINDOW_RUN_ID, + status="pending", + trigger="manual", + recalculate=True, + created_at=rerun_at, + started_at=None, + completed_at=None, + years=tuple( + budget_window_year_read( + year, + id=UUID(int=year), + baseline_simulation_run_id=UUID(int=year * 2 + 10_000), + reform_simulation_run_id=UUID(int=year * 2 + 10_001), + status="pending", + trigger="manual", + created_at=rerun_at, + started_at=None, + completed_at=None, + **current, + ) + for year in (2026, 2027) + ), + **current, + ), + budget_window_run_read(), + ), + current_versions=CURRENT_VERSIONS, + ) + + +@dataclass +class FakeReportRunService(FakeReportService): + """Answer with ``report`` or ``runs``, or raise ``error``, recording each call. + + A re-run answers ``rerun`` when it is set, and ``report`` otherwise. + """ + + runs: StoredReportRunsRead = field(default_factory=report_runs_read) + rerun: StoredReportRead | None = None + + def rerun_report( + self, *, country_id: str, report_id: UUID, force: bool + ) -> StoredReportRead: + self._answer( + "rerun_report", country_id=country_id, report_id=report_id, force=force + ) + return self.report if self.rerun is None else self.rerun + + def list_report_runs( + self, *, country_id: str, report_id: UUID + ) -> StoredReportRunsRead: + self._answer("list_report_runs", country_id=country_id, report_id=report_id) + return self.runs diff --git a/tests/unit/v2/test_metadata_routes.py b/tests/unit/v2/test_metadata_routes.py index c126e70b7..8e1cef0c5 100644 --- a/tests/unit/v2/test_metadata_routes.py +++ b/tests/unit/v2/test_metadata_routes.py @@ -339,6 +339,7 @@ def test_openapi_references_explicit_resource_response_schemas() -> None: "/v2/policies/{policy_id}", "/v2/reports", "/v2/reports/{report_id}", + "/v2/reports/{report_id}/runs", "/v2/simulations", "/v2/simulations/{simulation_id}", "/v2/user-households", diff --git a/tests/unit/v2/test_report_rerun_service.py b/tests/unit/v2/test_report_rerun_service.py new file mode 100644 index 000000000..91cb50571 --- /dev/null +++ b/tests/unit/v2/test_report_rerun_service.py @@ -0,0 +1,202 @@ +"""Run reports again and list their runs in the report service, without a database.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import TYPE_CHECKING, get_args +from uuid import UUID, uuid4 + +import pytest +from sqlalchemy.exc import OperationalError + +from policyengine_api.data.v2.models import ( + BudgetWindowRun, + Report, + ReportRunStatus, + ReportRunTrigger, +) +from policyengine_api.services.v2.simulation_service.types import PackageVersions +from policyengine_api.services.v2.stored_reports.services import ( + SubmissionsNotConfiguredError, + V2StoredReportService, +) +from policyengine_api.services.v2.stored_reports.transformations import ( + budget_window_runs_read, +) +from policyengine_api.services.v2.stored_reports.types import ( + ReportKind, + RunRouteKind, + StoredReportRows, +) +from policyengine_api.services.v2.stored_reports.validators import ( + REPORT_KIND_NAMES, + ReportKindUnsupportedError, + ReportStorageError, + require_run_route_kind, +) + +if TYPE_CHECKING: + from contextlib import AbstractContextManager + + from sqlmodel import Session + + from policyengine_api.services.v2.simulation_service.types import ( + ReportRerunInput, + ) + +REPORT_ID = UUID("00000000-0000-0000-0000-00000000f001") +STORED_AT = datetime(2026, 11, 13, 9, tzinfo=UTC) +CURRENT_VERSIONS = PackageVersions( + country_package_version="1.5.0", policyengine_version="4.2.0" +) + + +@dataclass +class FailingDatabaseSession: + """A database that cannot be read: opening a session fails.""" + + error: Exception = field( + default_factory=lambda: OperationalError( + "SELECT 1", {}, Exception("connection refused") + ) + ) + opened: int = 0 + + def read(self) -> AbstractContextManager[Session]: + self.opened += 1 + raise self.error + + +@dataclass +class RecordingSimulationService: + reruns: list[ReportRerunInput] = field(default_factory=list) + + def rerun_report(self, rerun: ReportRerunInput) -> None: + self.reruns.append(rerun) + + +def stored_rows(**report_columns: object) -> StoredReportRows: + report = Report( + id=REPORT_ID, + label="Reform", + country="us", + tax_benefit_model_id=uuid4(), + year=2026, + inputs={}, + **report_columns, + ) + return StoredReportRows( + report=report, + region_code="us", + dataset_name=None, + baseline_simulation=None, + reform_simulation=None, + newest_run=None, + newest_budget_window_run=None, + budget_window_years=(), + ) + + +def current_versions(country_id: str) -> PackageVersions: + return CURRENT_VERSIONS + + +def test_a_rerun_without_the_simulation_service_reads_nothing() -> None: + database_session = FailingDatabaseSession() + service = V2StoredReportService( + database_session, + current_versions=current_versions, + ) + + with pytest.raises(SubmissionsNotConfiguredError): + service.rerun_report(country_id="us", report_id=REPORT_ID, force=False) + + assert database_session.opened == 0 + + +def test_a_rerun_whose_report_cannot_be_read_is_storage_unavailable() -> None: + database_session = FailingDatabaseSession() + simulation_service = RecordingSimulationService() + service = V2StoredReportService( + database_session, + current_versions=current_versions, + submissions=simulation_service, + ) + + with pytest.raises(ReportStorageError) as unavailable: + service.rerun_report(country_id="us", report_id=REPORT_ID, force=True) + + assert unavailable.value.__cause__ is database_session.error + assert simulation_service.reruns == [] + + +def test_a_run_list_whose_report_cannot_be_read_is_storage_unavailable() -> None: + database_session = FailingDatabaseSession() + service = V2StoredReportService( + database_session, + current_versions=current_versions, + ) + + with pytest.raises(ReportStorageError) as unavailable: + service.list_report_runs(country_id="us", report_id=REPORT_ID) + + assert unavailable.value.__cause__ is database_session.error + + +# The report columns that make a stored report of each kind. +KIND_COLUMNS: dict[ReportKind, dict[str, object]] = { + "economy": {}, + "budget_window": {"budget_window_size": 3}, + "household": {"household_id": UUID("00000000-0000-0000-0000-00000000f002")}, +} + + +@pytest.mark.parametrize("kind", get_args(RunRouteKind)) +def test_each_run_route_kinds_runs_can_be_run_and_listed(kind: ReportKind) -> None: + assert require_run_route_kind(stored_rows(**KIND_COLUMNS[kind])) == kind + + +def test_a_household_reports_runs_are_not_run_or_listed_yet() -> None: + with pytest.raises(ReportKindUnsupportedError) as unsupported: + require_run_route_kind(stored_rows(household_id=uuid4())) + + assert str(unsupported.value) == ( + f"Report {REPORT_ID} is a household report, whose runs these routes do not " + "start or list yet" + ) + + +def test_every_report_kind_has_a_name_and_the_run_routes_take_only_kinds() -> None: + # A kind added later is refused with its name (422), not a KeyError (500). + assert set(REPORT_KIND_NAMES) == set(get_args(ReportKind)) + assert set(KIND_COLUMNS) == set(get_args(ReportKind)) + assert set(get_args(RunRouteKind)).issubset(get_args(ReportKind)) + + +def test_a_window_run_with_no_year_lists_no_years_and_no_versions() -> None: + window_run = BudgetWindowRun( + id=uuid4(), + created_at=STORED_AT, + updated_at=STORED_AT, + report_id=REPORT_ID, + tax_benefit_model_version_id=uuid4(), + trigger=ReportRunTrigger.MANUAL, + recalculate=True, + status=ReportRunStatus.PENDING, + ) + + runs = budget_window_runs_read( + stored_rows(budget_window_size=2), + [window_run], + {}, + current_versions=CURRENT_VERSIONS, + ) + + assert runs.kind == "budget_window" + (run,) = runs.runs + assert run.id == window_run.id + assert (run.trigger, run.recalculate, run.status) == ("manual", True, "pending") + assert run.years == () + assert run.country_package_version is None + assert run.policyengine_version is None diff --git a/tests/unit/v2/test_report_rerun_submissions.py b/tests/unit/v2/test_report_rerun_submissions.py new file mode 100644 index 000000000..b9dff2e7b --- /dev/null +++ b/tests/unit/v2/test_report_rerun_submissions.py @@ -0,0 +1,161 @@ +"""Check a stored report's country and policies before running it again.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest +from sqlalchemy.exc import OperationalError + +from policyengine_api.services.v2.countries.types import CountryRead +from policyengine_api.services.v2.policies.types import CheckedPolicyRead +from policyengine_api.services.v2.policies.validators import ( + PolicyNotFoundError, + PolicyParametersMissingError, +) +from policyengine_api.services.v2.simulation_service.services import ( + SimulationSubmissionService, +) +from policyengine_api.services.v2.simulation_service.types import ReportRerunInput +from policyengine_api.services.v2.simulation_service.validators import ( + EconomyCalculationsUnsupportedError, + SubmissionStorageError, +) + +REPORT_ID = UUID("00000000-0000-0000-0000-00000000e001") +STORED_POLICY_ID = UUID("00000000-0000-0000-0000-00000000e002") +OUTDATED_POLICY_ID = UUID("00000000-0000-0000-0000-00000000e003") +REMOVED_PARAMETERS = ("gov.example.removed_amount",) +# Each stored policy's parameters the running catalog version lacks. +MISSING_PARAMETERS: dict[UUID, tuple[str, ...]] = { + STORED_POLICY_ID: (), + OUTDATED_POLICY_ID: REMOVED_PARAMETERS, +} +STORED_AT = datetime(2026, 11, 13, tzinfo=UTC) + + +@dataclass +class StoredCountries: + society_wide_enabled: bool = True + + def list_countries(self) -> list[CountryRead]: + return [ + CountryRead( + id="us", + household_enabled=True, + society_wide_enabled=self.society_wide_enabled, + ) + ] + + +@dataclass +class StoredPolicies: + """The policies in MISSING_PARAMETERS, stored in every country.""" + + error: Exception | None = None + asked: list[UUID] = field(default_factory=list) + + def get_policy(self, *, country_id: str, policy_id: UUID) -> CheckedPolicyRead: + self.asked.append(policy_id) + if self.error is not None: + raise self.error + if policy_id not in MISSING_PARAMETERS: + msg = f"Policy {policy_id} was not found" + raise PolicyNotFoundError(msg) + return CheckedPolicyRead( + id=policy_id, + country_id=country_id, + tax_benefit_model_id=uuid4(), + tax_benefit_model_version_id=uuid4(), + created_at=STORED_AT, + updated_at=STORED_AT, + parameter_values=(), + missing_parameters=MISSING_PARAMETERS[policy_id], + ) + + +@dataclass +class RecordingSimulationService: + reruns: list[ReportRerunInput] = field(default_factory=list) + + def rerun_report(self, rerun: ReportRerunInput) -> None: + self.reruns.append(rerun) + + +def unconfigured_client() -> RecordingSimulationService: + msg = "the simulation service must not be asked" + raise AssertionError(msg) + + +def rerun_input(*policy_ids: UUID | None) -> ReportRerunInput: + return ReportRerunInput( + country_id="us", report_id=REPORT_ID, force=True, policy_ids=policy_ids + ) + + +def test_a_rerun_is_sent_once_its_country_and_policies_are_checked() -> None: + client = RecordingSimulationService() + policies = StoredPolicies() + service = SimulationSubmissionService( + countries=StoredCountries(), policies=policies, client_factory=lambda: client + ) + rerun = rerun_input(None, STORED_POLICY_ID) + + service.rerun_report(rerun) + + assert client.reruns == [rerun] + assert policies.asked == [STORED_POLICY_ID] + + +def test_a_rerun_in_a_household_only_country_is_refused_before_sending() -> None: + service = SimulationSubmissionService( + countries=StoredCountries(society_wide_enabled=False), + policies=StoredPolicies(), + client_factory=unconfigured_client, + ) + + with pytest.raises(EconomyCalculationsUnsupportedError, match="'us'"): + service.rerun_report(rerun_input(None, STORED_POLICY_ID)) + + +def test_a_rerun_naming_a_policy_no_longer_stored_is_refused() -> None: + missing_policy_id = uuid4() + service = SimulationSubmissionService( + countries=StoredCountries(), + policies=StoredPolicies(), + client_factory=unconfigured_client, + ) + + with pytest.raises(PolicyNotFoundError, match=str(missing_policy_id)): + service.rerun_report(rerun_input(None, missing_policy_id)) + + +def test_a_rerun_whose_policy_sets_parameters_the_catalog_lacks_is_refused() -> None: + service = SimulationSubmissionService( + countries=StoredCountries(), + policies=StoredPolicies(), + client_factory=unconfigured_client, + ) + + with pytest.raises(PolicyParametersMissingError) as refusal: + service.rerun_report(rerun_input(STORED_POLICY_ID, OUTDATED_POLICY_ID)) + + assert refusal.value.missing_parameters == REMOVED_PARAMETERS + + +def test_a_rerun_whose_policy_check_cannot_read_storage_is_storage_unavailable() -> ( + None +): + error = OperationalError("SELECT 1", {}, Exception("connection refused")) + service = SimulationSubmissionService( + countries=StoredCountries(), + policies=StoredPolicies(error=error), + client_factory=unconfigured_client, + ) + + with pytest.raises(SubmissionStorageError) as unavailable: + service.rerun_report(rerun_input(None, STORED_POLICY_ID)) + + assert unavailable.value.__cause__ is error diff --git a/tests/unit/v2/test_report_run_routes.py b/tests/unit/v2/test_report_run_routes.py new file mode 100644 index 000000000..d0c03e2b4 --- /dev/null +++ b/tests/unit/v2/test_report_run_routes.py @@ -0,0 +1,443 @@ +"""Run v2 reports again and list their runs over stand-in services.""" + +from __future__ import annotations + +from dataclasses import replace +from uuid import UUID + +import pytest + +from policyengine_api.services.v2.policies.validators import ( + PolicyNotFoundError, + PolicyParametersMissingError, +) +from policyengine_api.services.v2.simulation_service.types import ( + SimulationServiceAnswerUnreadableError, + SimulationServiceConflictError, + SimulationServiceInternalError, + SimulationServiceInvalidRequestError, + SimulationServiceUnavailableError, +) +from policyengine_api.services.v2.simulation_service.validators import ( + EconomyCalculationsUnsupportedError, + SubmissionStorageError, +) +from policyengine_api.services.v2.stored_reports.validators import ( + ReportKindUnsupportedError, + ReportStorageError, + RerunReportUnreadableError, + StoredReportNotFoundError, +) +from tests.unit.v2.budget_window_report_support import ( + BUDGET_WINDOW_REPORT_ID, + budget_window_report_read, +) +from tests.unit.v2.report_run_support import ( + NEW_RUN_ID, + NEW_WINDOW_RUN_ID, + FakeReportRunService, + budget_window_runs_read, +) +from tests.unit.v2.stored_report_support import ( + POLICY_ID, + REPORT_ID, + REPORT_RUN_ID, + report_route_client, + report_run_read, +) + +RUNS_PATH = f"/v2/reports/{REPORT_ID}/runs?country_id=us" +RUNNING_RUN_ID = UUID("00000000-0000-0000-0000-00000000a00c") + + +def test_a_reports_runs_are_answered_newest_first_with_versions_and_current_ones() -> ( + None +): + service = FakeReportRunService() + + response = report_route_client(reports=service).get(RUNS_PATH) + + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ok" + assert body["message"] is None + result = body["result"] + assert result["report_id"] == str(REPORT_ID) + assert result["kind"] == "economy" + assert [run["id"] for run in result["items"]] == [ + str(NEW_RUN_ID), + str(REPORT_RUN_ID), + ] + assert [run["trigger"] for run in result["items"]] == ["manual", "initial"] + assert [ + (run["country_package_version"], run["policyengine_version"]) + for run in result["items"] + ] == [("1.5.0", "4.2.0"), ("1.4.0", "4.1.0")] + assert result["current_versions"] == { + "country_package_version": "1.5.0", + "policyengine_version": "4.2.0", + } + assert service.calls == [ + ("list_report_runs", {"country_id": "us", "report_id": REPORT_ID}) + ] + + +def test_a_budget_windows_runs_are_answered_with_their_years() -> None: + service = FakeReportRunService(runs=budget_window_runs_read()) + + response = report_route_client(reports=service).get( + f"/v2/reports/{BUDGET_WINDOW_REPORT_ID}/runs?country_id=us" + ) + + assert response.status_code == 200, response.text + result = response.json()["result"] + assert result["report_id"] == str(BUDGET_WINDOW_REPORT_ID) + assert result["kind"] == "budget_window" + newest, first = result["items"] + assert newest["id"] == str(NEW_WINDOW_RUN_ID) + assert (newest["trigger"], newest["recalculate"]) == ("manual", True) + assert newest["policyengine_version"] == "4.2.0" + assert [year["year"] for year in first["years"]] == [2026, 2027] + assert first["years"][0]["country_package_version"] == "1.4.0" + assert result["current_versions"]["policyengine_version"] == "4.2.0" + + +def test_running_a_budget_window_again_answers_the_window_with_its_new_run() -> None: + window = budget_window_report_read(status="pending", result=None) + service = FakeReportRunService(rerun=window) + + response = report_route_client(reports=service).post( + f"/v2/reports/{BUDGET_WINDOW_REPORT_ID}/runs?country_id=us", + json={"force": True}, + ) + + assert response.status_code == 201, response.text + item = response.json()["result"]["item"] + assert item["kind"] == "budget_window" + assert item["budget_window_size"] == 2 + assert [year["year"] for year in item["run"]["years"]] == [2026, 2027] + assert item["result"] is None + + +def test_a_report_with_no_run_lists_none() -> None: + service = FakeReportRunService() + service.runs = replace(service.runs, runs=()) + + response = report_route_client(reports=service).get(RUNS_PATH) + + assert response.status_code == 200 + assert response.json()["result"]["items"] == [] + + +@pytest.mark.parametrize( + ("error", "status_code", "message"), + [ + ( + StoredReportNotFoundError("Report x was not found in country 'us'"), + 404, + "Report x was not found in country 'us'", + ), + ( + ReportKindUnsupportedError("Report x is a budget window"), + 422, + "Report x is a budget window", + ), + ( + ReportStorageError("V2 report storage is unavailable"), + 503, + "V2 report storage is unavailable", + ), + (RuntimeError("unexpected"), 500, "V2 report operation failed"), + ], + ids=["not-found", "other-kind", "storage", "unexpected"], +) +def test_run_list_failures_answer_typed_errors( + error: Exception, status_code: int, message: str +) -> None: + client = report_route_client(reports=FakeReportRunService(error=error)) + + response = client.get(RUNS_PATH) + + assert response.status_code == status_code + assert response.json() == {"status": "error", "message": message} + + +@pytest.mark.parametrize( + "query", + ["", "?country_id=us&unknown=1", "?country_id=us&country_id=uk"], + ids=["no-country", "unknown-parameter", "repeated-country"], +) +def test_a_run_list_refuses_a_malformed_query(query: str) -> None: + service = FakeReportRunService() + + response = report_route_client(reports=service).get( + f"/v2/reports/{REPORT_ID}/runs{query}" + ) + + assert response.status_code == 422 + assert service.calls == [] + + +@pytest.mark.parametrize( + ("body", "force"), + [({}, False), ({"force": False}, False), ({"force": True}, True)], + ids=["empty", "reuse", "force"], +) +def test_running_a_report_again_answers_the_report_with_its_new_run( + body: dict[str, bool], + force: bool, # noqa: FBT001 # pytest passes parameters positionally +) -> None: + service = FakeReportRunService() + service.report = replace( + service.report, + status="pending", + run=report_run_read(id=NEW_RUN_ID, status="pending", trigger="manual"), + result=None, + ) + + response = report_route_client(reports=service).post(RUNS_PATH, json=body) + + assert response.status_code == 201 + item = response.json()["result"]["item"] + assert item["id"] == str(REPORT_ID) + assert item["status"] == "pending" + assert item["run"]["id"] == str(NEW_RUN_ID) + assert item["run"]["trigger"] == "manual" + assert item["result"] is None + assert item["current_versions"]["policyengine_version"] == "4.2.0" + assert service.calls == [ + ("rerun_report", {"country_id": "us", "report_id": REPORT_ID, "force": force}) + ] + + +@pytest.mark.parametrize( + "body", + [ + {"force": "yes"}, + {"force": 1}, + {"force": None}, + {"force": True, "country_id": "us"}, + [], + "force", + ], + ids=["text", "number", "null", "extra-field", "list", "string"], +) +def test_a_rerun_refuses_a_malformed_body(body: object) -> None: + service = FakeReportRunService() + + response = report_route_client(reports=service).post(RUNS_PATH, json=body) + + assert response.status_code == 422 + assert service.calls == [] + + +def test_a_rerun_needs_a_body() -> None: + service = FakeReportRunService() + + response = report_route_client(reports=service).post(RUNS_PATH) + + assert response.status_code == 422 + assert service.calls == [] + + +def test_a_rerun_needs_a_country() -> None: + service = FakeReportRunService() + + response = report_route_client(reports=service).post( + f"/v2/reports/{REPORT_ID}/runs", json={} + ) + + assert response.status_code == 422 + assert service.calls == [] + + +def test_a_rerun_body_over_16_kib_is_refused() -> None: + service = FakeReportRunService() + + response = report_route_client(reports=service).post( + RUNS_PATH, json={"force": True, "padding": "x" * (17 * 1024)} + ) + + assert response.status_code == 413 + assert service.calls == [] + + +@pytest.mark.parametrize( + ("run_id", "answered_run_id"), + [(RUNNING_RUN_ID, str(RUNNING_RUN_ID)), (None, None)], + ids=["run-in-progress", "not-run-again"], +) +def test_a_report_the_service_will_not_run_again_now_answers_409( + run_id: UUID | None, answered_run_id: str | None +) -> None: + message = ( + "The simulation service will not run the report again now " + "(run_in_progress): Report r has run x running." + ) + service = FakeReportRunService( + error=SimulationServiceConflictError(message, run_id=run_id) + ) + + response = report_route_client(reports=service).post(RUNS_PATH, json={}) + + assert response.status_code == 409 + assert response.json() == { + "status": "error", + "message": message, + "run_id": answered_run_id, + } + + +def test_a_new_run_whose_report_cannot_be_read_back_answers_the_reports_id() -> None: + service = FakeReportRunService( + error=RerunReportUnreadableError(report_id=REPORT_ID, country_id="us") + ) + + response = report_route_client(reports=service).post(RUNS_PATH, json={}) + + assert response.status_code == 502 + assert response.json() == { + "status": "error", + "message": ( + f"The simulation service started a new run of report {REPORT_ID}, " + "but the report could not be read back. Read it with GET " + f"/v2/reports/{REPORT_ID}?country_id=us rather than running it again." + ), + "report_id": str(REPORT_ID), + } + + +@pytest.mark.parametrize( + ("error", "status_code", "message"), + [ + ( + StoredReportNotFoundError("Report x was not found in country 'us'"), + 404, + "Report x was not found in country 'us'", + ), + ( + PolicyNotFoundError("Policy p was not found in country 'us'"), + 404, + "Policy p was not found in country 'us'", + ), + ( + ReportKindUnsupportedError("Report x is a household report"), + 422, + "Report x is a household report", + ), + ( + EconomyCalculationsUnsupportedError("No economy calculations for 'uk'"), + 422, + "No economy calculations for 'uk'", + ), + ( + SimulationServiceInvalidRequestError( + "The simulation service cannot run the report again " + "(unknown_region): no region state/ca" + ), + 422, + "The simulation service cannot run the report again " + "(unknown_region): no region state/ca", + ), + ( + SimulationServiceInternalError("may have started a new run"), + 502, + "may have started a new run", + ), + ( + SimulationServiceAnswerUnreadableError("its answer was lost"), + 502, + "its answer was lost", + ), + ( + SimulationServiceUnavailableError( + "The simulation service is unavailable (503)." + ), + 503, + "The simulation service is unavailable (503).", + ), + ( + SubmissionStorageError("V2 storage is unavailable to check"), + 503, + "V2 storage is unavailable", + ), + ( + ReportStorageError("V2 report storage is unavailable"), + 503, + "V2 report storage is unavailable", + ), + (RuntimeError("unexpected"), 500, "V2 report operation failed"), + ], + ids=[ + "report-not-found", + "policy-not-found", + "other-kind", + "household-only-country", + "service-refusal", + "service-failure", + "service-answer-lost", + "service-unavailable", + "check-storage", + "report-storage", + "unexpected", + ], +) +def test_rerun_failures_answer_typed_errors( + error: Exception, status_code: int, message: str +) -> None: + client = report_route_client(reports=FakeReportRunService(error=error)) + + response = client.post(RUNS_PATH, json={"force": True}) + + assert response.status_code == status_code + assert response.json() == {"status": "error", "message": message} + + +def test_a_rerun_of_a_policy_setting_missing_parameters_answers_422() -> None: + error = PolicyParametersMissingError(POLICY_ID, ("gov.example.removed_amount",)) + client = report_route_client(reports=FakeReportRunService(error=error)) + + response = client.post(RUNS_PATH, json={}) + + assert response.status_code == 422 + assert response.json() == { + "status": "error", + "message": str(error), + "code": "missing_parameters", + "missing_parameters": ["gov.example.removed_amount"], + } + + +def test_the_openapi_document_describes_both_run_routes() -> None: + document = report_route_client().get("/v2/openapi.json").json() + + runs = document["paths"]["/v2/reports/{report_id}/runs"] + assert set(runs) == {"get", "post"} + assert {"200", "404", "422", "500", "503"} <= set(runs["get"]["responses"]) + assert {"201", "404", "409", "413", "422", "500", "502", "503"} <= set( + runs["post"]["responses"] + ) + conflict_schema = runs["post"]["responses"]["409"]["content"]["application/json"][ + "schema" + ] + assert conflict_schema == { + "$ref": "#/components/schemas/ReportRerunConflictResponse" + } + assert {"$ref": "#/components/schemas/RerunReportUnreadableResponse"} in runs[ + "post" + ]["responses"]["502"]["content"]["application/json"]["schema"]["anyOf"] + assert {"$ref": "#/components/schemas/V2MissingParametersErrorResponse"} in runs[ + "post" + ]["responses"]["422"]["content"]["application/json"]["schema"]["anyOf"] + + +def test_the_openapi_document_gives_each_kinds_run_list() -> None: + document = report_route_client().get("/v2/openapi.json").json() + + schemas = document["components"]["schemas"] + result = schemas["ReportRunListResponse"]["properties"]["result"] + assert result["discriminator"]["propertyName"] == "kind" + assert {option["$ref"] for option in result["oneOf"]} == { + "#/components/schemas/ReportRunListResult", + "#/components/schemas/BudgetWindowRunListResult", + }