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

"""P3 unit edge/exception/empty-value tests (engineering rules).

Covers boundary, exception-input and empty/zero-value paths for the P3
platform batch modules, complementing the integration-level tests.

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

All behaviour below was confirmed by probing the actual implementation;
no behaviour is assumed.
"""
import os
import sys
import tempfile

_BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, _BASE)
_TMP = os.path.join(tempfile.mkdtemp(), "edge.db")
os.environ["AFM_DB_PATH"] = _TMP
os.environ["KIMI_API_KEY"] = ""

# ---------------------------------------------------------------- strategies
from src.afmcore.strategies import (  # noqa: E402
    get_strategy, normalize_method, register_strategy, is_registered,
)
from src.afmcore.strategies.full_factorial import FullFactorialStrategy  # noqa: E402
from src.afmcore.strategies.lhs import LHSStrategy  # noqa: E402

# [S1] registry exception paths
try:
    register_strategy("", object)
    raise SystemExit("empty kind should raise ValueError")
except ValueError:
    pass
try:
    register_strategy("x", 123)
    raise SystemExit("non-subclass should raise TypeError")
except TypeError:
    pass
try:
    get_strategy("no-such-strategy")
    raise SystemExit("unknown kind should raise KeyError")
except KeyError:
    pass
assert is_registered("full_factorial") and is_registered("lhs") and is_registered("adaptive")
print("[S1] registry exception paths OK")

# [S2] normalize_method
assert normalize_method(None) == "full_factorial"
assert normalize_method("") == "full_factorial"
assert normalize_method("active_learning") == "adaptive"
assert normalize_method("constrained") == "adaptive"
assert normalize_method("  ADAPTIVE ") == "adaptive"
assert normalize_method("ABC") == "abc"  # unknown kept lowercase, reported by caller
print("[S2] normalize_method OK")

# [S3] FullFactorial empty / batching / converged
ff_empty = FullFactorialStrategy(points=[])
assert ff_empty.select_next() == []
assert ff_empty.is_converged() is True
assert ff_empty.next_batch_ready() is False
ff = FullFactorialStrategy(points=[{"x": 1}, {"x": 2}, {"x": 3}, {"x": 4}, {"x": 5}], batch_size=2)
b1 = ff.select_next()
b2 = ff.select_next()
b3 = ff.select_next()
assert [len(b1), len(b2), len(b3)] == [2, 2, 1], (b1, b2, b3)
assert ff.is_converged() is True
assert ff.select_next() == []  # exhausted
print("[S3] FullFactorial empty/batch/converged OK")

# [S4] LHS empty vs normal
lhs_empty = LHSStrategy(parameters=[], n_samples=4)
assert lhs_empty.select_next() == []
lhs = LHSStrategy(
    parameters=[
        {"name": "a", "min_value": 1, "max_value": 2},
        {"name": "b", "min_value": 10, "max_value": 20},
    ],
    n_samples=3,
)
pts = lhs.select_next()
assert len(pts) == 3, pts
for p in pts:
    assert p["point_id"] is not None
    assert set(p["params"].keys()) == {"a", "b"}
print("[S4] LHS empty/normal OK")

# [S5] batch_size boundary
ff0 = FullFactorialStrategy(points=[{"x": 1}, {"x": 2}], batch_size=0)
assert ff0.batch_size == 1  # max(1, 0)
print("[S5] batch_size boundary OK")

# --------------------------------------------------------------- orchestrator
sys.path.insert(0, os.path.join(_BASE, "web", "backend"))
from app.database import init_db  # noqa: E402
init_db()
from app.services.strategy_orchestrator import AdaptiveOrchestrator  # noqa: E402
from app.services.task_manager import get_task_manager  # noqa: E402
from app.services.adaptive_loop import AdaptiveLoop  # noqa: E402

tm = get_task_manager()
orch = AdaptiveOrchestrator(state_dir=os.path.join(tempfile.mkdtemp(), "loops"))

# [O1] start_loop empty parameters -> ValueError
try:
    orch.start_loop("e1", parameters=[])
    raise SystemExit("start with empty parameters should raise ValueError")
except ValueError:
    pass

# [O2] duplicate loop -> ValueError
orch.start_loop("dup", parameters=[{"name": "airgap_mm", "min_value": 1, "max_value": 2}])
try:
    orch.start_loop("dup", parameters=[{"name": "airgap_mm", "min_value": 1, "max_value": 2}])
    raise SystemExit("duplicate loop should raise ValueError")
except ValueError:
    pass

# [O3] advance/status on missing loop -> ValueError
for fn in (orch.advance_loop, orch.get_loop_status):
    try:
        fn("missing-loop")
        raise SystemExit("missing loop should raise ValueError")
    except ValueError:
        pass

# [O4] advance while batch still running -> no error, stays running
res = orch.start_loop(
    "l1",
    parameters=[{"name": "airgap_mm", "min_value": 1, "max_value": 2}],
    total_budget=8, batch_size=4, initial_samples=4,
)
view = orch.advance_loop("l1")
assert view["phase"] == "running", view
assert view.get("message") == "batch still running", view
print("[O1-O4] orchestrator boundary/exception paths OK")

# [O5] AdaptiveLoop.submit_batch_to_executor before init -> RuntimeError
loop = AdaptiveLoop(user_requirement="probe")
try:
    loop.submit_batch_to_executor()
    raise SystemExit("submit before init should raise RuntimeError")
except RuntimeError:
    pass
print("[O5] AdaptiveLoop submit-before-init raises RuntimeError OK")

# ----------------------------------------------------------------- task_manager
# [T1] get_task on missing id -> None
assert tm.get_task("no-such-task") is None
# [T2] create_task with empty parameters -> still creates a task (no crash)
t = tm.create_task(plan_id=None, plan_data={}, parameters=[])
assert t.get("task_id"), t
print("[T1-T2] task_manager empty/missing paths OK")

print("\nALL P3 UNIT EDGE/EXCEPTION/EMPTY TESTS PASSED")