HartOMat/backend/tests/integration/test_shadow_render_parity.py

"""Shadow render parity test.

Compares a shadow-mode render output against the authoritative legacy output
for the same order line to verify that graph-path and legacy-path produce
visually equivalent results.

Usage (manual integration test, requires running stack):
    python -m pytest backend/tests/integration/test_shadow_render_parity.py -v -s

    # Or run the standalone pixel-diff helper directly:
    python backend/tests/integration/test_shadow_render_parity.py \\
        --legacy /path/to/legacy.png --shadow /path/to/shadow.png

Thresholds (tuneable via env vars or CLI flags):
    MAX_CHANGED_PCT   — max fraction of pixels that may differ (default 0.5 %)
    MAX_CHANNEL_DIFF  — max per-channel absolute difference allowed (default 3)
"""
from __future__ import annotations

import argparse
import os
import sys
from pathlib import Path
from typing import NamedTuple

# ---------------------------------------------------------------------------
# Core comparison logic (no pytest dependency)
# ---------------------------------------------------------------------------

class DiffResult(NamedTuple):
    legacy_size: tuple[int, int]
    shadow_size: tuple[int, int]
    total_pixels: int
    changed_pixels: int
    changed_pct: float
    mean_diff: float
    max_diff: int
    p99_diff: float
    size_mismatch: bool

    @property
    def passed(self) -> bool:
        return (
            not self.size_mismatch
            and self.changed_pct <= float(os.environ.get("MAX_CHANGED_PCT", "0.5"))
            and self.max_diff <= int(os.environ.get("MAX_CHANNEL_DIFF", "3"))
        )

    def summary(self) -> str:
        lines = [
            f"  Size:            {self.legacy_size[0]}x{self.legacy_size[1]} px",
            f"  Changed pixels:  {self.changed_pixels:,} / {self.total_pixels:,} ({self.changed_pct:.3f}%)",
            f"  Mean diff:       {self.mean_diff:.4f}",
            f"  Max diff:        {self.max_diff}",
            f"  p99 diff:        {self.p99_diff:.1f}",
            f"  Result:          {'PASS' if self.passed else 'FAIL'}",
        ]
        if self.size_mismatch:
            lines.insert(0, f"  Shadow size:     {self.shadow_size[0]}x{self.shadow_size[1]} px  ← MISMATCH")
        return "\n".join(lines)


def compare_renders(legacy_path: Path, shadow_path: Path) -> DiffResult:
    """Compare two PNG render outputs. Returns a DiffResult with statistics."""
    try:
        import numpy as np
        from PIL import Image
    except ImportError as exc:
        raise RuntimeError(
            "Pillow and numpy are required for render parity tests. "
            "pip install Pillow numpy"
        ) from exc

    legacy_img = Image.open(legacy_path).convert("RGB")
    shadow_img = Image.open(shadow_path).convert("RGB")
    size_mismatch = legacy_img.size != shadow_img.size

    if size_mismatch:
        shadow_img = shadow_img.resize(legacy_img.size, Image.LANCZOS)

    la = np.array(legacy_img, dtype=np.int32)
    sa = np.array(shadow_img, dtype=np.int32)
    diff = np.abs(la - sa)

    total_pixels = legacy_img.size[0] * legacy_img.size[1]
    changed_mask = np.any(diff > 0, axis=2)
    changed_pixels = int(changed_mask.sum())
    changed_pct = 100.0 * changed_pixels / total_pixels
    mean_diff = float(diff.mean())
    max_diff = int(diff.max())
    nonzero_vals = diff[diff > 0]
    p99_diff = float(np.percentile(nonzero_vals, 99)) if nonzero_vals.size > 0 else 0.0

    return DiffResult(
        legacy_size=legacy_img.size,
        shadow_size=shadow_img.size,
        total_pixels=total_pixels,
        changed_pixels=changed_pixels,
        changed_pct=changed_pct,
        mean_diff=mean_diff,
        max_diff=max_diff,
        p99_diff=p99_diff,
        size_mismatch=size_mismatch,
    )


def save_diff_images(
    legacy_path: Path,
    shadow_path: Path,
    out_dir: Path,
) -> tuple[Path, Path]:
    """Save amplified diff image and side-by-side comparison. Returns (diff, comparison)."""
    import numpy as np
    from PIL import Image

    legacy_img = Image.open(legacy_path).convert("RGB")
    shadow_img = Image.open(shadow_path).convert("RGB")
    if legacy_img.size != shadow_img.size:
        shadow_img = shadow_img.resize(legacy_img.size, Image.LANCZOS)

    la = np.array(legacy_img, dtype=np.int32)
    sa = np.array(shadow_img, dtype=np.int32)
    diff = np.abs(la - sa)
    diff_amplified = (diff.astype(np.float32) * 10).clip(0, 255).astype(np.uint8)
    diff_img = Image.fromarray(diff_amplified)

    out_dir.mkdir(parents=True, exist_ok=True)
    diff_path = out_dir / "diff_amplified.png"
    diff_img.save(diff_path)

    THUMB_W = 600
    ratio = THUMB_W / legacy_img.size[0]
    THUMB_H = int(legacy_img.size[1] * ratio)
    leg_t = legacy_img.resize((THUMB_W, THUMB_H), Image.LANCZOS)
    sha_t = shadow_img.resize((THUMB_W, THUMB_H), Image.LANCZOS)
    diff_t = diff_img.resize((THUMB_W, THUMB_H), Image.LANCZOS)

    comparison = Image.new("RGB", (THUMB_W * 3, THUMB_H))
    comparison.paste(leg_t, (0, 0))
    comparison.paste(sha_t, (THUMB_W, 0))
    comparison.paste(diff_t, (THUMB_W * 2, 0))
    comparison_path = out_dir / "comparison_legacy_shadow_diff.png"
    comparison.save(comparison_path)

    return diff_path, comparison_path


# ---------------------------------------------------------------------------
# DB helpers — find shadow render pairs from the live database
# ---------------------------------------------------------------------------

def _find_shadow_pairs(limit: int = 10) -> list[tuple[Path, Path, str]]:
    """Query the live DB for completed shadow renders and return (legacy, shadow, line_id) triples."""
    import sqlalchemy as sa

    db_url = os.environ.get(
        "DATABASE_URL_SYNC",
        "postgresql://hartomat:hartomat@localhost:5432/hartomat",
    )
    engine = sa.create_engine(db_url)
    with engine.connect() as conn:
        rows = conn.execute(
            sa.text("""
                SELECT
                    ol.id::text,
                    ol.result_path,
                    wnr.output->>'observer_result_path' AS shadow_path
                FROM order_lines ol
                JOIN output_types ot ON ol.output_type_id = ot.id
                JOIN workflow_runs wr ON wr.order_line_id = ol.id
                JOIN workflow_node_results wnr ON wnr.run_id = wr.id
                WHERE ot.workflow_rollout_mode = 'shadow'
                  AND ol.render_status = 'completed'
                  AND wr.status = 'completed'
                  AND wnr.node_name = 'output'
                  AND wnr.output->>'observer_result_path' IS NOT NULL
                ORDER BY ol.created_at DESC
                LIMIT :limit
            """),
            {"limit": limit},
        ).fetchall()

    pairs: list[tuple[Path, Path, str]] = []
    upload_dir = Path(os.environ.get("UPLOAD_DIR", "/app/uploads"))

    for row in rows:
        line_id, legacy_rel, shadow_rel = row
        if not legacy_rel or not shadow_rel:
            continue
        legacy_abs = _resolve_upload_path(legacy_rel, upload_dir)
        shadow_abs = _resolve_upload_path(shadow_rel, upload_dir)
        if legacy_abs.exists() and shadow_abs.exists():
            pairs.append((legacy_abs, shadow_abs, line_id))

    return pairs


def _resolve_upload_path(raw: str, upload_dir: Path) -> Path:
    """Convert an /app/uploads/... path to the host-side absolute path."""
    p = Path(raw)
    try:
        relative = p.relative_to("/app/uploads")
        return upload_dir / relative
    except ValueError:
        return p


# ---------------------------------------------------------------------------
# pytest integration
# ---------------------------------------------------------------------------

def pytest_collect_file(parent, file_path):
    pass


