govoplan-core/tests/test_tenant_erasure_contract.py

from __future__ import annotations

from datetime import UTC, datetime
from types import SimpleNamespace

import pytest

from govoplan_core.core.tenant_erasure import (
    TENANT_ERASURE_PROVIDER_CAPABILITY_PREFIX,
    TenantErasurePreview,
    TenantErasureResource,
    TenantErasureStep,
    TenantErasureStepResult,
    collect_tenant_erasure_inventory,
    tenant_erasure_providers,
)


class _Provider:
    module_id = "files"

    def preview_tenant_erasure(self, session, tenant_id: str) -> TenantErasurePreview:
        del session
        assert tenant_id == "tenant-1"
        return TenantErasurePreview(
            module_id=self.module_id,
            complete=True,
            resources=(
                TenantErasureResource(
                    resource_type="file_blobs",
                    count=2,
                    disposition="erase",
                    summary="Two tenant-owned file blobs will be erased.",
                ),
            ),
            steps=(
                TenantErasureStep(
                    step_id="erase-blobs",
                    kind="erase",
                    summary="Erase tenant-owned file blobs.",
                    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, step_id, idempotency_key
        return TenantErasureStepResult(
            state="completed",
            summary="Tenant file blobs erased.",
            metrics={"deleted": 2},
        )

    def reconcile_tenant_erasure_step(
        self, session, tenant_id: str, step_id: str, idempotency_key: str
    ) -> TenantErasureStepResult:
        return self.execute_tenant_erasure_step(
            session, tenant_id, step_id, idempotency_key
        )


class _Registry:
    def __init__(self, *, provider: object | None = None, counts: dict[str, int] | None = None):
        self._provider = provider
        self._counts = counts

    def manifests(self):
        return (
            SimpleNamespace(id="core"),
            SimpleNamespace(id="files"),
            SimpleNamespace(id="wiki"),
        )

    def capability_names(self):
        if self._provider is None:
            return ()
        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):
        if self._counts is None:
            return {}
        return {"files": lambda _session, _tenant_id: self._counts}


def test_contract_rejects_unsafe_irreversible_step() -> None:
    with pytest.raises(ValueError, match="must be destructive"):
        TenantErasureStep(
            step_id="unsafe",
            kind="erase",
            summary="Invalid step.",
            destructive=False,
            irreversible=True,
        )


def test_contract_rejects_cyclic_step_dependencies() -> None:
    with pytest.raises(ValueError, match="contain a cycle"):
        TenantErasurePreview(
            module_id="files",
            complete=True,
            steps=(
                TenantErasureStep(
                    step_id="first",
                    kind="erase",
                    summary="First.",
                    destructive=True,
                    irreversible=True,
                    depends_on=("second",),
                ),
                TenantErasureStep(
                    step_id="second",
                    kind="verify",
                    summary="Second.",
                    destructive=False,
                    irreversible=False,
                    depends_on=("first",),
                ),
            ),
        )


def test_contract_requires_action_or_blocker_for_tenant_data() -> None:
    resource = TenantErasureResource(
        resource_type="files",
        count=1,
        disposition="erase",
        summary="One file exists.",
    )
    with pytest.raises(ValueError, match="steps or an explicit blocker"):
        TenantErasurePreview(
            module_id="files",
            complete=True,
            resources=(resource,),
        )


def test_inventory_collects_provider_and_marks_non_data_modules() -> None:
    inventory = collect_tenant_erasure_inventory(
        _Registry(provider=_Provider()),
        object(),
        "tenant-1",
        observed_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC),
    )

    assert inventory.complete
    assert inventory.allowed
    assert [item.module_id for item in inventory.modules] == ["core", "files", "wiki"]
    assert inventory.modules[1].steps[0].irreversible
    assert inventory.to_dict()["generated_at"] == "2026-08-24T12:00:00+00:00"


def test_summary_fallback_blocks_when_data_exists() -> None:
    inventory = collect_tenant_erasure_inventory(
        _Registry(counts={"file_blobs": 3}),
        object(),
        "tenant-1",
    )

    files = next(item for item in inventory.modules if item.module_id == "files")
    assert inventory.complete
    assert not inventory.allowed
    assert files.resources[0].disposition == "unavailable"
    assert files.blockers == (
        "Tenant-owned data exists but the module has no erasure provider.",
    )


def test_provider_identity_must_match_capability_suffix() -> None:
    provider = _Provider()
    provider.module_id = "mail"

    with pytest.raises(ValueError, match="identity"):
        tenant_erasure_providers(_Registry(provider=provider))