vee
107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
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
|
|
|
|
|
|
def test_bridge_manager_does_not_restart_a_live_busy_request(tmp_path: Path) -> None:
|
|
status_path = tmp_path / "bridge.json"
|
|
status_path.write_text(
|
|
json.dumps({
|
|
"heartbeat_epoch": time.time() - 5.0,
|
|
"bridge_state": "RUNNING",
|
|
"current_request_id": "SELF_HEAL_BENCH_1",
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
manager = ZcanBridgeManager(
|
|
status_path,
|
|
tmp_path / "bridge.py",
|
|
heartbeat_stale_s=3.0,
|
|
busy_heartbeat_grace_s=45.0,
|
|
)
|
|
|
|
assert manager._heartbeat_online() is False
|
|
assert manager._bridge_busy_within_grace() is True
|