Merge pull request #461 from AIOSAI/system/devpulse-testdrone-cover-all-untested-functions-per-tdplan-
feat(system): test(drone): cover all untested functions per TDPLAN-0003
This commit is contained in:
AIPass
@@ -430,3 +430,45 @@ class TestModuleOrchestrator:
|
||||
result = handle_command("unknown_sub", ["arg"])
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
# ===================================================================
|
||||
# 8. list_all module function
|
||||
# ===================================================================
|
||||
|
||||
|
||||
class TestListAll:
|
||||
"""Tests for the list_all() module-level function."""
|
||||
|
||||
def test_list_all_empty(self, isolated_registry: Path) -> None:
|
||||
"""list_all returns empty list when no commands registered."""
|
||||
from aipass.drone.apps.modules.commands import list_all
|
||||
|
||||
result = list_all()
|
||||
assert result == []
|
||||
|
||||
def test_list_all_returns_sorted(self, isolated_registry: Path) -> None:
|
||||
"""list_all returns commands sorted by name."""
|
||||
ops.add_command("zebra", "@t", "c1")
|
||||
ops.add_command("alpha", "@t", "c2")
|
||||
ops.add_command("middle", "@t", "c3")
|
||||
|
||||
from aipass.drone.apps.modules.commands import list_all
|
||||
|
||||
result = list_all()
|
||||
names = [c["name"] for c in result]
|
||||
assert names == ["alpha", "middle", "zebra"]
|
||||
|
||||
def test_list_all_returns_all_fields(self, isolated_registry: Path) -> None:
|
||||
"""list_all returns full command dicts with expected keys."""
|
||||
ops.add_command("audit", "@seedgo", "audit", ["aipass"], "Run audit", "seedgo")
|
||||
|
||||
from aipass.drone.apps.modules.commands import list_all
|
||||
|
||||
result = list_all()
|
||||
assert len(result) == 1
|
||||
cmd = result[0]
|
||||
assert cmd["name"] == "audit"
|
||||
assert cmd["target"] == "@seedgo"
|
||||
assert cmd["command"] == "audit"
|
||||
assert cmd["args"] == ["aipass"]
|
||||
|
||||
@@ -684,3 +684,85 @@ def test_log_operation_survives_empty_log_file(tmp_path: Path) -> None: # JH-04
|
||||
data = json.loads(log_path.read_text(encoding="utf-8"))
|
||||
assert len(data) == 1, "Should have exactly one log entry after recovery"
|
||||
assert data[0]["operation"] == "test_op"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 9. increment_counter
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestIncrementCounter:
|
||||
"""Tests for increment_counter()."""
|
||||
|
||||
def test_increment_creates_counter(self, tmp_path: Path) -> None:
|
||||
"""Incrementing a non-existent counter creates it at the given amount."""
|
||||
json_handler.ensure_module_jsons("incr_test")
|
||||
result = json_handler.increment_counter("incr_test", "hits")
|
||||
assert result is True
|
||||
data = json_handler.load_json("incr_test", "data")
|
||||
assert data["hits"] == 1
|
||||
|
||||
def test_increment_adds_to_existing(self, tmp_path: Path) -> None:
|
||||
"""Incrementing an existing counter adds to its current value."""
|
||||
json_handler.ensure_module_jsons("incr_test2")
|
||||
json_handler.increment_counter("incr_test2", "hits")
|
||||
json_handler.increment_counter("incr_test2", "hits")
|
||||
json_handler.increment_counter("incr_test2", "hits", amount=5)
|
||||
data = json_handler.load_json("incr_test2", "data")
|
||||
assert data["hits"] == 7
|
||||
|
||||
def test_increment_custom_amount(self, tmp_path: Path) -> None:
|
||||
"""Custom amount parameter is respected."""
|
||||
json_handler.ensure_module_jsons("incr_test3")
|
||||
result = json_handler.increment_counter("incr_test3", "visits", amount=42)
|
||||
assert result is True
|
||||
data = json_handler.load_json("incr_test3", "data")
|
||||
assert data["visits"] == 42
|
||||
|
||||
def test_increment_returns_false_on_load_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Returns False when data file cannot be loaded."""
|
||||
json_handler.ensure_module_jsons("incr_fail")
|
||||
monkeypatch.setattr(json_handler, "load_json", lambda *a, **kw: None)
|
||||
result = json_handler.increment_counter("incr_fail", "hits")
|
||||
assert result is False
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 10. update_data_metrics
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestUpdateDataMetrics:
|
||||
"""Tests for update_data_metrics()."""
|
||||
|
||||
def test_update_single_metric(self, tmp_path: Path) -> None:
|
||||
"""Updating a single metric writes it to the data file."""
|
||||
json_handler.ensure_module_jsons("metric_test")
|
||||
result = json_handler.update_data_metrics("metric_test", uptime=99.5)
|
||||
assert result is True
|
||||
data = json_handler.load_json("metric_test", "data")
|
||||
assert data["uptime"] == 99.5
|
||||
|
||||
def test_update_multiple_metrics(self, tmp_path: Path) -> None:
|
||||
"""Multiple keyword arguments are all written."""
|
||||
json_handler.ensure_module_jsons("metric_test2")
|
||||
json_handler.update_data_metrics("metric_test2", cpu=80, memory=60, disk=45)
|
||||
data = json_handler.load_json("metric_test2", "data")
|
||||
assert data["cpu"] == 80
|
||||
assert data["memory"] == 60
|
||||
assert data["disk"] == 45
|
||||
|
||||
def test_update_overwrites_existing(self, tmp_path: Path) -> None:
|
||||
"""Existing keys are overwritten by new values."""
|
||||
json_handler.ensure_module_jsons("metric_test3")
|
||||
json_handler.update_data_metrics("metric_test3", score=10)
|
||||
json_handler.update_data_metrics("metric_test3", score=20)
|
||||
data = json_handler.load_json("metric_test3", "data")
|
||||
assert data["score"] == 20
|
||||
|
||||
def test_update_returns_false_on_load_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Returns False when data file cannot be loaded."""
|
||||
json_handler.ensure_module_jsons("metric_fail")
|
||||
monkeypatch.setattr(json_handler, "load_json", lambda *a, **kw: None)
|
||||
result = json_handler.update_data_metrics("metric_fail", x=1)
|
||||
assert result is False
|
||||
|
||||
@@ -2,13 +2,14 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Module path prefix for patching
|
||||
# ---------------------------------------------------------------------------
|
||||
_MOD = "aipass.drone.apps.modules.module_registry"
|
||||
_HANDLER = "aipass.drone.apps.handlers.module_registry_handler"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -396,3 +397,323 @@ class TestHandleCommandUnknown:
|
||||
mock_jh.log_operation.assert_called_once_with(
|
||||
"handle_command", {"module": "module_registry", "command": "bogus"}
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 7. refresh_external_modules (handler-level)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestRefreshExternalModules:
|
||||
"""refresh_external_modules() reloads from routing_config.json."""
|
||||
|
||||
def test_refresh_reloads_config(self) -> None:
|
||||
"""After refresh, _EXTERNAL_MODULES reflects current config."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original = dict(mrh._EXTERNAL_MODULES)
|
||||
with patch.object(mrh, "_load_external_modules", return_value={}):
|
||||
mrh.refresh_external_modules()
|
||||
assert mrh._EXTERNAL_MODULES == {}
|
||||
mrh._EXTERNAL_MODULES = original
|
||||
|
||||
def test_refresh_picks_up_new_module(self) -> None:
|
||||
"""A module added to config appears after refresh."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
from aipass.drone.apps.handlers.module_registry_handler import (
|
||||
_ExternalModuleConfig,
|
||||
)
|
||||
|
||||
original = dict(mrh._EXTERNAL_MODULES)
|
||||
fake = {"newmod": _ExternalModuleConfig("newmod", "some.entry", "New", "1.0")}
|
||||
with patch.object(mrh, "_load_external_modules", return_value=fake):
|
||||
mrh.refresh_external_modules()
|
||||
assert "newmod" in mrh._EXTERNAL_MODULES
|
||||
mrh._EXTERNAL_MODULES = original
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 8. route_module_command (handler-level)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestRouteModuleCommand:
|
||||
"""route_module_command() routes to external or internal modules."""
|
||||
|
||||
def test_routes_external_module_via_capture(self) -> None:
|
||||
"""External modules route through capture_main."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
from aipass.drone.apps.handlers.module_registry_handler import (
|
||||
_ExternalModuleConfig,
|
||||
)
|
||||
|
||||
original_ext = dict(mrh._EXTERNAL_MODULES)
|
||||
mrh._EXTERNAL_MODULES["testext"] = _ExternalModuleConfig("testext", "fake.entry", "Test external", "1.0")
|
||||
try:
|
||||
with (
|
||||
patch(
|
||||
f"{_HANDLER}.capture_main",
|
||||
return_value={
|
||||
"stdout": "ok",
|
||||
"stderr": "",
|
||||
"exit_code": 0,
|
||||
},
|
||||
) as mock_cap,
|
||||
patch(f"{_HANDLER}.json_handler"),
|
||||
):
|
||||
result = mrh.route_module_command("testext", "run", ["--flag"])
|
||||
assert result["stdout"] == "ok"
|
||||
assert result["exit_code"] == 0
|
||||
mock_cap.assert_called_once_with("fake.entry", "testext", "run", ["--flag"])
|
||||
finally:
|
||||
mrh._EXTERNAL_MODULES = original_ext
|
||||
|
||||
def test_routes_internal_module_via_import(self) -> None:
|
||||
"""Internal modules route through importlib + handle_command."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
mrh._INTERNAL_MODULES["fakeint"] = "fake.internal.mod"
|
||||
try:
|
||||
mock_mod = MagicMock()
|
||||
mock_mod.handle_command.return_value = {
|
||||
"stdout": "done",
|
||||
"stderr": "",
|
||||
"exit_code": 0,
|
||||
}
|
||||
with (
|
||||
patch(
|
||||
f"{_HANDLER}.importlib.import_module",
|
||||
return_value=mock_mod,
|
||||
),
|
||||
patch(f"{_HANDLER}.json_handler"),
|
||||
):
|
||||
result = mrh.route_module_command("fakeint", "status")
|
||||
assert result["stdout"] == "done"
|
||||
mock_mod.handle_command.assert_called_once_with("status", None)
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
def test_internal_bool_true_converted_to_dict(self) -> None:
|
||||
"""Internal module returning True converts to dict with exit_code 0."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
mrh._INTERNAL_MODULES["boolmod"] = "fake.bool.mod"
|
||||
try:
|
||||
mock_mod = MagicMock()
|
||||
mock_mod.handle_command.return_value = True
|
||||
with (
|
||||
patch(
|
||||
f"{_HANDLER}.importlib.import_module",
|
||||
return_value=mock_mod,
|
||||
),
|
||||
patch(f"{_HANDLER}.json_handler"),
|
||||
):
|
||||
result = mrh.route_module_command("boolmod", "check")
|
||||
assert result["exit_code"] == 0
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
def test_internal_bool_false_converted_to_exit_code_1(self) -> None:
|
||||
"""Internal module returning False gets exit_code 1."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
mrh._INTERNAL_MODULES["failmod"] = "fake.fail.mod"
|
||||
try:
|
||||
mock_mod = MagicMock()
|
||||
mock_mod.handle_command.return_value = False
|
||||
with (
|
||||
patch(
|
||||
f"{_HANDLER}.importlib.import_module",
|
||||
return_value=mock_mod,
|
||||
),
|
||||
patch(f"{_HANDLER}.json_handler"),
|
||||
):
|
||||
result = mrh.route_module_command("failmod", "broken")
|
||||
assert result["exit_code"] == 1
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
def test_logs_operation_for_external(self) -> None:
|
||||
"""External module routing logs via json_handler."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
from aipass.drone.apps.handlers.module_registry_handler import (
|
||||
_ExternalModuleConfig,
|
||||
)
|
||||
|
||||
original_ext = dict(mrh._EXTERNAL_MODULES)
|
||||
mrh._EXTERNAL_MODULES["logext"] = _ExternalModuleConfig("logext", "fake.entry", "Log test", "1.0")
|
||||
try:
|
||||
with (
|
||||
patch(
|
||||
f"{_HANDLER}.capture_main",
|
||||
return_value={
|
||||
"stdout": "",
|
||||
"stderr": "",
|
||||
"exit_code": 0,
|
||||
},
|
||||
),
|
||||
patch(f"{_HANDLER}.json_handler") as mock_jh,
|
||||
):
|
||||
mrh.route_module_command("logext", "ping")
|
||||
mock_jh.log_operation.assert_called_once_with(
|
||||
"route_module_command", {"module": "logext", "command": "ping"}
|
||||
)
|
||||
finally:
|
||||
mrh._EXTERNAL_MODULES = original_ext
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 9. get_module_help (handler-level)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestGetModuleHelp:
|
||||
"""get_module_help() retrieves help text from modules."""
|
||||
|
||||
def test_external_module_help_no_command(self) -> None:
|
||||
"""External module help without command captures --help output."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
from aipass.drone.apps.handlers.module_registry_handler import (
|
||||
_ExternalModuleConfig,
|
||||
)
|
||||
|
||||
original_ext = dict(mrh._EXTERNAL_MODULES)
|
||||
mrh._EXTERNAL_MODULES["helpext"] = _ExternalModuleConfig("helpext", "fake.entry", "Help test", "1.0")
|
||||
try:
|
||||
with patch(
|
||||
f"{_HANDLER}.capture_main",
|
||||
return_value={
|
||||
"stdout": "Usage: helpext",
|
||||
"stderr": "",
|
||||
},
|
||||
) as mock_cap:
|
||||
result = mrh.get_module_help("helpext")
|
||||
assert result == "Usage: helpext"
|
||||
mock_cap.assert_called_once_with("fake.entry", "helpext", "--help")
|
||||
finally:
|
||||
mrh._EXTERNAL_MODULES = original_ext
|
||||
|
||||
def test_external_module_help_with_command(self) -> None:
|
||||
"""External module help with command passes command + --help."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
from aipass.drone.apps.handlers.module_registry_handler import (
|
||||
_ExternalModuleConfig,
|
||||
)
|
||||
|
||||
original_ext = dict(mrh._EXTERNAL_MODULES)
|
||||
mrh._EXTERNAL_MODULES["helpext2"] = _ExternalModuleConfig("helpext2", "fake.entry", "Help test", "1.0")
|
||||
try:
|
||||
with patch(
|
||||
f"{_HANDLER}.capture_main",
|
||||
return_value={"stdout": "Sub help", "stderr": ""},
|
||||
) as mock_cap:
|
||||
result = mrh.get_module_help("helpext2", "subcmd")
|
||||
assert result == "Sub help"
|
||||
mock_cap.assert_called_once_with("fake.entry", "helpext2", "subcmd", ["--help"])
|
||||
finally:
|
||||
mrh._EXTERNAL_MODULES = original_ext
|
||||
|
||||
def test_internal_module_help(self) -> None:
|
||||
"""Internal module help calls get_help() on the module."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
mrh._INTERNAL_MODULES["helpint"] = "fake.help.mod"
|
||||
try:
|
||||
mock_mod = MagicMock()
|
||||
mock_mod.get_help.return_value = "Internal help text"
|
||||
with patch(
|
||||
f"{_HANDLER}.importlib.import_module",
|
||||
return_value=mock_mod,
|
||||
):
|
||||
result = mrh.get_module_help("helpint", "status")
|
||||
assert result == "Internal help text"
|
||||
mock_mod.get_help.assert_called_once_with("status")
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
def test_internal_module_without_get_help(self) -> None:
|
||||
"""Internal module without get_help() returns empty string."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
mrh._INTERNAL_MODULES["nohelp"] = "fake.nohelp.mod"
|
||||
try:
|
||||
mock_mod = MagicMock(spec=[])
|
||||
with patch(
|
||||
f"{_HANDLER}.importlib.import_module",
|
||||
return_value=mock_mod,
|
||||
):
|
||||
result = mrh.get_module_help("nohelp")
|
||||
assert result == ""
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
def test_unknown_module_returns_empty(self) -> None:
|
||||
"""Unknown module name returns empty string."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
result = mrh.get_module_help("nonexistent_mod_xyz")
|
||||
assert result == ""
|
||||
|
||||
def test_import_error_returns_empty(self) -> None:
|
||||
"""ImportError during internal module load returns empty string."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
mrh._INTERNAL_MODULES["broken"] = "fake.broken.mod"
|
||||
try:
|
||||
with patch(
|
||||
f"{_HANDLER}.importlib.import_module",
|
||||
side_effect=ImportError("nope"),
|
||||
):
|
||||
result = mrh.get_module_help("broken")
|
||||
assert result == ""
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 10. register_module (handler-level)
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestRegisterModule:
|
||||
"""register_module() adds internal modules dynamically."""
|
||||
|
||||
def test_register_adds_to_internal(self) -> None:
|
||||
"""register_module makes the module available."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
try:
|
||||
mrh.register_module("dynamic", "aipass.dynamic.mod")
|
||||
assert mrh._INTERNAL_MODULES["dynamic"] == "aipass.dynamic.mod"
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
def test_register_overwrites_existing(self) -> None:
|
||||
"""Registering an existing name overwrites the adapter path."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
try:
|
||||
mrh.register_module("overwrite", "path.v1")
|
||||
mrh.register_module("overwrite", "path.v2")
|
||||
assert mrh._INTERNAL_MODULES["overwrite"] == "path.v2"
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
def test_registered_module_appears_in_list(self) -> None:
|
||||
"""A dynamically registered module appears in list_modules()."""
|
||||
import aipass.drone.apps.handlers.module_registry_handler as mrh
|
||||
|
||||
original_int = dict(mrh._INTERNAL_MODULES)
|
||||
try:
|
||||
mrh.register_module("newdyn", "aipass.newdyn.mod")
|
||||
assert "newdyn" in mrh.list_modules()
|
||||
finally:
|
||||
mrh._INTERNAL_MODULES = original_int
|
||||
|
||||
@@ -155,7 +155,7 @@
|
||||
"content_hash": "a4cf0a8e3b4f",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f026": {
|
||||
"f015": {
|
||||
"path": "apps/modules/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "e3b0c44298fc",
|
||||
@@ -263,7 +263,7 @@
|
||||
"content_hash": "28e9ae373563",
|
||||
"has_branch_placeholder": false
|
||||
},
|
||||
"f015": {
|
||||
"f026": {
|
||||
"path": "apps/plugins/__init__.py",
|
||||
"name": "__init__.py",
|
||||
"content_hash": "e3b0c44298fc",
|
||||
|
||||
Reference in New Issue
Block a user