from __future__ import annotations from datetime import UTC, datetime from types import SimpleNamespace import unittest from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.db.base import Base from govoplan_core.core.files import ( CAPABILITY_FILES_ARTIFACT_STORE, ManagedArtifactRef, ) from govoplan_core.security.module_permissions import scopes_grant_compatible from govoplan_reporting.backend.definitions import ( ADMIN_SCOPE, READ_SCOPE, WRITE_SCOPE, ReportingDefinitionError, create_definition, definition_history, get_definition, list_definitions, update_definition, ) from govoplan_reporting.backend.execution import ( QUALITY_SCOPE, RUN_SCOPE, ReportingExecutionFailure, execute_report, run_quality_plan, ) from govoplan_reporting.backend.drilldown import ( ReportingDrillError, create_drill_context, resolve_drill_context, ) from govoplan_reporting.backend.operations import ( IMPORT_SCOPE, PUBLISH_SCOPE, SCHEDULE_SCOPE, ReportingOperationError, assess_import, dispatch_due_schedules, export_execution, list_publications, publish_execution, upsert_saved_view, upsert_schedule, ) from govoplan_reporting.backend.schemas import ReportQuery from govoplan_reporting.backend.postgres_planner import compile_postgres_query from govoplan_reporting.backend.contracts import ( CAPABILITY_REPORTING_PUBLICATION_FILES, ) from govoplan_reporting.backend.publication_targets import ( FilesReportingPublicationTarget, publication_target_catalog, ) from govoplan_reporting.backend.schemas import ( DatasetDefinition, SemanticModelDefinition, ) NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC) ALL_SCOPES = ( READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE, RUN_SCOPE, QUALITY_SCOPE, PUBLISH_SCOPE, SCHEDULE_SCOPE, IMPORT_SCOPE, ) class Principal: def __init__( self, tenant_id: str = "tenant-1", account_id: str = "analyst-1", *, scopes: tuple[str, ...] = ALL_SCOPES, group_ids: tuple[str, ...] = (), ) -> None: self.tenant_id = tenant_id self.account_id = account_id self.identity_id = f"identity-{account_id}" self.membership_id = f"membership-{account_id}" self.group_ids = frozenset(group_ids) self.role_ids = frozenset() self.function_assignment_ids = frozenset() self.acting_assignment_id = None self.service_account_id = None self.scopes = frozenset(scopes) self.user = SimpleNamespace(id=f"user-{account_id}") def has(self, scope: str) -> bool: return scopes_grant_compatible(self.scopes, scope) class CapabilityRegistry: def __init__(self) -> None: self.providers: dict[str, object] = {} def has_capability(self, name: str) -> bool: return name in self.providers def capability(self, name: str) -> object | None: return self.providers.get(name) class ArtifactStore: def __init__(self) -> None: self.requests: list[object] = [] def store_artifact(self, session, principal, *, request): del session, principal self.requests.append(request) return ManagedArtifactRef( file_asset_id="asset-1", file_version_id="version-1", filename=request.filename, display_path=f"{request.folder}/{request.filename}", content_type=request.content_type, size_bytes=len(request.payload), sha256="a" * 64, provenance={"stored": True}, ) class ReportingServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(self.engine) self.session = Session(self.engine, expire_on_commit=False) self.principal = Principal() def tearDown(self) -> None: self.session.close() self.engine.dispose() def test_definition_graph_is_pinned_versioned_and_occ_guarded(self) -> None: dataset, semantic, report = self._create_report_graph() self.assertEqual( ("dataset", dataset.definition_id, 1), ( semantic.parent_kind, semantic.parent_id, semantic.parent_revision, ), ) replay = self._create( "report", "report-1", report_payload(), idempotency_key="create-report", ) self.assertEqual(report.content_hash, replay.content_hash) updated = update_definition( self.session, self.principal, definition_kind="report", definition_id="report-1", expected_revision=1, recorded_at=NOW.replace(hour=11), change_reason="Clarify the institutional question.", idempotency_key="update-report", changes={"description": "Regional workload and value."}, ) self.assertEqual(2, updated.revision) self.assertEqual( [2, 1], [ item.revision for item in definition_history( self.session, self.principal, definition_kind="report", definition_id="report-1", ) ], ) with self.assertRaisesRegex(ReportingDefinitionError, "stale"): update_definition( self.session, self.principal, definition_kind="report", definition_id="report-1", expected_revision=1, recorded_at=NOW.replace(hour=12), change_reason="Stale edit.", idempotency_key="stale-report", changes={"name": "Stale report"}, ) def test_active_children_require_exact_active_parent_revision(self) -> None: self._create("dataset", "dataset-draft", dataset_payload(), status="draft") payload = semantic_payload(dataset_id="dataset-draft") with self.assertRaisesRegex(ReportingDefinitionError, "active dataset"): self._create("semantic_model", "semantic-invalid", payload) payload["dataset_id"] = "missing" with self.assertRaisesRegex(ReportingDefinitionError, "missing dataset"): self._create("semantic_model", "semantic-missing", payload) def test_semantic_execution_replay_pivot_saved_view_and_schedule(self) -> None: self._create_report_graph() result = execute_report( self.session, self.principal, registry=None, report_id="report-1", report_revision=1, parameters={}, query=None, idempotency_key="run-report-1", ) self.assertEqual("succeeded", result["status"]) self.assertEqual( [ {"region": "North", "amount": 150, "cases": 15, "value_per_case": 10}, {"region": "South", "amount": 40, "cases": 8, "value_per_case": 5}, ], result["rows"], ) self.assertEqual( result["execution_id"], execute_report( self.session, self.principal, registry=None, report_id="report-1", report_revision=1, parameters={}, query=None, idempotency_key="run-report-1", )["execution_id"], ) self.assertEqual("bar", result["visualization"]["kind"]) self.assertIsNotNone(result["visualization"]["tabular_fallback"]) pivot = execute_report( self.session, self.principal, registry=None, report_id="report-1", report_revision=1, parameters={}, query=ReportQuery.model_validate( { "mode": "pivot", "pivot": { "rows": ["region"], "columns": ["category"], "measures": ["amount"], }, } ), idempotency_key="run-report-pivot", ) self.assertEqual(2, pivot["total_rows"]) self.assertEqual(100, pivot["rows"][0]["A.amount"]) view = upsert_saved_view( self.session, self.principal, view_id="view-1", report_id="report-1", report_revision=1, name="Regional pivot", state={"query": pivot["query"]}, shared=True, access={}, expected_revision=None, ) self.assertEqual(1, view["revision"]) schedule = upsert_schedule( self.session, self.principal, schedule_id="schedule-1", report_id="report-1", report_revision=1, name="Daily regional report", trigger_kind="interval", trigger_config={"seconds": 86_400}, parameters={}, query=ReportQuery(), publication_target={}, enabled=True, next_run_at=NOW, expected_revision=None, ) self.assertEqual(1, schedule["revision"]) dispatched = dispatch_due_schedules( self.session, self.principal, registry=None, now=NOW, limit=10, ) self.assertEqual( (1, 1, 0), ( dispatched["claimed"], dispatched["succeeded"], dispatched["failed"], ), ) def test_quality_gate_blocks_and_preserves_failure_evidence(self) -> None: self._create_report_graph() self._create( "quality_plan", "quality-1", { "dataset_id": "dataset-1", "dataset_revision": 1, "block_report_execution": True, "assertions": [ { "key": "region-unique", "kind": "unique", "field": "region", "severity": "blocker", } ], }, ) quality = run_quality_plan( self.session, self.principal, registry=None, quality_plan_id="quality-1", quality_plan_revision=1, parameters={}, ) self.assertEqual("failed", quality["status"]) with self.assertRaises(ReportingExecutionFailure) as raised: execute_report( self.session, self.principal, registry=None, report_id="report-1", report_revision=1, parameters={}, query=None, idempotency_key="quality-blocked-run", ) self.assertTrue(raised.exception.execution_id) def test_restricted_access_and_service_scope_guards(self) -> None: self._create_report_graph( report_access={ "subjects": [ { "kind": "group", "id": "group-auditors", "permissions": ["read"], } ] }, report_visibility="restricted", ) regular_scopes = (READ_SCOPE, WRITE_SCOPE, RUN_SCOPE) outsider = Principal(account_id="outsider", scopes=regular_scopes) auditor = Principal( account_id="auditor", scopes=regular_scopes, group_ids=("group-auditors",), ) self.assertIsNone( get_definition( self.session, outsider, definition_kind="report", definition_id="report-1", ) ) self.assertIsNotNone( get_definition( self.session, auditor, definition_kind="report", definition_id="report-1", ) ) self.assertEqual( 0, list_definitions( self.session, outsider, definition_kinds=("report",), )[1], ) without_permissions = Principal(account_id="none", scopes=()) with self.assertRaises(PermissionError): list_definitions(self.session, without_permissions) with self.assertRaises(PermissionError): self._create( "dataset", "unauthorized", dataset_payload(), principal=without_permissions, ) def test_safe_export_and_import_activation_assessment(self) -> None: self._create_report_graph() detail = execute_report( self.session, self.principal, registry=None, report_id="report-1", report_revision=1, parameters={}, query=ReportQuery(mode="detail", dimensions=["note"]), idempotency_key="detail-export", ) content, content_type, _filename = export_execution( self.session, self.principal, execution_id=str(detail["execution_id"]), format="csv", ) self.assertEqual("text/csv; charset=utf-8", content_type) self.assertIn("'=cmd", content.decode("utf-8-sig")) blocked = assess_import( self.session, self.principal, source_system="SuperX", source_id="legacy-report", metadata={"features": ["dataset", "raw_sql", "dashboard_layout"]}, accepted_approximations=[], ) self.assertEqual("blocked", blocked["status"]) self.assertFalse(blocked["mapping_report"]["activation_allowed"]) ready = assess_import( self.session, self.principal, source_system="SuperX", source_id="mapped-report", metadata={"features": ["dataset", "dashboard_layout"]}, accepted_approximations=["dashboard_layout"], ) self.assertEqual("ready", ready["status"]) with self.assertRaisesRegex(ReportingOperationError, "must be a list"): assess_import( self.session, self.principal, source_system="invalid", source_id="invalid", metadata={"features": "raw_sql"}, accepted_approximations=[], ) def test_drill_context_is_bounded_actor_bound_and_reauthorized(self) -> None: self._create_report_graph() execution = execute_report( self.session, self.principal, registry=None, report_id="report-1", report_revision=1, parameters={}, query=None, idempotency_key="drill-source", ) north = next(row for row in execution["rows"] if row["region"] == "North") context = create_drill_context( self.session, self.principal, registry=None, execution_id=str(execution["execution_id"]), aggregate_row=north, limit=50, ) detail = resolve_drill_context( self.session, self.principal, registry=None, token=str(context["token"]), ) self.assertEqual(2, detail["total_rows"]) self.assertEqual({"North"}, {row["region"] for row in detail["rows"]}) self.assertEqual("region", detail["dimension_path"][0]["dimension"]) with self.assertRaises(PermissionError): resolve_drill_context( self.session, Principal(account_id="another-analyst"), registry=None, token=str(context["token"]), ) with self.assertRaises(ReportingDrillError): create_drill_context( self.session, self.principal, registry=None, execution_id=str(execution["execution_id"]), aggregate_row={"region": "Not an execution row"}, limit=50, ) def test_governance_scope_inheritance_never_broadens_parent_limits(self) -> None: system = Principal( scopes=(*ALL_SCOPES, "system:governance:write"), group_ids=("group-reporters",), ) dataset = dataset_payload() dataset["governance"] = { "scope_type": "system", "inherit_to_lower_scopes": True, "allow_run": True, "allow_reuse": True, "allow_automation": False, } self._create("dataset", "dataset-governed", dataset, principal=system) semantic = semantic_payload(dataset_id="dataset-governed") semantic["governance"] = { "scope_type": "tenant", "inherit_to_lower_scopes": True, "allow_run": True, "allow_reuse": True, "allow_automation": True, } with self.assertRaisesRegex(ValueError, "cannot broaden inherited limits"): self._create( "semantic_model", "semantic-broadened", semantic, principal=system, ) semantic["governance"]["allow_automation"] = False semantic_record = self._create( "semantic_model", "semantic-governed", semantic, principal=system, ) semantic_governance = semantic_record.payload["governance"] self.assertEqual("system", semantic_governance["source_scope"]["scope_type"]) self.assertFalse( semantic_governance["source_effective_limits"]["allow_automation"] ) report = report_payload() report["semantic_model_id"] = "semantic-governed" report["governance"] = { "scope_type": "group", "scope_id": "group-reporters", "inherit_to_lower_scopes": False, "allow_run": True, "allow_reuse": False, "allow_automation": False, } self._create("report", "report-governed", report, principal=system) self.assertIsNotNone( get_definition( self.session, system, definition_kind="report", definition_id="report-governed", ) ) self.assertIsNone( get_definition( self.session, Principal(account_id="outsider"), definition_kind="report", definition_id="report-governed", ) ) def test_postgres_plan_is_bounded_and_parameterized(self) -> None: dataset = DatasetDefinition.model_validate(dataset_payload()) semantic = SemanticModelDefinition.model_validate(semantic_payload()) query = ReportQuery.model_validate( { "mode": "summary", "dimensions": ["region"], "measures": ["amount", "value_per_case"], "filters": [ { "dimension": "region", "operator": "contains", "value": "North%' OR TRUE --", } ], "sort": [{"key": "amount", "direction": "desc"}], "limit": 25, } ) plan = compile_postgres_query(dataset, semantic, query) self.assertIn("GROUP BY", plan.sql) self.assertIn("LIMIT :result_limit OFFSET :result_offset", plan.sql) self.assertNotIn("North%' OR TRUE --", plan.sql) self.assertIn("North", str(plan.parameters["filter_0"])) calculated_only = compile_postgres_query( dataset, semantic, ReportQuery( mode="summary", dimensions=["region"], measures=["value_per_case"], ), ) self.assertIn('SUM(NULLIF(source_row ->> :measure_0, \'\')::numeric)', calculated_only.sql) self.assertIn('AS "value_per_case"', calculated_only.sql) def test_files_publication_is_idempotent_and_retains_evidence(self) -> None: self._create_report_graph() registry = CapabilityRegistry() store = ArtifactStore() registry.providers[CAPABILITY_FILES_ARTIFACT_STORE] = store registry.providers[CAPABILITY_REPORTING_PUBLICATION_FILES] = ( FilesReportingPublicationTarget(registry) ) execution = execute_report( self.session, self.principal, registry=registry, report_id="report-1", report_revision=1, parameters={}, query=None, idempotency_key="publish-source", ) first = publish_execution( self.session, self.principal, registry=registry, execution_id=str(execution["execution_id"]), target_capability=CAPABILITY_REPORTING_PUBLICATION_FILES, target_ref="Reports/Monthly", format="csv", idempotency_key="publish-files-once", options={"filename": "regional workload.csv"}, ) replay = publish_execution( self.session, self.principal, registry=registry, execution_id=str(execution["execution_id"]), target_capability=CAPABILITY_REPORTING_PUBLICATION_FILES, target_ref="Reports/Monthly", format="csv", idempotency_key="publish-files-once", options={"filename": "regional workload.csv"}, ) self.assertEqual(first["publication_id"], replay["publication_id"]) self.assertEqual(1, len(store.requests)) self.assertEqual("version-1", first["evidence"]["file_version_id"]) self.assertEqual( 1, len( list_publications( self.session, self.principal, execution_id=str(execution["execution_id"]), ) ), ) targets = publication_target_catalog(registry) self.assertTrue(targets[0]["available"]) self.assertFalse(targets[1]["available"]) self.assertIn("Enable Mail", str(targets[1]["reason"])) def _create_report_graph( self, *, report_access: dict[str, object] | None = None, report_visibility: str = "tenant", ): dataset = self._create("dataset", "dataset-1", dataset_payload()) semantic = self._create( "semantic_model", "semantic-1", semantic_payload(), ) report = self._create( "report", "report-1", report_payload(access_policy=report_access), visibility=report_visibility, idempotency_key="create-report", ) return dataset, semantic, report def _create( self, kind: str, definition_id: str, payload: dict[str, object], *, status: str = "active", visibility: str = "tenant", idempotency_key: str | None = None, principal: Principal | None = None, ): return create_definition( self.session, principal or self.principal, definition_kind=kind, definition_id=definition_id, definition_key=definition_id, name=definition_id.replace("-", " ").title(), description="Governed reporting test definition.", status=status, visibility=visibility, recorded_at=NOW, change_reason="Initial governed baseline.", payload=payload, idempotency_key=idempotency_key or f"create-{definition_id}", ) def dataset_payload() -> dict[str, object]: return { "source_kind": "static", "source_ref": "fixture.regional-workload", "purpose": "Verify governed institutional reporting.", "schema": [ {"name": "region", "type": "string", "nullable": False}, {"name": "category", "type": "string", "nullable": False}, {"name": "amount", "type": "number", "nullable": False}, {"name": "cases", "type": "integer", "nullable": False}, {"name": "note", "type": "string", "nullable": False}, ], "static_rows": [ { "region": "North", "category": "A", "amount": 100, "cases": 10, "note": "=cmd", }, { "region": "North", "category": "B", "amount": 50, "cases": 5, "note": "ordinary", }, { "region": "South", "category": "A", "amount": 40, "cases": 8, "note": "ordinary", }, ], } def semantic_payload(*, dataset_id: str = "dataset-1") -> dict[str, object]: return { "dataset_id": dataset_id, "dataset_revision": 1, "dimensions": [ {"key": "region", "field": "region", "label": "Region"}, {"key": "category", "field": "category", "label": "Category"}, {"key": "note", "field": "note", "label": "Note"}, ], "hierarchies": [ { "key": "regional-category", "label": "Region and category", "levels": [{"dimension": "region"}, {"dimension": "category"}], } ], "measures": [ { "key": "amount", "label": "Amount", "aggregation": "sum", "field": "amount", }, {"key": "cases", "label": "Cases", "aggregation": "sum", "field": "cases"}, { "key": "value_per_case", "label": "Value per case", "aggregation": "calculated", "expression": { "op": "divide", "args": [ {"op": "measure", "ref": "amount"}, {"op": "measure", "ref": "cases"}, ], }, }, ], "default_dimensions": ["region"], "default_measures": ["amount", "cases", "value_per_case"], } def report_payload( *, access_policy: dict[str, object] | None = None, ) -> dict[str, object]: return { "semantic_model_id": "semantic-1", "semantic_model_revision": 1, "default_query": { "mode": "summary", "dimensions": ["region"], "measures": ["amount", "cases", "value_per_case"], "sort": [{"key": "region", "direction": "asc"}], }, "visualization": { "kind": "bar", "category_dimension": "region", "measures": ["amount"], "tabular_fallback": True, }, "access_policy": access_policy or {}, "institutional_references": [ { "owner_module": "projects", "resource_type": "outcome", "resource_id": "faster-decisions", "revision": "1", "relationship": "measures", } ], } if __name__ == "__main__": unittest.main()