govoplan/tools/checks/check-dsar-coverage.py

#!/usr/bin/env python3
"""Require DSAR coverage or a reviewed no-store rationale for every module."""

from __future__ import annotations

import argparse
import importlib
import json
import re
import sys
from dataclasses import dataclass
from pathlib import Path


META_ROOT = Path(__file__).resolve().parents[2]
EXEMPTIONS_PATH = Path(__file__).with_name("dsar-coverage-exemptions.json")
REPORT_PATH = (
    META_ROOT
    / "docs"
    / "evidence"
    / "snapshots"
    / "DSAR_PROVIDER_COVERAGE.generated.md"
)
MODULE_NAME_PATTERN = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*")
REQUIRED_DOCUMENTATION_TYPES = frozenset({"admin"})


@dataclass(frozen=True, slots=True)
class CoverageRow:
    module_id: str
    repository: str
    migration_owned: bool
    capability: str | None
    rationale: str


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--workspace-root",
        type=Path,
        default=None,
        help="Directory containing GovOPlaN repositories.",
    )
    parser.add_argument(
        "--render",
        action="store_true",
        help="Print the current matrix instead of comparing the checked-in report.",
    )
    args = parser.parse_args()

    catalog = json.loads((META_ROOT / "repositories.json").read_text(encoding="utf-8"))
    workspace_root = (args.workspace_root or Path(catalog["default_parent"])).resolve()
    exemptions = _exemptions()
    manifests, load_errors = _load_manifests(
        workspace_root=workspace_root,
        repositories=tuple(catalog["repositories"]),
    )

    errors = list(load_errors)
    rows: list[CoverageRow] = []
    manifest_ids = {manifest.id for _, manifest in manifests}
    stale_exemptions = sorted(set(exemptions) - manifest_ids)
    if stale_exemptions:
        errors.append(
            "DSAR coverage exemptions reference unknown modules: "
            + ", ".join(stale_exemptions)
        )

    for repository, manifest in manifests:
        expected = f"privacy.dsar.{manifest.id}"
        provided = {
            item.name
            for item in manifest.provides_interfaces
            if item.name.startswith("privacy.dsar.")
        }
        factories = {
            name
            for name in manifest.capability_factories
            if name.startswith("privacy.dsar.")
        }
        migration_owned = manifest.migration_spec is not None
        rationale = exemptions.get(manifest.id)

        if provided != factories:
            errors.append(
                f"{repository}: DSAR interface/factory mismatch: "
                f"interfaces={sorted(provided)!r}, factories={sorted(factories)!r}"
            )
        if provided and provided != {expected}:
            errors.append(
                f"{repository}: expected only {expected!r}, found {sorted(provided)!r}"
            )

        capability = (
            expected if expected in provided and expected in factories else None
        )
        if migration_owned and capability is None:
            errors.append(
                f"{repository}: migration-owning module {manifest.id!r} must provide "
                f"and register {expected!r}"
            )
        if migration_owned and rationale is not None:
            errors.append(
                f"{repository}: migration-owning module {manifest.id!r} cannot use a "
                "no-store DSAR exemption"
            )
        if not migration_owned and capability is None and rationale is None:
            errors.append(
                f"{repository}: module {manifest.id!r} needs a DSAR provider or an "
                "explicit reviewed no-store rationale"
            )
        if capability is not None and rationale is not None:
            errors.append(
                f"{repository}: module {manifest.id!r} has both DSAR coverage and a "
                "stale exemption"
            )
        if capability is not None:
            if capability not in manifest.capability_documentation:
                errors.append(
                    f"{repository}: {capability!r} lacks capability documentation"
                )
            matching_topics = tuple(
                topic
                for topic in manifest.documentation
                if "data-subject-request" in topic.id
            )
            if not matching_topics or not any(
                REQUIRED_DOCUMENTATION_TYPES.issubset(topic.documentation_types)
                for topic in matching_topics
            ):
                errors.append(
                    f"{repository}: DSAR coverage needs a static administrator "
                    "data-subject-requests DocumentationTopic"
                )

        rows.append(
            CoverageRow(
                module_id=manifest.id,
                repository=repository,
                migration_owned=migration_owned,
                capability=capability,
                rationale=(
                    f"Provider `{capability}` is registered and documented."
                    if capability
                    else rationale or "MISSING"
                ),
            )
        )

    report = _report(rows)
    if args.render:
        print(report, end="")
    elif not REPORT_PATH.is_file():
        errors.append(f"DSAR coverage report is missing: {REPORT_PATH}")
    elif REPORT_PATH.read_text(encoding="utf-8") != report:
        errors.append(
            "DSAR coverage report is stale; review changes and replace it with "
            "the output of tools/checks/check-dsar-coverage.py --render"
        )

    if errors:
        print("\n".join(errors), file=sys.stderr)
        return 1

    provider_count = sum(row.capability is not None for row in rows)
    print(
        "DSAR coverage check passed: "
        f"{provider_count} providers, {len(rows) - provider_count} reviewed "
        f"no-store rationales, {len(rows)} active modules."
    )
    return 0


