912 lines
30 KiB
Python
912 lines
30 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.core.dataflows import (
|
|
CAPABILITY_DATAFLOW_DATASET_OUTPUT,
|
|
DataflowDatasetResult,
|
|
)
|
|
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 DataflowOutput:
|
|
def __init__(self, rows: list[dict[str, object]]) -> None:
|
|
self.rows = tuple(dict(item) for item in rows)
|
|
self.last_request = None
|
|
|
|
def list_outputs(self, *_args, **_kwargs):
|
|
return ()
|
|
|
|
def read_output(self, _session, _principal, *, request):
|
|
self.last_request = request
|
|
return DataflowDatasetResult(
|
|
pipeline_ref=request.pipeline_ref,
|
|
revision=request.revision,
|
|
definition_hash=request.expected_definition_hash or "pipeline-hash",
|
|
rows=self.rows,
|
|
total_rows=len(self.rows),
|
|
truncated=False,
|
|
output_hash="d" * 64,
|
|
executor_version="duckdb-v1",
|
|
run_ref=request.run_ref,
|
|
source_fingerprints=(
|
|
{"node_id": "source", "fingerprint": "source-v1"},
|
|
),
|
|
generated_at=NOW,
|
|
provenance={"immutable_run": bool(request.run_ref)},
|
|
)
|
|
|
|
|
|
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_dataflow_run_can_be_exported_as_formula_safe_csv(self) -> None:
|
|
payload = dataset_payload()
|
|
rows = list(payload.pop("static_rows"))
|
|
payload.update(
|
|
{
|
|
"source_kind": "dataflow",
|
|
"source_ref": "pipeline:monthly-comparison",
|
|
"source_revision": 4,
|
|
"source_run_ref": "dataflow-run:published-july",
|
|
"definition_hash": "pipeline-hash",
|
|
}
|
|
)
|
|
self._create("dataset", "dataset-1", payload)
|
|
self._create("semantic_model", "semantic-1", semantic_payload())
|
|
self._create("report", "report-1", report_payload())
|
|
provider = DataflowOutput(rows)
|
|
registry = CapabilityRegistry()
|
|
registry.providers[CAPABILITY_DATAFLOW_DATASET_OUTPUT] = provider
|
|
|
|
result = execute_report(
|
|
self.session,
|
|
self.principal,
|
|
registry=registry,
|
|
report_id="report-1",
|
|
report_revision=1,
|
|
parameters={},
|
|
query=ReportQuery(mode="detail", dimensions=["note"]),
|
|
idempotency_key="published-dataflow-run",
|
|
)
|
|
|
|
self.assertEqual("succeeded", result["status"])
|
|
self.assertIsNotNone(provider.last_request)
|
|
self.assertEqual(
|
|
"dataflow-run:published-july",
|
|
provider.last_request.run_ref,
|
|
)
|
|
self.assertTrue(result["provenance"]["source"]["immutable_run"])
|
|
content, content_type, filename = export_execution(
|
|
self.session,
|
|
self.principal,
|
|
execution_id=str(result["execution_id"]),
|
|
format="csv",
|
|
)
|
|
self.assertEqual("text/csv; charset=utf-8", content_type)
|
|
self.assertTrue(filename.endswith(".csv"))
|
|
self.assertIn("'=cmd", content.decode("utf-8-sig"))
|
|
|
|
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()
|