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

# -*- coding: utf-8 -*-
"""P5-M3: unit tests for get_state_summary() batch_summary / l0_summary.

Covers: happy path, post-report aggregation, empty search boundary,
infeasible-point L0 reasons, objective direction (max/min), and
unknown point_id report resilience.

All source is ASCII only. Run: python scripts/test_search_state_summary.py
exit 0 = PASS.
"""
import os
import sys
import unittest

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

from app.services.feasibility_search import (  # noqa: E402
    FeasibilityFirstSearch,
    ParameterRange,
)


def _make_search(direction="maximize", metric="tavg_nm"):
    """Build a minimal search with one parameter range."""
    return FeasibilityFirstSearch(
        parameters=[
            ParameterRange(name="airgap_mm", min_value=0.5, max_value=2.0, step=0.1),
        ],
        total_budget=20,
        batch_size=4,
        initial_samples=8,
        objective_metric=metric,
        objective_direction=direction,
        seed=42,
    )


class TestStateSummaryNewFields(unittest.TestCase):
    """P5-M3: batch_summary / l0_summary / infeasible / failed fields."""

    def test_initial_batch_has_new_fields(self):
        """Happy path: after generate_initial_batch, all new fields present."""
        s = _make_search()
        s.generate_initial_batch()
        state = s.get_state_summary()
        for key in ("infeasible_points", "failed_points", "batch_summary", "l0_summary"):
            self.assertIn(key, state, "missing key: %s" % key)
        self.assertIsInstance(state["batch_summary"], list)
        self.assertGreater(len(state["batch_summary"]), 0, "at least batch 0")
        b0 = state["batch_summary"][0]
        for key in ("batch_id", "total", "pending", "ok", "infeasible", "failed", "best_objective"):
            self.assertIn(key, b0, "batch_summary entry missing: %s" % key)
        self.assertEqual(b0["total"], b0["pending"] + b0["ok"] + b0["infeasible"] + b0["failed"])
        l0 = state["l0_summary"]
        for key in ("sampled", "feasible", "infeasible", "pass_rate", "top_infeasible_reasons"):
            self.assertIn(key, l0, "l0_summary missing: %s" % key)
        self.assertEqual(l0["sampled"], state["total_points"])
        self.assertEqual(l0["feasible"] + l0["infeasible"], l0["sampled"])

    def test_report_results_updates_batch_summary(self):
        """Report ok results: batch_summary ok count rises, best_objective set."""
        s = _make_search(direction="maximize")
        pts = s.generate_initial_batch()
        pending = [p for p in pts if p.status == "pending"]
        self.assertGreater(len(pending), 0, "need at least one pending point")
        target = pending[0]
        s.report_result(target.id, {"tavg_nm": 1.5}, "ok")
        state = s.get_state_summary()
        b0 = state["batch_summary"][0]
        self.assertEqual(b0["ok"], 1)
        self.assertEqual(b0["best_objective"], 1.5)
        self.assertEqual(state["completed_points"], 1)

    def test_minimize_direction_best_objective(self):
        """Minimize: best_objective is the minimum reported value."""
        s = _make_search(direction="minimize", metric="total_losses_w")
        pts = s.generate_initial_batch()
        pending = [p for p in pts if p.status == "pending"]
        self.assertGreaterEqual(len(pending), 2, "need 2 pending points")
        s.report_result(pending[0].id, {"total_losses_w": 50.0}, "ok")
        s.report_result(pending[1].id, {"total_losses_w": 30.0}, "ok")
        state = s.get_state_summary()
        self.assertEqual(state["batch_summary"][0]["best_objective"], 30.0)

    def test_failed_point_counted(self):
        """Report failed: failed_points increments, batch_summary failed count."""
        s = _make_search()
        pts = s.generate_initial_batch()
        pending = [p for p in pts if p.status == "pending"]
        self.assertGreater(len(pending), 0)
        s.report_result(pending[0].id, {}, "failed")
        state = s.get_state_summary()
        self.assertEqual(state["failed_points"], 1)
        self.assertEqual(state["batch_summary"][0]["failed"], 1)

    def test_empty_search_boundary(self):
        """Boundary: search created but no batch generated -> empty summaries."""
        s = _make_search()
        state = s.get_state_summary()
        self.assertEqual(state["total_points"], 0)
        self.assertEqual(state["batch_summary"], [])
        self.assertEqual(state["l0_summary"]["sampled"], 0)
        self.assertEqual(state["l0_summary"]["pass_rate"], 0.0)
        self.assertEqual(state["infeasible_points"], 0)
        self.assertEqual(state["failed_points"], 0)

    def test_infeasible_points_l0_reasons_structure(self):
        """Infeasible points (if any) carry top_infeasible_reasons with name/count/category."""
        s = _make_search()
        s.generate_initial_batch()
        state = s.get_state_summary()
        reasons = state["l0_summary"]["top_infeasible_reasons"]
        self.assertIsInstance(reasons, list)
        for r in reasons:
            self.assertIn("name", r)
            self.assertIn("count", r)
            self.assertIn("category", r)
            self.assertGreaterEqual(r["count"], 1)

    def test_unknown_point_id_report_no_crash(self):
        """Resilience: report_result for unknown point_id does not raise."""
        s = _make_search()
        s.generate_initial_batch()
        # Should not raise; implementation may silently ignore or log.
        try:
            s.report_result(999999, {"tavg_nm": 1.0}, "ok")
        except Exception as exc:
            self.fail("report_result unknown id raised: %s" % exc)
        state = s.get_state_summary()
        # Unknown point must not inflate completed counts.
        self.assertEqual(state["completed_points"], 0)

    def test_batch_summary_sorted_by_batch_id(self):
        """batch_summary entries sorted ascending by batch_id."""
        s = _make_search()
        s.generate_initial_batch()
        s.select_next_batch()
        state = s.get_state_summary()
        ids = [b["batch_id"] for b in state["batch_summary"]]
        self.assertEqual(ids, sorted(ids))


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