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

# One-off: P3-M3 regression - executor point_id passthrough + claim check.
import json
import os
import sys
import tempfile

_SCRIPTS = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _SCRIPTS)

import task_executor as te  # noqa: E402

# force local-file mode (no web calls)
te.requests = None

tmp = tempfile.mkdtemp()
captured = {}


def make_task(tid, params):
    path = os.path.join(tmp, tid + "_task.json")
    with open(path, "w", encoding="utf-8") as f:
        json.dump({"task_id": tid, "task_name": "m3", "parameters": params}, f)
    return {"task_id": tid, "_local_file": path, "parameters": params}


# ---------------- 1) point_id passthrough (claim succeeds) ----------------
ex = te.TaskExecutor(task_dir=tmp, enable_mock=True)
ex.dispatch_task = lambda tid: True  # claim ok
ex.on_complete = lambda tid, res, met: captured.setdefault(tid, res)
task = make_task("t1", [{"airgap_mm": 1.0, "current_a": 10.0, "point_id": 7},
                        {"airgap_mm": 2.0, "current_a": 5.0, "point_id": 8}])
ex.execute_task(task)
res1 = captured.get("t1")
assert res1 and len(res1) == 2, res1
for r in res1:
    assert r["status"] == "OK", r
    assert r["point_id"] in (7, 8), r
    assert r["params"].get("point_id") == r["point_id"], r
print("[1] point_id passthrough OK:", [(r["point_id"], r["status"]) for r in res1])

# ---------------- 2) FAILED point keeps point_id ----------------
captured2 = {}


def _boom(params, idx):
    raise RuntimeError("simulation failed")


ex2 = te.TaskExecutor(task_dir=tmp, enable_mock=True)
ex2.dispatch_task = lambda tid: True
ex2._run_simulation_point = _boom
ex2.on_complete = lambda tid, res, met: captured2.setdefault(tid, res)
task2 = make_task("t2", [{"airgap_mm": 1.0, "point_id": 42}])
ex2.execute_task(task2)
r2 = captured2["t2"][0]
assert r2["status"] == "FAILED", r2
assert r2["point_id"] == 42, r2
print("[2] FAILED point keeps point_id OK:", r2["point_id"], r2["status"])

# ---------------- 3) claim rejected -> task skipped ----------------
captured3 = {}
ex3 = te.TaskExecutor(task_dir=tmp, enable_mock=True)
ex3.dispatch_task = lambda tid: False  # already claimed by another instance
ex3.on_complete = lambda tid, res, met: captured3.setdefault(tid, res)
task3 = make_task("t3", [{"airgap_mm": 1.0, "point_id": 99}])
ex3.execute_task(task3)
assert "t3" not in captured3, "claimed task must be skipped"
print("[3] claim-rejected task skipped OK")

# ---------------- 4) unique executor ids ----------------
a = te.TaskExecutor(task_dir=tmp)
b = te.TaskExecutor(task_dir=tmp)
c = te.TaskExecutor(task_dir=tmp, executor_id="motorcad-executor-9")
assert a.executor_id != b.executor_id, (a.executor_id, b.executor_id)
assert c.executor_id == "motorcad-executor-9"
print("[4] unique executor ids OK:", a.executor_id, "|", b.executor_id, "|", c.executor_id)

print("\nALL P3-M3 EXECUTOR TESTS PASSED")