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