LEDMatrix/scripts/check_plugin.py

#!/usr/bin/env python3
"""
Plugin safety checker.

Renders a plugin across every declared screen (mode) and every supported matrix
size, and fails if any screen crashes, overflows the panel, or (for plugins with
committed golden images) drifts visually.

Usage:
    # Functional + bounds check across all sizes/modes:
    python scripts/check_plugin.py --plugin clock-simple

    # Every discovered plugin:
    python scripts/check_plugin.py --all

    # Dump PNGs for each size/mode so you can eyeball them:
    python scripts/check_plugin.py --plugin ledmatrix-weather --out-dir /tmp/preview

    # Refresh committed golden images after an intentional visual change:
    python scripts/check_plugin.py --plugin clock-simple --update-golden \
        --mock-data plugins/clock-simple/test/fixtures/mock.json

Exit code is non-zero if any (plugin, size, mode) fails.
"""

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Dict, List, Optional

PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

os.environ['EMULATOR'] = 'true'


def _make_output_encoding_safe() -> None:
    """Stop an unencodable character from killing the run.

    This script's own report is ASCII, but it echoes text it does not control
    -- plugin ids, mode names and exception messages -- and a Windows console
    is cp1252, which cannot encode most of what a plugin might put there. The
    default 'strict' error handler turns that into a UnicodeEncodeError from
    inside `print`, so a rendering run that had already succeeded exited
    non-zero with a traceback instead of printing its results.

    'replace' degrades the offending character to '?' and keeps going; the
    encoding itself is left alone so output still matches the terminal.
    """
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(errors='replace')
        except (AttributeError, ValueError, OSError):
            # Not a reconfigurable TextIOWrapper (redirected, wrapped by a
            # test harness). Nothing to do -- this is best-effort hardening.
            pass


_make_output_encoding_safe()

from src.logging_config import get_logger  # noqa: E402
from src.plugin_system.testing.loading import (  # noqa: E402
    build_full_config, find_plugin_dir, load_harness_spec, load_manifest,
)
from src.plugin_system.testing.harness import (  # noqa: E402
    RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
    check_empty_claimed,
    check_scale_up,
)
from src.plugin_system.testing.sizes import (  # noqa: E402
    parse_size_token, resolve_test_sizes, safe_mode_filename, size_label,
)
from src.plugin_system.testing.vegas import check_plugin_vegas_elements  # noqa: E402

logger = get_logger("[Check Plugin]")

DEFAULT_SEARCH_DIRS = [
    str(PROJECT_ROOT / 'plugins'),
    str(PROJECT_ROOT / 'plugin-repos'),
    # The scoreboards live in the sibling ledmatrix-plugins checkout, not
    # in this repo. Without this, --all silently skips every one of them.
    str(PROJECT_ROOT.parent / 'ledmatrix-plugins' / 'plugins'),
]


def discover_plugins(search_dirs: List[str]) -> List[str]:
    """All plugin ids found across the search dirs (dirs containing manifest.json)."""
    found = []
    for d in search_dirs:
        base = Path(d)
        if not base.exists():
            continue
        for child in sorted(base.iterdir()):
            if (child / 'manifest.json').exists() and child.name not in found:
                found.append(child.name)
    return found


def parse_sizes(spec: Optional[str]):
    if not spec:
        return None
    sizes = []
    for token in spec.split(','):
        if not token.strip():
            continue
        try:
            sizes.append(parse_size_token(token))
        except ValueError as exc:
            raise SystemExit(str(exc)) from exc
    return sizes


