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

"""P3-M4 regression: unified task contract + batch_scheduler field alignment.

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

Uses an isolated temp state file for the BatchScheduler (never touches the
real output/scheduler_state.json). No web server or Motor-CAD required.
"""
import os
import sys
import tempfile

sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "web", "backend"))
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# ---- 1) task_contract: status normalization ----
from app.services.task_contract import (  # noqa: E402
    normalize_status,
    is_terminal,
    merge_adaptive_fields,
    TASK_STATUS_PENDING,
    TASK_STATUS_COMPLETED,
)

assert normalize_status("queued") == TASK_STATUS_PENDING, normalize_status("queued")
assert normalize_status("pending") == TASK_STATUS_PENDING
assert normalize_status("completed_with_errors") == TASK_STATUS_COMPLETED
assert normalize_status("canceled") == "cancelled"
assert normalize_status("mystery") == "mystery"  # pass through
assert is_terminal("completed") and is_terminal("failed") and is_terminal("cancelled")
assert not is_terminal("pending") and not is_terminal("queued")
print("[1] task_contract status normalization OK")

# ---- 2) task_contract: adaptive field merge ----
merged = merge_adaptive_fields({}, task_type="adaptive_batch", loop_id="L1",
                               batch_id=2, point_ids=[1, 2, 3], dynamic=True)
assert merged["task_type"] == "adaptive_batch"
assert merged["loop_id"] == "L1" and merged["batch_id"] == 2
assert merged["point_ids"] == [1, 2, 3] and merged["dynamic"] is True
defaults = merge_adaptive_fields({})
assert defaults["task_type"] == "scan" and defaults["dynamic"] is False
assert defaults["point_ids"] == [] and defaults["loop_id"] is None
print("[2] task_contract adaptive field merge OK")

# ---- 3) BatchScheduler: adaptive fields on add_task + summary ----
from app.services.batch_scheduler import BatchScheduler  # noqa: E402

tmp_state = os.path.join(tempfile.mkdtemp(), "sched.json")
sch = BatchScheduler(state_file=tmp_state)
t = sch.add_task(
    task_id="batch-1", task_name="adaptive-L1-b0", task_type="adaptive_batch",
    loop_id="L1", batch_id=0, point_ids=[10, 11], dynamic=True,
    parameters=[{"airgap_mm": 1.0, "point_id": 10}],
)
assert t["task_type"] == "adaptive_batch", t
assert t["loop_id"] == "L1" and t["batch_id"] == 0, t
assert t["point_ids"] == [10, 11] and t["dynamic"] is True, t
assert t["status"] == "queued", t

# legacy call still works (no new args)
t2 = sch.add_task(task_id="scan-1", task_name="plain")
assert t2["task_type"] == "scan" and t2["dynamic"] is False, t2

# get_next_task transitions queued -> running
nxt = sch.get_next_task()
assert nxt is not None and nxt["task_id"] == "batch-1"
assert nxt["status"] == "running" and nxt["loop_id"] == "L1"
sch.complete_task("batch-1")

# get_next_task picks the legacy scan task (wait_for already satisfied)
nxt2 = sch.get_next_task()
assert nxt2 is not None and nxt2["task_id"] == "scan-1", nxt2
sch.complete_task("scan-1")

# statistics summary exposes new fields
stats = sch.get_statistics()
summaries = stats["recent_completed"] + stats["running_tasks"] + stats["queued_tasks"]
assert any(s.get("task_id") == "batch-1" and s.get("task_type") == "adaptive_batch"
           and s.get("loop_id") == "L1" for s in summaries), stats
print("[3] BatchScheduler adaptive fields aligned OK")

# ---- 4) persist / reload keeps fields ----
sch2 = BatchScheduler(state_file=tmp_state)
assert os.path.exists(tmp_state)
print("[4] scheduler state file exists:", os.path.exists(tmp_state))

print("\nALL P3-M4 CONTRACT TESTS PASSED")