from __future__ import annotations from datetime import UTC, datetime, timedelta from types import SimpleNamespace from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.core.reporting import ( REPORT_PROVIDER_CAPABILITY_PREFIX, REPORT_PROVIDER_CONTRACT_VERSION, ReportDescriptor, ReportParameterDescriptor, ReportParameterOption, ReportPrivacyTransform, ReportProviderRequest, ReportProviderResult, ReportResultField, ) from govoplan_core.db.base import Base from govoplan_reporting.backend.db.models import ( ReportingProviderExecution, ReportingProviderExport, ) from govoplan_reporting.backend.provider_reports import ( ProviderReportError, execute_provider_report, export_provider_execution, list_provider_reports, ) from govoplan_reporting.backend.retention import ReportingRetentionService class _Principal: tenant_id = "tenant-1" user = SimpleNamespace(id="analyst-1") api_key = None class _Provider: provider_id = "example" contract_version = REPORT_PROVIDER_CONTRACT_VERSION def __init__(self) -> None: self.result_access = True def list_reports(self, session, principal): del session, principal return ( ReportDescriptor( provider_id=self.provider_id, report_id="summary", revision="example.summary.v1", title="Example summary", summary="A minimized example report.", parameters=( ReportParameterDescriptor( key="source_id", label="Source", type="reference", required=True, options_from_provider=True, ), ), result_schema=( ReportResultField( path="metric", label="Metric", type="integer", group="Outcome", ), ), privacy_transforms=( ReportPrivacyTransform(id="aggregate", label="Aggregate"), ), export_formats=("json",), reidentification_risk="low", ), ) def parameter_options(self, session, principal, **kwargs): del session, principal, kwargs return (ReportParameterOption(value="source-1", label="Source 1"),) def execute_report( self, session, principal, *, request: ReportProviderRequest, ): del session return ReportProviderResult( report_id=request.report_id, generated_at=datetime.now(UTC), payload={"metric": 12}, source_revisions=( { "module_id": "example", "resource_id": request.parameters["source_id"], "revision_id": "7", }, ), effective_scope={ "tenant_id": principal.tenant_id, "source_id": request.parameters["source_id"], }, applied_privacy_transforms=("aggregate",), provenance={"executor": "example-v1"}, ) def authorize_result(self, session, principal, **kwargs): del session, principal, kwargs return self.result_access class _Registry: def __init__(self, provider=None): self.provider = provider or _Provider() def capability_names(self): return (REPORT_PROVIDER_CAPABILITY_PREFIX + "example",) def has_capability(self, name): return name == REPORT_PROVIDER_CAPABILITY_PREFIX + "example" def capability(self, name): return self.provider if self.has_capability(name) else None def test_provider_report_execution_is_governed_replayable_and_audited() -> None: engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) principal = _Principal() registry = _Registry() with Session(engine, expire_on_commit=False) as session: catalogue = list_provider_reports(session, principal, registry=registry) assert catalogue["reports"][0]["available"] is True execution = execute_provider_report( session, principal, registry=registry, provider_id="example", report_id="summary", parameters={"source_id": "source-1"}, purpose="Operational overview", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, idempotency_key="provider-run-1", ) replay = execute_provider_report( session, principal, registry=registry, provider_id="example", report_id="summary", parameters={"source_id": "source-1"}, purpose="Operational overview", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, idempotency_key="provider-run-1", ) assert replay["execution_id"] == execution["execution_id"] assert execution["source_revisions"][0]["revision_id"] == "7" assert execution["privacy_transforms"] == ["aggregate"] assert execution["retention_days"] == 30 content, media_type, filename = export_provider_execution( session, principal, registry=registry, execution_id=str(execution["execution_id"]), format="json", purpose="Archive approved aggregate", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, ) assert b'"metric": 12' in content assert media_type == "application/json" assert filename.endswith(".json") assert session.query(ReportingProviderExport).count() == 1 engine.dispose() def test_provider_result_missing_a_required_transform_is_rejected() -> None: class _UnsafeProvider(_Provider): def execute_report(self, session, principal, *, request): result = super().execute_report( session, principal, request=request, ) return ReportProviderResult( report_id=result.report_id, generated_at=result.generated_at, payload=result.payload, source_revisions=result.source_revisions, effective_scope=result.effective_scope, applied_privacy_transforms=(), provenance=result.provenance, ) engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as session: try: execute_provider_report( session, _Principal(), registry=_Registry(_UnsafeProvider()), provider_id="example", report_id="summary", parameters={"source_id": "source-1"}, purpose="Operational overview", audience_scope={"scope_type": "tenant"}, idempotency_key="unsafe-provider-run", ) except ProviderReportError as exc: assert "privacy transformations" in str(exc) else: raise AssertionError("unsafe provider report was accepted") engine.dispose() def test_expired_provider_result_is_minimized_but_evidence_is_retained() -> None: engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine, expire_on_commit=False) as session: execution = execute_provider_report( session, _Principal(), registry=_Registry(), provider_id="example", report_id="summary", parameters={"source_id": "source-1"}, purpose="Operational overview", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, idempotency_key="expired-provider-run", ) row = session.query(ReportingProviderExecution).one() original_hash = row.output_hash row.expires_at = datetime.now(UTC) - timedelta(seconds=1) session.flush() preview = ReportingRetentionService().apply_retention( session, dry_run=True, now=datetime.now(UTC), ) assert preview["eligible"] == 1 assert row.result_payload == {"metric": 12} applied = ReportingRetentionService().apply_retention( session, dry_run=False, now=datetime.now(UTC), ) assert applied["redacted"] == 1 assert row.result_payload == {} assert row.retention_redacted_at is not None assert row.output_hash == original_hash == execution["output_hash"] assert row.source_revisions[0]["revision_id"] == "7" try: execute_provider_report( session, _Principal(), registry=_Registry(), provider_id="example", report_id="summary", parameters={"source_id": "source-1"}, purpose="Operational overview", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, idempotency_key="expired-provider-run", ) except ProviderReportError as exc: assert "expired" in str(exc) else: raise AssertionError("expired provider-report detail was replayed") engine.dispose() def test_provider_parameters_are_checked_against_the_declared_type() -> None: engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as session: try: execute_provider_report( session, _Principal(), registry=_Registry(), provider_id="example", report_id="summary", parameters={"source_id": 123}, purpose="Operational overview", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, idempotency_key="invalid-provider-parameters", ) except ProviderReportError as exc: assert "parameter types" in str(exc) else: raise AssertionError("invalid provider-report parameters were accepted") engine.dispose() def test_source_access_is_rechecked_before_retained_result_export() -> None: engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) provider = _Provider() registry = _Registry(provider) with Session(engine) as session: execution = execute_provider_report( session, _Principal(), registry=registry, provider_id="example", report_id="summary", parameters={"source_id": "source-1"}, purpose="Operational overview", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, idempotency_key="source-access-provider-run", ) provider.result_access = False try: export_provider_execution( session, _Principal(), registry=registry, execution_id=str(execution["execution_id"]), format="json", purpose="Archive aggregate", audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"}, ) except PermissionError as exc: assert "source module" in str(exc) else: raise AssertionError("revoked source result was exported") engine.dispose()