def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
              config: Dict, run_update: bool, out_dir: Optional[Path],
              update_golden: bool, golden_dir_override: Optional[Path],
              freeze_time: Optional[str]) -> List[RenderResult]:
    plugin_dir = find_plugin_dir(plugin_id, search_dirs)
    if not plugin_dir:
        logger.error("Plugin '%s' not found in: %s", plugin_id, search_dirs)
        return [RenderResult(plugin_id, 0, 0, "<not-found>", error="plugin directory not found")]

    # Per-plugin test/harness.json holds the deterministic settings the committed
    # goldens were generated with (config, mock data, frozen time, sizes). Load
    # them so the CLI/CI render reproduces the golden the same way the pytest
    # matrix path does; explicit CLI flags still override the file.
    spec = load_harness_spec(plugin_dir)

    # config_schema defaults (real-install behavior, with enabled forced True
    # so a plugin's own enabled:false default can't accidentally disable
    # testing), then harness.json config, then CLI --config — most specific
    # wins.
    full_config = build_full_config(plugin_dir, spec, config)

    # Precedence: CLI flag > LEDMATRIX_TEST_SIZES env > harness.json > default.
    effective_sizes = sizes if sizes else resolve_test_sizes(spec.get("sizes"))
    # CLI value wins when provided, else fall back to the harness.json setting.
    effective_mock_data = mock_data or spec.get("mock_data_contents", {})
    effective_freeze = freeze_time or spec.get("freeze_time")
    effective_run_update = run_update and not spec.get("skip_update", False)

    # The plugin's declared design size drives the scale-up fill check
    # (panels >= 2x the design size must not be left mostly empty).
    declared = load_manifest(plugin_dir).get("display", {}).get("design_size", {})
    design_size = (int(declared.get("width", 128)), int(declared.get("height", 32)))
    fill_strict = spec.get("fill_check") == "strict"
    # A mode that renders nothing without returning False is never skipped by
    # the display controller, so it holds a blank panel for its whole duration.
    # Warn-only by default: a scroll mode's first frame is legitimately its
    # blank scroll-in buffer.
    empty_strict = spec.get("empty_check") == "strict"

    # Every run: the base config, plus one per harness.json "variant" —
    # a config overlay with its own golden dir (e.g. adaptive layout mode
    # tested alongside the classic default).
    runs = [(None, {}, golden_dir_override or (plugin_dir / 'test' / 'golden'))]
    for variant in spec.get("variants", []):
        name = variant.get("name") or "variant"
        vdir = plugin_dir / variant.get("golden_dir", f"test/golden-{name}")
        runs.append((name, variant.get("config", {}), vdir))

    all_run_results: List[RenderResult] = []
    for variant_name, overlay, golden_dir in runs:
        run_config = {**full_config, **overlay}
        results = render_plugin_matrix(
            plugin_id=plugin_id, plugin_dir=plugin_dir, config=run_config,
            mock_data=effective_mock_data, sizes=effective_sizes,
            run_update=effective_run_update, freeze_time=effective_freeze,
        )

        if update_golden:
            written = write_goldens(results, golden_dir)
            logger.info("Wrote %d golden image(s) for %s%s to %s", written, plugin_id,
                        f" [{variant_name}]" if variant_name else "", golden_dir)
        else:
            compare_to_goldens(results, golden_dir)

        check_scale_up(results, design_size=design_size, strict=fill_strict)
        check_empty_claimed(results, strict=empty_strict)

        # Tag variant runs so the report and PNG dumps stay distinguishable.
        if variant_name:
            for r in results:
                r.mode = f"{r.mode}@{variant_name}"

        if out_dir:
            for r in results:
                if r.image is None:
                    continue
                dest = out_dir / plugin_id / size_label(r.width, r.height)
                dest.mkdir(parents=True, exist_ok=True)
                r.image.save(dest / f"{safe_mode_filename(r.mode)}.png", format="PNG")

        all_run_results.extend(results)

    # Live Vegas elements, for a plugin that has them: checked once, at the
    # first size, with the base config.
    width, height = effective_sizes[0]
    try:
        vegas = check_plugin_vegas_elements(
            plugin_id, plugin_dir, full_config, effective_mock_data, width, height,
            run_update=effective_run_update)
    except Exception as exc:  # noqa: BLE001 - one plugin must not end an --all run
        all_run_results.append(RenderResult(
            plugin_id, width, height, "vegas elements",
            error=f"the element check itself failed: {exc!r}"))
        return all_run_results
    if vegas.implemented:
        all_run_results.append(RenderResult(
            plugin_id, width, height, "vegas elements",
            error="; ".join(vegas.errors) or None,
            notes=[f"{vegas.elements} element(s), {vegas.live} live"] + vegas.warnings))

    return all_run_results


