570 lines
18 KiB
Python
570 lines
18 KiB
Python
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.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.operations import (
|
|
IMPORT_SCOPE,
|
|
PUBLISH_SCOPE,
|
|
SCHEDULE_SCOPE,
|
|
ReportingOperationError,
|
|
assess_import,
|
|
dispatch_due_schedules,
|
|
export_execution,
|
|
upsert_saved_view,
|
|
upsert_schedule,
|
|
)
|
|
from govoplan_reporting.backend.schemas import ReportQuery
|
|
|
|
|
|
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 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 _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()
|