97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.dsar import DsarSubjectRef
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_tenancy.backend.db.models import TenantErasureOperation
|
|
from govoplan_tenancy.backend.dsar_provider import TenancyDsarProvider
|
|
|
|
|
|
def _operation(now: datetime) -> TenantErasureOperation:
|
|
return TenantErasureOperation(
|
|
tenant_id="tenant-1",
|
|
state="ready",
|
|
idempotency_key="request-1234",
|
|
request_digest="a" * 64,
|
|
preview_digest="b" * 64,
|
|
preview={"schema_version": 1},
|
|
previewed_at=now,
|
|
preview_expires_at=now + timedelta(minutes=15),
|
|
policy={"required_approvals": 2},
|
|
approvals=[
|
|
{"account_id": "account-1", "approved_at": now.isoformat()},
|
|
{"account_id": "account-2", "approved_at": now.isoformat()},
|
|
],
|
|
steps=[],
|
|
requested_by_account_id="account-1",
|
|
reason="Contract ended",
|
|
destructive_started=False,
|
|
revision=3,
|
|
)
|
|
|
|
|
|
def test_dsar_exports_only_subject_actor_evidence_and_retains_it() -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
now = datetime(2026, 8, 24, 12, 0, tzinfo=UTC)
|
|
with Session(engine) as session:
|
|
operation = _operation(now)
|
|
session.add(operation)
|
|
session.commit()
|
|
provider = TenancyDsarProvider()
|
|
|
|
requester_records = provider.search_subject(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id="account-1"),
|
|
)
|
|
approver_records = provider.search_subject(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id="account-2"),
|
|
)
|
|
|
|
assert requester_records[0].data["actor_roles"] == ["requester", "approver"]
|
|
assert requester_records[0].data["request_reason"] == "Contract ended"
|
|
assert approver_records[0].data["actor_roles"] == ["approver"]
|
|
assert "request_reason" not in approver_records[0].data
|
|
assert requester_records[0].immutable_evidence
|
|
|
|
actions = provider.plan_erasure(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id="account-1"),
|
|
records=requester_records,
|
|
)
|
|
assert actions[0].kind == "retain"
|
|
assert not actions[0].executable
|
|
results = provider.execute_erasure(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(account_id="account-1"),
|
|
actions=actions,
|
|
request_id="dsar-1",
|
|
)
|
|
assert results[0].status == "blocked"
|
|
|
|
|
|
def test_dsar_rejects_conflicting_account_selectors() -> None:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
assert (
|
|
TenancyDsarProvider().search_subject(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
subject=DsarSubjectRef(
|
|
account_id="account-1",
|
|
external_references={"access.account": "account-2"},
|
|
),
|
|
)
|
|
== ()
|
|
)
|