try:
    import pytest

    @pytest.fixture(scope="module")
    def shadow_pairs():
        pairs = _find_shadow_pairs(limit=5)
        if not pairs:
            pytest.skip("No completed shadow renders found in the database.")
        return pairs

    @pytest.mark.integration
    def test_shadow_render_parity_all_pairs(shadow_pairs, tmp_path):
        """All found shadow renders must be within visual parity thresholds."""
        failures: list[str] = []

        for legacy_path, shadow_path, line_id in shadow_pairs:
            result = compare_renders(legacy_path, shadow_path)
            diff_path, comparison_path = save_diff_images(legacy_path, shadow_path, tmp_path / line_id[:8])
            status = "PASS" if result.passed else "FAIL"
            print(f"\n[{status}] order_line={line_id[:8]}...")
            print(result.summary())
            print(f"  Diff image → {diff_path}")
            print(f"  Comparison → {comparison_path}")
            if not result.passed:
                failures.append(
                    f"order_line={line_id[:8]} changed={result.changed_pct:.3f}% max_diff={result.max_diff}"
                )

        assert not failures, "Shadow render parity failures:\n" + "\n".join(failures)

    @pytest.mark.integration
    def test_shadow_render_parity_first_pair(shadow_pairs, tmp_path):
        """Quick smoke-test: the most recent shadow render must be within parity thresholds."""
        legacy_path, shadow_path, line_id = shadow_pairs[0]
        result = compare_renders(legacy_path, shadow_path)
        save_diff_images(legacy_path, shadow_path, tmp_path)
        assert not result.size_mismatch, f"Size mismatch: {result.legacy_size} vs {result.shadow_size}"
        assert result.changed_pct <= float(os.environ.get("MAX_CHANGED_PCT", "0.5")), (
            f"Too many changed pixels: {result.changed_pct:.3f}% > {os.environ.get('MAX_CHANGED_PCT', '0.5')}%\n"
            + result.summary()
        )
        assert result.max_diff <= int(os.environ.get("MAX_CHANNEL_DIFF", "3")), (
            f"Max channel diff too large: {result.max_diff} > {os.environ.get('MAX_CHANNEL_DIFF', '3')}\n"
            + result.summary()
        )

except ImportError:
    pass


# ---------------------------------------------------------------------------
# Standalone CLI
# ---------------------------------------------------------------------------

def _cli() -> None:
    parser = argparse.ArgumentParser(
        description="Compare a legacy render and a shadow render for visual parity."
    )
    parser.add_argument("--legacy", type=Path, help="Path to authoritative legacy render PNG")
    parser.add_argument("--shadow", type=Path, help="Path to shadow render PNG")
    parser.add_argument("--out-dir", type=Path, default=Path("/tmp/shadow_parity"), help="Output directory for diff images")
    parser.add_argument("--from-db", action="store_true", help="Fetch most recent shadow pair from the live DB")
    parser.add_argument("--max-changed-pct", type=float, default=0.5, help="Threshold for changed pixel fraction (default 0.5)")
    parser.add_argument("--max-channel-diff", type=int, default=3, help="Threshold for max channel diff (default 3)")
    args = parser.parse_args()

    os.environ["MAX_CHANGED_PCT"] = str(args.max_changed_pct)
    os.environ["MAX_CHANNEL_DIFF"] = str(args.max_channel_diff)

    if args.from_db:
        print("Querying DB for most recent shadow render pair...")
        pairs = _find_shadow_pairs(limit=1)
        if not pairs:
            print("ERROR: No completed shadow renders found in the database.")
            sys.exit(1)
        legacy_path, shadow_path, line_id = pairs[0]
        print(f"Using order_line: {line_id}")
    elif args.legacy and args.shadow:
        legacy_path, shadow_path = args.legacy, args.shadow
    else:
        parser.print_help()
        sys.exit(1)

    print(f"Legacy: {legacy_path}")
    print(f"Shadow: {shadow_path}")

    result = compare_renders(legacy_path, shadow_path)
    diff_path, comparison_path = save_diff_images(legacy_path, shadow_path, args.out_dir)

    print(f"\nResult:\n{result.summary()}")
    print(f"Diff image (10× amplified) → {diff_path}")
    print(f"Side-by-side comparison    → {comparison_path}")

    sys.exit(0 if result.passed else 1)


if __name__ == "__main__":
    _cli()