tests/test_forge_bridge.py

"""Tests for the vendored-forge subprocess bridge (app/forge_bridge.py).

These exercise the REAL scripts end-to-end: probe, strict gate, honest hosted
preview manifest construction, and pass-gated factory generation.
"""

from __future__ import annotations

import copy
import json
import subprocess
import sys
from pathlib import Path

import pytest

from app import forge_bridge
from app.forge_bridge import ForgeError

REPO_ROOT = Path(__file__).resolve().parents[1]
FIXTURES = Path(__file__).resolve().parent / "fixtures"


def write_png(path: Path, w: int = 64, h: int = 64) -> Path:
    import struct
    import zlib

    raw = bytearray()
    for y in range(h):
        raw.append(0)
        for x in range(w):
            raw += bytes(((x * 4) % 256, (y * 4) % 256, ((x + y) * 2) % 256))

    def chunk(tag, data):
        c = struct.pack(">I", len(data)) + tag + data
        return c + struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)

    ihdr = struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)
    path.write_bytes(b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr)
                     + chunk(b"IDAT", zlib.compress(bytes(raw), 9)) + chunk(b"IEND", b""))
    return path


class TestProbe:
    def test_probe_parses_json(self, tmp_path):
        image = write_png(tmp_path / "ref.png")
        payload = forge_bridge.probe_image(image)
        assert payload["width"] == 64 and payload["height"] == 64
        assert payload["type"] == "png"

    def test_probe_missing_file_raises(self, tmp_path):
        with pytest.raises(ForgeError):
            forge_bridge.probe_image(tmp_path / "nope.png")

    def test_probe_env_is_scrubbed(self, tmp_path, monkeypatch):
        monkeypatch.setenv("LLM_API_KEY", "should-not-leak")
        image = write_png(tmp_path / "ref.png")
        # If env leaked this would still pass; assert the scrub map directly.
        env = forge_bridge._scrubbed_env()
        assert "LLM_API_KEY" not in env
        assert "ANTHROPIC_API_KEY" not in env
        assert env["PATH"]


class TestValidate:
    def test_canned_spec_passes_strict(self, tmp_path):
        """Drift guard: the committed fixture must pass the vendored gate."""
        spec_path = tmp_path / "spec.json"
        spec_path.write_text((FIXTURES / "canned_spec.json").read_text(encoding="utf-8"),
                             encoding="utf-8")
        result = forge_bridge.validate_spec(spec_path, strict=True)
        assert result["ok"] is True, result.get("errors")

    def test_shallow_starter_spec_fails_strict(self, tmp_path):
        """Mirror of the upstream gate test: an unassessed starter spec must
        be blocked by --strict-quality."""
        assessment = tmp_path / "a.json"
        spec = tmp_path / "s.json"
        subprocess.run(
            [sys.executable, str(forge_bridge.FORGE / "stage2_spec" / "new_pre_spec_assessment.py"),
             "Widget", "--complexity", "simple", "--out", str(assessment)],
            check=True, capture_output=True)
        subprocess.run(
            [sys.executable, str(forge_bridge.FORGE / "stage2_spec" / "new_sculpt_spec.py"),
             "Widget", "--assessment", str(assessment), "--out", str(spec)],
            check=True, capture_output=True)
        # Non-strict validation passes (structure is fine)…
        relaxed = forge_bridge.validate_spec(spec, strict=False)
        assert relaxed["ok"] is True
        # …but the strict-quality gate blocks it as too shallow.
        strict = forge_bridge.validate_spec(spec, strict=True)
        assert strict["ok"] is False
        assert strict["errors"]

    @pytest.mark.parametrize(
        "invalid", [True, float("nan"), float("inf"), 10**400]
    )
    def test_validator_rejects_bool_and_non_finite_vectors(
        self, tmp_path, canned_spec, invalid
    ):
        spec = copy.deepcopy(canned_spec)
        spec["componentTree"][0]["transform"]["position"][0] = invalid
        spec_path = tmp_path / "spec.json"
        spec_path.write_text(json.dumps(spec), encoding="utf-8")
        result = forge_bridge.validate_spec(spec_path, strict=False)
        assert result["ok"] is False
        assert any("transform.position must be [number, number, number]" in error
                   for error in result["errors"])

    def test_validator_requires_renderable_mirrored_color_gradient(
        self, tmp_path, canned_spec
    ):
        gradient = {
            "type": "linear",
            "axis": [1.0, 0.0],
            "stops": [
                {"offset": 0.0, "color": "rgba(30, 40, 50, 1.0)"},
                {"offset": 1.0, "color": "rgba(130, 140, 150, 1.0)"},
            ],
        }
        spec = copy.deepcopy(canned_spec)
        component = spec["componentTree"][0]
        component["colorMaterialRecipe"]["colorGradient"] = gradient
        material = next(
            item for item in spec["materials"]
            if item["id"] == component["material"]
        )
        material["colorGradient"] = copy.deepcopy(gradient)
        path = tmp_path / "valid-gradient.json"
        path.write_text(json.dumps(spec), encoding="utf-8")
        assert forge_bridge.validate_spec(path, strict=True)["ok"] is True

        del material["colorGradient"]
        path.write_text(json.dumps(spec), encoding="utf-8")
        unmirrored = forge_bridge.validate_spec(path, strict=True)
        assert unmirrored["ok"] is False
        assert any("mirrored exactly" in error for error in unmirrored["errors"])

        material["colorGradient"] = {
            **gradient,
            "axis": [True, 0.0],
        }
        component["colorMaterialRecipe"]["colorGradient"] = copy.deepcopy(
            material["colorGradient"]
        )
        path.write_text(json.dumps(spec), encoding="utf-8")
        malformed = forge_bridge.validate_spec(path, strict=True)
        assert malformed["ok"] is False
        assert any("axis must be two finite numbers" in error
                   for error in malformed["errors"])


