mirror of
https://github.com/smittix/intercept.git
synced 2026-07-05 16:18:12 -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
|
# Detect interfaces via shared capability detection
|
||||||
capabilities["interfaces"] = detect_interfaces()
|
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
|
# 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():
|
for mode, ready in raw_modes.items():
|
||||||
# Apply this agent's per-mode config gating on top of tool readiness
|
# 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
|
capabilities["modes"][mode] = ready if config.modes_enabled.get(mode, True) else False
|
||||||
|
|
||||||
# Populate detailed tool info for modes tracked in dependencies.py
|
# Populate detailed tool info for modes tracked in dependencies.py
|
||||||
if HAS_DEPENDENCIES_MODULE:
|
if dep_status is not None:
|
||||||
try:
|
for dep_mode, cap_mode in MODE_DEPENDENCY_MAP.items():
|
||||||
dep_status = check_all_dependencies()
|
if dep_mode in dep_status:
|
||||||
for dep_mode, cap_mode in MODE_DEPENDENCY_MAP.items():
|
mode_info = dep_status[dep_mode]
|
||||||
if dep_mode in dep_status:
|
capabilities["tool_details"][cap_mode] = {
|
||||||
mode_info = dep_status[dep_mode]
|
"name": mode_info["name"],
|
||||||
capabilities["tool_details"][cap_mode] = {
|
"ready": mode_info["ready"],
|
||||||
"name": mode_info["name"],
|
"missing_required": mode_info["missing_required"],
|
||||||
"ready": mode_info["ready"],
|
"tools": mode_info["tools"],
|
||||||
"missing_required": mode_info["missing_required"],
|
}
|
||||||
"tools": mode_info["tools"],
|
|
||||||
}
|
|
||||||
except Exception as e:
|
|
||||||
logger.warning(f"Tool detail collection failed: {e}")
|
|
||||||
|
|
||||||
# Use Intercept's SDR detection
|
# Use Intercept's SDR detection
|
||||||
sdr_factory = self._get_sdr_factory()
|
sdr_factory = self._get_sdr_factory()
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
"""Tests for shared capability detection."""
|
"""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
|
from utils.capabilities import detect_interfaces, detect_mode_availability
|
||||||
|
|
||||||
@@ -34,10 +34,90 @@ class TestModeAvailability:
|
|||||||
modes = detect_mode_availability()
|
modes = detect_mode_availability()
|
||||||
assert modes.get("sensor") is False
|
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:
|
class TestInterfaceDetection:
|
||||||
def test_returns_expected_shape(self, fake_process):
|
def test_darwin_wifi_parsing(self):
|
||||||
with patch("subprocess.Popen", return_value=fake_process()):
|
"""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()
|
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.
|
"""Detect mode availability from tool dependencies.
|
||||||
|
|
||||||
Returns a ``{cap_mode: bool}`` map of raw tool readiness. Falls back to
|
Returns a ``{cap_mode: bool}`` map of raw tool readiness. Falls back to
|
||||||
direct tool checks if :func:`check_all_dependencies` raises.
|
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] = {}
|
modes: dict[str, bool] = {}
|
||||||
try:
|
try:
|
||||||
dep_status = check_all_dependencies()
|
if dep_status is None:
|
||||||
|
dep_status = check_all_dependencies()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Dependency check failed, using fallback: {e}")
|
logger.warning(f"Dependency check failed, using fallback: {e}")
|
||||||
return _detect_mode_availability_fallback()
|
return _detect_mode_availability_fallback()
|
||||||
|
|||||||
Reference in New Issue
Block a user