def _exemptions() -> dict[str, str]:
    values = json.loads(EXEMPTIONS_PATH.read_text(encoding="utf-8"))
    if not isinstance(values, dict) or any(
        not isinstance(key, str) or not isinstance(value, str) or not value.strip()
        for key, value in values.items()
    ):
        raise ValueError("DSAR coverage exemptions must be non-empty string mappings.")
    return {key: value.strip() for key, value in values.items()}


def _load_manifests(*, workspace_root: Path, repositories: tuple[dict, ...]):
    sources: list[Path] = []
    candidates: list[tuple[str, Path, Path]] = []
    for repository in repositories:
        source = workspace_root / repository["path"] / "src"
        if not source.is_dir():
            continue
        sources.append(source)
        candidates.extend(
            (repository["name"], source, path)
            for path in sorted(source.glob("*/backend/manifest.py"))
        )
    core_source = workspace_root / "govoplan-core" / "src"
    sys.path[:0] = [
        str(core_source),
        *(str(source) for source in sources if source != core_source),
    ]

    manifests = []
    errors = []
    for repository, source, path in candidates:
        module_name = ".".join(path.relative_to(source).with_suffix("").parts)
        if MODULE_NAME_PATTERN.fullmatch(module_name) is None:
            errors.append(f"{repository}: unsafe manifest module name {module_name!r}")
            continue
        try:
            module = importlib.import_module(module_name)
            manifests.append((repository, module.get_manifest()))
        except Exception as exc:  # pragma: no cover - emitted as check evidence
            errors.append(f"{repository}: could not load {module_name}: {exc}")
    return manifests, errors


def _report(rows: list[CoverageRow]) -> str:
    ordered = sorted(rows, key=lambda row: row.module_id)
    providers = sum(row.capability is not None for row in ordered)
    lines = [
        "# DSAR Provider Coverage",
        "",
        "This generated matrix is enforced by `tools/checks/check-dsar-coverage.py`.",
        "A migration-owning module must register and document its canonical DSAR provider.",
        "Every other active module requires a reviewed explanation of why it owns no",
        "persistent subject-data store. Adding a migration invalidates that explanation.",
        "",
        f"- Active modules: {len(ordered)}",
        f"- Registered and documented DSAR providers: {providers}",
        f"- Reviewed no-store rationales: {len(ordered) - providers}",
        "- Unexplained coverage gaps: 0",
        "",
        "| Module | Repository | Persistence | Coverage | Rationale |",
        "| --- | --- | --- | --- | --- |",
    ]
    for row in ordered:
        lines.append(
            "| "
            + " | ".join(
                (
                    f"`{row.module_id}`",
                    f"`{row.repository}`",
                    "Migration-owned" if row.migration_owned else "No module migration",
                    "Provider" if row.capability else "Reviewed no-store rationale",
                    row.rationale.replace("|", "\\|"),
                )
            )
            + " |"
        )
    lines.extend(
        (
            "",
            "Provider search, export minimization, retention, and erasure behavior remains",
            "documented and tested by each owning module. This matrix verifies adoption and",
            "ownership coverage; Core continues to test disabled providers, partial failure,",
            "retry, authorization evidence, and horizontally coordinated execution.",
            "",
        )
    )
    return "\n".join(lines)


if __name__ == "__main__":
    raise SystemExit(main())