govoplan/tools/checks/check-contracts.py

#!/usr/bin/env python3
"""Static cross-repository module contract check."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

META_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(META_ROOT / "tools" / "release"))

from govoplan_release.contracts import (  # noqa: E402
    collect_contracts,
    format_version_range,
    interface_impact_map,
    provided_interface_map,
    validate_contracts,
)
from govoplan_release.model import RepositorySnapshot, to_jsonable  # noqa: E402
from govoplan_release.workspace import load_repository_specs, resolve_repo_path, resolve_workspace_root  # noqa: E402


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--workspace-root",
        type=Path,
        default=None,
        help="Workspace root containing the repositories. Defaults to repositories.json default_parent.",
    )
    parser.add_argument("--include-website", action="store_true", help="Also scan website repositories.")
    parser.add_argument("--json", action="store_true", help="Print machine-readable JSON.")
    parser.add_argument("--no-impact", action="store_true", help="Omit human-readable provider impact output.")
    parser.add_argument("--strict-warnings", action="store_true", help="Fail when warnings are present.")
    args = parser.parse_args()

    workspace_root = resolve_workspace_root(args.workspace_root)
    repositories = repository_snapshots(workspace_root=workspace_root, include_website=args.include_website)
    contracts = collect_contracts(repositories)
    issues = validate_contracts(contracts)
    providers = provided_interface_map(contracts)
    consumers = interface_impact_map(contracts)

    payload = {
        "workspace_root": str(workspace_root),
        "repository_count": len(repositories),
        "contract_count": len(contracts),
        "provider_count": sum(len(contract.provides_interfaces) for contract in contracts),
        "requirement_count": sum(len(contract.requires_interfaces) for contract in contracts),
        "providers": provider_payload(providers, consumers),
        "issues": to_jsonable(issues),
    }

    blockers = [issue for issue in issues if issue.severity == "blocker"]
    warnings = [issue for issue in issues if issue.severity == "warning"]
    exit_code = 1 if blockers or (args.strict_warnings and warnings) else 0

    if args.json:
        print(json.dumps(payload, indent=2, sort_keys=True))
        return exit_code

    print("Static module contract scan")
    print(f"Workspace: {workspace_root}")
    print(
        f"Contracts: {payload['contract_count']} modules, "
        f"{payload['provider_count']} providers, {payload['requirement_count']} requirements"
    )

    if not args.no_impact:
        print()
        print("Provider impact:")
        if providers:
            for interface_name, provider_items in providers.items():
                provider_text = ", ".join(
                    f"{contract.module_id}@{provider.version} ({contract.repo})"
                    for contract, provider in provider_items
                )
                consumer_text = ", ".join(
                    f"{contract.module_id} ({format_version_range(version_min=req.version_min, version_max_exclusive=req.version_max_exclusive)})"
                    for contract, req in consumers.get(interface_name, ())
                )
                if not consumer_text:
                    consumer_text = "no consumers"
                print(f"- {interface_name}: {provider_text}; consumers: {consumer_text}")
        else:
            print("- no provided interfaces found")

    if issues:
        print()
        print("Contract issues:")
        for issue in issues:
            location = issue.repo or "workspace"
            if issue.module_id:
                location += f"/{issue.module_id}"
            print(f"- [{issue.severity}] {issue.code}: {location}: {issue.message}")
    else:
        print()
        print("Contract check passed.")

    return exit_code


def repository_snapshots(*, workspace_root: Path, include_website: bool) -> tuple[RepositorySnapshot, ...]:
    snapshots: list[RepositorySnapshot] = []
    for spec in load_repository_specs(include_website=include_website):
        path = resolve_repo_path(spec, workspace_root)
        snapshots.append(
            RepositorySnapshot(
                spec=spec,
                absolute_path=str(path),
                exists=path.exists(),
                is_git=(path / ".git").exists(),
            )
        )
    return tuple(snapshots)


def provider_payload(providers, consumers):
    payload: dict[str, dict[str, list[dict[str, str | bool | None]]]] = {}
    for interface_name, provider_items in providers.items():
        payload[interface_name] = {
            "providers": [
                {
                    "repo": contract.repo,
                    "module_id": contract.module_id,
                    "version": provider.version,
                    "manifest_path": contract.manifest_path,
                }
                for contract, provider in provider_items
            ],
            "consumers": [
                {
                    "repo": contract.repo,
                    "module_id": contract.module_id,
                    "version_min": requirement.version_min,
                    "version_max_exclusive": requirement.version_max_exclusive,
                    "optional": requirement.optional,
                    "manifest_path": contract.manifest_path,
                }
                for contract, requirement in consumers.get(interface_name, ())
            ],
        }
    return payload


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