feat: integrate calibration extension and self-repair modules
This commit is contained in:
@@ -0,0 +1,83 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from cognitive_agent.application.self_repair.healing_model import (
|
||||
PHASE_DIAGNOSED,
|
||||
PHASE_INJECTED,
|
||||
PHASE_PASSED,
|
||||
SCENARIOS,
|
||||
SelfHealingEngine,
|
||||
public_scenarios,
|
||||
)
|
||||
from cognitive_agent.application.self_repair.service import SelfRepairService
|
||||
from cognitive_agent.application.self_repair.signal_diagnostic import SignalDiagnosticModel
|
||||
from cognitive_agent.application.self_repair.zcan_bridge import ZcanBridgeManager
|
||||
|
||||
|
||||
def test_six_production_cases_complete_the_shadow_repair_loop() -> None:
|
||||
assert len(SCENARIOS) == 6
|
||||
assert len(public_scenarios()) == 6
|
||||
|
||||
for scenario in SCENARIOS:
|
||||
engine = SelfHealingEngine(clock=lambda: 1000.0)
|
||||
started = engine.start(scenario["id"], now=1000.0)
|
||||
assert started["execution_mode"] == "shadow"
|
||||
assert started["injection_scope"] == "SHADOW_TELEMETRY_ONLY"
|
||||
assert engine.snapshot(1006.2)["phase"] == PHASE_INJECTED
|
||||
assert engine.snapshot(1008.7)["phase"] == PHASE_DIAGNOSED
|
||||
completed = engine.snapshot(1018.2)
|
||||
assert completed["phase"] == PHASE_PASSED
|
||||
assert completed["pass"] is True
|
||||
assert completed["telemetry"]["safety_protection"] is True
|
||||
assert completed["telemetry"]["task_restored"] is True
|
||||
|
||||
|
||||
def test_connector_inventory_matches_physical_housings() -> None:
|
||||
snapshot = SignalDiagnosticModel().analyze(None)
|
||||
connectors = snapshot["connectors"]
|
||||
|
||||
def pin(pin_id: str) -> dict[str, object]:
|
||||
connector, number = pin_id.split("-", maxsplit=1)
|
||||
return next(
|
||||
item
|
||||
for item in connectors[connector]
|
||||
if item["number"] == int(number)
|
||||
)
|
||||
|
||||
assert [pin["number"] for pin in connectors["JA1"]] == list(range(1, 15))
|
||||
assert [pin["number"] for pin in connectors["JB1"]] == list(range(1, 19))
|
||||
assert [pin["number"] for pin in connectors["JC1"]] == list(range(1, 41))
|
||||
assert sum(len(pins) for pins in connectors.values()) == 72
|
||||
assert snapshot["connector_enabled_count"] == 21
|
||||
assert pin("JA1-14")["enabled"] is False
|
||||
assert pin("JC1-21")["enabled"] is True
|
||||
|
||||
|
||||
def test_service_is_safe_without_machine_specific_tool_paths() -> None:
|
||||
service = SelfRepairService()
|
||||
health = service.health()
|
||||
state = service.snapshot()
|
||||
|
||||
assert health["ok"] is True
|
||||
assert health["service"] == "seat-self-repair"
|
||||
assert health["configuration"]["diagnostic_config_available"] is True
|
||||
assert health["configuration"]["bridge_script_available"] is True
|
||||
assert len(state["scenarios"]) == 6
|
||||
assert state["integration"]["auto_start_bridge"] is False
|
||||
assert state["model"]["type"] == "PROCEDURAL_SEAT"
|
||||
assert service.run_self_test() is True
|
||||
|
||||
|
||||
def test_bridge_manager_does_not_arm_visible_tooling_by_default(tmp_path: Path) -> None:
|
||||
manager = ZcanBridgeManager(
|
||||
tmp_path / "bridge.json",
|
||||
tmp_path / "bridge.py",
|
||||
zcanpro_exe=tmp_path / "ZCANPRO.exe",
|
||||
script_config_path=tmp_path / "zcanpro.ini",
|
||||
)
|
||||
|
||||
snapshot = manager.snapshot()
|
||||
assert snapshot["state"] == "IDLE"
|
||||
assert snapshot["retry_armed"] is False
|
||||
assert snapshot["attempts"] == 0
|
||||
Reference in New Issue
Block a user