class TestGeneration:
    def _write_spec(self, tmp_path: Path, spec: dict) -> Path:
        path = tmp_path / "spec.json"
        path.write_text(json.dumps(spec), encoding="utf-8")
        return path

    def test_locked_pass_refused_without_reviews(self, tmp_path, canned_spec):
        spec_path = self._write_spec(tmp_path, dict(canned_spec))
        out = tmp_path / "out.ts"
        with pytest.raises(ForgeError):
            forge_bridge.generate_factory(spec_path, out, pass_id="material-pass")

    def test_hosted_preview_does_not_approve_and_generates(self, tmp_path, canned_spec):
        spec = copy.deepcopy(canned_spec)
        original = copy.deepcopy(spec)
        preview = forge_bridge.prepare_hosted_preview(spec)
        assert spec == original
        assert spec["reviewHistory"] == []
        assert preview["reviewHistory"] == []
        assert preview["sculptPipeline"]["passGateMode"] == "hosted-unreviewed-preview"
        assert preview["sculptPipeline"]["completedPasses"] == []
        assert set(preview["buildPasses"][0]["componentRefs"]) == {"body", "handle"}

        spec_path = self._write_spec(tmp_path, preview)

        out = tmp_path / "createMugModel.ts"
        forge_bridge.generate_factory(
            spec_path, out, pass_id=forge_bridge.HOSTED_PREVIEW_PASS)
        source = out.read_text(encoding="utf-8")
        assert "import * as THREE from 'three'" in source
        assert "sculptRuntime" in source
        assert "Compile-only hosted preview: unreviewed" in source
        assert "TODO:" not in source
        assert forge_bridge.factory_export_name(source) == "createMugModel"

    def test_generator_does_not_self_validate(self, tmp_path):
        """Regression guard: the generator exits 0 on garbage specs, so the
        adapter must never call it before the strict gate passes."""
        bad = self._write_spec(tmp_path, {"bad": True})
        out = tmp_path / "bad.ts"
        forge_bridge.generate_factory(bad, out, pass_id=None)
        assert out.exists()  # exits 0 and emits — validation is OUR job

    def test_factory_export_name(self):
        assert forge_bridge.factory_export_name(
            "export function createTeaPotModel(options) {}") == "createTeaPotModel"
        assert forge_bridge.factory_export_name("const x = 1;") is None


