"""P5-M6: unit tests for metrics extension (thermal + structural) and domain-grouped report generation. Covers: - New thermal/structural metrics present in METRIC_DEFINITIONS with domain - parse_export + extract_all_metrics picks up new metrics automatically - check_required_metrics unaffected (new metrics required=False) - report_generator._group_metrics_by_domain correct grouping - report_generator JSON fallback includes metrics_by_domain - Boundary: empty metrics, unknown keys, mixed domains - robust_motorcad enable_thermal parameter exists (signature check) All source is ASCII only. Run: python scripts/test_metrics_extension.py exit 0 = PASS. """ import os import sys import tempfile import unittest _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) _SRC = os.path.join(_ROOT, "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) from afmcore.metrics import ( # noqa: E402 METRIC_DEFINITIONS, METRIC_KEYS, REQUIRED_METRICS, check_required_metrics, extract_all_metrics, parse_export, ) # report_generator lives under web/backend; add its dir to path _REPORT_DIR = os.path.join(_ROOT, "web", "backend", "app", "services") if _REPORT_DIR not in sys.path: sys.path.insert(0, _REPORT_DIR) from report_generator import ( # noqa: E402 ReportGenerator, _DOMAIN_ORDER, _group_metrics_by_domain, _metric_display, ) THERMAL_KEYS = [ "winding_hotspot_temp_c", "magnet_temp_c", "stator_temp_c", "bearing_temp_c", "temp_rise_c", "thermal_resistance_k_w", ] STRUCTURAL_KEYS = [ "axial_force_n", "radial_force_n", "max_stress_mpa", "deformation_mm", ] class TestMetricDefinitions(unittest.TestCase): def test_thermal_metrics_present(self): keys = {m["key"] for m in METRIC_DEFINITIONS} for k in THERMAL_KEYS: self.assertIn(k, keys, "missing thermal metric: %s" % k) def test_structural_metrics_present(self): keys = {m["key"] for m in METRIC_DEFINITIONS} for k in STRUCTURAL_KEYS: self.assertIn(k, keys, "missing structural metric: %s" % k) def test_thermal_metrics_have_domain(self): for m in METRIC_DEFINITIONS: if m["key"] in THERMAL_KEYS: self.assertEqual(m.get("domain"), "thermal", "%s should have domain=thermal" % m["key"]) def test_structural_metrics_have_domain(self): for m in METRIC_DEFINITIONS: if m["key"] in STRUCTURAL_KEYS: self.assertEqual(m.get("domain"), "structural", "%s should have domain=structural" % m["key"]) def test_new_metrics_not_required(self): for k in THERMAL_KEYS + STRUCTURAL_KEYS: self.assertNotIn(k, REQUIRED_METRICS, "%s should be required=False" % k) def test_total_metric_count(self): # original 25 + 6 thermal + 4 structural = 35 self.assertEqual(len(METRIC_DEFINITIONS), 35) class TestMetricExtraction(unittest.TestCase): """Construct a mock Motor-CAD export CSV with thermal/structural fields and verify extract_all_metrics picks them up automatically.""" def _make_export(self, fields): """Write a mock semicolon-CSV export and return its path.""" lines = ["E-Magnetics"] for field, value in fields: lines.append("%s;%s" % (field, value)) fd, path = tempfile.mkstemp(suffix=".csv") with os.fdopen(fd, "w", encoding="utf-8") as f: f.write("\n".join(lines)) self.addCleanup(os.unlink, path) return path def test_extract_thermal_metrics(self): path = self._make_export([ ("Average torque (virtual work)", "1.5"), ("Magnet Temperature", "62.5"), ("Winding Hotspot Temperature", "88.3"), ("Stator Temperature", "70.1"), ("Bearing Temperature", "55.0"), ("Temperature Rise", "48.2"), ("Thermal Resistance", "0.85"), ]) parsed = parse_export(path) metrics = extract_all_metrics(parsed) self.assertAlmostEqual(metrics["magnet_temp_c"], 62.5, places=2) self.assertAlmostEqual(metrics["winding_hotspot_temp_c"], 88.3, places=2) self.assertAlmostEqual(metrics["stator_temp_c"], 70.1, places=2) self.assertAlmostEqual(metrics["bearing_temp_c"], 55.0, places=2) self.assertAlmostEqual(metrics["temp_rise_c"], 48.2, places=2) self.assertAlmostEqual(metrics["thermal_resistance_k_w"], 0.85, places=2) def test_extract_structural_metrics(self): path = self._make_export([ ("Axial Force", "125.5"), ("Radial Force", "45.2"), ("Maximum Stress", "180.3"), ("Max Deformation", "0.12"), ]) parsed = parse_export(path) metrics = extract_all_metrics(parsed) self.assertAlmostEqual(metrics["axial_force_n"], 125.5, places=2) self.assertAlmostEqual(metrics["radial_force_n"], 45.2, places=2) self.assertAlmostEqual(metrics["max_stress_mpa"], 180.3, places=2) self.assertAlmostEqual(metrics["deformation_mm"], 0.12, places=2) def test_chinese_alias_thermal(self): path = self._make_export([ ("\u6c38\u78c1\u4f53\u6e29\u5ea6", "70.0"), # magnet temp ("\u8f74\u5411\u529b", "200.0"), # axial force ]) parsed = parse_export(path) metrics = extract_all_metrics(parsed) self.assertAlmostEqual(metrics["magnet_temp_c"], 70.0, places=2) self.assertAlmostEqual(metrics["axial_force_n"], 200.0, places=2) def test_required_check_unaffected(self): # Only tavg/ripple/efficiency/total_losses are required; # missing thermal/structural metrics must not fail the check. ok, missing = check_required_metrics({"tavg_nm": 1.0, "ripple_pct": 2.0, "efficiency_pct": 90.0, "total_losses_w": 10.0}) self.assertTrue(ok) self.assertEqual(missing, []) def test_required_check_fails_on_missing_core(self): ok, missing = check_required_metrics({"tavg_nm": 1.0}) self.assertFalse(ok) self.assertIn("ripple_pct", missing) class TestDomainGrouping(unittest.TestCase): def test_group_by_domain(self): metrics = { "tavg_nm": 1.5, "ripple_pct": 5.0, # electromagnetic (default) "magnet_temp_c": 60.0, "temp_rise_c": 40.0, # thermal "axial_force_n": 100.0, "max_stress_mpa": 150.0, # structural } grouped = _group_metrics_by_domain(metrics) self.assertIn("electromagnetic", grouped) self.assertIn("thermal", grouped) self.assertIn("structural", grouped) self.assertEqual(set(grouped["thermal"].keys()), {"magnet_temp_c", "temp_rise_c"}) self.assertEqual(set(grouped["structural"].keys()), {"axial_force_n", "max_stress_mpa"}) def test_empty_metrics(self): self.assertEqual(_group_metrics_by_domain({}), {}) def test_unknown_key_defaults_electromagnetic(self): grouped = _group_metrics_by_domain({"unknown_metric_xyz": 42.0}) self.assertIn("electromagnetic", grouped) self.assertEqual(grouped["electromagnetic"]["unknown_metric_xyz"], 42.0) def test_domain_order(self): self.assertEqual(_DOMAIN_ORDER, ["electromagnetic", "thermal", "structural"]) def test_metric_display(self): label, value = _metric_display("tavg_nm", 1.5) self.assertIn("Average Torque", label) self.assertEqual(value, "1.5") def test_metric_display_float_format(self): _, value = _metric_display("efficiency_pct", 92.3456789) self.assertEqual(value, "92.35") # %.4g class TestReportJsonFallback(unittest.TestCase): def test_json_report_includes_metrics_by_domain(self): rg = ReportGenerator(output_dir=tempfile.mkdtemp()) task_data = { "task_id": "test-001", "task_name": "Test Task", "status": "completed", "result_metrics": { "tavg_nm": 1.5, "magnet_temp_c": 60.0, "axial_force_n": 100.0, }, } path = rg._generate_json_report(task_data, None, None) self.addCleanup(os.unlink, path) import json with open(path, "r", encoding="utf-8") as f: report = json.load(f) self.assertIn("metrics_by_domain", report) self.assertIn("thermal", report["metrics_by_domain"]) self.assertIn("structural", report["metrics_by_domain"]) self.assertEqual(report["metrics_by_domain"]["thermal"]["magnet_temp_c"], 60.0) class TestRobustMotorcadThermalParam(unittest.TestCase): """Verify enable_thermal parameter exists in RobustMotorCADSolver.""" def test_init_has_enable_thermal(self): sys.path.insert(0, _ROOT) from scripts.robust_motorcad import RobustMotorCADSolver import inspect sig = inspect.signature(RobustMotorCADSolver.__init__) self.assertIn("enable_thermal", sig.parameters) self.assertFalse(sig.parameters["enable_thermal"].default) def test_run_single_point_has_enable_thermal(self): sys.path.insert(0, _ROOT) from scripts.robust_motorcad import RobustMotorCADSolver import inspect sig = inspect.signature(RobustMotorCADSolver.run_single_point) self.assertIn("enable_thermal", sig.parameters) self.assertIsNone(sig.parameters["enable_thermal"].default) if __name__ == "__main__": unittest.main(verbosity=2)