def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
    """Print a per-plugin grid. Returns True if everything passed."""
    everything_ok = True
    for plugin_id, results in all_results.items():
        print(f"\n=== {plugin_id} ===")
        for r in results:
            if r.ok:
                status = "PASS"
                detail = ""
                if r.golden_checked:
                    detail = " (golden ok)"
                if r.update_error is not None:
                    detail += f" (update warn: {r.update_error})"
                if r.fill_checked and r.fill_ok is None and r.fill_extent:
                    # warn-only underfill: big panel left mostly empty
                    ex, ey = r.fill_extent
                    detail += f" (fill warn: extent {ex:.0%}x{ey:.0%})"
                if r.empty_claimed and r.empty_ok is None:
                    detail += (f" (empty warn: drew nothing but display() returned"
                               f" {r.display_returned!r}, so the mode is not skipped)")
            else:
                everything_ok = False
                if r.error is not None:
                    status, detail = "FAIL", f" error={r.error}"
                elif r.overflow is not None:
                    status, detail = "FAIL", f" overflow bbox={r.overflow}"
                elif r.golden_ok is False:
                    status = "FAIL"
                    detail = f" golden drift: {r.golden_diff_pixels}px (max delta={r.golden_max_delta})"
                elif r.fill_ok is False:
                    ex, ey = r.fill_extent or (0.0, 0.0)
                    status = "FAIL"
                    detail = f" fill: extent {ex:.0%}x{ey:.0%} below required coverage"
                elif r.empty_ok is False:
                    status = "FAIL"
                    detail = (f" drew nothing but display() returned"
                              f" {r.display_returned!r}; return False so the"
                              f" controller skips the mode")
                else:
                    status, detail = "FAIL", ""
            if r.notes:
                detail += f" ({'; '.join(r.notes)})"
            print(f"  [{status}] {r.size_label:>7}  {r.mode}{detail}")
    print()
    return everything_ok


def main() -> int:
    parser = argparse.ArgumentParser(description="Check a plugin renders safely across sizes & screens")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument('--plugin', '-p', help='Plugin id to check')
    group.add_argument('--all', action='store_true', help='Check every discovered plugin')
    parser.add_argument('--plugin-dir', '-d', default=None, help='Directory to search for plugins')
    parser.add_argument('--sizes', default=None, help='Comma-separated WxH list (default: all supported)')
    parser.add_argument('--config', '-c', default='{}', help='Plugin config overrides as JSON')
    parser.add_argument('--mock-data', '-m', default=None, help='Path to JSON file with mock cache data')
    parser.add_argument('--out-dir', '-o', default=None, help='Also dump rendered PNGs here')
    parser.add_argument('--skip-update', action='store_true', help='Skip calling update()')
    parser.add_argument('--update-golden', action='store_true', help='Write/refresh golden images')
    parser.add_argument('--golden-dir', default=None, help='Override golden dir (default: <plugin>/test/golden)')
    parser.add_argument('--freeze-time', default=None,
                        help='Freeze wall clock, e.g. "2025-08-01 15:25:00" (for time-dependent plugins)')
    args = parser.parse_args()

    search_dirs = [args.plugin_dir] if args.plugin_dir else DEFAULT_SEARCH_DIRS
    sizes = parse_sizes(args.sizes)

    try:
        config = json.loads(args.config)
    except json.JSONDecodeError as e:
        logger.error("Invalid --config JSON: %s", e)
        return 2
    if not isinstance(config, dict):
        logger.error("--config must be a JSON object, got %s", type(config).__name__)
        return 2

    mock_data = {}
    if args.mock_data:
        mock_path = Path(args.mock_data)
        if not mock_path.exists():
            logger.error("Mock data file not found: %s", args.mock_data)
            return 2
        with open(mock_path) as f:
            mock_data = json.load(f)
        if not isinstance(mock_data, dict):
            logger.error("--mock-data must be a JSON object (key -> cache value), got %s",
                         type(mock_data).__name__)
            return 2

    plugin_ids = discover_plugins(search_dirs) if args.all else [args.plugin]
    if not plugin_ids:
        logger.error("No plugins found in: %s", search_dirs)
        return 2

    out_dir = Path(args.out_dir) if args.out_dir else None
    golden_dir_override = Path(args.golden_dir) if args.golden_dir else None

    all_results: Dict[str, List[RenderResult]] = {}
    for plugin_id in plugin_ids:
        all_results[plugin_id] = check_one(
            plugin_id=plugin_id, search_dirs=search_dirs, sizes=sizes,
            mock_data=mock_data, config=config, run_update=not args.skip_update,
            out_dir=out_dir, update_golden=args.update_golden,
            golden_dir_override=golden_dir_override, freeze_time=args.freeze_time,
        )

    # When refreshing goldens we skip drift comparison, but a crash or overflow
    # still means the plugin is broken — never let --update-golden mask that.
    ok = print_report(all_results)
    return 0 if ok else 1


if __name__ == '__main__':
    sys.exit(main())