class TestPassHelpers:
    def test_pass_order(self, canned_spec):
        assert forge_bridge.pass_order(canned_spec) == [
            "blockout", "structural-pass", "material-pass"]

    def test_pass_order_default_on_garbage(self):
        assert forge_bridge.pass_order({"bad": True}) == ["blockout"]

    @pytest.mark.parametrize(
        "primitive",
        [
            "tube", "lathe", "extrude", "ground-blade", "curve-sweep",
            "instanced-cluster",
        ],
    )
    def test_v13_primitive_is_supported(self, canned_spec, primitive):
        spec = copy.deepcopy(canned_spec)
        spec["componentTree"][0]["primitive"] = primitive
        errors = forge_bridge.hosted_spec_errors(spec)
        assert not any("unsupported primitive" in error for error in errors)

    @pytest.mark.parametrize(
        ("primitive", "descriptor"),
        [
            (
                "extrude",
                {
                    "profile2D": {
                        "points": [[-0.4, -0.2], [0.4, -0.2], [0.0, 0.5]],
                        "depth": 0.12,
                        "holes": [[[-0.1, -0.1], [0.1, -0.1], [0.0, 0.1]]],
                        "ovalHoles": [{"cx": 0.0, "cy": 0.2, "rx": 0.04, "ry": 0.06}],
                    }
                },
            ),
            (
                "lathe",
                {
                    "latheProfile": {
                        "points": [[0.0, -0.5], [0.3, 0.0], [0.1, 0.5]],
                        "segments": 24,
                    }
                },
            ),
            (
                "tube",
                {
                    "tubePath": {
                        "points": [[0.0, -0.5, 0.0], [0.2, 0.0, 0.1], [0.0, 0.5, 0.0]],
                        "radius": 0.05,
                        "radialSegments": 8,
                        "closed": False,
                    }
                },
            ),
            (
                "ground-blade",
                {
                    "bladeSpec": {
                        "stations": [[0.0, 0.3, -0.2], [0.5, 0.25, -0.15]],
                        "thickness": 0.05,
                        "grindFrac": 0.55,
                        "swedgeFromTipFrac": 0.34,
                        "spineFlat": 0.3,
                    }
                },
            ),
            (
                "curve-sweep",
                {
                    "curveSweep": {
                        "spine": [[-0.5, 0.0, 0.0], [0.0, 0.2, 0.1], [0.5, 0.0, 0.0]],
                        "crossSection": {
                            "points": [[-0.05, -0.02], [0.05, -0.02], [0.0, 0.04]]
                        },
                        "closed": False,
                    }
                },
            ),
            ("instanced-cluster", {"baseGeometry": "box"}),
        ],
    )
    def test_advanced_geometry_descriptor_accepts_valid_payload(
        self, canned_spec, primitive, descriptor
    ):
        spec = copy.deepcopy(canned_spec)
        component = spec["componentTree"][0]
        component["primitive"] = primitive
        component["geometryDescriptor"] = descriptor
        assert forge_bridge.hosted_spec_errors(spec) == []

    @pytest.mark.parametrize(
        ("primitive", "descriptor", "message"),
        [
            (
                "extrude",
                {"profile2D": {"points": [[0, 0], [1, 0], [True, 1]], "depth": 0.1}},
                "profile2D.points",
            ),
            (
                "extrude",
                {"profile2D": {"points": [[0, 0], [1, 0], [0, 1]], "depth": float("inf")}},
                "profile2D.depth",
            ),
            (
                "extrude",
                {
                    "profile2D": {
                        "points": [[0, 0], [1, 0], [0, 1]],
                        "depth": 0.1,
                        "ovalHoles": [{"cx": 0, "cy": 0, "rx": False, "ry": 0.1}],
                    }
                },
                "profile2D.ovalHoles",
            ),
            (
                "lathe",
                {"latheProfile": {"points": [[-0.1, 0], [0.2, 1]], "segments": 24}},
                "latheProfile.points radii",
            ),
            (
                "lathe",
                {"latheProfile": {"points": [[0.1, 0], [0.2, 1]]}},
                "latheProfile.segments",
            ),
            (
                "tube",
                {"tubePath": {"points": [[0, 0, 0]], "radius": 0.05}},
                "tubePath.points",
            ),
            (
                "tube",
                {
                    "tubePath": {
                        "points": [[0, 0, 0], [0, 1, 0]],
                        "radius": False,
                    }
                },
                "tubePath.radius",
            ),
            (
                "tube",
                {
                    "tubePath": {
                        "points": [[0, 0, 0], [0, 1, 0]],
                        "radius": 0.05,
                    }
                },
                "tubePath.closed",
            ),
            (
                "tube",
                {
                    "tubePath": {
                        "points": [[0, 0, 0], [0, 1, 0]],
                        "radius": 0.05,
                        "radialSegments": 2,
                    }
                },
                "tubePath.radialSegments",
            ),
            (
                "ground-blade",
                {
                    "bladeSpec": {
                        "stations": [[0, 0.2, -0.2], [1, 0.2]],
                        "thickness": 0.05,
                    }
                },
                "bladeSpec.stations",
            ),
            (
                "ground-blade",
                {
                    "bladeSpec": {
                        "stations": [[0, 0.2, -0.2], [1, 0.3, -0.1]],
                        "grindFrac": 0.55,
                        "swedgeFromTipFrac": 0.34,
                        "spineFlat": 0.3,
                    }
                },
                "bladeSpec.thickness",
            ),
            (
                "ground-blade",
                {
                    "bladeSpec": {
                        "stations": [[0, 0.2, -0.2], [1, 0.3, -0.1]],
                        "thickness": 0.05,
                        "grindFrac": 0.7,
                        "spineFlat": 0.3,
                    }
                },
                "grindFrac + spineFlat must be less than 1",
            ),
            (
                "ground-blade",
                {
                    "bladeSpec": {
                        "stations": [[0, 0.2, -0.2], [1, 0.3, -0.1]],
                        "thickness": 0.05,
                        "grindFrac": 0.8,
                    }
                },
                "grindFrac + spineFlat must be less than 1",
            ),
            (
                "curve-sweep",
                {
                    "curveSweep": {
                        "spine": [[0, 0, 0], [1, 1, 1]],
                        "crossSection": {"points": [[0, 0], [1, 0]]},
                    }
                },
                "curveSweep.crossSection.points",
            ),
            (
                "curve-sweep",
                {
                    "curveSweep": {
                        "spine": [[0, 0, 0], [1, float("nan"), 1]],
                        "crossSection": {"points": [[0, 0], [1, 0], [0, 1]]},
                        "closed": False,
                    }
                },
                "curveSweep.spine",
            ),
            (
                "curve-sweep",
                {
                    "curveSweep": {
                        "spine": [[0, 0, 0], [1, 1, 1]],
                        "crossSection": {"points": [[0, 0], [1, 0], [0, 1]]},
                    }
                },
                "curveSweep.closed",
            ),
            (
                "instanced-cluster",
                {"baseGeometry": "instanced-cluster"},
                "cannot reference instanced-cluster itself",
            ),
            (
                "instanced-cluster",
                {"baseGeometry": "metaball"},
                "baseGeometry uses unsupported primitive",
            ),
        ],
    )
    def test_advanced_geometry_descriptor_rejects_malformed_payload(
        self, canned_spec, primitive, descriptor, message
    ):
        spec = copy.deepcopy(canned_spec)
        component = spec["componentTree"][0]
        component["primitive"] = primitive
        component["geometryDescriptor"] = descriptor
        errors = forge_bridge.hosted_spec_errors(spec)
        assert any(message in error for error in errors), errors

    def test_numeric_helpers_reject_bool_and_non_finite_values(self):
        assert forge_bridge._is_finite_number(0)
        assert forge_bridge._is_positive_number(0.1)
        for value in (True, False, float("nan"), float("inf"), float("-inf")):
            assert not forge_bridge._is_finite_number(value)
            assert not forge_bridge._is_positive_number(value)

    def test_parent_cycle_rejected(self, canned_spec):
        spec = copy.deepcopy(canned_spec)
        spec["componentTree"][0]["parent"] = "handle"
        errors = forge_bridge.hosted_spec_errors(spec)
        assert any("parent cycle" in error for error in errors)

    def test_missing_dimensions_rejected(self, canned_spec):
        spec = copy.deepcopy(canned_spec)
        spec["componentTree"][0]["dimensions"] = {"width": 0.08}
        errors = forge_bridge.hosted_spec_errors(spec)
        assert any("positive width, height" in error for error in errors)

    def test_child_before_parent_is_emitted_after_parent(self, tmp_path, canned_spec):
        spec = copy.deepcopy(canned_spec)
        spec["componentTree"].reverse()
        preview = forge_bridge.prepare_hosted_preview(spec)
        spec_path = tmp_path / "spec.json"
        spec_path.write_text(json.dumps(preview), encoding="utf-8")
        out = tmp_path / "out.ts"
        forge_bridge.generate_factory(
            spec_path, out, pass_id=forge_bridge.HOSTED_PREVIEW_PASS)
        source = out.read_text(encoding="utf-8")
        assert source.index("const node_body_0") < source.index("const node_handle_1")