m3d-core / tests /test_e2e.py
jkorstad's picture
contract tests
456e769 verified
Raw History Blame Contribute Delete
3.26 kB
"""M3D v0 contract tests: module isolation, rig detection, full DAG with human gate."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from m3d.modules.gltf_mcp import build_module as build_gltf
from m3d.modules.mock_gen import build_module as build_gen_module
from m3d.runner import Runner
from m3d.store import Store
from tests.fixtures import make_character_glb, make_rigged_glb
@pytest.fixture()
def env(tmp_path):
store = Store(root=tmp_path / "cas")
modules = {
"std/mock-generate": build_gen_module(store),
"std/gltf-mcp": build_gltf(store),
}
runner = Runner(store, modules, gates_dir=tmp_path / "gates")
return store, modules, runner, tmp_path
def test_module_isolation(env, tmp_path):
"""Isolation principle: gltf-mcp works standalone on a plain file path."""
store, modules, runner, _ = env
glb = tmp_path / "char.glb"
make_character_glb(glb)
mod = modules["std/gltf-mcp"]
assert mod.run_tool("validate", {"uri": str(glb)})["valid"] is True
assert mod.run_tool("rig_check", {"uri": str(glb)})["rigged"] is False
def test_rig_check_detects_skin(env, tmp_path):
store, modules, runner, _ = env
glb = tmp_path / "rigged.glb"
make_rigged_glb(glb)
rep = modules["std/gltf-mcp"].run_tool("rig_check", {"uri": str(glb)})
assert rep["rigged"] is True
skin = rep["skins"][0]
assert skin["joints"] == 2
assert skin["ibm_matches_joints"] is True
def test_optimize_refuses_rigged(env, tmp_path):
store, modules, runner, _ = env
glb = tmp_path / "rigged.glb"
make_rigged_glb(glb)
rep = modules["std/gltf-mcp"].run_tool("optimize", {"uri": str(glb)})
assert rep["ok"] is False and "rigged" in rep["reason"]
def test_dag_end_to_end(env, tmp_path):
"""Full workflow: generate -> validate/rig/stats -> human-gate(approve) -> inspect.
Proves content addressing, provenance sidecars, and the approval gate."""
store, modules, runner, tmp = env
fixture = tmp / "fixture_char.glb"
make_character_glb(fixture)
wf = json.loads(Path(__file__).parents[1].joinpath("workflows/sample_character.json").read_text())
stem = Runner._gate_stem(wf["id"], "review")
gate = tmp / "gates" / f"{stem}.approved.json"
gate.write_text(json.dumps({"decision": "approve"}))
result = runner.run(
wf,
inputs={"prompt": "hero knight", "fixture_uri": str(fixture)},
)
out = result["outputs"]
assert out["gen"]["output_uri"].startswith("cas://")
assert out["validate"]["valid"] is True
assert out["review"]["decision"] == "approve"
assert out["post"]["counts"]["meshes"] == 1 # conditioned edge taken
# provenance sidecar exists and records the producing node
blob = store.resolve(out["gen"]["output_uri"])
side = blob.with_suffix(".asset.json")
assert side.exists()
prov = json.loads(side.read_text())["provenance"]
assert any(p["node"] == "gen" for p in prov)
# resumability hook: the same inputs must re-address to the same blob
again = modules["std/mock-generate"].run_tool(
"run", {"prompt": "hero knight", "fixture_uri": str(fixture)}
)
assert again["output_uri"] == out["gen"]["output_uri"]