feat: add governed cross-module reporting
This commit is contained in:
@@ -0,0 +1,336 @@
|
||||
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(2026, 8, 2, 10, 0, tzinfo=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()
|
||||
Reference in New Issue
Block a user