136 lines
4.4 KiB
Python
136 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.reporting import ReportingGovernanceRequest
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.tenancy.scope import Tenant, create_scope_tables
|
|
from govoplan_policy.backend.reporting_governance import (
|
|
ReportingGovernancePolicyProvider,
|
|
)
|
|
from govoplan_policy.backend import retention as retention_module
|
|
|
|
|
|
def _request(action: str, *, export_format: str | None = None):
|
|
return ReportingGovernanceRequest(
|
|
action=action, # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
provider_id="campaigns",
|
|
report_id="delivery-outcomes",
|
|
purpose="Operational review",
|
|
audience_scope={"scope_type": "tenant", "scope_id": "tenant-1"},
|
|
retention_class="stored_report_detail",
|
|
export_format=export_format,
|
|
reidentification_risk="low",
|
|
declared_privacy_transforms=("small_cell_suppression",),
|
|
applied_privacy_transforms=("small_cell_suppression",),
|
|
)
|
|
|
|
|
|
def test_tenant_reporting_policy_can_tighten_export_and_retention() -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
create_scope_tables(engine)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
session.add(
|
|
Tenant(
|
|
id="tenant-1",
|
|
slug="tenant-1",
|
|
name="Tenant 1",
|
|
settings={
|
|
"privacy_retention_policy": {
|
|
"stored_report_detail_retention_days": 10
|
|
},
|
|
"reporting_governance_policy": {
|
|
"allow_exports": False,
|
|
"required_privacy_transforms": ["explicit_denominator"],
|
|
},
|
|
},
|
|
)
|
|
)
|
|
session.commit()
|
|
provider = ReportingGovernancePolicyProvider()
|
|
|
|
execute = provider.decide_reporting_action(
|
|
session,
|
|
object(),
|
|
request=_request("execute"),
|
|
)
|
|
export = provider.decide_reporting_action(
|
|
session,
|
|
object(),
|
|
request=_request("export", export_format="json"),
|
|
)
|
|
|
|
assert execute.allowed is True
|
|
assert execute.retention_days == 10
|
|
assert set(execute.required_privacy_transforms) == {
|
|
"small_cell_suppression",
|
|
"explicit_denominator",
|
|
}
|
|
assert export.allowed is False
|
|
assert export.export_formats == ()
|
|
engine.dispose()
|
|
|
|
|
|
def test_malformed_reporting_policy_fails_closed() -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
create_scope_tables(engine)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
session.add(
|
|
Tenant(
|
|
id="tenant-1",
|
|
slug="tenant-1",
|
|
name="Tenant 1",
|
|
settings={"reporting_governance_policy": {"allow_exports": "yes"}},
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
decision = ReportingGovernancePolicyProvider().decide_reporting_action(
|
|
session,
|
|
object(),
|
|
request=_request("execute"),
|
|
)
|
|
|
|
assert decision.allowed is False
|
|
assert decision.provenance["decision"] == "fail_closed"
|
|
engine.dispose()
|
|
|
|
|
|
def test_shared_retention_run_invokes_reporting_without_model_imports(
|
|
monkeypatch,
|
|
) -> None:
|
|
class _ReportingRetention:
|
|
def apply_retention(self, session, *, dry_run, now, limit=500):
|
|
del session, now, limit
|
|
return {
|
|
"eligible": 2,
|
|
"redacted": 0 if dry_run else 2,
|
|
"remaining_in_batch": 0,
|
|
}
|
|
|
|
class _Registry:
|
|
def has_capability(self, name):
|
|
return name == "reporting.retention"
|
|
|
|
def require_capability(self, name):
|
|
assert name == "reporting.retention"
|
|
return _ReportingRetention()
|
|
|
|
monkeypatch.setattr(retention_module, "get_registry", lambda: _Registry())
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
create_scope_tables(engine)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
result = retention_module.apply_retention_policy(session, dry_run=False)
|
|
|
|
assert result["counts"]["stored_report_detail"]["provider_reports"] == {
|
|
"eligible": 2,
|
|
"redacted": 2,
|
|
"remaining_in_batch": 0,
|
|
}
|
|
engine.dispose()
|