slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(prax): slim devpulse dashboard — drop duplicated todos[] bodies, keep count (startup-context fix)

Browse Source
This commit is contained in:
AIOSAI
2026-06-12 23:09:42 -07:00
parent c5a96cbdcf
commit 1057be65a4
18 changed files with 310 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+8
View File
@@ -13,6 +13,14 @@ PyPI version — not the changelog header.
### Changed
- **Devpulse dashboard slimmed — todos no longer duplicated (startup-context
fix).** `DASHBOARD.local.json` was embedding the full `todos[]` bodies that
already live in `.trinity/local.json`; since both files are read at every
startup, that was pure duplication. The dashboard now emits `todo_count` only
(the glance value) — the bodies are commented out in the prax
`devpulse_dashboard` plugin's `todo_section.py` (revivable). Dashboard
`DASHBOARD.local.json` 6.8 KB → 3.0 KB. Devpulse-only (plugin, not templated).
Verified: seedgo 100%, 17/17 plugin tests.
- **`.backupignore` is now a true `.gitignore` for the backup system — a single
source of truth (FPLAN-0269).** Replaced the hand-rolled `fnmatch`+part-loop
matcher (which broke leading-slash anchoring, `*`-crossing-`/`, dir-only `foo/`,
src/aipass/daemon/tests/conftest.py
+5 -7
View File
@@ -10,11 +10,12 @@
#
# CODE STANDARDS:
# - Error handling: Use error handler system (apps/handlers/error/)
# =============================================
#
#@Meta header not seedgo standards
# =============================================
#
# @Meta header not seedgo standards
"""Shared pytest fixtures for daemon tests"""
import os
import tempfile
@@ -45,10 +46,7 @@ def sample_test_data() -> dict:
Customize this fixture for your module's needs
"""
return {
"test_key": "test_value",
"sample_data": "example"
}
return {"test_key": "test_value", "sample_data": "example"}
@pytest.fixture()
src/aipass/daemon/tests/test_actions_registry.py
+9 -8
View File
@@ -50,6 +50,7 @@ def clean_registry(tmp_path):
# CRUD TESTS
# =============================================
class TestCreate:
def test_create_action_basic(self, clean_registry):
"""Create a simple schedule action and verify fields."""
@@ -128,8 +129,7 @@ class TestList:
def test_list_excludes_completed(self, clean_registry):
"""Completed reminders should be excluded by default."""
create_action(name="done", action_type="reminder", schedule_type="once",
due_date="2026-01-01")
create_action(name="done", action_type="reminder", schedule_type="once", due_date="2026-01-01")
mark_reminder_completed("0001")
assert len(list_actions()) == 0
assert len(list_actions(include_completed=True)) == 1
@@ -146,8 +146,7 @@ class TestToggle:
def test_toggle_on(self, clean_registry):
"""Toggle an action back on."""
create_action(name="toggleme", action_type="schedule", schedule_type="daily",
enabled=False)
create_action(name="toggleme", action_type="schedule", schedule_type="daily", enabled=False)
assert toggle_action("0001", True) is True
action = get_action("0001")
assert action is not None
@@ -174,6 +173,7 @@ class TestDelete:
# DUE CHECKING TESTS
# =============================================
class TestIsDue:
def test_daily_due_at_correct_time(self, clean_registry):
"""Daily action is due when current time matches."""
@@ -190,6 +190,7 @@ class TestIsDue:
def test_daily_not_due_wrong_time(self, clean_registry):
"""Daily action is not due at wrong time (12 hours away from now)."""
from datetime import datetime
now = datetime.now()
# Pick a time 12 hours away — always outside the 15-min fuzzy window
far_hour = (now.hour + 12) % 24
@@ -295,6 +296,7 @@ class TestIsDue:
# NEXT RUN TESTS
# =============================================
class TestCalcNextRun:
def test_daily_next_run(self, clean_registry):
"""Daily action calculates next run correctly."""
@@ -334,11 +336,11 @@ class TestNextDueStr:
# UPDATE TESTS
# =============================================
class TestUpdateLastRun:
def test_update_last_run(self, clean_registry):
"""Update last_run sets timestamp and recalculates next_run."""
create_action(name="test", action_type="schedule", schedule_type="interval",
interval_minutes=60)
create_action(name="test", action_type="schedule", schedule_type="interval", interval_minutes=60)
ts = "2026-03-02T12:00:00"
assert update_last_run("0001", ts) is True
action = get_action("0001")
@@ -350,8 +352,7 @@ class TestUpdateLastRun:
class TestMarkCompleted:
def test_mark_reminder_completed(self, clean_registry):
"""Marking a reminder completed sets completed timestamp and disables it."""
create_action(name="reminder", action_type="reminder", schedule_type="once",
due_date="2026-03-01")
create_action(name="reminder", action_type="reminder", schedule_type="once", due_date="2026-03-01")
assert mark_reminder_completed("0001") is True
action = get_action("0001")
assert action is not None
src/aipass/daemon/tests/test_activity_report.py
+7 -1
View File
@@ -8,7 +8,7 @@
"""Tests for the activity_report CLI module (apps/modules/activity_report.py)."""
from unittest.mock import patch, MagicMock
from unittest.mock import patch
MODULE = "aipass.daemon.apps.modules.activity_report"
@@ -17,6 +17,7 @@ MODULE = "aipass.daemon.apps.modules.activity_report"
# handle_command -- routing basics
# =============================================
@patch(f"{MODULE}.json_handler")
@patch(f"{MODULE}.console")
@patch(f"{MODULE}.error")
@@ -64,6 +65,7 @@ class TestHandleCommandRouting:
# handle_command -- activity-report
# =============================================
@patch(f"{MODULE}.json_handler")
@patch(f"{MODULE}.console")
@patch(f"{MODULE}.error")
@@ -114,6 +116,7 @@ class TestActivityReportCommand:
# handle_command -- activity_report alias
# =============================================
@patch(f"{MODULE}.json_handler")
@patch(f"{MODULE}.console")
@patch(f"{MODULE}.error")
@@ -144,6 +147,7 @@ class TestActivityReportAlias:
# handle_command -- branch-health
# =============================================
@patch(f"{MODULE}.json_handler")
@patch(f"{MODULE}.console")
@patch(f"{MODULE}.error")
@@ -198,6 +202,7 @@ class TestBranchHealthCommand:
# _parse_hours_arg
# =============================================
@patch(f"{MODULE}.logger")
class TestParseHoursArg:
"""Tests for _parse_hours_arg helper."""
@@ -229,6 +234,7 @@ class TestParseHoursArg:
# _extract_branch_name
# =============================================
class TestExtractBranchName:
"""Tests for _extract_branch_name helper."""
src/aipass/daemon/tests/test_cli_routing.py
+2 -1
View File
@@ -25,7 +25,7 @@ Covers 9 tests:
"""
import sys
from unittest.mock import patch, MagicMock
from unittest.mock import patch
import pytest
@@ -49,6 +49,7 @@ def _mock_log_operation():
# CLI Routing Tests
# ============================================================================
def test_help_flag() -> None:
"""--help flag triggers help and returns exit code 0."""
with patch.object(sys, "argv", ["daemon", "--help"]):
src/aipass/daemon/tests/test_contracts.py
+14 -23
View File
@@ -42,9 +42,7 @@ _json_mod_path = f"aipass.{BRANCH_MODULE}.apps.handlers.json.json_handler"
if _handler_pkg not in sys.modules:
_stub = types.ModuleType(_handler_pkg)
_handlers_dir = (
Path(__file__).resolve().parents[3] / "aipass" / BRANCH_MODULE / "apps" / "handlers"
)
_handlers_dir = Path(__file__).resolve().parents[3] / "aipass" / BRANCH_MODULE / "apps" / "handlers"
_stub.__path__ = [str(_handlers_dir)]
sys.modules[_handler_pkg] = _stub
@@ -72,8 +70,7 @@ for _candidate in _JSON_DIR_CANDIDATES:
if _JSON_DIR_ATTR is None:
pytest.skip(
f"Cannot find JSON_DIR attribute on {BRANCH_MODULE}.json_handler -- "
f"tried: {_JSON_DIR_CANDIDATES}",
f"Cannot find JSON_DIR attribute on {BRANCH_MODULE}.json_handler -- tried: {_JSON_DIR_CANDIDATES}",
allow_module_level=True,
)
@@ -82,6 +79,7 @@ if _JSON_DIR_ATTR is None:
# Isolation fixture
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def isolate_json_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Redirect JSON operations to tmp_path for test isolation."""
@@ -98,6 +96,7 @@ def isolate_json_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
# Default factory helpers
# ---------------------------------------------------------------------------
def _get_default_for_type(json_type: str, module_name: str = "test_mod") -> Any:
"""Call whichever default factory the branch exposes."""
for fn_name in (
@@ -145,6 +144,7 @@ def _default_factory_raises_on_unknown() -> bool:
# Group 1 -- Return type contracts (4 tests)
# ============================================================================
def test_command_returns_bool() -> None:
"""command_returns_bool: route_command returns a bool."""
from aipass.daemon.apps import daemon as _daemon_mod
@@ -167,40 +167,33 @@ def test_command_returns_bool() -> None:
def test_paths_return_path() -> None:
"""paths_return_path: get_json_path returns a Path."""
result = json_handler.get_json_path("contract_mod", "config")
assert isinstance(result, Path), (
f"get_json_path must return Path, got {type(result)}"
)
assert isinstance(result, Path), f"get_json_path must return Path, got {type(result)}"
def test_paths_return_path_for_data() -> None:
"""paths_return_path: get_json_path returns Path for data type too."""
result = json_handler.get_json_path("contract_mod", "data")
assert isinstance(result, Path), (
f"get_json_path('data') must return Path, got {type(result)}"
)
assert isinstance(result, Path), f"get_json_path('data') must return Path, got {type(result)}"
def test_ensure_json_exists_returns_bool(tmp_path: Path) -> None:
"""ensure_json_exists must return a bool."""
result = json_handler.ensure_json_exists("contract_mod", "data")
assert isinstance(result, bool), (
f"ensure_json_exists must return bool, got {type(result)}"
)
assert isinstance(result, bool), f"ensure_json_exists must return bool, got {type(result)}"
assert result is True
def test_load_json_returns_dict_for_config(tmp_path: Path) -> None:
"""load_json for config type must return a dict."""
result = json_handler.load_json("contract_mod", "config")
assert isinstance(result, dict), (
f"load_json('...', 'config') must return dict, got {type(result)}"
)
assert isinstance(result, dict), f"load_json('...', 'config') must return dict, got {type(result)}"
# ============================================================================
# Group 2 -- Data structure contracts (3 tests)
# ============================================================================
def test_config_keys(tmp_path: Path) -> None:
"""config_keys: config data structure contains module_name and version."""
json_handler.ensure_json_exists("struct_mod", "config")
@@ -227,9 +220,7 @@ def test_log_entry_has_operation(tmp_path: Path) -> None:
val = getattr(_mod, _JSON_DIR_ATTR)
json_dir = Path(val) if isinstance(val, str) else val
log = json.loads(
(json_dir / "struct_mod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "struct_mod_log.json").read_text(encoding="utf-8"))
assert len(log) >= 1, "log_operation must append at least one entry"
assert "operation" in log[-1], "Log entry must have 'operation' key"
assert log[-1]["operation"] == "contract_test"
@@ -239,6 +230,7 @@ def test_log_entry_has_operation(tmp_path: Path) -> None:
# Group 3 -- Success/failure paths (4 tests)
# ============================================================================
def test_known_routes_true() -> None:
"""known_routes_true: a module that handles a command causes route_command to return True."""
from aipass.daemon.apps import daemon as _daemon_mod
@@ -287,6 +279,7 @@ def test_no_args_triggers() -> None:
# Group 4 -- Infrastructure mocking (3 tests)
# ============================================================================
def test_log_operation_mocked(tmp_path: Path) -> None:
"""Infrastructure: log_operation can be mocked without side effects."""
with patch.object(_mod, "log_operation", return_value=True) as mock_log:
@@ -306,8 +299,6 @@ def test_sys_modules_mock() -> None:
def test_reimport_after_mock(tmp_path: Path) -> None:
"""reimport_after_mock: module can be reloaded cleanly."""
handler_module = sys.modules.get(
f"aipass.{BRANCH_MODULE}.apps.handlers.json.json_handler"
)
handler_module = sys.modules.get(f"aipass.{BRANCH_MODULE}.apps.handlers.json.json_handler")
if handler_module:
importlib.reload(handler_module)
src/aipass/daemon/tests/test_data_loader.py
+6
View File
@@ -33,6 +33,7 @@ get_escalations = _dl_mod.get_escalations
# FIXTURES
# =============================================
@pytest.fixture(autouse=True)
def isolate_paths(tmp_path, monkeypatch):
"""Redirect INBOX_PATH and LOCAL_PATH to tmp_path for every test."""
@@ -79,6 +80,7 @@ def _write_json(path: Path, data: object) -> None:
# LOAD INBOX TESTS
# =============================================
class TestLoadInbox:
def test_load_valid_inbox(self, isolate_paths, sample_inbox_data, monkeypatch):
"""Loading a well-formed inbox.json returns its full contents."""
@@ -116,6 +118,7 @@ class TestLoadInbox:
# LOAD LOCAL TESTS
# =============================================
class TestLoadLocal:
def test_load_valid_local(self, isolate_paths, sample_local_data):
"""Loading a well-formed local.json returns its full contents."""
@@ -149,6 +152,7 @@ class TestLoadLocal:
# CATEGORIZE MESSAGES TESTS
# =============================================
class TestCategorizeMessages:
def test_new_and_opened_split(self):
"""Messages are split into new and opened buckets by status."""
@@ -218,6 +222,7 @@ class TestCategorizeMessages:
# GET SESSION SUMMARY TESTS
# =============================================
class TestGetSessionSummary:
def test_summary_with_sessions(self, sample_local_data):
"""Session summary extracts totals and latest session."""
@@ -260,6 +265,7 @@ class TestGetSessionSummary:
# GET ESCALATIONS TESTS
# =============================================
class TestGetEscalations:
def test_urgent_message_detected(self):
"""Messages with URGENT in subject are escalated."""
src/aipass/daemon/tests/test_error_resilience.py
+4 -5
View File
@@ -38,9 +38,7 @@ _json_mod_path = f"aipass.{BRANCH_MODULE}.apps.handlers.json.json_handler"
if _handler_pkg not in sys.modules:
_stub = types.ModuleType(_handler_pkg)
_handlers_dir = (
Path(__file__).resolve().parents[3] / "aipass" / BRANCH_MODULE / "apps" / "handlers"
)
_handlers_dir = Path(__file__).resolve().parents[3] / "aipass" / BRANCH_MODULE / "apps" / "handlers"
_stub.__path__ = [str(_handlers_dir)]
sys.modules[_handler_pkg] = _stub
@@ -68,8 +66,7 @@ for _candidate in _JSON_DIR_CANDIDATES:
if _JSON_DIR_ATTR is None:
pytest.skip(
f"Cannot find JSON_DIR attribute on {BRANCH_MODULE}.json_handler -- "
f"tried: {_JSON_DIR_CANDIDATES}",
f"Cannot find JSON_DIR attribute on {BRANCH_MODULE}.json_handler -- tried: {_JSON_DIR_CANDIDATES}",
allow_module_level=True,
)
@@ -78,6 +75,7 @@ if _JSON_DIR_ATTR is None:
# Isolation fixture
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def isolate_json_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Redirect JSON operations to tmp_path for test isolation."""
@@ -103,6 +101,7 @@ def _json_dir_as_path(tmp_path: Path) -> Path:
# Error Resilience Tests (4 tests)
# ============================================================================
def test_missing_file(tmp_path: Path) -> None:
"""Loading a non-existent file returns a graceful default, not a crash."""
json_dir = _json_dir_as_path(tmp_path)
src/aipass/daemon/tests/test_json_handler.py
+25 -44
View File
@@ -44,9 +44,7 @@ _json_mod_path = f"aipass.{BRANCH_MODULE}.apps.handlers.json.json_handler"
if _handler_pkg not in sys.modules:
_stub = types.ModuleType(_handler_pkg)
_handlers_dir = (
Path(__file__).resolve().parents[3] / "aipass" / BRANCH_MODULE / "apps" / "handlers"
)
_handlers_dir = Path(__file__).resolve().parents[3] / "aipass" / BRANCH_MODULE / "apps" / "handlers"
_stub.__path__ = [str(_handlers_dir)]
sys.modules[_handler_pkg] = _stub
@@ -74,8 +72,7 @@ for _candidate in _JSON_DIR_CANDIDATES:
if _JSON_DIR_ATTR is None:
pytest.skip(
f"Cannot find JSON_DIR attribute on {BRANCH_MODULE}.json_handler -- "
f"tried: {_JSON_DIR_CANDIDATES}",
f"Cannot find JSON_DIR attribute on {BRANCH_MODULE}.json_handler -- tried: {_JSON_DIR_CANDIDATES}",
allow_module_level=True,
)
@@ -84,6 +81,7 @@ if _JSON_DIR_ATTR is None:
# Default factory discovery
# ---------------------------------------------------------------------------
def _get_default_for_type(json_type: str, module_name: str = "test_mod") -> Any:
"""Call whichever default factory the branch exposes."""
for fn_name in (
@@ -146,6 +144,7 @@ def _default_factory_raises_on_unknown() -> bool:
# Isolation fixture
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def isolate_json_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Redirect JSON operations to tmp_path for test isolation."""
@@ -171,6 +170,7 @@ def _json_dir_as_path(tmp_path: Path) -> Path:
# Group 1 -- default_factory (4 tests)
# ============================================================================
def test_default_factory_config_returns_dict() -> None:
"""default_factory: config template returns a dict with required keys."""
if not _has_default_factory():
@@ -213,6 +213,7 @@ def test_default_factory_unknown_type_raises() -> None:
# Group 2 -- validate (10 tests)
# ============================================================================
def test_validate_valid_config() -> None:
"""validate: valid config structure passes."""
data = {"module_name": "x", "version": "1.0.0", "config": {}}
@@ -274,6 +275,7 @@ def test_validate_none_input_returns_false() -> None:
# Group 3 -- get_path (3 tests)
# ============================================================================
def test_get_path_returns_path_type(tmp_path: Path) -> None:
"""get_path: returns Path or str."""
result = json_handler.get_json_path("mymod", "config")
@@ -298,6 +300,7 @@ def test_get_path_different_combos_differ(tmp_path: Path) -> None:
# Group 4 -- ensure_exists (5 tests)
# ============================================================================
def test_ensure_exists_creates_file(tmp_path: Path) -> None:
"""ensure_exists: creates file when missing."""
result = json_handler.ensure_json_exists("ens_mod", "config")
@@ -364,6 +367,7 @@ def test_ensure_exists_returns_bool(tmp_path: Path) -> None:
# Group 5 -- load (4 tests)
# ============================================================================
def test_load_creates_default_when_missing(tmp_path: Path) -> None:
"""load: auto-creates default when file is missing."""
result = json_handler.load_json("fresh_mod", "log")
@@ -400,6 +404,7 @@ def test_load_returns_list_for_log(tmp_path: Path) -> None:
# Group 6 -- save (5 tests)
# ============================================================================
def test_save_roundtrip(tmp_path: Path) -> None:
"""save: data survives save-then-load roundtrip."""
json_dir = _json_dir_as_path(tmp_path)
@@ -437,9 +442,7 @@ def test_save_data_updates_last_updated(tmp_path: Path) -> None:
data = {"created": "2025-01-01", "last_updated": "2025-01-01"}
json_handler.save_json("ts", "data", data)
on_disk = json.loads(
(json_dir / "ts_data.json").read_text(encoding="utf-8")
)
on_disk = json.loads((json_dir / "ts_data.json").read_text(encoding="utf-8"))
assert on_disk["last_updated"] == today, "Saving data type must auto-stamp last_updated"
@@ -460,13 +463,12 @@ def test_save_writes_valid_json(tmp_path: Path) -> None:
# Group 7 -- log_operation (7 tests)
# ============================================================================
def test_log_operation_appends_entry(tmp_path: Path) -> None:
"""log_operation appends an entry to the log file."""
json_handler.log_operation("deploy", module_name="logmod")
json_dir = _json_dir_as_path(tmp_path)
log = json.loads(
(json_dir / "logmod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "logmod_log.json").read_text(encoding="utf-8"))
assert len(log) >= 1, "log_operation must append at least one entry"
assert log[-1]["operation"] == "deploy"
@@ -482,21 +484,15 @@ def test_log_operation_entry_has_timestamp(tmp_path: Path) -> None:
"""log_operation entries include a timestamp field."""
json_handler.log_operation("check_ts", module_name="tsmod")
json_dir = _json_dir_as_path(tmp_path)
log = json.loads(
(json_dir / "tsmod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "tsmod_log.json").read_text(encoding="utf-8"))
assert "timestamp" in log[-1], "Log entry must have a timestamp field"
def test_log_operation_includes_data(tmp_path: Path) -> None:
"""log_operation includes data dict when provided."""
json_handler.log_operation(
"with_data", data={"count": 5}, module_name="datamod"
)
json_handler.log_operation("with_data", data={"count": 5}, module_name="datamod")
json_dir = _json_dir_as_path(tmp_path)
log = json.loads(
(json_dir / "datamod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "datamod_log.json").read_text(encoding="utf-8"))
assert "data" in log[-1], "Log entry must include data dict when provided"
assert log[-1]["data"]["count"] == 5
@@ -507,9 +503,7 @@ def test_log_operation_multiple_calls_accumulate(tmp_path: Path) -> None:
json_handler.log_operation("second", module_name="accmod")
json_handler.log_operation("third", module_name="accmod")
json_dir = _json_dir_as_path(tmp_path)
log = json.loads(
(json_dir / "accmod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "accmod_log.json").read_text(encoding="utf-8"))
assert len(log) >= 3, "Multiple log_operation calls must accumulate entries"
ops = [e["operation"] for e in log[-3:]]
assert ops == ["first", "second", "third"]
@@ -530,9 +524,7 @@ def test_log_operation_fifo_rotation(tmp_path: Path) -> None:
json_handler.log_operation(f"op_{i}", module_name="fifomod")
json_dir = _json_dir_as_path(tmp_path)
log = json.loads(
(json_dir / "fifomod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "fifomod_log.json").read_text(encoding="utf-8"))
assert len(log) <= max_entries, f"Log must not exceed {max_entries} entries"
assert log[-1]["operation"] == f"op_{max_entries + 4}", "Most recent entry must be last"
@@ -541,9 +533,7 @@ def test_log_operation_empty_dict_not_attached(tmp_path: Path) -> None:
"""log_operation: empty dict data should not create non-empty data field."""
json_handler.log_operation("no_data", data={}, module_name="emptymod")
json_dir = _json_dir_as_path(tmp_path)
log = json.loads(
(json_dir / "emptymod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "emptymod_log.json").read_text(encoding="utf-8"))
entry = log[-1]
if "data" in entry:
assert entry["data"] == {} or entry["data"] is None
@@ -553,6 +543,7 @@ def test_log_operation_empty_dict_not_attached(tmp_path: Path) -> None:
# Group 8 -- ensure_module (5 tests)
# ============================================================================
def test_ensure_module_creates_all_three(tmp_path: Path) -> None:
"""ensure_module: creates config, data, and log files."""
if not hasattr(json_handler, "ensure_module_jsons"):
@@ -579,19 +570,13 @@ def test_ensure_module_files_pass_validation(tmp_path: Path) -> None:
json_handler.ensure_module_jsons("valid_mod")
json_dir = _json_dir_as_path(tmp_path)
config = json.loads(
(json_dir / "valid_mod_config.json").read_text(encoding="utf-8")
)
config = json.loads((json_dir / "valid_mod_config.json").read_text(encoding="utf-8"))
assert json_handler.validate_json_structure(config, "config") is True
data = json.loads(
(json_dir / "valid_mod_data.json").read_text(encoding="utf-8")
)
data = json.loads((json_dir / "valid_mod_data.json").read_text(encoding="utf-8"))
assert json_handler.validate_json_structure(data, "data") is True
log = json.loads(
(json_dir / "valid_mod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "valid_mod_log.json").read_text(encoding="utf-8"))
assert json_handler.validate_json_structure(log, "log") is True
@@ -601,9 +586,7 @@ def test_ensure_module_data_has_correct_keys(tmp_path: Path) -> None:
pytest.skip("Branch does not have ensure_module_jsons")
json_handler.ensure_module_jsons("keymod")
json_dir = _json_dir_as_path(tmp_path)
data = json.loads(
(json_dir / "keymod_data.json").read_text(encoding="utf-8")
)
data = json.loads((json_dir / "keymod_data.json").read_text(encoding="utf-8"))
assert "created" in data, "Data file must have 'created' key"
assert "last_updated" in data, "Data file must have 'last_updated' key"
@@ -614,8 +597,6 @@ def test_ensure_module_log_is_empty_list(tmp_path: Path) -> None:
pytest.skip("Branch does not have ensure_module_jsons")
json_handler.ensure_module_jsons("listmod")
json_dir = _json_dir_as_path(tmp_path)
log = json.loads(
(json_dir / "listmod_log.json").read_text(encoding="utf-8")
)
log = json.loads((json_dir / "listmod_log.json").read_text(encoding="utf-8"))
assert isinstance(log, list), "Log file must be a list"
assert len(log) == 0, "Initial log file must be an empty list"
src/aipass/daemon/tests/test_memory_health.py
+30 -14
View File
@@ -30,10 +30,11 @@ from aipass.daemon.apps.handlers.monitoring import memory_health as mh
# HELPERS
# =============================================
def _write_json(path: Path, data: dict) -> None:
"""Write a dict to a JSON file, creating parent dirs."""
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'w', encoding='utf-8') as f:
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f)
@@ -67,6 +68,7 @@ def _setup_full_branch(tmp_path: Path) -> Path:
# FILE EXISTENCE TESTS
# =============================================
class TestCheckMemoryFilesExist:
"""Tests for check_memory_files_exist()."""
@@ -157,6 +159,7 @@ class TestCheckMemoryFilesExist:
# STRUCTURE VALIDATION TESTS
# =============================================
class TestValidateMemoryStructure:
"""Tests for validate_memory_structure()."""
@@ -177,12 +180,15 @@ class TestValidateMemoryStructure:
def test_valid_structure_with_metadata_key(self, tmp_path: Path) -> None:
"""File using 'metadata' key (instead of 'document_metadata') is valid."""
f = tmp_path / "alt_meta.json"
_write_json(f, {
"metadata": {
"version": "1.0.0",
"limits": {"max_entries": 100},
_write_json(
f,
{
"metadata": {
"version": "1.0.0",
"limits": {"max_entries": 100},
},
},
})
)
result = mh.validate_memory_structure(str(f))
@@ -193,12 +199,15 @@ class TestValidateMemoryStructure:
def test_missing_limits_field(self, tmp_path: Path) -> None:
"""Metadata present but no limits field should report issue."""
f = tmp_path / "no_limits.json"
_write_json(f, {
"document_metadata": {
"document_type": "session_history",
"version": "1.0.0",
_write_json(
f,
{
"document_metadata": {
"document_type": "session_history",
"version": "1.0.0",
},
},
})
)
result = mh.validate_memory_structure(str(f))
@@ -253,6 +262,7 @@ class TestValidateMemoryStructure:
# FRESHNESS TESTS
# =============================================
class TestCheckFreshness:
"""Tests for check_freshness()."""
@@ -346,6 +356,7 @@ class TestCheckFreshness:
# OVERALL HEALTH STATUS TESTS
# =============================================
class TestGetMemoryHealthStatus:
"""Tests for get_memory_health_status()."""
@@ -468,9 +479,14 @@ class TestGetMemoryHealthStatus:
result = mh.get_memory_health_status(str(branch), "KEYS")
expected_keys = {
"branch_name", "branch_path", "overall_status",
"file_check", "structure_checks", "freshness_checks",
"issues", "check_time",
"branch_name",
"branch_path",
"overall_status",
"file_check",
"structure_checks",
"freshness_checks",
"issues",
"check_time",
}
assert expected_keys == set(result.keys())
assert isinstance(result["overall_status"], str)
src/aipass/daemon/tests/test_red_flag_detector.py
+28 -11
View File
@@ -15,12 +15,10 @@
"""Tests for the red flag detection engine."""
from datetime import datetime, timedelta
from unittest.mock import patch, MagicMock
from datetime import datetime
from unittest.mock import patch
import pytest
from aipass.daemon.apps.handlers.monitoring import red_flag_detector
from aipass.daemon.apps.handlers.monitoring.red_flag_detector import (
_parse_iso_datetime,
get_branch_status,
@@ -68,6 +66,7 @@ def _make_activity(
# _parse_iso_datetime TESTS
# =============================================
class TestParseIsoDatetime:
"""Tests for ISO datetime string parsing."""
@@ -116,6 +115,7 @@ class TestParseIsoDatetime:
# get_branch_status TESTS
# =============================================
class TestGetBranchStatus:
"""Tests for single-branch status detection."""
@@ -239,10 +239,18 @@ class TestGetBranchStatus:
mock_scan.return_value = _make_activity(code_files=[], memory_files=[])
result = get_branch_status("TEST", "/fake/path/test")
required_keys = {
"branch_name", "branch_path", "status", "code_changes",
"code_change_count", "latest_code_change", "memory_files_modified",
"memory_last_update", "hours_since_code", "threshold_hours",
"reason", "check_time",
"branch_name",
"branch_path",
"status",
"code_changes",
"code_change_count",
"latest_code_change",
"memory_files_modified",
"memory_last_update",
"hours_since_code",
"threshold_hours",
"reason",
"check_time",
}
assert required_keys.issubset(result.keys())
@@ -285,6 +293,7 @@ class TestGetBranchStatus:
# detect_red_flags (scan all branches) TESTS
# =============================================
class TestScanAllBranches:
"""Tests for multi-branch scanning and sorting via detect_red_flags."""
@@ -383,6 +392,7 @@ class TestScanAllBranches:
# get_red_flag_summary TESTS
# =============================================
class TestGetRedFlagSummary:
"""Tests for the get_red_flag_summary aggregation function."""
@@ -477,9 +487,16 @@ class TestGetRedFlagSummary:
summary = get_red_flag_summary(since_timestamp=datetime(2026, 3, 1))
expected_keys = {
"total_branches", "red_flags", "ok", "no_activity",
"violations", "scan_time", "threshold_hours",
"time_window_hours", "errors", "all_branches",
"total_branches",
"red_flags",
"ok",
"no_activity",
"violations",
"scan_time",
"threshold_hours",
"time_window_hours",
"errors",
"all_branches",
}
assert expected_keys.issubset(set(summary.keys()))
src/aipass/daemon/tests/test_scheduler_cron.py
+21 -4
View File
@@ -28,6 +28,7 @@ MODULE = "aipass.daemon.apps.scheduler_cron"
# FIXTURES
# =============================================
def _make_task(
task_id: str = "abc12345-6789",
recipient: str = "@devpulse",
@@ -58,6 +59,7 @@ def _silence_logging():
# _send_email_via_drone
# =============================================
class TestSendEmailViaDrone:
"""Tests for _send_email_via_drone subprocess wrapper."""
@@ -109,6 +111,7 @@ class TestSendEmailViaDrone:
# _next_cron_run
# =============================================
class TestNextCronRun:
"""Tests for next cron run time calculation."""
@@ -147,6 +150,7 @@ class TestNextCronRun:
# _process_single_task
# =============================================
class TestProcessSingleTask:
"""Tests for the single-task dispatch function."""
@@ -242,6 +246,7 @@ class TestProcessSingleTask:
# process_due_tasks
# =============================================
class TestProcessDueTasks:
"""Tests for the top-level due-task processor."""
@@ -349,6 +354,7 @@ class TestProcessDueTasks:
# _run_locked
# =============================================
class TestRunLocked:
"""Tests for the locked orchestration function."""
@@ -357,8 +363,13 @@ class TestRunLocked:
task_results = {"due": 0, "success": 0, "failed": 0, "recovered": 0, "errors": []}
action_results = {
"total": 0, "enabled": 0, "executed": 0, "failed": 0,
"errors": [], "executed_actions": [], "skipped_actions": [],
"total": 0,
"enabled": 0,
"executed": 0,
"failed": 0,
"errors": [],
"executed_actions": [],
"skipped_actions": [],
}
with (
@@ -375,8 +386,13 @@ class TestRunLocked:
task_results = {"due": 1, "success": 0, "failed": 1, "recovered": 0, "errors": ["fail"]}
action_results = {
"total": 0, "enabled": 0, "executed": 0, "failed": 0,
"errors": [], "executed_actions": [], "skipped_actions": [],
"total": 0,
"enabled": 0,
"executed": 0,
"failed": 0,
"errors": [],
"executed_actions": [],
"skipped_actions": [],
}
with (
@@ -416,6 +432,7 @@ class TestRunLocked:
# main
# =============================================
class TestMain:
"""Tests for the main entry point."""
src/aipass/daemon/tests/test_scheduler_ops.py
+3
View File
@@ -17,6 +17,7 @@ MODULE = "aipass.daemon.apps.modules.scheduler_ops"
# handle_command — routing
# =============================================
@patch(f"{MODULE}.json_handler")
@patch(f"{MODULE}.console")
@patch(f"{MODULE}.logger")
@@ -84,6 +85,7 @@ class TestHandleCommand:
# Module-level availability flags
# =============================================
class TestRegistryAvailability:
"""Verify that registry imports succeed in the test environment."""
@@ -102,6 +104,7 @@ class TestRegistryAvailability:
# print_introspection
# =============================================
@patch(f"{MODULE}.console")
class TestPrintIntrospection:
"""Tests for print_introspection output."""
src/aipass/daemon/tests/test_task_registry.py
+133 -77
View File
@@ -52,6 +52,7 @@ def isolate_registry(tmp_path):
# DATE PARSING TESTS
# =============================================
class TestParseDueDate:
def test_days_format(self):
"""'7d' should resolve to 7 days from today."""
@@ -126,6 +127,7 @@ class TestParseDueDate:
# LOAD / SAVE TESTS
# =============================================
class TestLoadSave:
def test_load_creates_file_if_missing(self, isolate_registry):
"""load_tasks should create schedule.json if it does not exist."""
@@ -167,6 +169,7 @@ class TestLoadSave:
# CREATE TASK TESTS
# =============================================
class TestCreateTask:
@patch.object(_mod.json_handler, "log_operation")
def test_create_basic(self, mock_log):
@@ -219,16 +222,21 @@ class TestCreateTask:
# DUE TASKS TESTS
# =============================================
class TestDueTasks:
def test_overdue_task_returned(self, isolate_registry):
"""A pending task with a past due_date should be returned."""
yesterday = (datetime.now().date() - timedelta(days=1)).isoformat()
save_tasks([{
"id": "past01",
"due_date": yesterday,
"status": "pending",
"task": "overdue",
}])
save_tasks(
[
{
"id": "past01",
"due_date": yesterday,
"status": "pending",
"task": "overdue",
}
]
)
due = get_due_tasks()
assert len(due) == 1
assert due[0]["id"] == "past01"
@@ -236,12 +244,16 @@ class TestDueTasks:
def test_today_task_returned(self, isolate_registry):
"""A pending task due today should be returned."""
today = datetime.now().date().isoformat()
save_tasks([{
"id": "today01",
"due_date": today,
"status": "pending",
"task": "due today",
}])
save_tasks(
[
{
"id": "today01",
"due_date": today,
"status": "pending",
"task": "due today",
}
]
)
due = get_due_tasks()
assert len(due) == 1
assert due[0]["id"] == "today01"
@@ -249,36 +261,48 @@ class TestDueTasks:
def test_future_task_not_returned(self, isolate_registry):
"""A pending task with a future due_date should not be returned."""
future = (datetime.now().date() + timedelta(days=30)).isoformat()
save_tasks([{
"id": "future01",
"due_date": future,
"status": "pending",
"task": "future task",
}])
save_tasks(
[
{
"id": "future01",
"due_date": future,
"status": "pending",
"task": "future task",
}
]
)
due = get_due_tasks()
assert len(due) == 0
def test_dispatching_task_excluded(self, isolate_registry):
"""Tasks with status 'dispatching' should not be returned."""
yesterday = (datetime.now().date() - timedelta(days=1)).isoformat()
save_tasks([{
"id": "disp01",
"due_date": yesterday,
"status": "dispatching",
"task": "already dispatching",
}])
save_tasks(
[
{
"id": "disp01",
"due_date": yesterday,
"status": "dispatching",
"task": "already dispatching",
}
]
)
due = get_due_tasks()
assert len(due) == 0
def test_completed_task_excluded(self, isolate_registry):
"""Tasks with status 'completed' should not be returned."""
yesterday = (datetime.now().date() - timedelta(days=1)).isoformat()
save_tasks([{
"id": "done01",
"due_date": yesterday,
"status": "completed",
"task": "done",
}])
save_tasks(
[
{
"id": "done01",
"due_date": yesterday,
"status": "completed",
"task": "done",
}
]
)
due = get_due_tasks()
assert len(due) == 0
@@ -292,15 +316,20 @@ class TestDueTasks:
# STATUS TRANSITION TESTS
# =============================================
class TestStatusTransitions:
def _seed_task(self, task_id: str = "abc12345abcd1234", status: str = "pending"):
"""Helper to seed a single task."""
save_tasks([{
"id": task_id,
"task": "test",
"status": status,
"due_date": "2026-01-01",
}])
save_tasks(
[
{
"id": task_id,
"task": "test",
"status": status,
"due_date": "2026-01-01",
}
]
)
return task_id
def test_mark_dispatching_success(self):
@@ -369,17 +398,22 @@ class TestStatusTransitions:
# RECOVER STALE DISPATCHES TESTS
# =============================================
class TestRecoverStale:
def test_recovers_stale_task(self, isolate_registry):
"""Task stuck in dispatching beyond max_age should be reset."""
stale_time = (datetime.now() - timedelta(minutes=10)).isoformat()
save_tasks([{
"id": "stale01",
"task": "stale dispatch",
"status": "dispatching",
"dispatch_started": stale_time,
"due_date": "2026-01-01",
}])
save_tasks(
[
{
"id": "stale01",
"task": "stale dispatch",
"status": "dispatching",
"dispatch_started": stale_time,
"due_date": "2026-01-01",
}
]
)
recovered = recover_stale_dispatches(max_age_minutes=5)
assert recovered == 1
task = get_task_by_id("stale01")
@@ -390,13 +424,17 @@ class TestRecoverStale:
def test_does_not_recover_recent_dispatch(self, isolate_registry):
"""Task dispatching within max_age should not be recovered."""
recent_time = (datetime.now() - timedelta(minutes=1)).isoformat()
save_tasks([{
"id": "recent01",
"task": "recent dispatch",
"status": "dispatching",
"dispatch_started": recent_time,
"due_date": "2026-01-01",
}])
save_tasks(
[
{
"id": "recent01",
"task": "recent dispatch",
"status": "dispatching",
"dispatch_started": recent_time,
"due_date": "2026-01-01",
}
]
)
recovered = recover_stale_dispatches(max_age_minutes=5)
assert recovered == 0
task = get_task_by_id("recent01")
@@ -405,13 +443,17 @@ class TestRecoverStale:
def test_recovers_invalid_timestamp(self, isolate_registry):
"""Task with unparseable dispatch_started should be recovered."""
save_tasks([{
"id": "bad_ts01",
"task": "bad timestamp",
"status": "dispatching",
"dispatch_started": "not-a-date",
"due_date": "2026-01-01",
}])
save_tasks(
[
{
"id": "bad_ts01",
"task": "bad timestamp",
"status": "dispatching",
"dispatch_started": "not-a-date",
"due_date": "2026-01-01",
}
]
)
recovered = recover_stale_dispatches(max_age_minutes=5)
assert recovered == 1
task = get_task_by_id("bad_ts01")
@@ -420,12 +462,16 @@ class TestRecoverStale:
def test_pending_tasks_untouched(self, isolate_registry):
"""Pending tasks should not be affected by recovery."""
save_tasks([{
"id": "ok01",
"task": "normal pending",
"status": "pending",
"due_date": "2026-01-01",
}])
save_tasks(
[
{
"id": "ok01",
"task": "normal pending",
"status": "pending",
"due_date": "2026-01-01",
}
]
)
recovered = recover_stale_dispatches(max_age_minutes=5)
assert recovered == 0
task = get_task_by_id("ok01")
@@ -441,6 +487,7 @@ class TestRecoverStale:
# DELETE TASK TESTS
# =============================================
class TestDeleteTask:
def test_delete_existing(self, isolate_registry):
"""Deleting an existing task returns True and removes it."""
@@ -455,10 +502,12 @@ class TestDeleteTask:
def test_delete_preserves_other_tasks(self, isolate_registry):
"""Deleting one task should leave others intact."""
save_tasks([
{"id": "keep01", "task": "keep this", "status": "pending"},
{"id": "del02", "task": "delete this", "status": "pending"},
])
save_tasks(
[
{"id": "keep01", "task": "keep this", "status": "pending"},
{"id": "del02", "task": "delete this", "status": "pending"},
]
)
delete_task("del02")
remaining = load_tasks()
assert len(remaining) == 1
@@ -473,27 +522,32 @@ class TestDeleteTask:
# GET PENDING TASKS TESTS
# =============================================
class TestGetPendingTasks:
"""Tests for get_pending_tasks()."""
def test_returns_only_pending(self, isolate_registry):
"""Only tasks with status 'pending' are returned."""
save_tasks([
{"id": "pend01", "task": "pending one", "status": "pending"},
{"id": "pend02", "task": "pending two", "status": "pending"},
{"id": "done01", "task": "done", "status": "completed"},
])
save_tasks(
[
{"id": "pend01", "task": "pending one", "status": "pending"},
{"id": "pend02", "task": "pending two", "status": "pending"},
{"id": "done01", "task": "done", "status": "completed"},
]
)
result = get_pending_tasks()
assert len(result) == 2
assert all(t["status"] == "pending" for t in result)
def test_excludes_dispatching_and_completed(self, isolate_registry):
"""Tasks with dispatching or completed status are excluded."""
save_tasks([
{"id": "disp01", "task": "dispatching", "status": "dispatching"},
{"id": "done01", "task": "completed", "status": "completed"},
{"id": "pend01", "task": "pending", "status": "pending"},
])
save_tasks(
[
{"id": "disp01", "task": "dispatching", "status": "dispatching"},
{"id": "done01", "task": "completed", "status": "completed"},
{"id": "pend01", "task": "pending", "status": "pending"},
]
)
result = get_pending_tasks()
assert len(result) == 1
assert result[0]["id"] == "pend01"
@@ -508,6 +562,7 @@ class TestGetPendingTasks:
# ENSURE LOCK DIR TESTS
# =============================================
class TestEnsureLockDir:
"""Tests for ensure_lock_dir()."""
@@ -516,6 +571,7 @@ class TestEnsureLockDir:
lock_dir = isolate_registry.parent
if lock_dir.exists():
import shutil
shutil.rmtree(lock_dir)
assert not lock_dir.exists()
src/aipass/daemon/tests/test_update_and_errors.py
+8 -4
View File
@@ -48,16 +48,20 @@ class TestUpdateCommand:
def test_update_no_args_runs_digest(self) -> None:
"""update with no args should run the digest, not show introspection."""
with patch.object(_update_mod, "load_inbox", return_value={"messages": [], "total_messages": 0}), \
patch.object(_update_mod, "load_local", return_value={}):
with (
patch.object(_update_mod, "load_inbox", return_value={"messages": [], "total_messages": 0}),
patch.object(_update_mod, "load_local", return_value={}),
):
result = _update_mod.handle_command("update", [])
assert result is True
def test_update_no_args_calls_load_inbox(self) -> None:
"""update with no args should call load_inbox (proving it runs the digest)."""
mock_inbox = MagicMock(return_value={"messages": [], "total_messages": 0})
with patch.object(_update_mod, "load_inbox", mock_inbox), \
patch.object(_update_mod, "load_local", return_value={}):
with (
patch.object(_update_mod, "load_inbox", mock_inbox),
patch.object(_update_mod, "load_local", return_value={}),
):
_update_mod.handle_command("update", [])
mock_inbox.assert_called_once()
src/aipass/daemon/tests/test_wakeup_ops.py
+2
View File
@@ -17,6 +17,7 @@ MODULE = "aipass.daemon.apps.modules.wakeup_ops"
# handle_command — routing
# =============================================
@patch(f"{MODULE}.json_handler")
@patch(f"{MODULE}.console")
@patch(f"{MODULE}.logger")
@@ -82,6 +83,7 @@ class TestHandleCommand:
# print_introspection
# =============================================
@patch(f"{MODULE}.console")
class TestPrintIntrospection:
"""Tests for print_introspection output."""
src/aipass/prax/apps/plugins/devpulse_dashboard/todo_section.py
+1 -3
View File
@@ -35,7 +35,6 @@ def build_todo_section(branch_path: Path) -> bool:
section_data: Dict = {
"managed_by": "devpulse",
"todo_count": 0,
"todos": [],
}
return write_section(branch_path, "todo", section_data)
@@ -46,7 +45,6 @@ def build_todo_section(branch_path: Path) -> bool:
section_data = {
"managed_by": "devpulse",
"todo_count": 0,
"todos": [],
}
return write_section(branch_path, "todo", section_data)
@@ -55,7 +53,7 @@ def build_todo_section(branch_path: Path) -> bool:
section_data = {
"managed_by": "devpulse",
"todo_count": len(todos),
"todos": todos,
# "todos": todos, # bodies live in .trinity/local.json — count-only here
}
return write_section(branch_path, "todo", section_data)
src/aipass/prax/tests/test_devpulse_dashboard_plugin.py
+4 -5
View File
@@ -243,8 +243,7 @@ class TestTodoSection:
todo = dash["sections"]["todo"]
assert todo["managed_by"] == "devpulse"
assert todo["todo_count"] == 2
assert len(todo["todos"]) == 2
assert todo["todos"][0]["id"] == "t1"
assert "todos" not in todo
def test_build_todo_section_empty_todos(self, branch_path):
"""Test todo section with empty todos list."""
@@ -260,7 +259,7 @@ class TestTodoSection:
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
todo = dash["sections"]["todo"]
assert todo["todo_count"] == 0
assert todo["todos"] == []
assert "todos" not in todo
def test_build_todo_section_no_local_json(self, branch_path):
"""Test todo section when local.json doesn't exist."""
@@ -272,7 +271,7 @@ class TestTodoSection:
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
todo = dash["sections"]["todo"]
assert todo["todo_count"] == 0
assert todo["todos"] == []
assert "todos" not in todo
def test_build_todo_section_corrupt_json(self, branch_path):
"""Test todo section handles corrupt local.json."""
@@ -303,7 +302,7 @@ class TestTodoSection:
dash = json.loads((branch_path / "DASHBOARD.local.json").read_text())
todo = dash["sections"]["todo"]
assert todo["todo_count"] == 0
assert todo["todos"] == []
assert "todos" not in todo
class TestRefresh: