| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153 |
- """P3-M5 integration: real HTTP closed loop with a mock local executor.
- Run: python scripts/test_p3_closed_loop.py (exit 0 = PASS)
- Spins up the real FastAPI web backend on an isolated temp SQLite DB, drives
- the AdaptiveOrchestrator, and lets a real TaskExecutor (mock solver) poll and
- execute adaptive batches over HTTP. Verifies the full chain:
- orchestrator.start_loop -> task created (HTTP) -> executor claims ->
- mock simulation -> report_results (point_id) -> orchestrator.advance_loop
- feeds back -> next batch -> ... -> budget exhausted.
- Requires: requests, uvicorn, fastapi. No real Motor-CAD.
- """
- import json
- import os
- import subprocess
- import sys
- import tempfile
- import time
- import urllib.request
- _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
- _BACKEND = os.path.join(_ROOT, "web", "backend")
- _SCRIPTS = os.path.join(_ROOT, "scripts")
- _TMP = tempfile.mkdtemp(prefix="p3loop_")
- _DB = os.path.join(_TMP, "web.db")
- _STATE = os.path.join(_TMP, "loops")
- _PORT = 8137
- _BASE = "http://127.0.0.1:%d" % _PORT
- os.environ["AFM_DB_PATH"] = _DB
- os.environ["KIMI_API_KEY"] = "" # keep test hermetic: no AI calls
- sys.path.insert(0, _BACKEND)
- sys.path.insert(0, _SCRIPTS)
- sys.path.insert(0, _ROOT) # src.* (l0 re-export) resolves from repo root
- 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
- WEB_ENV = dict(os.environ, AFM_DB_PATH=_DB, AFM_PORT=str(_PORT))
- def wait_health(timeout=30):
- url = _BASE + "/api/monitor/health"
- deadline = time.time() + timeout
- while time.time() < deadline:
- try:
- with urllib.request.urlopen(url, timeout=3) as r:
- if r.status == 200:
- return True
- except Exception:
- time.sleep(0.5)
- return False
- # ---- start web ----
- print("[0] starting web backend on :%d (temp DB)" % _PORT, flush=True)
- web = subprocess.Popen(
- [sys.executable, "-m", "uvicorn", "app.main:app",
- "--host", "127.0.0.1", "--port", str(_PORT), "--log-level", "warning"],
- cwd=_BACKEND, env=WEB_ENV,
- stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
- )
- try:
- assert wait_health(), "web backend did not become healthy"
- print("[1] web backend healthy", flush=True)
- tm = get_task_manager()
- orch = AdaptiveOrchestrator(state_dir=_STATE)
- # ---- orchestrator start ----
- res = orch.start_loop(
- loop_id="loop-http",
- parameters=[
- {"name": "airgap_mm", "min_value": 0.8, "max_value": 2.0, "step": 0.1, "unit": "mm"},
- {"name": "current_a", "min_value": 5.0, "max_value": 20.0, "step": 0.5},
- ],
- total_budget=8, batch_size=4, initial_samples=4,
- objective_metric="tavg_nm", objective_direction="maximize",
- )
- assert res["phase"] == "running" and res["current_task_id"], res
- print("[2] orchestrator started: task=%s" % res["current_task_id"], flush=True)
- # ---- local executor (mock) over real HTTP ----
- import task_executor as te_mod
- te_mod.requests = None if not True else te_mod.requests # keep real requests
- from task_executor import TaskExecutor
- ex = TaskExecutor(web_base_url=_BASE, enable_mock=True,
- executor_id="motorcad-mock-m5")
- thread = ex.start_polling(interval=2)
- print("[3] mock executor polling started", flush=True)
- # ---- drive the loop ----
- steps = 0
- max_steps = 12
- batches = set()
- while steps < max_steps:
- steps += 1
- view = orch.get_loop_status("loop-http")
- phase = view["phase"]
- if phase in ("converged", "budget_exhausted", "failed"):
- print("[4] terminal phase=%s after %d advance steps" % (phase, steps), flush=True)
- break
- tid = view.get("current_task_id")
- if not tid:
- view = orch.advance_loop("loop-http")
- continue
- # wait for the executor to finish this batch over HTTP
- deadline = time.time() + 90
- while time.time() < deadline:
- t = tm.get_task(tid)
- if t and t["status"] in ("completed", "failed", "cancelled"):
- break
- time.sleep(1)
- else:
- raise RuntimeError("batch task %s did not finish in time" % tid)
- view = orch.advance_loop("loop-http")
- batches.add(view.get("current_batch"))
- print("[5] advance -> batch=%s task=%s phase=%s n_results=%s"
- % (view.get("current_batch"), view.get("current_task_id"),
- view["phase"], view["n_results"]), flush=True)
- ex.stop()
- thread.join(timeout=5)
- final = orch.get_loop_status("loop-http")
- print("[6] FINAL phase=%s batches=%s n_results=%s"
- % (final["phase"], sorted(batches), final["n_results"]), flush=True)
- assert final["phase"] in ("converged", "budget_exhausted"), final
- assert final["n_results"] >= 4, final
- ss = final.get("search_state") or {}
- assert ss.get("completed_points", 0) >= 4, ss
- # every reported point must carry point_id and be fed back
- assert ss.get("used_budget", 0) > 0, ss
- print("\nALL P3-M5 HTTP CLOSED LOOP TESTS PASSED", flush=True)
- finally:
- # cleanup
- try:
- ex.stop()
- except Exception:
- pass
- web.terminate()
- try:
- web.wait(timeout=10)
- except Exception:
- web.kill()
- print("[cleanup] web stopped", flush=True)
|