1from __future__ import annotations
  2
  3from collections.abc import Callable, Generator
  4from typing import Any, TypeVar
  5
  6from plain.runtime import settings
  7
  8from .checks import PreflightCheck
  9from .results import PreflightResult, unused_silenced_results
 10
 11T = TypeVar("T")
 12
 13UNUSED_SILENCES_CHECK_NAME = "preflight.unused_silences"
 14
 15
 16class CheckUnusedSilences(PreflightCheck):
 17    """Reports `PREFLIGHT_SILENCED_RESULTS` entries that matched nothing.
 18
 19    An unused entry is either a typo or stale — the issue it silenced has
 20    been fixed. Not registered like a normal check: it needs every other
 21    check's results, so the registry runs it last with the full run's
 22    results, and only on a full run (deploy checks included) — a partial
 23    run skips checks whose entries would then look unused.
 24    """
 25
 26    def __init__(self, run_results: list[PreflightResult]) -> None:
 27        self.run_results = run_results
 28
 29    def run(self) -> list[PreflightResult]:
 30        return [
 31            PreflightResult(
 32                fix=f"Silenced result {entry!r} matched nothing in this run. "
 33                "Remove it from PREFLIGHT_SILENCED_RESULTS or fix the typo.",
 34                obj=entry,
 35                id="preflight.unused_silence",
 36                warning=True,
 37            )
 38            for entry in unused_silenced_results(self.run_results)
 39        ]
 40
 41
 42class CheckRegistry:
 43    def __init__(self) -> None:
 44        self.checks: dict[
 45            str, tuple[type[Any], bool]
 46        ] = {}  # name -> (check_class, deploy)
 47
 48    def register_check(
 49        self, check_class: type[Any], name: str, deploy: bool = False
 50    ) -> None:
 51        """Register a check class with a unique name."""
 52        if name in self.checks:
 53            raise ValueError(f"Check {name} already registered")
 54        self.checks[name] = (check_class, deploy)
 55
 56    def run_checks(
 57        self,
 58        include_deploy_checks: bool = False,
 59    ) -> Generator[tuple[type[Any], str, list[PreflightResult]]]:
 60        """
 61        Run all registered checks and yield (check_class, name, results) tuples.
 62        """
 63        # Validate silenced check names (the unused-silences check isn't in
 64        # self.checks — the registry emits it itself — but it's silenceable
 65        # like any other check)
 66        silenced_checks = settings.PREFLIGHT_SILENCED_CHECKS
 67        known_checks = set(self.checks.keys()) | {UNUSED_SILENCES_CHECK_NAME}
 68        unknown_silenced = set(silenced_checks) - known_checks
 69        if unknown_silenced:
 70            unknown_names = ", ".join(sorted(unknown_silenced))
 71            raise ValueError(
 72                f"Unknown check names in PREFLIGHT_SILENCED_CHECKS: {unknown_names}. "
 73                "Check for typos or remove outdated check names."
 74            )
 75
 76        all_results: list[PreflightResult] = []
 77
 78        for name, (check_class, deploy) in sorted(self.checks.items()):
 79            # Skip silenced checks
 80            if name in silenced_checks:
 81                continue
 82
 83            # Skip deployment checks if not requested
 84            if deploy and not include_deploy_checks:
 85                continue
 86
 87            # Instantiate and run check
 88            check = check_class()
 89            results = check.run()
 90            all_results.extend(results)
 91            yield check_class, name, results
 92
 93        if include_deploy_checks and UNUSED_SILENCES_CHECK_NAME not in silenced_checks:
 94            check = CheckUnusedSilences(all_results)
 95            yield CheckUnusedSilences, UNUSED_SILENCES_CHECK_NAME, check.run()
 96
 97
 98checks_registry = CheckRegistry()
 99
100
101def register_check(name: str, *, deploy: bool = False) -> Callable[[type[T]], type[T]]:
102    """
103    Decorator to register a check class.
104
105    Usage:
106        @register_check("security.secret_key", deploy=True)
107        class CheckSecretKey(PreflightCheck):
108            pass
109
110        @register_check("files.upload_temp_dir")
111        class CheckUploadTempDir(PreflightCheck):
112            pass
113    """
114
115    def wrapper(cls: type[T]) -> type[T]:
116        checks_registry.register_check(cls, name=name, deploy=deploy)
117        return cls
118
119    return wrapper
120
121
122run_checks = checks_registry.run_checks
123
124# Cached error/warning counts — populated on first call, refreshed by
125# PreflightView when the full page is viewed.
126_check_counts: dict[str, int] | None = None
127
128
129def get_check_counts() -> dict[str, int]:
130    """Return ``{"errors": N, "warnings": N}``, caching for the process lifetime."""
131    global _check_counts
132
133    if _check_counts is not None:
134        return _check_counts
135
136    from plain.packages import packages_registry
137
138    packages_registry.autodiscover_modules("preflight", include_app=True)
139
140    warning_count = 0
141    error_count = 0
142
143    for _check_class, _name, results in run_checks(
144        include_deploy_checks=not settings.DEBUG
145    ):
146        visible = [r for r in results if not r.is_silenced()]
147        if not visible:
148            continue
149        if any(not r.warning for r in visible):
150            error_count += 1
151        else:
152            warning_count += 1
153
154    _check_counts = {"errors": error_count, "warnings": warning_count}
155    return _check_counts
156
157
158def set_check_counts(*, errors: int, warnings: int) -> None:
159    """Update the cached counts (called by PreflightView after running full checks)."""
160    global _check_counts
161    _check_counts = {"errors": errors, "warnings": warnings}