pcb-afm-simulation-system/scripts/test_p4_schema.py

"""P4-M1 regression: single-source plan schema (src.plan_schema) + web wiring.

Run:  python scripts/test_p4_schema.py   (exit 0 = PASS)

Covers:
  1. src.plan_schema.parse_plan / validate_plan_dict:
       - happy path (values + start/stop/step vars, topology, strategy)
       - alias tolerance (min_value/max_value -> start/stop)
       - empty / zero-value inputs
       - malformed input (non-dict, bad structure) -> structured error
       - require_model_path stage (draft vs execution)
       - point generation consistency
  2. Web wiring: POST /api/plans rejects invalid plan_data (400), accepts
     valid draft (201); PUT rejects invalid plan_data (400). Uses FastAPI
     TestClient on an isolated temp SQLite DB. No real Motor-CAD.
"""
import os
import sys
import tempfile

_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_BACKEND = os.path.join(_ROOT, "web", "backend")
_DB = os.path.join(tempfile.mkdtemp(prefix="p4schema_"), "web.db")

os.environ["AFM_DB_PATH"] = _DB
os.environ["KIMI_API_KEY"] = ""  # hermetic: no AI calls
sys.path.insert(0, _BACKEND)
sys.path.insert(0, _ROOT)

from src.plan_schema import (  # noqa: E402
    parse_plan, validate_plan_dict, SimulationPlan,
)

VALID_VARS = [
    {"name": "airgap_mm", "display_name": "Airgap", "unit": "mm",
     "start": 0.8, "stop": 2.0, "step": 0.1},
    {"name": "Magnet_Arc_[ED]", "values": [120.0, 130.0, 140.0]},
]
VALID_FPS = [{"name": "Outer_Rotor_Diameter", "value": 200.0,
              "category": "Geometry"}]


def make_plan(**over):
    d = {
        "plan_version": "2.0",
        "topology": "SSSR",
        "model_path": "models/MARS-12S10P_SSSR_D76-C150_V5.0-0819.mot",
        "fixed_params": VALID_FPS,
        "variables": VALID_VARS,
        "cases": [{"id": "default", "name": "Default"}],
        "search_strategy": {"method": "adaptive", "batch_size": 4,
                            "max_solver_calls": 80},
        "acceptance_criteria": {"hard_constraints": ["efficiency_pct >= 92"]},
    }
    d.update(over)
    return d


# ---------------- 1) happy path + point generation ----------------
p = parse_plan(make_plan())
assert p.topology == "SSSR"
assert len(p.variables) == 2
assert p.variables[0].get_values()[0] == 0.8
assert len(p.variables[0].get_values()) == 13  # 0.8..2.0 step 0.1
ok, errs = validate_plan_dict(make_plan())
assert ok, errs
pts, names = p.generate_points()
assert len(pts) == 13 * 3 and set(names) == {"airgap_mm", "Magnet_Arc_[ED]"}
print("[1] happy path + point generation OK (39 points)")

# ---------------- 2) alias tolerance ----------------
p2 = parse_plan(make_plan(variables=[
    {"name": "airgap_mm", "min_value": 0.8, "max_value": 2.0, "step": 0.1}]))
assert p2.variables[0].get_values()[0] == 0.8, p2.variables[0].get_values()
print("[2] min_value/max_value alias -> start/stop OK")

# ---------------- 3) empty / zero-value inputs ----------------
ok3, errs3 = validate_plan_dict(make_plan(variables=[]))
assert not ok3 and any("scan variable" in e for e in errs3), errs3
ok3b, errs3b = validate_plan_dict({})
assert not ok3b, errs3b
ok3c, errs3c = validate_plan_dict(make_plan(variables=[
    {"name": "x", "start": 5.0, "stop": 1.0, "step": 0.5}]))
assert not ok3c and any("no values" in e for e in errs3c), errs3c
print("[3] empty / zero-value inputs rejected OK")

# ---------------- 4) malformed input ----------------
ok4, errs4 = validate_plan_dict("not-a-dict")
assert not ok4 and any("malformed" in e for e in errs4), errs4
ok4b, errs4b = validate_plan_dict(make_plan(variables=[{"no_name": 1}]))
assert not ok4b, errs4b  # KeyError on name -> malformed
print("[4] malformed input -> structured error OK")

# ---------------- 5) require_model_path stage ----------------
ok5, _ = validate_plan_dict(
    make_plan(model_path=""), require_model_path=False)
assert ok5, "draft stage must not require a model"
ok5b, errs5b = validate_plan_dict(make_plan(model_path=""))
assert not ok5b and any("model_path" in e for e in errs5b)
print("[5] require_model_path stage OK")

# ---------------- 6) unsupported topology / strategy ----------------
ok6, errs6 = validate_plan_dict(make_plan(topology="NOPE"))
assert not ok6 and any("topology" in e for e in errs6), errs6
ok6b, errs6b = validate_plan_dict(
    make_plan(search_strategy={"method": "no_such_method"}))
assert not ok6b and any("strategy" in e for e in errs6b), errs6b
print("[6] unsupported topology/strategy rejected OK")

# ---------------- 7) web wiring (TestClient) ----------------
from fastapi.testclient import TestClient  # noqa: E402
from app.database import init_db as web_init_db  # noqa: E402
web_init_db()

from app.main import app  # noqa: E402
client = TestClient(app)

# create project
r = client.post("/api/projects", json={"name": "t", "topology": "SSSR"})
assert r.status_code == 201, r.text
pid = r.json()["id"]

# invalid plan_data -> 400
bad = make_plan(variables=[])
r = client.post("/api/plans", json={"project_id": pid, "name": "bad",
                                    "plan_data": bad})
assert r.status_code == 400, r.status_code
assert "Invalid plan_data" in r.json()["detail"], r.text

# valid draft (no model path) -> 201
good = make_plan(model_path="")
r = client.post("/api/plans", json={"project_id": pid, "name": "good",
                                    "plan_data": good})
assert r.status_code == 201, r.text
plan_id = r.json()["id"]

# invalid update -> 400
r = client.put("/api/plans/%d" % plan_id,
               json={"plan_data": make_plan(variables=[])})
assert r.status_code == 400, r.status_code

# valid update -> 200
r = client.put("/api/plans/%d" % plan_id,
               json={"plan_data": make_plan(model_path="")})
assert r.status_code == 200, r.text

# download round-trips through the same schema
r = client.get("/api/plans/%d/download" % plan_id)
assert r.status_code == 200
assert r.json()["plan_data"]["variables"]
print("[7] web create/update validation wired OK (400 on invalid)")

print("\nALL P4-M1 SCHEMA TESTS PASSED")