"""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()