mirror of
https://github.com/smittix/intercept.git
synced 2026-06-18 18:39:47 -07:00
refactor: single dependency probe in capability detection; real test coverage
detect_mode_availability accepts a pre-computed dep_status so the agent probes once; interface and fallback paths now have content-level tests. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
+19
-15
@@ -423,27 +423,31 @@ class ModeManager:
|
||||
# Detect interfaces via shared capability detection
|
||||
capabilities["interfaces"] = detect_interfaces()
|
||||
|
||||
# Probe dependencies once; reuse for both mode availability and tool_details.
|
||||
dep_status: dict | None = None
|
||||
if HAS_DEPENDENCIES_MODULE:
|
||||
try:
|
||||
dep_status = check_all_dependencies()
|
||||
except Exception as e:
|
||||
logger.warning(f"Tool detail collection failed: {e}")
|
||||
|
||||
# Detect mode availability via shared capability detection
|
||||
raw_modes = detect_mode_availability()
|
||||
raw_modes = detect_mode_availability(dep_status)
|
||||
for mode, ready in raw_modes.items():
|
||||
# Apply this agent's per-mode config gating on top of tool readiness
|
||||
capabilities["modes"][mode] = ready if config.modes_enabled.get(mode, True) else False
|
||||
|
||||
# Populate detailed tool info for modes tracked in dependencies.py
|
||||
if HAS_DEPENDENCIES_MODULE:
|
||||
try:
|
||||
dep_status = check_all_dependencies()
|
||||
for dep_mode, cap_mode in MODE_DEPENDENCY_MAP.items():
|
||||
if dep_mode in dep_status:
|
||||
mode_info = dep_status[dep_mode]
|
||||
capabilities["tool_details"][cap_mode] = {
|
||||
"name": mode_info["name"],
|
||||
"ready": mode_info["ready"],
|
||||
"missing_required": mode_info["missing_required"],
|
||||
"tools": mode_info["tools"],
|
||||
}
|
||||
except Exception as e:
|
||||
logger.warning(f"Tool detail collection failed: {e}")
|
||||
if dep_status is not None:
|
||||
for dep_mode, cap_mode in MODE_DEPENDENCY_MAP.items():
|
||||
if dep_mode in dep_status:
|
||||
mode_info = dep_status[dep_mode]
|
||||
capabilities["tool_details"][cap_mode] = {
|
||||
"name": mode_info["name"],
|
||||
"ready": mode_info["ready"],
|
||||
"missing_required": mode_info["missing_required"],
|
||||
"tools": mode_info["tools"],
|
||||
}
|
||||
|
||||
# Use Intercept's SDR detection
|
||||
sdr_factory = self._get_sdr_factory()
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Tests for shared capability detection."""
|
||||
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from utils.capabilities import detect_interfaces, detect_mode_availability
|
||||
|
||||
@@ -34,10 +34,90 @@ class TestModeAvailability:
|
||||
modes = detect_mode_availability()
|
||||
assert modes.get("sensor") is False
|
||||
|
||||
def test_pre_computed_dep_status_skips_probe(self):
|
||||
"""Passing dep_status must not trigger a second check_all_dependencies call."""
|
||||
pre_computed = {
|
||||
key: {"ready": True}
|
||||
for key in (
|
||||
"pager",
|
||||
"sensor",
|
||||
"aircraft",
|
||||
"ais",
|
||||
"acars",
|
||||
"aprs",
|
||||
"wifi",
|
||||
"bluetooth",
|
||||
"tscm",
|
||||
"satellite",
|
||||
)
|
||||
}
|
||||
with patch("utils.capabilities.check_all_dependencies") as mock_deps:
|
||||
modes = detect_mode_availability(dep_status=pre_computed)
|
||||
mock_deps.assert_not_called()
|
||||
assert modes.get("sensor") is True
|
||||
assert modes.get("adsb") is True
|
||||
|
||||
|
||||
class TestInterfaceDetection:
|
||||
def test_returns_expected_shape(self, fake_process):
|
||||
with patch("subprocess.Popen", return_value=fake_process()):
|
||||
def test_darwin_wifi_parsing(self):
|
||||
"""The Darwin branch must parse a Wi-Fi device out of networksetup output."""
|
||||
networksetup_output = (
|
||||
"Hardware Port: Wi-Fi\n"
|
||||
"Device: en0\n"
|
||||
"Ethernet Address: aa:bb:cc:dd:ee:ff\n"
|
||||
"\n"
|
||||
"Hardware Port: Thunderbolt Bridge\n"
|
||||
"Device: bridge0\n"
|
||||
)
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
result = MagicMock()
|
||||
result.stdout = networksetup_output
|
||||
result.stderr = ""
|
||||
result.returncode = 0
|
||||
return result
|
||||
|
||||
with (
|
||||
patch("utils.capabilities.platform.system", return_value="Darwin"),
|
||||
patch("subprocess.run", side_effect=fake_run),
|
||||
):
|
||||
interfaces = detect_interfaces()
|
||||
assert set(interfaces) == {"wifi_interfaces", "bt_adapters", "sdr_devices"}
|
||||
assert isinstance(interfaces["wifi_interfaces"], list)
|
||||
|
||||
names = [i["name"] for i in interfaces["wifi_interfaces"]]
|
||||
assert "en0" in names
|
||||
# Verify the full shape of the parsed entry
|
||||
en0 = next(i for i in interfaces["wifi_interfaces"] if i["name"] == "en0")
|
||||
assert "display_name" in en0
|
||||
assert "type" in en0
|
||||
assert "monitor_capable" in en0
|
||||
# Thunderbolt Bridge must not appear — it has no Wi-Fi/AirPort keyword
|
||||
assert "bridge0" not in names
|
||||
|
||||
|
||||
class TestFallback:
|
||||
def test_fallback_uses_check_tool(self):
|
||||
"""When check_all_dependencies raises, fall back to per-tool checks."""
|
||||
with (
|
||||
patch(
|
||||
"utils.capabilities.check_all_dependencies",
|
||||
side_effect=RuntimeError("module unavailable"),
|
||||
),
|
||||
patch("utils.capabilities.check_tool", return_value=False) as mock_check,
|
||||
):
|
||||
modes = detect_mode_availability()
|
||||
assert modes.get("sensor") is False
|
||||
assert mock_check.called
|
||||
|
||||
def test_fallback_extra_mode_tools(self):
|
||||
"""EXTRA_MODE_TOOLS modes (dsc, rtlamr, listening_post) reflect check_tool's return."""
|
||||
with (
|
||||
patch(
|
||||
"utils.capabilities.check_all_dependencies",
|
||||
side_effect=RuntimeError("module unavailable"),
|
||||
),
|
||||
patch("utils.capabilities.check_tool", return_value=False),
|
||||
):
|
||||
modes = detect_mode_availability()
|
||||
assert modes.get("dsc") is False
|
||||
assert modes.get("rtlamr") is False
|
||||
assert modes.get("listening_post") is False
|
||||
|
||||
@@ -60,15 +60,21 @@ FALLBACK_TOOL_CHECKS = {
|
||||
}
|
||||
|
||||
|
||||
def detect_mode_availability() -> dict[str, bool]:
|
||||
def detect_mode_availability(dep_status: dict | None = None) -> dict[str, bool]:
|
||||
"""Detect mode availability from tool dependencies.
|
||||
|
||||
Returns a ``{cap_mode: bool}`` map of raw tool readiness. Falls back to
|
||||
direct tool checks if :func:`check_all_dependencies` raises.
|
||||
|
||||
Args:
|
||||
dep_status: Pre-computed result of :func:`check_all_dependencies`. When
|
||||
supplied the probe is skipped entirely, avoiding a second call when
|
||||
the caller has already fetched it.
|
||||
"""
|
||||
modes: dict[str, bool] = {}
|
||||
try:
|
||||
dep_status = check_all_dependencies()
|
||||
if dep_status is None:
|
||||
dep_status = check_all_dependencies()
|
||||
except Exception as e:
|
||||
logger.warning(f"Dependency check failed, using fallback: {e}")
|
||||
return _detect_mode_availability_fallback()
|
||||
|
||||
Reference in New Issue
Block a user