294 lines
9.0 KiB
Python
294 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.tenant_erasure import (
|
|
TENANT_ERASURE_PROVIDER_CAPABILITY_PREFIX,
|
|
TenantErasurePreview,
|
|
TenantErasureResource,
|
|
TenantErasureStep,
|
|
TenantErasureStepResult,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_tenancy.backend.db.models import TenantErasureOperation
|
|
from govoplan_tenancy.backend.erasure import (
|
|
TenantErasureConflict,
|
|
TenantErasurePolicy,
|
|
approve_tenant_erasure_operation,
|
|
assert_tenant_erasure_executable,
|
|
cancel_tenant_erasure_operation,
|
|
create_tenant_erasure_operation,
|
|
run_tenant_erasure_steps,
|
|
tenant_erasure_recently_authenticated,
|
|
)
|
|
|
|
|
|
class _Provider:
|
|
module_id = "files"
|
|
|
|
def __init__(self) -> None:
|
|
self.fail = False
|
|
self.calls: list[str] = []
|
|
|
|
def preview_tenant_erasure(self, session, tenant_id: str) -> TenantErasurePreview:
|
|
del session, tenant_id
|
|
return TenantErasurePreview(
|
|
module_id="files",
|
|
complete=True,
|
|
resources=(
|
|
TenantErasureResource(
|
|
resource_type="files",
|
|
count=1,
|
|
disposition="erase",
|
|
summary="One file is in scope.",
|
|
),
|
|
),
|
|
steps=(
|
|
TenantErasureStep(
|
|
step_id="erase-files",
|
|
kind="erase",
|
|
summary="Erase files.",
|
|
destructive=True,
|
|
irreversible=True,
|
|
),
|
|
),
|
|
)
|
|
|
|
def execute_tenant_erasure_step(
|
|
self, session, tenant_id: str, step_id: str, idempotency_key: str
|
|
) -> TenantErasureStepResult:
|
|
del session, tenant_id, idempotency_key
|
|
self.calls.append(f"execute:{step_id}")
|
|
if self.fail:
|
|
raise TimeoutError("provider timed out")
|
|
return TenantErasureStepResult(
|
|
state="completed",
|
|
summary="Files erased.",
|
|
metrics={"deleted": 1},
|
|
)
|
|
|
|
def reconcile_tenant_erasure_step(
|
|
self, session, tenant_id: str, step_id: str, idempotency_key: str
|
|
) -> TenantErasureStepResult:
|
|
del session, tenant_id, idempotency_key
|
|
self.calls.append(f"reconcile:{step_id}")
|
|
return TenantErasureStepResult(
|
|
state="completed",
|
|
summary="File erasure reconciled.",
|
|
metrics={"deleted": 1},
|
|
)
|
|
|
|
|
|
class _Registry:
|
|
def __init__(self, provider: _Provider):
|
|
self.provider = provider
|
|
|
|
def manifests(self):
|
|
return (SimpleNamespace(id="files"),)
|
|
|
|
def capability_names(self):
|
|
return (f"{TENANT_ERASURE_PROVIDER_CAPABILITY_PREFIX}files",)
|
|
|
|
def capability(self, name: str):
|
|
assert name.endswith("files")
|
|
return self.provider
|
|
|
|
def tenant_summary_providers(self):
|
|
return {}
|
|
|
|
|
|
@pytest.fixture
|
|
def session() -> Session:
|
|
engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine, expire_on_commit=False) as current:
|
|
yield current
|
|
|
|
|
|
def _operation(
|
|
session: Session,
|
|
provider: _Provider,
|
|
*,
|
|
current_time: datetime,
|
|
) -> TenantErasureOperation:
|
|
with patch(
|
|
"govoplan_tenancy.backend.erasure.tenant_erasure_policy",
|
|
return_value=TenantErasurePolicy(
|
|
production_profile=True,
|
|
required_approvals=2,
|
|
preview_ttl_seconds=900,
|
|
recent_authentication_seconds=900,
|
|
),
|
|
):
|
|
operation, replayed = create_tenant_erasure_operation(
|
|
session,
|
|
registry=_Registry(provider),
|
|
tenant_id="tenant-1",
|
|
idempotency_key="request-1234",
|
|
requested_by_account_id="account-1",
|
|
reason="Contract ended",
|
|
current_time=current_time,
|
|
)
|
|
session.commit()
|
|
assert not replayed
|
|
return operation
|
|
|
|
|
|
def test_operation_requires_distinct_multi_party_approval_and_typed_confirmation(
|
|
session: Session,
|
|
) -> None:
|
|
now = datetime(2026, 8, 24, 12, 0, tzinfo=UTC)
|
|
operation = _operation(session, _Provider(), current_time=now)
|
|
|
|
with pytest.raises(TenantErasureConflict, match="does not match"):
|
|
approve_tenant_erasure_operation(
|
|
operation,
|
|
account_id="account-1",
|
|
confirmation="wrong",
|
|
tenant_slug="target",
|
|
current_time=now,
|
|
)
|
|
assert not approve_tenant_erasure_operation(
|
|
operation,
|
|
account_id="account-1",
|
|
confirmation="target",
|
|
tenant_slug="target",
|
|
current_time=now,
|
|
)
|
|
assert operation.state == "awaiting_approval"
|
|
assert approve_tenant_erasure_operation(
|
|
operation,
|
|
account_id="account-1",
|
|
confirmation="target",
|
|
tenant_slug="target",
|
|
current_time=now,
|
|
)
|
|
assert not approve_tenant_erasure_operation(
|
|
operation,
|
|
account_id="account-2",
|
|
confirmation="target",
|
|
tenant_slug="target",
|
|
current_time=now,
|
|
)
|
|
assert operation.state == "ready"
|
|
assert_tenant_erasure_executable(
|
|
operation,
|
|
confirmation="target",
|
|
tenant_slug="target",
|
|
current_time=now,
|
|
)
|
|
|
|
|
|
def test_preview_idempotency_replays_but_rejects_changed_request(
|
|
session: Session,
|
|
) -> None:
|
|
now = datetime(2026, 8, 24, 12, 0, tzinfo=UTC)
|
|
provider = _Provider()
|
|
operation = _operation(session, provider, current_time=now)
|
|
with patch(
|
|
"govoplan_tenancy.backend.erasure.tenant_erasure_policy",
|
|
return_value=TenantErasurePolicy(),
|
|
):
|
|
replay, replayed = create_tenant_erasure_operation(
|
|
session,
|
|
registry=_Registry(provider),
|
|
tenant_id="tenant-1",
|
|
idempotency_key="request-1234",
|
|
requested_by_account_id="account-1",
|
|
reason="Contract ended",
|
|
current_time=now,
|
|
)
|
|
assert replayed
|
|
assert replay.id == operation.id
|
|
with pytest.raises(TenantErasureConflict, match="another request"):
|
|
create_tenant_erasure_operation(
|
|
session,
|
|
registry=_Registry(provider),
|
|
tenant_id="tenant-1",
|
|
idempotency_key="request-1234",
|
|
requested_by_account_id="account-1",
|
|
reason="Changed",
|
|
current_time=now,
|
|
)
|
|
|
|
|
|
def test_provider_timeout_requires_reconciliation_before_completion(
|
|
session: Session,
|
|
) -> None:
|
|
now = datetime(2026, 8, 24, 12, 0, tzinfo=UTC)
|
|
provider = _Provider()
|
|
operation = _operation(session, provider, current_time=now)
|
|
operation.state = "ready"
|
|
operation.approvals = [
|
|
{"account_id": "account-1"},
|
|
{"account_id": "account-2"},
|
|
]
|
|
session.commit()
|
|
provider.fail = True
|
|
|
|
assert not run_tenant_erasure_steps(
|
|
session,
|
|
registry=_Registry(provider),
|
|
operation=operation,
|
|
current_time=now,
|
|
)
|
|
assert operation.state == "reconciliation_required"
|
|
assert operation.steps[0]["state"] == "outcome_unknown"
|
|
assert operation.destructive_started
|
|
|
|
provider.fail = False
|
|
assert run_tenant_erasure_steps(
|
|
session,
|
|
registry=_Registry(provider),
|
|
operation=operation,
|
|
current_time=now,
|
|
)
|
|
assert provider.calls == ["execute:erase-files", "reconcile:erase-files"]
|
|
assert operation.steps[0]["state"] == "completed"
|
|
|
|
|
|
def test_cancellation_stops_at_destructive_boundary(session: Session) -> None:
|
|
now = datetime(2026, 8, 24, 12, 0, tzinfo=UTC)
|
|
operation = _operation(session, _Provider(), current_time=now)
|
|
cancel_tenant_erasure_operation(operation)
|
|
assert operation.state == "cancelled"
|
|
|
|
operation.state = "running"
|
|
operation.destructive_started = True
|
|
with pytest.raises(TenantErasureConflict, match="destructive work"):
|
|
cancel_tenant_erasure_operation(operation)
|
|
|
|
|
|
def test_recent_authentication_is_policy_bounded() -> None:
|
|
now = datetime(2026, 8, 24, 12, 0, tzinfo=UTC)
|
|
policy = TenantErasurePolicy(recent_authentication_seconds=300)
|
|
|
|
assert tenant_erasure_recently_authenticated(
|
|
SimpleNamespace(created_at=now - timedelta(minutes=4)),
|
|
policy,
|
|
current_time=now,
|
|
)
|
|
assert not tenant_erasure_recently_authenticated(
|
|
SimpleNamespace(created_at=now - timedelta(minutes=6)),
|
|
policy,
|
|
current_time=now,
|
|
)
|
|
assert not tenant_erasure_recently_authenticated(None, policy, current_time=now)
|
|
|
|
|
|
def test_production_policy_cannot_disable_multi_party_approval() -> None:
|
|
with pytest.raises(ValueError, match="at least two approvals"):
|
|
TenantErasurePolicy(production_profile=True, required_approvals=1)
|
|
|
|
non_production = TenantErasurePolicy(
|
|
production_profile=False,
|
|
required_approvals=1,
|
|
)
|
|
assert non_production.required_approvals == 1
|