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

"""P5-M2 regression: MotorCADTaskExecutor mock-vs-real branch.

Run: python scripts/test_executor_p5m2.py
Verifies that enable_mock=True routes to the base-class mock solver (no
Motor-CAD, no model_path required) and that enable_mock=False keeps the
real-adapter contract (model_path required). exit 0 == PASS.

NOTE: All strings in this file are ASCII only.
"""
import json
import os
import sys
import tempfile
import unittest

_SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
if _SCRIPTS_DIR not in sys.path:
    sys.path.insert(0, _SCRIPTS_DIR)

import task_executor as te  # noqa: E402

# local-file mode: no real web calls during the test
te.requests = None


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


class MotorCADExecutorMockBranchTest(unittest.TestCase):
    """Exercise the enable_mock branch added in P5-M2."""

    def setUp(self):
        self.tmp = tempfile.mkdtemp(prefix="p5m2_exec_")

    def tearDown(self):
        import shutil
        shutil.rmtree(self.tmp, ignore_errors=True)

    def _run(self, executor, tid, params):
        executor.dispatch_task = lambda t: True
        captured = {}
        executor.on_complete = lambda t, res, met: captured.setdefault(t, res)
        executor.execute_task(_make_local_task(self.tmp, tid, params))
        return captured.get(tid)

    # ---------- happy path ----------
    def test_mock_point_source_and_no_model_required(self):
        ex = te.MotorCADTaskExecutor(task_dir=self.tmp, enable_mock=True,
                                     model_path=None)
        res = self._run(ex, "m1", [{"airgap_mm": 1.0, "point_id": 1}])
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0]["status"], "OK")
        self.assertEqual(res[0]["source"], "mock")
        self.assertIn("tavg_nm", res[0])

    def test_mock_multi_point(self):
        ex = te.MotorCADTaskExecutor(task_dir=self.tmp, enable_mock=True,
                                     model_path=None)
        res = self._run(ex, "m2", [
            {"airgap_mm": 0.8, "point_id": 1},
            {"airgap_mm": 1.0, "point_id": 2},
            {"airgap_mm": 1.2, "point_id": 3},
        ])
        self.assertEqual(len(res), 3)
        for r in res:
            self.assertEqual(r["status"], "OK")
            self.assertEqual(r["source"], "mock")

    # ---------- boundary ----------
    def test_mock_with_illegal_model_path_still_mocks(self):
        # mock must win even when model_path points at a nonexistent file,
        # proving no Motor-CAD instance / adapter is launched.
        ex = te.MotorCADTaskExecutor(task_dir=self.tmp, enable_mock=True,
                                     model_path="Z:/nonexistent/model.mot")
        res = self._run(ex, "m3", [{"airgap_mm": 1.0, "point_id": 1}])
        self.assertEqual(res[0]["source"], "mock")

    def test_default_mock_off(self):
        # P4-M3 default: real solver adapter, so enable_mock defaults False
        ex = te.MotorCADTaskExecutor(task_dir=self.tmp, model_path=None)
        self.assertIs(ex.enable_mock, False)

    # ---------- abnormal ----------
    def test_real_mode_requires_model_path(self):
        ex = te.MotorCADTaskExecutor(task_dir=self.tmp, enable_mock=False,
                                     model_path=None)
        with self.assertRaises(RuntimeError):
            ex._run_simulation_point({"airgap_mm": 1.0}, 0)

    # ---------- metadata key exclusion ----------
    def test_point_id_excluded_from_motorcad_params(self):
        # Regression (P5-M2 real E2E): point_id is a passthrough
        # metadata key and must never be sent to Motor-CAD set_variable.
        src_path = os.path.join(_SCRIPTS_DIR, "robust_motorcad.py")
        with open(src_path, "r", encoding="utf-8") as fh:
            src = fh.read()
        self.assertIn(
            'if var in ("point_index", "point_label", "point_id"):',
            src)

    # ---------- business alias ----------
    def test_business_alias_airgap_resolves(self):
        from robust_motorcad import resolve_variable_name
        self.assertEqual(resolve_variable_name("airgap_mm"), "Airgap")
        # unknown canonical names pass through untouched
        self.assertEqual(resolve_variable_name("SomeUnknown"), "SomeUnknown")

    # ---------- empty value ----------
    def test_mock_empty_model_path_string(self):
        ex = te.MotorCADTaskExecutor(task_dir=self.tmp, enable_mock=True,
                                     model_path="")
        res = self._run(ex, "m4", [{"airgap_mm": 1.0, "point_id": 1}])
        self.assertEqual(res[0]["source"], "mock")


if __name__ == "__main__":
    unittest.main(verbosity=2)