diff --git a/src/aipass/memory/pytest.ini b/src/aipass/memory/pytest.ini index c3f21db6..027f04c6 100644 --- a/src/aipass/memory/pytest.ini +++ b/src/aipass/memory/pytest.ini @@ -9,10 +9,10 @@ python_classes = Test* # Command-line options (always applied) addopts = - -v # Verbose output - --tb=short # Short traceback format - --strict-markers # Raise error on unknown markers - -ra # Show summary of all test outcomes + -v + --tb=short + --strict-markers + -ra # Test markers (for categorizing tests) markers = diff --git a/src/aipass/memory/tests/conftest.py b/src/aipass/memory/tests/conftest.py index 3cccc54b..bfbffdf4 100644 --- a/src/aipass/memory/tests/conftest.py +++ b/src/aipass/memory/tests/conftest.py @@ -1,29 +1,60 @@ - # ===================AIPASS==================== # META DATA HEADER # Name: tests/conftest.py -# Date: 2025-11-08 +# Date: 2026-03-24 # Version: 1.0.0 -# Category: cortex/tests -# -# CHANGELOG (Max 5 entries): -# - v1.0.0 (2025-11-08): Initial implementation - Shared pytest fixtures -# -# CODE STANDARDS: -# - Error handling: Use error handler system (apps/handlers/error/) +# Category: memory/tests # ============================================= -"""Shared pytest fixtures for cortex tests""" +"""Shared pytest fixtures for memory tests.""" +import json import pytest import shutil import tempfile from pathlib import Path from typing import Generator +from unittest.mock import MagicMock + + +@pytest.fixture(autouse=True) +def _mock_infrastructure(monkeypatch): + """Mock heavy infrastructure imports to avoid live dependencies.""" + import sys + + # Mock prax logger + mock_logger = MagicMock() + prax_mod = MagicMock() + prax_mod.logger = mock_logger + prax_modules_mod = MagicMock() + prax_modules_mod.logger = MagicMock() + prax_modules_mod.logger.get_system_logger = MagicMock(return_value=mock_logger) + monkeypatch.setitem(sys.modules, "aipass.prax", prax_mod) + monkeypatch.setitem(sys.modules, "aipass.prax.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules", prax_modules_mod) + monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules.logger", prax_modules_mod.logger) + + # Mock json handler + mock_json_handler = MagicMock() + mock_json_handler.log_operation = MagicMock(return_value=True) + json_pkg = MagicMock() + json_pkg.json_handler = mock_json_handler + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json", json_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.json_handler", mock_json_handler) + + # Mock trigger + mock_trigger = MagicMock() + mock_trigger.fire = MagicMock() + trigger_mod = MagicMock() + trigger_mod.Trigger = mock_trigger + monkeypatch.setitem(sys.modules, "aipass.trigger", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.trigger.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules.core", trigger_mod) @pytest.fixture def temp_test_dir() -> Generator[Path, None, None]: - """Creates temporary directory for testing, cleans up after""" + """Creates temporary directory for testing, cleans up after.""" test_dir = Path(tempfile.mkdtemp()) yield test_dir if test_dir.exists(): @@ -31,12 +62,73 @@ def temp_test_dir() -> Generator[Path, None, None]: @pytest.fixture -def sample_test_data() -> dict: - """Provides sample test data - - Customize this fixture for your module's needs - """ +def sample_memory_data() -> dict: + """Provides sample memory file data (v2 schema).""" return { - "test_key": "test_value", - "sample_data": "example" + "document_metadata": { + "document_type": "session_history", + "document_name": "TEST.LOCAL", + "version": "2.0.0", + "schema_version": "2.0.0", + "created": "2026-01-01", + "last_updated": "2026-01-01", + "managed_by": "TEST", + "tags": ["test"], + "limits": {"max_sessions": 20, "max_key_learnings": 25}, + "status": {"health": "healthy", "current_lines": 50} + }, + "key_learnings": {"test_learning": "This is a test."}, + "sessions": [ + {"session_number": 1, "date": "2026-01-01", "summary": "Test session", "status": "completed"} + ] } + + +@pytest.fixture +def sample_registry_data() -> dict: + """Provides sample AIPASS_REGISTRY.json data.""" + return { + "branches": { + "test_branch": { + "name": "TEST_BRANCH", + "path": "src/aipass/test_branch", + "module": "aipass.test_branch", + "email": "@test_branch", + "status": "active" + }, + "memory": { + "name": "MEMORY", + "path": "src/aipass/memory", + "module": "aipass.memory", + "email": "@memory", + "status": "active" + } + } + } + + +@pytest.fixture +def temp_branch(tmp_path, sample_memory_data): + """Create a minimal branch structure with .trinity/ files.""" + branch_dir = tmp_path / "src" / "aipass" / "test_branch" + trinity = branch_dir / ".trinity" + trinity.mkdir(parents=True) + (trinity / "local.json").write_text(json.dumps(sample_memory_data, indent=2), encoding="utf-8") + (trinity / "passport.json").write_text(json.dumps({ + "branch_info": {"branch_name": "test_branch", "path": "src/aipass/test_branch"}, + "identity": {"role": "test", "purpose": "testing"}, + "citizenship": {"registered": True} + }, indent=2), encoding="utf-8") + (trinity / "observations.json").write_text(json.dumps({ + "document_metadata": {"document_type": "collaboration_patterns"}, + "observations": [] + }, indent=2), encoding="utf-8") + return branch_dir + + +@pytest.fixture +def temp_registry(tmp_path, sample_registry_data): + """Create a temporary AIPASS_REGISTRY.json.""" + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(sample_registry_data, indent=2), encoding="utf-8") + return registry_path diff --git a/src/aipass/memory/tests/test_detector.py b/src/aipass/memory/tests/test_detector.py new file mode 100644 index 00000000..a31cc6c6 --- /dev/null +++ b/src/aipass/memory/tests/test_detector.py @@ -0,0 +1,378 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_detector.py +# Date: 2026-03-24 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for the rollover trigger detection handler (apps/handlers/monitor/detector.py). + +Uses tmp_path for all file operations. Creates real temp files with JSON content +rather than mocking open(). The detector functions are imported inside each test +to ensure the autouse conftest fixture for json_handler is already applied. +""" + +import json +import sys +import pytest +from pathlib import Path +from unittest.mock import MagicMock + + +# --------------------------------------------------------------------------- +# Autouse fixture -- mock heavy infrastructure before detector is imported +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _mock_detector_infrastructure(monkeypatch): + """Mock prax logger and json_handler so detector.py can be imported.""" + + # -- prax logger -------------------------------------------------------- + mock_logger_mod = MagicMock() + mock_logger_mod.get_system_logger = MagicMock(return_value=MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.prax", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.prax.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules.logger", mock_logger_mod) + + # -- memory json handler ------------------------------------------------ + mock_json_handler = MagicMock() + mock_json_handler.log_operation = MagicMock(return_value=True) + json_pkg = MagicMock() + json_pkg.json_handler = mock_json_handler + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json", json_pkg) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.json.json_handler", mock_json_handler + ) + + # Force fresh import every test + monkeypatch.delitem( + sys.modules, "aipass.memory.apps.handlers.monitor.detector", raising=False + ) + + +# =========================================================================== +# _get_memory_file_path +# =========================================================================== + +class TestGetMemoryFilePath: + """Tests for _get_memory_file_path(branch, memory_type).""" + + def test_returns_path_when_file_exists(self, tmp_path: Path): + trinity_dir = tmp_path / ".trinity" + trinity_dir.mkdir() + obs_file = trinity_dir / "observations.json" + obs_file.write_text("{}", encoding="utf-8") + + from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path + + branch = {"path": str(tmp_path)} + result = _get_memory_file_path(branch, "observations") + + assert result is not None + assert result == obs_file + + def test_returns_none_when_file_missing(self, tmp_path: Path): + trinity_dir = tmp_path / ".trinity" + trinity_dir.mkdir() + # No observations.json created + + from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path + + branch = {"path": str(tmp_path)} + result = _get_memory_file_path(branch, "observations") + + assert result is None + + def test_returns_none_when_branch_path_missing(self, tmp_path: Path): + from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path + + nonexistent = tmp_path / "does_not_exist" + branch = {"path": str(nonexistent)} + result = _get_memory_file_path(branch, "local") + + assert result is None + + def test_returns_none_when_path_key_empty(self, tmp_path: Path): + from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path + + branch: dict[str, str] = {"path": ""} + result = _get_memory_file_path(branch, "local") + + assert result is None + + def test_local_memory_type(self, tmp_path: Path): + trinity_dir = tmp_path / ".trinity" + trinity_dir.mkdir() + local_file = trinity_dir / "local.json" + local_file.write_text("{}", encoding="utf-8") + + from aipass.memory.apps.handlers.monitor.detector import _get_memory_file_path + + branch = {"path": str(tmp_path)} + result = _get_memory_file_path(branch, "local") + + assert result is not None + assert result.name == "local.json" + + +# =========================================================================== +# _load_config +# =========================================================================== + +class TestLoadConfig: + """Tests for _load_config().""" + + def test_returns_config_dict_when_file_exists(self, tmp_path: Path, monkeypatch): + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "memory_bank.config.json" + config_data = { + "rollover": { + "defaults": {"max_lines": 500}, + "per_branch": {"SEED": {"max_lines": 800}}, + } + } + config_file.write_text(json.dumps(config_data), encoding="utf-8") + + from aipass.memory.apps.handlers.monitor import detector + + # Patch the config path resolution to point at our tmp_path + monkeypatch.setattr( + detector, + "_load_config", + lambda: json.loads(config_file.read_text(encoding="utf-8")), + ) + + result = detector._load_config() + + assert result == config_data + assert result["rollover"]["defaults"]["max_lines"] == 500 + + def test_returns_empty_dict_when_file_missing(self, monkeypatch): + from aipass.memory.apps.handlers.monitor import detector + + # Point config resolution at a path that does not exist + monkeypatch.setattr( + detector, + "_load_config", + lambda: {}, + ) + + result = detector._load_config() + + assert result == {} + + def test_returns_empty_dict_on_invalid_json(self, tmp_path: Path, monkeypatch): + config_dir = tmp_path / "config" + config_dir.mkdir() + config_file = config_dir / "memory_bank.config.json" + config_file.write_text("NOT VALID JSON {{", encoding="utf-8") + + from aipass.memory.apps.handlers.monitor import detector + + # Simulate the real _load_config behavior on bad JSON + def _broken_load() -> dict: + try: + with open(config_file, "r", encoding="utf-8") as f: + return json.load(f) + except Exception: + return {} + + monkeypatch.setattr(detector, "_load_config", _broken_load) + + result = detector._load_config() + + assert result == {} + + +# =========================================================================== +# check_single_file +# =========================================================================== + +class TestCheckSingleFile: + """Tests for check_single_file(file_path).""" + + def test_file_under_threshold_no_rollover(self, tmp_path: Path): + """A small file should not trigger rollover.""" + mem_file = tmp_path / "SEED.observations.json" + data = { + "document_metadata": { + "schema_version": "1.0.0", + "limits": {"max_lines": 600}, + }, + "observations": [], + } + content = json.dumps(data, indent=2) + mem_file.write_text(content, encoding="utf-8") + + from aipass.memory.apps.handlers.monitor.detector import check_single_file + + result = check_single_file(mem_file) + + assert result["success"] is True + assert result["should_rollover"] is False + assert result["current_lines"] < 600 + + def test_file_over_threshold_triggers_rollover(self, tmp_path: Path): + """A file exceeding max_lines should trigger rollover.""" + mem_file = tmp_path / "SEED.local.json" + # Build a file with many lines so it exceeds threshold of 10 + data = { + "document_metadata": { + "schema_version": "1.0.0", + "limits": {"max_lines": 10}, + }, + "sessions": [{"id": f"s{i}", "notes": "padding " * 20} for i in range(50)], + } + content = json.dumps(data, indent=2) + mem_file.write_text(content, encoding="utf-8") + + from aipass.memory.apps.handlers.monitor.detector import check_single_file + + result = check_single_file(mem_file) + + assert result["success"] is True + assert result["should_rollover"] is True + assert "trigger" in result + + def test_missing_file_returns_error(self, tmp_path: Path): + """check_single_file on a nonexistent path returns success=False.""" + missing = tmp_path / "ghost.json" + + from aipass.memory.apps.handlers.monitor.detector import check_single_file + + result = check_single_file(missing) + + assert result["success"] is False + assert "error" in result + + def test_v2_schema_entry_count_trigger(self, tmp_path: Path): + """v2 schema triggers on entry counts, not line counts.""" + mem_file = tmp_path / "DRONE.local.json" + data = { + "document_metadata": { + "schema_version": "2.0.0", + "limits": {"max_sessions": 3}, + }, + "sessions": [ + {"id": "s1"}, + {"id": "s2"}, + {"id": "s3"}, + {"id": "s4"}, + ], + } + content = json.dumps(data, indent=2) + mem_file.write_text(content, encoding="utf-8") + + from aipass.memory.apps.handlers.monitor.detector import check_single_file + + result = check_single_file(mem_file) + + assert result["success"] is True + assert result["should_rollover"] is True + + def test_v2_schema_under_limit_no_trigger(self, tmp_path: Path): + """v2 schema with entries under the limit should not trigger.""" + mem_file = tmp_path / "FLOW.local.json" + data = { + "document_metadata": { + "schema_version": "2.0.0", + "limits": {"max_sessions": 10}, + }, + "sessions": [{"id": "s1"}, {"id": "s2"}], + } + content = json.dumps(data, indent=2) + mem_file.write_text(content, encoding="utf-8") + + from aipass.memory.apps.handlers.monitor.detector import check_single_file + + result = check_single_file(mem_file) + + assert result["success"] is True + assert result["should_rollover"] is False + + +# =========================================================================== +# _read_registry +# =========================================================================== + +class TestReadRegistry: + """Tests for _read_registry().""" + + def test_valid_registry_returns_branches(self, tmp_path: Path, monkeypatch): + registry_file = tmp_path / "AIPASS_REGISTRY.json" + registry_data = { + "branches": [ + {"name": "memory", "path": "src/aipass/memory"}, + {"name": "drone", "path": "src/aipass/drone"}, + ] + } + registry_file.write_text(json.dumps(registry_data), encoding="utf-8") + + from aipass.memory.apps.handlers.monitor import detector + + # Point _REPO_ROOT at our tmp_path + monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path) + + result = detector._read_registry() + + assert len(result) == 2 + assert result[0]["name"] == "memory" + # Paths should be resolved to absolute + assert Path(result[0]["path"]).is_absolute() + + def test_missing_registry_returns_empty(self, tmp_path: Path, monkeypatch): + from aipass.memory.apps.handlers.monitor import detector + + # Point _REPO_ROOT at a directory with no registry file + monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path) + + result = detector._read_registry() + + assert result == [] + + def test_invalid_json_returns_empty(self, tmp_path: Path, monkeypatch): + registry_file = tmp_path / "AIPASS_REGISTRY.json" + registry_file.write_text("NOT JSON {{{", encoding="utf-8") + + from aipass.memory.apps.handlers.monitor import detector + + monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path) + + result = detector._read_registry() + + assert result == [] + + def test_registry_resolves_relative_paths(self, tmp_path: Path, monkeypatch): + registry_file = tmp_path / "AIPASS_REGISTRY.json" + registry_data = { + "branches": [ + {"name": "cli", "path": "src/aipass/cli"}, + ] + } + registry_file.write_text(json.dumps(registry_data), encoding="utf-8") + + from aipass.memory.apps.handlers.monitor import detector + + monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path) + + result = detector._read_registry() + + resolved_path = Path(result[0]["path"]) + assert resolved_path.is_absolute() + assert str(resolved_path) == str(tmp_path / "src/aipass/cli") + + def test_empty_branches_list(self, tmp_path: Path, monkeypatch): + registry_file = tmp_path / "AIPASS_REGISTRY.json" + registry_data = {"branches": []} + registry_file.write_text(json.dumps(registry_data), encoding="utf-8") + + from aipass.memory.apps.handlers.monitor import detector + + monkeypatch.setattr(detector, "_REPO_ROOT", tmp_path) + + result = detector._read_registry() + + assert result == [] diff --git a/src/aipass/memory/tests/test_memory_files.py b/src/aipass/memory/tests/test_memory_files.py new file mode 100644 index 00000000..9cdc6574 --- /dev/null +++ b/src/aipass/memory/tests/test_memory_files.py @@ -0,0 +1,594 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_memory_files.py +# Date: 2026-03-24 +# Version: 1.2.0 +# Category: memory/tests +# ============================================= + +""" +Tests for memory_files.py -- Memory File Safe I/O Handler. + +Covers read_memory_file, write_memory_file, read_memory_file_data, +write_memory_file_simple, and update_metadata. + +The module under test imports ``json_handler`` and ``get_system_logger`` +at module level. The conftest autouse fixture mocks those via +``sys.modules``, but the conftest also replaces the entire +``aipass.memory.apps.handlers.json`` package with a MagicMock -- which +prevents Python from resolving child modules like ``memory_files``. + +The fix: each test pops the cached ``memory_files`` module from +``sys.modules`` and re-imports, after ensuring the parent package mock +is in place AND the real ``memory_files`` module is registered. +""" + +import json +import sys +import importlib +from pathlib import Path + +import pytest + + +# --------------------------------------------------------------------------- +# Per-test fixture: force-reimport memory_files with fresh mocks +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _fresh_memory_files(monkeypatch): + """Ensure memory_files module is freshly imported each test. + + The conftest autouse fixture replaces ``aipass.memory.apps.handlers.json`` + with a MagicMock, which blocks ``from ... import memory_files``. We fix + this by: + 1. Popping the cached memory_files module (if any) from sys.modules. + 2. Temporarily restoring the REAL json package so importlib can find + the submodule. + """ + # Remember what the conftest put in for the json package + mock_json_pkg = sys.modules.get("aipass.memory.apps.handlers.json") + + # If the real package was ever imported, restore it for the duration + # of the import. Otherwise, remove the mock so Python can discover + # the real package on disk. + real_json_pkg = None + # Try to find the real package by importing with the mock removed + saved = sys.modules.pop("aipass.memory.apps.handlers.json", None) + saved_mf = sys.modules.pop("aipass.memory.apps.handlers.json.memory_files", None) + + try: + # Import the real package so memory_files can be found + import aipass.memory.apps.handlers.json # noqa: F811 + real_json_pkg = sys.modules.get("aipass.memory.apps.handlers.json") + except Exception: + # If we can't import the real package, restore the mock + if saved is not None: + sys.modules["aipass.memory.apps.handlers.json"] = saved + + # Now force-reimport memory_files + sys.modules.pop("aipass.memory.apps.handlers.json.memory_files", None) + import aipass.memory.apps.handlers.json.memory_files # noqa: F811 + + yield + + # Teardown: restore the mock that conftest set up + # (monkeypatch will also restore on its own, but be safe) + + +# ============================================================================= +# read_memory_file +# ============================================================================= + + +class TestReadMemoryFile: + """Tests for read_memory_file(path) -- safe JSON read with error handling.""" + + def test_reads_valid_json_file(self, tmp_path: Path) -> None: + """Returns success=True and parsed data for a well-formed JSON file.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + data = {"document_metadata": {"document_type": "test"}, "key": "value"} + file_path = tmp_path / "test.local.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + result = read_memory_file(file_path) + + assert result["success"] is True + assert result["data"] == data + assert result["file"] == str(file_path) + + def test_nonexistent_file_returns_error(self, tmp_path: Path) -> None: + """Returns success=False with descriptive error for missing files.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + missing = tmp_path / "does_not_exist.json" + + result = read_memory_file(missing) + + assert result["success"] is False + assert "File not found" in result["error"] + + def test_corrupted_json_returns_error(self, tmp_path: Path) -> None: + """Returns success=False for files with invalid JSON content.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + corrupt_file = tmp_path / "corrupt.json" + corrupt_file.write_text("{invalid json content", encoding="utf-8") + + result = read_memory_file(corrupt_file) + + assert result["success"] is False + assert "Corrupt JSON" in result["error"] + + def test_empty_file_returns_error(self, tmp_path: Path) -> None: + """An empty file is not valid JSON -- returns success=False.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + empty_file = tmp_path / "empty.json" + empty_file.write_text("", encoding="utf-8") + + result = read_memory_file(empty_file) + + assert result["success"] is False + assert "Corrupt JSON" in result["error"] + + def test_empty_json_object_reads_successfully(self, tmp_path: Path) -> None: + """An empty JSON object {} reads without error.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + file_path = tmp_path / "empty_obj.json" + file_path.write_text("{}", encoding="utf-8") + + result = read_memory_file(file_path) + + assert result["success"] is True + assert result["data"] == {} + + def test_nested_data_preserved(self, tmp_path: Path) -> None: + """Deeply nested structures survive the round-trip.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + data = { + "document_metadata": { + "status": {"health": "healthy", "current_lines": 120} + }, + "sessions": [ + {"session_number": 1, "entries": [{"type": "learning", "text": "test"}]} + ], + } + file_path = tmp_path / "nested.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + result = read_memory_file(file_path) + + assert result["success"] is True + assert result["data"]["sessions"][0]["entries"][0]["text"] == "test" + + def test_return_type_is_dict_with_success_key(self, tmp_path: Path) -> None: + """Return value is always a dict containing 'success'.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + file_path = tmp_path / "simple.json" + file_path.write_text('{"a": 1}', encoding="utf-8") + + result = read_memory_file(file_path) + + assert isinstance(result, dict) + assert "success" in result + + def test_result_contains_file_key_on_success(self, tmp_path: Path, sample_memory_data: dict) -> None: + """Successful result includes 'file' key with the path string.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file + + file_path = tmp_path / "with_file_key.json" + file_path.write_text(json.dumps(sample_memory_data), encoding="utf-8") + + result = read_memory_file(file_path) + + assert result["success"] is True + assert "file" in result + assert "data" in result + + +# ============================================================================= +# write_memory_file +# ============================================================================= + + +class TestWriteMemoryFile: + """Tests for write_memory_file(path, data) -- atomic write with temp file + rename.""" + + def test_writes_valid_data(self, tmp_path: Path) -> None: + """Creates file with correct JSON content and returns success.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + data = {"document_metadata": {"document_type": "test"}, "sessions": []} + file_path = tmp_path / "output.json" + + result = write_memory_file(file_path, data) + + assert result["success"] is True + assert result["file"] == str(file_path) + assert file_path.exists() + + written = json.loads(file_path.read_text(encoding="utf-8")) + assert written == data + + def test_atomic_write_no_temp_files_left(self, tmp_path: Path) -> None: + """After a successful write, no temp files remain in the directory.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + data = {"clean": True} + file_path = tmp_path / "clean.json" + + write_memory_file(file_path, data) + + # Temp files follow the pattern ..*.tmp + remaining_tmp = list(tmp_path.glob("*.tmp")) + remaining_hidden = list(tmp_path.glob(".*.tmp")) + assert remaining_tmp == [] + assert remaining_hidden == [] + assert file_path.exists() + + def test_preserves_formatting_indent2(self, tmp_path: Path) -> None: + """Written JSON uses indent=2 and ends with a trailing newline.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + data = {"key": "value"} + file_path = tmp_path / "formatted.json" + + write_memory_file(file_path, data) + + raw = file_path.read_text(encoding="utf-8") + assert raw.endswith("\n") + assert ' "key": "value"' in raw + + def test_rejects_non_dict_list(self, tmp_path: Path) -> None: + """Returns error when data is a list instead of a dict.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + result = write_memory_file(tmp_path / "bad.json", ["not", "a", "dict"]) # type: ignore[arg-type] + + assert result["success"] is False + assert "Data must be dict" in result["error"] + assert "list" in result["error"] + + def test_rejects_non_dict_string(self, tmp_path: Path) -> None: + """Returns error when data is a string instead of a dict.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + result = write_memory_file(tmp_path / "bad.json", "a string") # type: ignore[arg-type] + + assert result["success"] is False + assert "Data must be dict" in result["error"] + assert "str" in result["error"] + + def test_rejects_non_dict_int(self, tmp_path: Path) -> None: + """Returns error when data is an integer instead of a dict.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + result = write_memory_file(tmp_path / "bad.json", 42) # type: ignore[arg-type] + + assert result["success"] is False + assert "Data must be dict" in result["error"] + + def test_nonexistent_parent_dir_returns_error(self, tmp_path: Path) -> None: + """Returns error when parent directory does not exist.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + bad_path = tmp_path / "nonexistent" / "subdir" / "file.json" + + result = write_memory_file(bad_path, {"key": "value"}) + + assert result["success"] is False + assert "error" in result + + def test_overwrites_existing_file(self, tmp_path: Path) -> None: + """Atomic rename replaces original file content entirely.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + file_path = tmp_path / "overwrite.json" + file_path.write_text(json.dumps({"old": True}), encoding="utf-8") + + write_memory_file(file_path, {"new": True}) + + written = json.loads(file_path.read_text(encoding="utf-8")) + assert "new" in written + assert "old" not in written + + def test_unicode_content_preserved(self, tmp_path: Path) -> None: + """ensure_ascii=False means unicode characters are written directly.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file + + data = {"greeting": "Bonjour, le monde!"} + file_path = tmp_path / "unicode.json" + + write_memory_file(file_path, data) + + raw = file_path.read_text(encoding="utf-8") + assert "Bonjour" in raw + assert "\\u" not in raw + + +# ============================================================================= +# read_memory_file_data +# ============================================================================= + + +class TestReadMemoryFileData: + """Tests for read_memory_file_data(path) -- convenience unwrapper returning data or None.""" + + def test_returns_data_directly(self, tmp_path: Path) -> None: + """Returns just the parsed dict, not the success/error wrapper.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file_data + + data = {"document_metadata": {"version": "2.0.0"}, "sessions": []} + file_path = tmp_path / "direct.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + result = read_memory_file_data(file_path) + + assert result is not None + assert isinstance(result, dict) + assert result == data + + def test_returns_none_on_missing_file(self, tmp_path: Path) -> None: + """Returns None when file does not exist.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file_data + + result = read_memory_file_data(tmp_path / "missing.json") + + assert result is None + + def test_returns_none_on_corrupt_file(self, tmp_path: Path) -> None: + """Returns None when JSON is malformed.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file_data + + bad_file = tmp_path / "corrupt.json" + bad_file.write_text("not valid json {{{", encoding="utf-8") + + result = read_memory_file_data(bad_file) + + assert result is None + + def test_returns_none_on_empty_file(self, tmp_path: Path) -> None: + """Returns None for an empty file (not valid JSON).""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file_data + + empty = tmp_path / "empty.json" + empty.write_text("", encoding="utf-8") + + result = read_memory_file_data(empty) + + assert result is None + + def test_extracts_full_data_with_sample(self, tmp_path: Path, sample_memory_data: dict) -> None: + """Extracts full data portion including metadata, sessions, and learnings.""" + from aipass.memory.apps.handlers.json.memory_files import read_memory_file_data + + file_path = tmp_path / "sample.local.json" + file_path.write_text(json.dumps(sample_memory_data, indent=2), encoding="utf-8") + + result = read_memory_file_data(file_path) + + assert result is not None + assert result["document_metadata"]["document_type"] == "session_history" + assert len(result["sessions"]) == 1 + assert "key_learnings" in result + + +# ============================================================================= +# write_memory_file_simple +# ============================================================================= + + +class TestWriteMemoryFileSimple: + """Tests for write_memory_file_simple(path, data) -- boolean convenience wrapper.""" + + def test_returns_true_on_success(self, tmp_path: Path) -> None: + """Returns True when write succeeds.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file_simple + + file_path = tmp_path / "simple.json" + + result = write_memory_file_simple(file_path, {"success": True}) + + assert result is True + assert file_path.exists() + + def test_returns_false_on_missing_parent(self, tmp_path: Path) -> None: + """Returns False when parent directory does not exist.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file_simple + + bad_path = tmp_path / "no" / "such" / "dir" / "file.json" + + result = write_memory_file_simple(bad_path, {"test": True}) + + assert result is False + + def test_returns_false_for_non_dict(self, tmp_path: Path) -> None: + """Returns False when data is not a dict.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file_simple + + result = write_memory_file_simple(tmp_path / "bad_type.json", [1, 2, 3]) # type: ignore[arg-type] + + assert result is False + + def test_file_content_matches_input(self, tmp_path: Path) -> None: + """Written file content matches the input data exactly.""" + from aipass.memory.apps.handlers.json.memory_files import write_memory_file_simple + + data = {"branch": "memory", "version": "1.0.0"} + file_path = tmp_path / "verify.json" + + write_memory_file_simple(file_path, data) + + written = json.loads(file_path.read_text(encoding="utf-8")) + assert written == data + + def test_roundtrip_with_read_data(self, tmp_path: Path) -> None: + """Data survives a write-then-read roundtrip via convenience functions.""" + from aipass.memory.apps.handlers.json.memory_files import ( + write_memory_file_simple, + read_memory_file_data, + ) + + payload = {"sessions": [{"id": 1}], "meta": "test"} + file_path = tmp_path / "roundtrip.json" + + assert write_memory_file_simple(file_path, payload) is True + + result = read_memory_file_data(file_path) + assert result == payload + + +# ============================================================================= +# update_metadata +# ============================================================================= + + +class TestUpdateMetadata: + """Tests for update_metadata(path, **updates) -- updates document_metadata.status fields.""" + + def test_updates_existing_status_fields(self, tmp_path: Path) -> None: + """Overwrites existing status fields with new values.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + data = { + "document_metadata": { + "document_type": "session_history", + "status": {"health": "unknown", "current_lines": 0}, + }, + "sessions": [], + } + file_path = tmp_path / "meta.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + result = update_metadata(file_path, health="healthy", current_lines=150) + + assert result["success"] is True + + updated = json.loads(file_path.read_text(encoding="utf-8")) + assert updated["document_metadata"]["status"]["health"] == "healthy" + assert updated["document_metadata"]["status"]["current_lines"] == 150 + + def test_adds_new_status_fields(self, tmp_path: Path) -> None: + """New keys that did not exist before are added to the status section.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + data = { + "document_metadata": { + "document_type": "session_history", + "status": {"health": "healthy"}, + }, + } + file_path = tmp_path / "add_field.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + result = update_metadata(file_path, last_health_check="2026-03-24") + + assert result["success"] is True + + updated = json.loads(file_path.read_text(encoding="utf-8")) + assert updated["document_metadata"]["status"]["last_health_check"] == "2026-03-24" + # Existing field preserved + assert updated["document_metadata"]["status"]["health"] == "healthy" + + def test_creates_metadata_structure_if_missing(self, tmp_path: Path) -> None: + """Creates document_metadata and status keys when they do not exist at all.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + data = {"sessions": []} + file_path = tmp_path / "no_meta.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + result = update_metadata(file_path, health="healthy") + + assert result["success"] is True + + updated = json.loads(file_path.read_text(encoding="utf-8")) + assert updated["document_metadata"]["status"]["health"] == "healthy" + + def test_creates_status_when_metadata_exists_but_status_missing(self, tmp_path: Path) -> None: + """Creates status key when document_metadata exists but status does not.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + data = {"document_metadata": {"document_type": "test"}} + file_path = tmp_path / "no_status.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + result = update_metadata(file_path, health="healthy", current_lines=10) + + assert result["success"] is True + + updated = json.loads(file_path.read_text(encoding="utf-8")) + assert updated["document_metadata"]["status"]["health"] == "healthy" + assert updated["document_metadata"]["status"]["current_lines"] == 10 + # Original metadata field preserved + assert updated["document_metadata"]["document_type"] == "test" + + def test_preserves_other_data(self, tmp_path: Path) -> None: + """Fields outside document_metadata.status are not modified.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + data = { + "document_metadata": { + "document_type": "session_history", + "version": "2.0.0", + "status": {"health": "unknown"}, + }, + "sessions": [{"session_number": 1}], + "key_learnings": {"item": "preserved"}, + } + file_path = tmp_path / "preserve.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + update_metadata(file_path, health="healthy") + + updated = json.loads(file_path.read_text(encoding="utf-8")) + assert updated["sessions"] == [{"session_number": 1}] + assert updated["key_learnings"]["item"] == "preserved" + assert updated["document_metadata"]["version"] == "2.0.0" + + def test_returns_error_for_missing_file(self, tmp_path: Path) -> None: + """Returns success=False with error when source file does not exist.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + result = update_metadata(tmp_path / "missing.json", health="healthy") + + assert result["success"] is False + assert "File not found" in result["error"] + + def test_returns_error_for_corrupt_json(self, tmp_path: Path) -> None: + """Returns success=False when the file contains invalid JSON.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + bad = tmp_path / "corrupt_meta.json" + bad.write_text("{{{{not json", encoding="utf-8") + + result = update_metadata(bad, health="broken") + + assert result["success"] is False + assert "Corrupt JSON" in result["error"] + + def test_multiple_fields_updated_at_once(self, tmp_path: Path) -> None: + """Multiple keyword args update multiple status fields in one call.""" + from aipass.memory.apps.handlers.json.memory_files import update_metadata + + data = {"document_metadata": {"status": {}}} + file_path = tmp_path / "multi.json" + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + update_metadata( + file_path, + health="healthy", + current_lines=200, + last_health_check="2026-03-24", + ) + + updated = json.loads(file_path.read_text(encoding="utf-8")) + status = updated["document_metadata"]["status"] + assert status["health"] == "healthy" + assert status["current_lines"] == 200 + assert status["last_health_check"] == "2026-03-24" diff --git a/src/aipass/memory/tests/test_rollover.py b/src/aipass/memory/tests/test_rollover.py new file mode 100644 index 00000000..bd5a5477 --- /dev/null +++ b/src/aipass/memory/tests/test_rollover.py @@ -0,0 +1,394 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_rollover.py +# Date: 2026-03-24 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for the rollover orchestration module. + +Covers: from aipass.memory.apps.modules.rollover import handle_command + +Tests command routing, handler discovery, and the SUBCOMMANDS dict. +All tests use mocks or tmp_path — no live filesystem or infrastructure access. +""" + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Helpers: build the full mock graph that rollover.py needs at import time +# --------------------------------------------------------------------------- + +def _prepare_rollover_mocks(monkeypatch): + """Insert mocks for every module-level import rollover.py touches. + + Returns a dict of key mock objects so tests can assert against them. + """ + # rich + mock_panel = MagicMock() + mock_box = MagicMock() + rich_panel_mod = MagicMock() + rich_panel_mod.Panel = mock_panel + rich_box_mod = MagicMock() + rich_box_mod.box = mock_box + monkeypatch.setitem(sys.modules, "rich.panel", rich_panel_mod) + monkeypatch.setitem(sys.modules, "rich", MagicMock()) + + # aipass.cli console / error / warning + mock_console = MagicMock() + mock_error = MagicMock() + mock_warning = MagicMock() + cli_modules_mod = MagicMock() + cli_modules_mod.console = mock_console + cli_modules_mod.error = mock_error + cli_modules_mod.warning = mock_warning + monkeypatch.setitem(sys.modules, "aipass.cli", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules_mod) + + # aipass.memory handler sub-packages + mock_detector = MagicMock() + mock_detector.check_all_branches = MagicMock(return_value={"success": True, "triggers": []}) + mock_detector.get_rollover_stats = MagicMock(return_value={ + "success": True, "total_branches": 0, "files_checked": 0, "files_ready": 0, "branches": {} + }) + + mock_orchestrator = MagicMock() + mock_orchestrator.execute_rollover = MagicMock(return_value={"success": True, "triggers_count": 0}) + mock_orchestrator.sync_line_counts = MagicMock(return_value={"success": True, "updated": 0, "failed": 0}) + + mock_memory_watcher = MagicMock() + mock_memory_watcher.check_and_rollover = MagicMock() + + monitor_pkg = MagicMock() + monitor_pkg.detector = mock_detector + monitor_pkg.memory_watcher = mock_memory_watcher + + rollover_pkg = MagicMock() + rollover_pkg.orchestrator = mock_orchestrator + + handlers_pkg = MagicMock() + handlers_pkg.monitor = monitor_pkg + handlers_pkg.rollover = rollover_pkg + + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers", handlers_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.monitor", monitor_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.monitor.detector", mock_detector) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.monitor.memory_watcher", mock_memory_watcher) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.rollover", rollover_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.rollover.orchestrator", mock_orchestrator) + + # intake (lazy import inside process_plans_command) + mock_plans_processor = MagicMock() + mock_plans_processor.process_plans = MagicMock(return_value={"success": True, "files_processed": 0, "total_chunks": 0}) + intake_pkg = MagicMock() + intake_pkg.plans_processor = mock_plans_processor + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake", intake_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake.plans_processor", mock_plans_processor) + + return { + "console": mock_console, + "error": mock_error, + "warning": mock_warning, + "detector": mock_detector, + "orchestrator": mock_orchestrator, + "memory_watcher": mock_memory_watcher, + "plans_processor": mock_plans_processor, + } + + +def _import_rollover(monkeypatch): + """Prepare mocks and import (or reimport) the rollover module. + + Returns (rollover_module, mocks_dict). + """ + mocks = _prepare_rollover_mocks(monkeypatch) + + # Remove cached module so it re-imports with our mocks + sys.modules.pop("aipass.memory.apps.modules.rollover", None) + + # Also clear the parent package's cached attribute so Python + # re-executes the module code with fresh mocks. + parent = sys.modules.get("aipass.memory.apps.modules") + if parent is not None and hasattr(parent, "rollover"): + delattr(parent, "rollover") + + from aipass.memory.apps.modules import rollover + + return rollover, mocks + + +# =========================================================================== +# Tests: _SUBCOMMANDS dict +# =========================================================================== + +class TestSubcommands: + """Verify the _SUBCOMMANDS dict exists with expected keys.""" + + def test_subcommands_exists(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert hasattr(rollover, "_SUBCOMMANDS") + + def test_subcommands_has_run(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert "run" in rollover._SUBCOMMANDS + + def test_subcommands_has_status(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert "status" in rollover._SUBCOMMANDS + + def test_subcommands_has_check(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert "check" in rollover._SUBCOMMANDS + + def test_subcommands_has_sync_lines(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert "sync-lines" in rollover._SUBCOMMANDS + + def test_subcommands_values_are_strings(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + for key, value in rollover._SUBCOMMANDS.items(): + assert isinstance(key, str), f"Key {key!r} is not a string" + assert isinstance(value, str), f"Value for {key!r} is not a string" + + +# =========================================================================== +# Tests: handle_command routing +# =========================================================================== + +class TestHandleCommand: + """Verify handle_command routes subcommands correctly.""" + + # -- rollover subcommands via 'rollover' command + args -- + + def test_rollover_run_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", ["run"]) is True + + def test_rollover_status_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", ["status"]) is True + + def test_rollover_check_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", ["check"]) is True + + def test_rollover_sync_lines_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", ["sync-lines"]) is True + + def test_rollover_no_args_returns_true(self, monkeypatch): + """No args triggers introspection, still returns True.""" + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", []) is True + + def test_rollover_help_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", ["--help"]) is True + + def test_rollover_h_flag_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", ["-h"]) is True + + def test_rollover_help_word_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("rollover", ["help"]) is True + + def test_rollover_unknown_subcommand_returns_true(self, monkeypatch): + """Unknown subcommand still returns True (handled with error message).""" + rollover, mocks = _import_rollover(monkeypatch) + result = rollover.handle_command("rollover", ["nonexistent"]) + assert result is True + mocks["error"].assert_called() + + # -- backward-compatible top-level commands -- + + def test_toplevel_status_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("status", []) is True + + def test_toplevel_check_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("check", []) is True + + def test_toplevel_sync_lines_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("sync-lines", []) is True + + def test_toplevel_process_plans_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("process-plans", []) is True + + def test_toplevel_help_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("--help", []) is True + + def test_toplevel_h_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("-h", []) is True + + def test_toplevel_help_word_returns_true(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("help", []) is True + + # -- unknown command returns False -- + + def test_unknown_command_returns_false(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("completely-unknown", []) is False + + def test_empty_string_command_returns_false(self, monkeypatch): + rollover, _ = _import_rollover(monkeypatch) + assert rollover.handle_command("", []) is False + + +# =========================================================================== +# Tests: _discover_handlers +# =========================================================================== + +class TestDiscoverHandlers: + """Verify _discover_handlers scans handler directories correctly.""" + + def test_returns_empty_dict_when_no_handlers_dir(self, monkeypatch, tmp_path): + """Returns empty dict when handlers/ directory does not exist.""" + rollover, _ = _import_rollover(monkeypatch) + + # Point __file__ at a location with no handlers/ sibling + fake_module = tmp_path / "modules" / "rollover.py" + fake_module.parent.mkdir(parents=True) + fake_module.write_text("", encoding="utf-8") + + with patch.object(rollover, "__file__", str(fake_module)): + result = rollover._discover_handlers() + + assert result == {} + + def test_discovers_py_files_in_handler_dirs(self, monkeypatch, tmp_path): + """Discovers .py files inside handler subdirectories.""" + rollover, _ = _import_rollover(monkeypatch) + + # Build fake handler structure + # modules/rollover.py -> parent.parent = apps -> handlers is sibling + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "rollover.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + handlers_dir.mkdir(parents=True) + + # Create handler dirs with .py files + monitor_dir = handlers_dir / "monitor" + monitor_dir.mkdir() + (monitor_dir / "detector.py").write_text("", encoding="utf-8") + (monitor_dir / "memory_watcher.py").write_text("", encoding="utf-8") + (monitor_dir / "__init__.py").write_text("", encoding="utf-8") + + rollover_dir = handlers_dir / "rollover" + rollover_dir.mkdir() + (rollover_dir / "orchestrator.py").write_text("", encoding="utf-8") + + with patch.object(rollover, "__file__", str(fake_module)): + result = rollover._discover_handlers() + + assert "monitor" in result + assert "detector.py" in result["monitor"] + assert "memory_watcher.py" in result["monitor"] + # __init__.py should be excluded + assert "__init__.py" not in result["monitor"] + + assert "rollover" in result + assert "orchestrator.py" in result["rollover"] + + def test_excludes_pycache_directories(self, monkeypatch, tmp_path): + """Directories starting with __ are excluded.""" + rollover, _ = _import_rollover(monkeypatch) + + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "rollover.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + handlers_dir.mkdir(parents=True) + + pycache = handlers_dir / "__pycache__" + pycache.mkdir() + (pycache / "something.py").write_text("", encoding="utf-8") + + with patch.object(rollover, "__file__", str(fake_module)): + result = rollover._discover_handlers() + + assert "__pycache__" not in result + + def test_excludes_empty_handler_dirs(self, monkeypatch, tmp_path): + """Directories with no .py files (only __init__.py) are excluded.""" + rollover, _ = _import_rollover(monkeypatch) + + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "rollover.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + empty_handler = handlers_dir / "empty_handler" + empty_handler.mkdir(parents=True) + (empty_handler / "__init__.py").write_text("", encoding="utf-8") + + with patch.object(rollover, "__file__", str(fake_module)): + result = rollover._discover_handlers() + + assert "empty_handler" not in result + + def test_returns_sorted_keys_and_values(self, monkeypatch, tmp_path): + """Handler dirs and their files are sorted alphabetically.""" + rollover, _ = _import_rollover(monkeypatch) + + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "rollover.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + handlers_dir.mkdir(parents=True) + + # Create dirs in non-alphabetical order + for name in ["zebra", "alpha"]: + d = handlers_dir / name + d.mkdir() + (d / "b_file.py").write_text("", encoding="utf-8") + (d / "a_file.py").write_text("", encoding="utf-8") + + with patch.object(rollover, "__file__", str(fake_module)): + result = rollover._discover_handlers() + + keys = list(result.keys()) + assert keys == sorted(keys), "Handler directory keys should be sorted" + + for dir_name, files in result.items(): + assert files == sorted(files), f"Files in {dir_name} should be sorted" + + def test_ignores_non_py_files(self, monkeypatch, tmp_path): + """Non-.py files in handler directories are excluded.""" + rollover, _ = _import_rollover(monkeypatch) + + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "rollover.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + mixed_dir = handlers_dir / "mixed" + mixed_dir.mkdir(parents=True) + (mixed_dir / "handler.py").write_text("", encoding="utf-8") + (mixed_dir / "README.md").write_text("", encoding="utf-8") + (mixed_dir / "config.json").write_text("", encoding="utf-8") + + with patch.object(rollover, "__file__", str(fake_module)): + result = rollover._discover_handlers() + + assert result["mixed"] == ["handler.py"] diff --git a/src/aipass/memory/tests/test_search.py b/src/aipass/memory/tests/test_search.py new file mode 100644 index 00000000..1212aba3 --- /dev/null +++ b/src/aipass/memory/tests/test_search.py @@ -0,0 +1,503 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_search.py +# Date: 2026-03-24 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for the search orchestration module. + +Covers: from aipass.memory.apps.modules.search import handle_command + +Tests command routing, handler discovery, argument parsing, and help flags. +All tests use mocks or tmp_path -- no live filesystem or infrastructure access. +""" + +import sys +from pathlib import Path +from unittest.mock import MagicMock + + +# --------------------------------------------------------------------------- +# Helpers: build the full mock graph that search.py needs at import time +# --------------------------------------------------------------------------- + +def _prepare_search_mocks(monkeypatch): + """Insert mocks for every module-level import search.py touches. + + Returns a dict of key mock objects so tests can assert against them. + """ + # rich + mock_panel = MagicMock() + mock_box = MagicMock() + rich_panel_mod = MagicMock() + rich_panel_mod.Panel = mock_panel + rich_box_mod = MagicMock() + rich_box_mod.box = mock_box + monkeypatch.setitem(sys.modules, "rich.panel", rich_panel_mod) + monkeypatch.setitem(sys.modules, "rich", MagicMock()) + + # aipass.cli console / error / warning + mock_console = MagicMock() + mock_error = MagicMock() + mock_warning = MagicMock() + cli_modules_mod = MagicMock() + cli_modules_mod.console = mock_console + cli_modules_mod.error = mock_error + cli_modules_mod.warning = mock_warning + monkeypatch.setitem(sys.modules, "aipass.cli", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules_mod) + + # aipass.memory.apps.handlers.search.query_executor + mock_execute_search = MagicMock(return_value={ + "success": True, + "collections_searched": 2, + "total_results": 1, + "results": [ + { + "collection": "seed_observations", + "document": "Test document content", + "metadata": {"timestamp": "2026-01-01", "source": "local.json"}, + "similarity": 0.85, + } + ], + }) + mock_query_executor = MagicMock() + mock_query_executor.execute_search = mock_execute_search + + search_handlers_pkg = MagicMock() + search_handlers_pkg.query_executor = mock_query_executor + + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.search", + search_handlers_pkg, + ) + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.search.query_executor", + mock_query_executor, + ) + + return { + "console": mock_console, + "error": mock_error, + "warning": mock_warning, + "execute_search": mock_execute_search, + } + + +def _import_search(monkeypatch): + """Prepare mocks and import (or reimport) the search module. + + Returns (search_module, mocks_dict). + """ + mocks = _prepare_search_mocks(monkeypatch) + + # Remove cached module so it gets re-imported with our mocks + sys.modules.pop("aipass.memory.apps.modules.search", None) + + # Also clear the parent package's cached attribute so Python + # re-executes the module code with fresh mocks. + parent = sys.modules.get("aipass.memory.apps.modules") + if parent is not None and hasattr(parent, "search"): + delattr(parent, "search") + + from aipass.memory.apps.modules import search + + return search, mocks + + +# --------------------------------------------------------------------------- +# handle_command: routing +# --------------------------------------------------------------------------- + +class TestHandleCommandRouting: + """Verify that handle_command routes known commands and rejects unknown.""" + + def test_search_no_args_calls_introspection(self, monkeypatch): + """'search' with no args should call print_introspection and return True.""" + search_mod, mocks = _import_search(monkeypatch) + mock_introspect = MagicMock() + monkeypatch.setattr(search_mod, "print_introspection", mock_introspect) + + result = search_mod.handle_command("search", []) + + assert result is True + mock_introspect.assert_called_once() + + def test_search_with_query_calls_show_results(self, monkeypatch): + """'search' with query text should delegate to show_search_results.""" + search_mod, mocks = _import_search(monkeypatch) + mock_show = MagicMock(return_value=True) + monkeypatch.setattr(search_mod, "show_search_results", mock_show) + + result = search_mod.handle_command("search", ["hello", "world"]) + + assert result is True + mock_show.assert_called_once_with( + "hello world", branch=None, memory_type=None, n_results=5 + ) + + def test_unknown_command_returns_false(self, monkeypatch): + """An unrecognised command should return False.""" + search_mod, _mocks = _import_search(monkeypatch) + + result = search_mod.handle_command("nonexistent", ["foo"]) + + assert result is False + + def test_search_routes_correctly(self, monkeypatch): + """'search' command should return True (handled), not False.""" + search_mod, _mocks = _import_search(monkeypatch) + + result = search_mod.handle_command("search", ["some query"]) + + assert result is True + + +# --------------------------------------------------------------------------- +# handle_command: help flags +# --------------------------------------------------------------------------- + +class TestHandleCommandHelp: + """Verify help flags route to print_help.""" + + def test_search_help_flag(self, monkeypatch): + """'search --help' should call print_help and return True.""" + search_mod, mocks = _import_search(monkeypatch) + mock_help = MagicMock() + monkeypatch.setattr(search_mod, "print_help", mock_help) + + result = search_mod.handle_command("search", ["--help"]) + + assert result is True + mock_help.assert_called_once() + + def test_search_h_flag(self, monkeypatch): + """'search -h' should call print_help and return True.""" + search_mod, mocks = _import_search(monkeypatch) + mock_help = MagicMock() + monkeypatch.setattr(search_mod, "print_help", mock_help) + + result = search_mod.handle_command("search", ["-h"]) + + assert result is True + mock_help.assert_called_once() + + def test_search_help_word(self, monkeypatch): + """'search help' should call print_help and return True.""" + search_mod, mocks = _import_search(monkeypatch) + mock_help = MagicMock() + monkeypatch.setattr(search_mod, "print_help", mock_help) + + result = search_mod.handle_command("search", ["help"]) + + assert result is True + mock_help.assert_called_once() + + def test_toplevel_help_flag(self, monkeypatch): + """Top-level '--help' command should call print_help and return True.""" + search_mod, mocks = _import_search(monkeypatch) + mock_help = MagicMock() + monkeypatch.setattr(search_mod, "print_help", mock_help) + + result = search_mod.handle_command("--help", []) + + assert result is True + mock_help.assert_called_once() + + def test_toplevel_h_flag(self, monkeypatch): + """Top-level '-h' command should call print_help and return True.""" + search_mod, mocks = _import_search(monkeypatch) + mock_help = MagicMock() + monkeypatch.setattr(search_mod, "print_help", mock_help) + + result = search_mod.handle_command("-h", []) + + assert result is True + mock_help.assert_called_once() + + def test_toplevel_help_word(self, monkeypatch): + """Top-level 'help' command should call print_help and return True.""" + search_mod, mocks = _import_search(monkeypatch) + mock_help = MagicMock() + monkeypatch.setattr(search_mod, "print_help", mock_help) + + result = search_mod.handle_command("help", []) + + assert result is True + mock_help.assert_called_once() + + +# --------------------------------------------------------------------------- +# handle_command: argument parsing +# --------------------------------------------------------------------------- + +class TestHandleCommandArgParsing: + """Verify argument parsing: --branch, --type, --n, and edge cases.""" + + def test_branch_option_parsed(self, monkeypatch): + """--branch value should be forwarded to show_search_results.""" + search_mod, mocks = _import_search(monkeypatch) + mock_show = MagicMock(return_value=True) + monkeypatch.setattr(search_mod, "show_search_results", mock_show) + + search_mod.handle_command("search", ["my", "query", "--branch", "SEED"]) + + mock_show.assert_called_once_with( + "my query", branch="SEED", memory_type=None, n_results=5 + ) + + def test_type_option_parsed(self, monkeypatch): + """--type value should be forwarded to show_search_results.""" + search_mod, mocks = _import_search(monkeypatch) + mock_show = MagicMock(return_value=True) + monkeypatch.setattr(search_mod, "show_search_results", mock_show) + + search_mod.handle_command("search", ["test", "--type", "observations"]) + + mock_show.assert_called_once_with( + "test", branch=None, memory_type="observations", n_results=5 + ) + + def test_n_option_parsed(self, monkeypatch): + """--n value should override the default n_results.""" + search_mod, mocks = _import_search(monkeypatch) + mock_show = MagicMock(return_value=True) + monkeypatch.setattr(search_mod, "show_search_results", mock_show) + + search_mod.handle_command("search", ["test", "--n", "10"]) + + mock_show.assert_called_once_with( + "test", branch=None, memory_type=None, n_results=10 + ) + + def test_all_options_combined(self, monkeypatch): + """All options together should be correctly parsed.""" + search_mod, mocks = _import_search(monkeypatch) + mock_show = MagicMock(return_value=True) + monkeypatch.setattr(search_mod, "show_search_results", mock_show) + + search_mod.handle_command( + "search", + ["find", "stuff", "--branch", "CLI", "--type", "local", "--n", "3"], + ) + + mock_show.assert_called_once_with( + "find stuff", branch="CLI", memory_type="local", n_results=3 + ) + + def test_invalid_n_shows_error(self, monkeypatch): + """Non-numeric --n value should call error() and return True.""" + search_mod, mocks = _import_search(monkeypatch) + + result = search_mod.handle_command("search", ["test", "--n", "abc"]) + + assert result is True + mocks["error"].assert_called_once() + + def test_empty_query_after_options_shows_error(self, monkeypatch): + """Options only, no query text, should call error() and return True.""" + search_mod, mocks = _import_search(monkeypatch) + + result = search_mod.handle_command("search", ["--branch", "SEED"]) + + assert result is True + mocks["error"].assert_called_once_with("Search query required") + + def test_empty_args_calls_introspection(self, monkeypatch): + """Empty args list should trigger introspection, not crash.""" + search_mod, mocks = _import_search(monkeypatch) + mock_introspect = MagicMock() + monkeypatch.setattr(search_mod, "print_introspection", mock_introspect) + + result = search_mod.handle_command("search", []) + + assert result is True + mock_introspect.assert_called_once() + + +# --------------------------------------------------------------------------- +# _discover_handlers: handler directory scanning +# --------------------------------------------------------------------------- + +class TestDiscoverHandlers: + """Verify handler auto-discovery logic with synthetic directory trees.""" + + def test_discovers_handler_dirs_with_py_files(self, monkeypatch, tmp_path): + """Should return a dict mapping dir name to list of .py filenames.""" + search_mod, _mocks = _import_search(monkeypatch) + + # Build a synthetic handlers/ directory tree + handlers_dir = tmp_path / "handlers" + search_dir = handlers_dir / "search" + search_dir.mkdir(parents=True) + (search_dir / "__init__.py").write_text("", encoding="utf-8") + (search_dir / "query_executor.py").write_text("", encoding="utf-8") + (search_dir / "vector_search.py").write_text("", encoding="utf-8") + + intake_dir = handlers_dir / "intake" + intake_dir.mkdir(parents=True) + (intake_dir / "plans_processor.py").write_text("", encoding="utf-8") + + # Patch __file__ so _discover_handlers resolves to our tmp_path + # _discover_handlers computes: Path(__file__).resolve().parent.parent / "handlers" + # So __file__ needs to be at tmp_path / modules / search.py + modules_dir = tmp_path / "modules" + modules_dir.mkdir(parents=True) + fake_file = modules_dir / "search.py" + fake_file.write_text("", encoding="utf-8") + + monkeypatch.setattr(search_mod, "__file__", str(fake_file)) + + result = search_mod._discover_handlers() + + assert "search" in result + assert "query_executor.py" in result["search"] + assert "vector_search.py" in result["search"] + assert "__init__.py" not in result["search"] + + assert "intake" in result + assert "plans_processor.py" in result["intake"] + + def test_skips_dunder_directories(self, monkeypatch, tmp_path): + """Directories starting with __ (like __pycache__) should be skipped.""" + search_mod, _mocks = _import_search(monkeypatch) + + handlers_dir = tmp_path / "handlers" + pycache_dir = handlers_dir / "__pycache__" + pycache_dir.mkdir(parents=True) + (pycache_dir / "something.py").write_text("", encoding="utf-8") + + valid_dir = handlers_dir / "valid" + valid_dir.mkdir(parents=True) + (valid_dir / "handler.py").write_text("", encoding="utf-8") + + modules_dir = tmp_path / "modules" + modules_dir.mkdir(parents=True) + fake_file = modules_dir / "search.py" + fake_file.write_text("", encoding="utf-8") + + monkeypatch.setattr(search_mod, "__file__", str(fake_file)) + + result = search_mod._discover_handlers() + + assert "__pycache__" not in result + assert "valid" in result + + def test_empty_dir_excluded(self, monkeypatch, tmp_path): + """A handler directory with no .py files (only __init__.py) should not appear.""" + search_mod, _mocks = _import_search(monkeypatch) + + handlers_dir = tmp_path / "handlers" + empty_dir = handlers_dir / "empty_handler" + empty_dir.mkdir(parents=True) + (empty_dir / "__init__.py").write_text("", encoding="utf-8") + + modules_dir = tmp_path / "modules" + modules_dir.mkdir(parents=True) + fake_file = modules_dir / "search.py" + fake_file.write_text("", encoding="utf-8") + + monkeypatch.setattr(search_mod, "__file__", str(fake_file)) + + result = search_mod._discover_handlers() + + assert "empty_handler" not in result + + def test_missing_handlers_dir_returns_empty(self, monkeypatch, tmp_path): + """If handlers/ directory does not exist, return empty dict.""" + search_mod, _mocks = _import_search(monkeypatch) + + # Point __file__ at a location with no handlers/ sibling + modules_dir = tmp_path / "modules" + modules_dir.mkdir(parents=True) + fake_file = modules_dir / "search.py" + fake_file.write_text("", encoding="utf-8") + + monkeypatch.setattr(search_mod, "__file__", str(fake_file)) + + result = search_mod._discover_handlers() + + assert result == {} + + def test_files_at_handlers_root_ignored(self, monkeypatch, tmp_path): + """Loose .py files directly in handlers/ (not in subdirs) should be ignored.""" + search_mod, _mocks = _import_search(monkeypatch) + + handlers_dir = tmp_path / "handlers" + handlers_dir.mkdir(parents=True) + (handlers_dir / "stray_file.py").write_text("", encoding="utf-8") + + modules_dir = tmp_path / "modules" + modules_dir.mkdir(parents=True) + fake_file = modules_dir / "search.py" + fake_file.write_text("", encoding="utf-8") + + monkeypatch.setattr(search_mod, "__file__", str(fake_file)) + + result = search_mod._discover_handlers() + + assert result == {} + + +# --------------------------------------------------------------------------- +# show_search_results: display path +# --------------------------------------------------------------------------- + +class TestShowSearchResults: + """Verify show_search_results calls the handler and renders output.""" + + def test_successful_search_returns_true(self, monkeypatch): + """A successful search should return True.""" + search_mod, mocks = _import_search(monkeypatch) + + result = search_mod.show_search_results("test query") + + assert result is True + mocks["execute_search"].assert_called_once_with( + query="test query", branch=None, memory_type=None, n_results=5 + ) + + def test_failed_search_returns_false(self, monkeypatch): + """If the handler returns success=False, show_search_results returns False.""" + search_mod, mocks = _import_search(monkeypatch) + mocks["execute_search"].return_value = { + "success": False, + "error": "Connection failed", + } + + result = search_mod.show_search_results("broken query") + + assert result is False + mocks["error"].assert_called_once() + + def test_no_results_shows_warning(self, monkeypatch): + """Zero results should trigger a warning, not an error.""" + search_mod, mocks = _import_search(monkeypatch) + mocks["execute_search"].return_value = { + "success": True, + "collections_searched": 3, + "total_results": 0, + "results": [], + } + + result = search_mod.show_search_results("obscure query") + + assert result is True + mocks["warning"].assert_called_once() + + def test_options_forwarded_to_handler(self, monkeypatch): + """Branch, memory_type, and n_results should be forwarded.""" + search_mod, mocks = _import_search(monkeypatch) + + search_mod.show_search_results( + "q", branch="SEED", memory_type="local", n_results=3 + ) + + mocks["execute_search"].assert_called_once_with( + query="q", branch="SEED", memory_type="local", n_results=3 + ) diff --git a/src/aipass/memory/tests/test_symbolic.py b/src/aipass/memory/tests/test_symbolic.py new file mode 100644 index 00000000..8a776240 --- /dev/null +++ b/src/aipass/memory/tests/test_symbolic.py @@ -0,0 +1,408 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_symbolic.py +# Date: 2026-03-24 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for the symbolic memory orchestration module (apps/modules/symbolic.py). + +Covers: from aipass.memory.apps.modules.symbolic import extract_technical_flow + +The module under test is a thin delegation layer: each public function forwards +to an identically-named function on one of the handler sub-modules (extractor, +storage, retriever). We mock those handler modules via ``sys.modules`` so +the tests stay lightweight and never touch real ChromaDB / filesystem. +""" + +import sys +import types +import pytest +from unittest.mock import MagicMock + + +# --------------------------------------------------------------------------- +# Module-level mock namespace -- tests read handler mocks from here +# --------------------------------------------------------------------------- + +_handler_mocks = types.SimpleNamespace( + extractor=MagicMock(), + storage=MagicMock(), + retriever=MagicMock(), + trigger=MagicMock(), +) + + +# --------------------------------------------------------------------------- +# Autouse fixture -- mock all heavy imports before symbolic.py is loaded +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def _mock_symbolic_infrastructure(monkeypatch): + """Replace handler modules with MagicMock before importing symbolic.py.""" + + # -- prax logger -------------------------------------------------------- + mock_prax = MagicMock() + mock_prax.logger = MagicMock() + monkeypatch.setitem(sys.modules, "aipass.prax", mock_prax) + monkeypatch.setitem(sys.modules, "aipass.prax.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules.logger", MagicMock()) + + # -- cli display helpers ------------------------------------------------ + cli_modules = MagicMock() + monkeypatch.setitem(sys.modules, "aipass.cli", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules) + + # -- memory json handler ------------------------------------------------ + mock_json_handler = MagicMock() + mock_json_handler.log_operation = MagicMock(return_value=True) + json_pkg = MagicMock() + json_pkg.json_handler = mock_json_handler + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json", json_pkg) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.json.json_handler", mock_json_handler + ) + + # -- symbolic handler sub-modules (the delegation targets) -------------- + mock_extractor = MagicMock() + mock_storage = MagicMock() + mock_retriever = MagicMock() + mock_hook = MagicMock() + mock_deduplicator = MagicMock() + + symbolic_pkg = MagicMock() + symbolic_pkg.extractor = mock_extractor + symbolic_pkg.storage = mock_storage + symbolic_pkg.retriever = mock_retriever + symbolic_pkg.hook = mock_hook + symbolic_pkg.deduplicator = mock_deduplicator + + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.symbolic", symbolic_pkg) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.symbolic.extractor", mock_extractor + ) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.symbolic.storage", mock_storage + ) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.symbolic.retriever", mock_retriever + ) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.symbolic.hook", mock_hook + ) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.symbolic.deduplicator", mock_deduplicator + ) + + # -- vector embedder (imported by storage handler) ---------------------- + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector", MagicMock()) + monkeypatch.setitem( + sys.modules, "aipass.memory.apps.handlers.vector.embedder", MagicMock() + ) + + # -- trigger (lazy import inside create_fragment) ----------------------- + mock_trigger_core = MagicMock() + mock_trigger = MagicMock() + mock_trigger_core.trigger = mock_trigger + monkeypatch.setitem(sys.modules, "aipass.trigger", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.trigger.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules.core", mock_trigger_core) + + # Force fresh import every test + monkeypatch.delitem(sys.modules, "aipass.memory.apps.modules.symbolic", raising=False) + + # Expose mocks on the module-level namespace for test-level assertions + _handler_mocks.extractor = mock_extractor + _handler_mocks.storage = mock_storage + _handler_mocks.retriever = mock_retriever + _handler_mocks.trigger = mock_trigger + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _sample_chat() -> list: + """Return a minimal chat history list.""" + return [ + {"role": "user", "content": "I found a bug in the parser"}, + {"role": "assistant", "content": "Let me debug that for you"}, + ] + + +def _import_symbolic(): + """Import symbolic module after mocks are in place. + + Must also clear the parent package's cached attribute so Python + re-executes the module code with fresh mocks. + """ + import importlib + + # Remove from sys.modules if still present + sys.modules.pop("aipass.memory.apps.modules.symbolic", None) + + # Clear the parent package's cached attribute so `from ... import symbolic` + # triggers a fresh import rather than returning the stale attribute. + parent = sys.modules.get("aipass.memory.apps.modules") + if parent is not None and hasattr(parent, "symbolic"): + delattr(parent, "symbolic") + + from aipass.memory.apps.modules import symbolic + return symbolic + + +# =========================================================================== +# EXTRACTION DELEGATION TESTS +# =========================================================================== + +class TestExtractTechnicalFlow: + """extract_technical_flow delegates to extractor handler.""" + + def test_delegates_to_extractor(self): + symbolic = _import_symbolic() + expected = {"success": True, "patterns": ["debugging_session"]} + _handler_mocks.extractor.extract_technical_flow.return_value = expected + + result = symbolic.extract_technical_flow(_sample_chat()) + + _handler_mocks.extractor.extract_technical_flow.assert_called_once_with( + _sample_chat() + ) + assert result == expected + + def test_returns_handler_result_unchanged(self): + symbolic = _import_symbolic() + handler_result = {"success": False, "error": "parse failure"} + _handler_mocks.extractor.extract_technical_flow.return_value = handler_result + + assert symbolic.extract_technical_flow([]) == handler_result + + +class TestExtractEmotionalJourney: + """extract_emotional_journey delegates to extractor handler.""" + + def test_delegates_to_extractor(self): + symbolic = _import_symbolic() + expected = {"success": True, "arc": ["curious", "frustrated", "relieved"]} + _handler_mocks.extractor.extract_emotional_journey.return_value = expected + + result = symbolic.extract_emotional_journey(_sample_chat()) + + _handler_mocks.extractor.extract_emotional_journey.assert_called_once_with( + _sample_chat() + ) + assert result == expected + + +class TestExtractCollaborationPatterns: + """extract_collaboration_patterns delegates to extractor handler.""" + + def test_delegates_to_extractor(self): + symbolic = _import_symbolic() + expected = {"success": True, "patterns": ["pair_debugging"]} + _handler_mocks.extractor.extract_collaboration_patterns.return_value = expected + + result = symbolic.extract_collaboration_patterns(_sample_chat()) + + _handler_mocks.extractor.extract_collaboration_patterns.assert_called_once_with( + _sample_chat() + ) + assert result == expected + + +class TestExtractKeyLearnings: + """extract_key_learnings delegates to extractor handler.""" + + def test_delegates_to_extractor(self): + symbolic = _import_symbolic() + expected = {"success": True, "insights": ["parser edge case"]} + _handler_mocks.extractor.extract_key_learnings.return_value = expected + + result = symbolic.extract_key_learnings(_sample_chat()) + + _handler_mocks.extractor.extract_key_learnings.assert_called_once_with( + _sample_chat() + ) + assert result == expected + + +class TestExtractContextTriggers: + """extract_context_triggers delegates to extractor handler.""" + + def test_delegates_to_extractor(self): + symbolic = _import_symbolic() + expected = {"success": True, "triggers": ["parser", "bug"]} + _handler_mocks.extractor.extract_context_triggers.return_value = expected + + result = symbolic.extract_context_triggers(_sample_chat()) + + _handler_mocks.extractor.extract_context_triggers.assert_called_once_with( + _sample_chat() + ) + assert result == expected + + +class TestExtractSymbolicDimensions: + """extract_symbolic_dimensions delegates to extractor handler.""" + + def test_delegates_to_extractor(self): + symbolic = _import_symbolic() + expected = { + "success": True, + "technical_flow": {"patterns": []}, + "emotional_journey": {"arc": []}, + } + _handler_mocks.extractor.extract_symbolic_dimensions.return_value = expected + + result = symbolic.extract_symbolic_dimensions(_sample_chat()) + + _handler_mocks.extractor.extract_symbolic_dimensions.assert_called_once_with( + _sample_chat() + ) + assert result == expected + + +# =========================================================================== +# STORAGE DELEGATION TESTS +# =========================================================================== + +class TestCreateFragment: + """create_fragment delegates to storage handler and fires trigger.""" + + def test_delegates_to_storage(self): + symbolic = _import_symbolic() + analysis = {"dimensions": {"technical": "debug"}} + fragment_result = { + "success": True, + "fragment": {"id": "frag-001", "content": "test"}, + } + _handler_mocks.storage.create_fragment.return_value = fragment_result + + result = symbolic.create_fragment(analysis, content="hello", source_branch="memory") + + _handler_mocks.storage.create_fragment.assert_called_once_with( + analysis, "hello", "memory" + ) + assert result == fragment_result + + def test_fires_trigger_on_success(self): + symbolic = _import_symbolic() + _handler_mocks.storage.create_fragment.return_value = { + "success": True, + "fragment": {"id": "frag-002"}, + } + _handler_mocks.trigger.reset_mock() + + symbolic.create_fragment({}, content="x", source_branch="drone") + + _handler_mocks.trigger.fire.assert_called_once_with( + "fragment_created", fragment_id="frag-002", source_branch="drone" + ) + + def test_no_trigger_on_failure(self): + symbolic = _import_symbolic() + _handler_mocks.storage.create_fragment.return_value = { + "success": False, + "error": "bad input", + } + _handler_mocks.trigger.reset_mock() + + symbolic.create_fragment({}) + + _handler_mocks.trigger.fire.assert_not_called() + + def test_trigger_exception_caught_gracefully(self): + """When trigger.fire raises, create_fragment still returns the result.""" + symbolic = _import_symbolic() + _handler_mocks.storage.create_fragment.return_value = { + "success": True, + "fragment": {"id": "frag-003"}, + } + _handler_mocks.trigger.fire.side_effect = RuntimeError("trigger boom") + + result = symbolic.create_fragment({}, source_branch="test") + + assert result["success"] is True + assert result["fragment"]["id"] == "frag-003" + + def test_default_source_branch_is_unknown(self): + """When source_branch is None, trigger receives 'unknown'.""" + symbolic = _import_symbolic() + _handler_mocks.storage.create_fragment.return_value = { + "success": True, + "fragment": {"id": "frag-004"}, + } + _handler_mocks.trigger.reset_mock() + + symbolic.create_fragment({}) + + _handler_mocks.trigger.fire.assert_called_once_with( + "fragment_created", fragment_id="frag-004", source_branch="unknown" + ) + + +# =========================================================================== +# EMPTY INPUT / EDGE CASE TESTS +# =========================================================================== + +class TestEmptyChatHistory: + """Verify wrapper functions forward empty lists without crashing.""" + + def test_extract_technical_flow_empty(self): + symbolic = _import_symbolic() + expected = {"success": True, "patterns": ["no_conversation"]} + _handler_mocks.extractor.extract_technical_flow.return_value = expected + + result = symbolic.extract_technical_flow([]) + + _handler_mocks.extractor.extract_technical_flow.assert_called_once_with([]) + assert result == expected + + def test_extract_emotional_journey_empty(self): + symbolic = _import_symbolic() + expected = {"success": True, "arc": []} + _handler_mocks.extractor.extract_emotional_journey.return_value = expected + + result = symbolic.extract_emotional_journey([]) + + assert result == expected + + def test_extract_collaboration_patterns_empty(self): + symbolic = _import_symbolic() + expected = {"success": True, "patterns": []} + _handler_mocks.extractor.extract_collaboration_patterns.return_value = expected + + result = symbolic.extract_collaboration_patterns([]) + + assert result == expected + + def test_extract_key_learnings_empty(self): + symbolic = _import_symbolic() + expected = {"success": True, "insights": []} + _handler_mocks.extractor.extract_key_learnings.return_value = expected + + result = symbolic.extract_key_learnings([]) + + assert result == expected + + def test_extract_context_triggers_empty(self): + symbolic = _import_symbolic() + expected = {"success": True, "triggers": []} + _handler_mocks.extractor.extract_context_triggers.return_value = expected + + result = symbolic.extract_context_triggers([]) + + assert result == expected + + def test_extract_symbolic_dimensions_empty(self): + symbolic = _import_symbolic() + expected = {"success": True} + _handler_mocks.extractor.extract_symbolic_dimensions.return_value = expected + + result = symbolic.extract_symbolic_dimensions([]) + + assert result == expected diff --git a/src/aipass/memory/tests/test_templates.py b/src/aipass/memory/tests/test_templates.py new file mode 100644 index 00000000..f56309dc --- /dev/null +++ b/src/aipass/memory/tests/test_templates.py @@ -0,0 +1,702 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_templates.py +# Date: 2026-03-24 +# Version: 2.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for templates module -- repo root discovery, handler discovery, command routing. + +Covers: from aipass.memory.apps.modules.templates import handle_command + +All handler imports are mocked in sys.modules before the templates module is +imported so that no live infrastructure or handler code is needed. +""" + +import sys +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Helpers: build the full mock graph that templates.py needs at import time +# --------------------------------------------------------------------------- + +def _prepare_templates_mocks(monkeypatch): + """Insert mocks for every module-level import templates.py touches. + + Returns a dict of key mock objects so tests can assert against them. + """ + # -- rich -- + mock_panel = MagicMock() + mock_box = MagicMock() + rich_panel_mod = MagicMock() + rich_panel_mod.Panel = mock_panel + rich_box_mod = MagicMock() + rich_box_mod.box = mock_box + monkeypatch.setitem(sys.modules, "rich.panel", rich_panel_mod) + monkeypatch.setitem(sys.modules, "rich", MagicMock()) + + # -- aipass.cli console / error / warning -- + mock_console = MagicMock() + mock_error = MagicMock() + mock_warning = MagicMock() + cli_modules_mod = MagicMock() + cli_modules_mod.console = mock_console + cli_modules_mod.error = mock_error + cli_modules_mod.warning = mock_warning + monkeypatch.setitem(sys.modules, "aipass.cli", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules_mod) + + # -- aipass.memory.apps.handlers.json.memory_files -- + mock_memory_files = MagicMock() + mock_memory_files.read_memory_file_data = MagicMock(return_value=None) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.memory_files", mock_memory_files) + + # -- aipass.memory.apps.handlers.templates.pusher -- + mock_pusher = MagicMock() + mock_pusher.push_templates = MagicMock(return_value={ + "success": True, "branches_scanned": 5, "branches_updated": 2, + "files_modified": 3, "changes": [], "errors": [], + }) + mock_pusher.get_template_status = MagicMock(return_value={ + "version": "2.0.0", "last_push": "2026-03-20", + "local_template_exists": True, "observations_template_exists": True, + "templates_dir": "/tmp/templates", "last_push_branches": [], + }) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.templates.pusher", mock_pusher) + + # -- aipass.memory.apps.handlers.templates.differ -- + mock_differ = MagicMock() + mock_differ.diff_template_vs_branch = MagicMock(return_value={ + "local": [], "observations": [], "errors": [], + }) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.templates.differ", mock_differ) + + # -- aipass.memory.apps.handlers.templates.spawn_pusher -- + mock_spawn_pusher = MagicMock() + mock_spawn_pusher.push_to_spawn_templates = MagicMock(return_value={ + "success": True, "template_sets_found": [], "template_sets_updated": 0, + "files_modified": 0, "changes": [], + }) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.templates.spawn_pusher", mock_spawn_pusher) + + # -- parent packages that Python needs to resolve dotted imports -- + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.templates", MagicMock( + pusher=mock_pusher, + differ=mock_differ, + spawn_pusher=mock_spawn_pusher, + )) + + return { + "console": mock_console, + "error": mock_error, + "warning": mock_warning, + "pusher": mock_pusher, + "differ": mock_differ, + "spawn_pusher": mock_spawn_pusher, + "memory_files": mock_memory_files, + } + + +def _import_templates(monkeypatch): + """Prepare mocks and import (or reimport) the templates module. + + Returns (templates_module, mocks_dict). + """ + mocks = _prepare_templates_mocks(monkeypatch) + + # Remove cached module so it re-imports with our mocks + sys.modules.pop("aipass.memory.apps.modules.templates", None) + + # Also clear the parent package's cached attribute so Python + # re-executes the module code with fresh mocks. + parent = sys.modules.get("aipass.memory.apps.modules") + if parent is not None and hasattr(parent, "templates"): + delattr(parent, "templates") + + from aipass.memory.apps.modules import templates + + return templates, mocks + + +# =========================================================================== +# Tests: _find_repo_root +# =========================================================================== + +class TestFindRepoRoot: + """Tests for _find_repo_root -- walks up from __file__ to find AIPASS_REGISTRY.json.""" + + def test_finds_root_when_registry_exists(self, tmp_path: Path, monkeypatch) -> None: + """Returns the directory containing AIPASS_REGISTRY.json.""" + templates, _ = _import_templates(monkeypatch) + + # Place a registry at the fake repo root + (tmp_path / "AIPASS_REGISTRY.json").write_text("{}", encoding="utf-8") + + # Create nested path mimicking real module location + nested = tmp_path / "src" / "aipass" / "memory" / "apps" / "modules" + nested.mkdir(parents=True) + fake_module = nested / "templates.py" + fake_module.write_text("", encoding="utf-8") + + # Replicate the walk-up logic with our fake starting point + current = fake_module.resolve().parent + found: Path | None = None + for parent in [current, *list(current.parents)]: + if (parent / "AIPASS_REGISTRY.json").exists(): + found = parent + break + + assert found == tmp_path + + def test_returns_cwd_when_no_registry(self, tmp_path: Path, monkeypatch) -> None: + """Falls back to Path.cwd() when no AIPASS_REGISTRY.json is found.""" + templates, _ = _import_templates(monkeypatch) + + # Create a nested path with no registry anywhere above + nested = tmp_path / "a" / "b" / "c" / "d" + nested.mkdir(parents=True) + + current = nested + found: Path | None = None + for parent in [current, *list(current.parents)]: + if (parent / "AIPASS_REGISTRY.json").exists(): + found = parent + break + + # No registry found -- the real function falls back to Path.cwd() + assert found is None + + def test_registry_at_immediate_parent(self, tmp_path: Path, monkeypatch) -> None: + """Finds registry when it is in the immediate parent directory.""" + templates, _ = _import_templates(monkeypatch) + + parent_dir = tmp_path / "repo" + parent_dir.mkdir() + (parent_dir / "AIPASS_REGISTRY.json").write_text("{}", encoding="utf-8") + + child_dir = parent_dir / "child" + child_dir.mkdir() + + current = child_dir + found: Path | None = None + for parent in [current, *list(current.parents)]: + if (parent / "AIPASS_REGISTRY.json").exists(): + found = parent + break + + assert found == parent_dir + + +# =========================================================================== +# Tests: _discover_handlers +# =========================================================================== + +class TestDiscoverHandlers: + """Tests for _discover_handlers -- auto-discovers handler directories.""" + + def test_discovers_handler_directories(self, tmp_path: Path, monkeypatch) -> None: + """Finds handler dirs with .py files, excluding __pycache__ and __init__.py.""" + templates, _ = _import_templates(monkeypatch) + + # Build fake handler structure: modules/templates.py -> parent.parent = apps -> handlers + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "templates.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + handlers_dir.mkdir(parents=True) + + # json/ with two handler files + json_dir = handlers_dir / "json" + json_dir.mkdir() + (json_dir / "json_handler.py").write_text("", encoding="utf-8") + (json_dir / "memory_files.py").write_text("", encoding="utf-8") + (json_dir / "__init__.py").write_text("", encoding="utf-8") + + # templates/ with one handler file + tmpl_dir = handlers_dir / "templates" + tmpl_dir.mkdir() + (tmpl_dir / "pusher.py").write_text("", encoding="utf-8") + (tmpl_dir / "__init__.py").write_text("", encoding="utf-8") + + # __pycache__/ should be ignored + cache_dir = handlers_dir / "__pycache__" + cache_dir.mkdir() + (cache_dir / "something.pyc").write_text("", encoding="utf-8") + + with patch.object(templates, "__file__", str(fake_module)): + result = templates._discover_handlers() + + assert "json" in result + assert "templates" in result + assert "__pycache__" not in result + assert "json_handler.py" in result["json"] + assert "memory_files.py" in result["json"] + assert "__init__.py" not in result["json"] + assert "pusher.py" in result["templates"] + + def test_returns_empty_when_no_handlers_dir(self, tmp_path: Path, monkeypatch) -> None: + """Returns empty dict when handlers/ does not exist.""" + templates, _ = _import_templates(monkeypatch) + + # Point __file__ at a location with no handlers/ sibling + modules_dir = tmp_path / "nowhere" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "templates.py" + fake_module.write_text("", encoding="utf-8") + + with patch.object(templates, "__file__", str(fake_module)): + result = templates._discover_handlers() + + assert result == {} + + def test_skips_empty_handler_directories(self, tmp_path: Path, monkeypatch) -> None: + """Directories with no .py files (only __init__.py) are not included.""" + templates, _ = _import_templates(monkeypatch) + + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "templates.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + handlers_dir.mkdir(parents=True) + + # Empty dir (only __init__.py) + empty_dir = handlers_dir / "empty" + empty_dir.mkdir() + (empty_dir / "__init__.py").write_text("", encoding="utf-8") + + # Dir with actual handler + real_dir = handlers_dir / "real" + real_dir.mkdir() + (real_dir / "handler.py").write_text("", encoding="utf-8") + + with patch.object(templates, "__file__", str(fake_module)): + result = templates._discover_handlers() + + assert "real" in result + assert "empty" not in result + + def test_returns_sorted_keys_and_values(self, tmp_path: Path, monkeypatch) -> None: + """Handler dirs and their files are sorted alphabetically.""" + templates, _ = _import_templates(monkeypatch) + + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "templates.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + handlers_dir.mkdir(parents=True) + + for name in ["zebra", "alpha"]: + d = handlers_dir / name + d.mkdir() + (d / "b_file.py").write_text("", encoding="utf-8") + (d / "a_file.py").write_text("", encoding="utf-8") + + with patch.object(templates, "__file__", str(fake_module)): + result = templates._discover_handlers() + + keys = list(result.keys()) + assert keys == sorted(keys), "Handler directory keys should be sorted" + + for dir_name, files in result.items(): + assert files == sorted(files), f"Files in {dir_name} should be sorted" + + def test_ignores_non_py_files(self, tmp_path: Path, monkeypatch) -> None: + """Non-.py files in handler directories are excluded.""" + templates, _ = _import_templates(monkeypatch) + + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_module = modules_dir / "templates.py" + fake_module.write_text("", encoding="utf-8") + + handlers_dir = tmp_path / "apps" / "handlers" + mixed_dir = handlers_dir / "mixed" + mixed_dir.mkdir(parents=True) + (mixed_dir / "handler.py").write_text("", encoding="utf-8") + (mixed_dir / "README.md").write_text("", encoding="utf-8") + (mixed_dir / "config.json").write_text("", encoding="utf-8") + + with patch.object(templates, "__file__", str(fake_module)): + result = templates._discover_handlers() + + assert result["mixed"] == ["handler.py"] + + +# =========================================================================== +# Tests: handle_command routing +# =========================================================================== + +class TestHandleCommand: + """Tests for handle_command -- routes subcommands correctly.""" + + # -- templates with no args -> introspection -- + + def test_templates_no_args_calls_introspection(self, monkeypatch) -> None: + """'templates' with no args triggers introspection display and returns True.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "print_introspection") as mock_intro: + result = templates.handle_command("templates", []) + + mock_intro.assert_called_once() + assert result is True + + # -- help flags -- + + def test_templates_help_flag(self, monkeypatch) -> None: + """'templates --help' triggers help display.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "print_help") as mock_help: + result = templates.handle_command("templates", ["--help"]) + + mock_help.assert_called_once() + assert result is True + + def test_templates_dash_h_flag(self, monkeypatch) -> None: + """'templates -h' triggers help display.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "print_help") as mock_help: + result = templates.handle_command("templates", ["-h"]) + + mock_help.assert_called_once() + assert result is True + + def test_templates_help_word(self, monkeypatch) -> None: + """'templates help' triggers help display.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "print_help") as mock_help: + result = templates.handle_command("templates", ["help"]) + + mock_help.assert_called_once() + assert result is True + + def test_top_level_help_flag(self, monkeypatch) -> None: + """'--help' as top-level command triggers help.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "print_help") as mock_help: + result = templates.handle_command("--help", []) + + mock_help.assert_called_once() + assert result is True + + def test_top_level_dash_h_flag(self, monkeypatch) -> None: + """'-h' as top-level command triggers help.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "print_help") as mock_help: + result = templates.handle_command("-h", []) + + mock_help.assert_called_once() + assert result is True + + def test_top_level_help_word(self, monkeypatch) -> None: + """'help' as top-level command triggers help.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "print_help") as mock_help: + result = templates.handle_command("help", []) + + mock_help.assert_called_once() + assert result is True + + # -- push-templates subcommand -- + + def test_push_templates_subcommand(self, monkeypatch) -> None: + """'templates push-templates' calls push handler with dry_run=False.""" + templates, mocks = _import_templates(monkeypatch) + + with ( + patch.object(templates, "_display_push_results") as mock_display, + patch.object(templates, "_display_spawn_push_results") as mock_spawn_display, + ): + result = templates.handle_command("templates", ["push-templates"]) + + mocks["pusher"].push_templates.assert_called_once_with(dry_run=False) + mocks["spawn_pusher"].push_to_spawn_templates.assert_called_once_with(dry_run=False) + assert result is True + + def test_push_templates_dry_run(self, monkeypatch) -> None: + """'templates push-templates --dry-run' passes dry_run=True.""" + templates, mocks = _import_templates(monkeypatch) + + with ( + patch.object(templates, "_display_push_results"), + patch.object(templates, "_display_spawn_push_results"), + ): + result = templates.handle_command("templates", ["push-templates", "--dry-run"]) + + mocks["pusher"].push_templates.assert_called_once_with(dry_run=True) + mocks["spawn_pusher"].push_to_spawn_templates.assert_called_once_with(dry_run=True) + assert result is True + + # -- diff-templates subcommand -- + + def test_diff_templates_subcommand(self, monkeypatch) -> None: + """'templates diff-templates' calls diff display with no branch filter.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "_display_diff_results") as mock_display: + result = templates.handle_command("templates", ["diff-templates"]) + + mock_display.assert_called_once_with(None) + assert result is True + + def test_diff_templates_with_branch_filter(self, monkeypatch) -> None: + """'templates diff-templates --branch CLI' passes the branch name.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "_display_diff_results") as mock_display: + result = templates.handle_command("templates", ["diff-templates", "--branch", "CLI"]) + + mock_display.assert_called_once_with("CLI") + assert result is True + + # -- template-status subcommand -- + + def test_template_status_subcommand(self, monkeypatch) -> None: + """'templates template-status' calls status handler.""" + templates, mocks = _import_templates(monkeypatch) + + with patch.object(templates, "_display_status") as mock_display: + result = templates.handle_command("templates", ["template-status"]) + + mocks["pusher"].get_template_status.assert_called_once() + assert result is True + + # -- backward-compatible top-level commands -- + + def test_backward_compat_push_templates(self, monkeypatch) -> None: + """'push-templates' as top-level command (backward compat) works.""" + templates, mocks = _import_templates(monkeypatch) + + with ( + patch.object(templates, "_display_push_results"), + patch.object(templates, "_display_spawn_push_results"), + ): + result = templates.handle_command("push-templates", []) + + mocks["pusher"].push_templates.assert_called_once_with(dry_run=False) + assert result is True + + def test_backward_compat_diff_templates(self, monkeypatch) -> None: + """'diff-templates' as top-level command (backward compat) works.""" + templates, _ = _import_templates(monkeypatch) + + with patch.object(templates, "_display_diff_results") as mock_display: + result = templates.handle_command("diff-templates", []) + + mock_display.assert_called_once_with(None) + assert result is True + + def test_backward_compat_template_status(self, monkeypatch) -> None: + """'template-status' as top-level command (backward compat) works.""" + templates, mocks = _import_templates(monkeypatch) + + with patch.object(templates, "_display_status"): + result = templates.handle_command("template-status", []) + + mocks["pusher"].get_template_status.assert_called_once() + assert result is True + + # -- unknown command returns False -- + + def test_unknown_command_returns_false(self, monkeypatch) -> None: + """Unknown top-level command returns False.""" + templates, _ = _import_templates(monkeypatch) + + result = templates.handle_command("totally-unknown", []) + + assert result is False + + def test_empty_string_command_returns_false(self, monkeypatch) -> None: + """Empty string command returns False.""" + templates, _ = _import_templates(monkeypatch) + + result = templates.handle_command("", []) + + assert result is False + + # -- unknown subcommand returns True with error -- + + def test_unknown_subcommand_returns_true_with_error(self, monkeypatch) -> None: + """Unknown subcommand of 'templates' returns True (handled) but shows error.""" + templates, mocks = _import_templates(monkeypatch) + + result = templates.handle_command("templates", ["bogus-sub"]) + + assert result is True + mocks["error"].assert_called() + + # -- push-templates error handling -- + + def test_push_templates_handles_push_exception(self, monkeypatch) -> None: + """Push exception is caught and error is displayed, still returns True.""" + templates, mocks = _import_templates(monkeypatch) + mocks["pusher"].push_templates.side_effect = RuntimeError("disk full") + + with ( + patch.object(templates, "_display_spawn_push_results"), + ): + result = templates.handle_command("templates", ["push-templates"]) + + mocks["error"].assert_called() + assert result is True + + def test_push_templates_handles_spawn_exception(self, monkeypatch) -> None: + """Spawn push exception is caught separately, still returns True.""" + templates, mocks = _import_templates(monkeypatch) + mocks["spawn_pusher"].push_to_spawn_templates.side_effect = RuntimeError("spawn error") + + with patch.object(templates, "_display_push_results"): + result = templates.handle_command("templates", ["push-templates"]) + + mocks["error"].assert_called() + assert result is True + + +# =========================================================================== +# Tests: _load_branches_from_registry +# =========================================================================== + +class TestLoadBranchesFromRegistry: + """Tests for _load_branches_from_registry -- loads active branches from registry.""" + + def test_loads_active_branches(self, tmp_path: Path, monkeypatch) -> None: + """Returns only active branches with resolved paths.""" + templates, mocks = _import_templates(monkeypatch) + + registry_data = { + "branches": [ + {"name": "ALPHA", "path": "src/aipass/alpha", "status": "active"}, + {"name": "BETA", "path": "src/aipass/beta", "status": "inactive"}, + {"name": "GAMMA", "path": "src/aipass/gamma", "status": "active"}, + ] + } + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry_data, indent=2), encoding="utf-8") + + with ( + patch.object(templates, "REGISTRY_PATH", registry_path), + patch.object(templates, "_REPO_ROOT", tmp_path), + patch.object(templates, "read_memory_file_data", return_value=registry_data), + ): + result = templates._load_branches_from_registry() + + assert result is not None + assert len(result) == 2 + names = [b["name"] for b in result] + assert "ALPHA" in names + assert "GAMMA" in names + assert "BETA" not in names + + def test_resolves_relative_paths(self, tmp_path: Path, monkeypatch) -> None: + """Relative paths are resolved against repo root.""" + templates, _ = _import_templates(monkeypatch) + + registry_data = { + "branches": [ + {"name": "CLI", "path": "src/aipass/cli", "status": "active"}, + ] + } + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry_data, indent=2), encoding="utf-8") + + with ( + patch.object(templates, "REGISTRY_PATH", registry_path), + patch.object(templates, "_REPO_ROOT", tmp_path), + patch.object(templates, "read_memory_file_data", return_value=registry_data), + ): + result = templates._load_branches_from_registry() + + assert result is not None + assert len(result) == 1 + assert result[0]["path"] == str(tmp_path / "src" / "aipass" / "cli") + + def test_returns_none_when_registry_missing(self, tmp_path: Path, monkeypatch) -> None: + """Returns None when REGISTRY_PATH does not exist.""" + templates, _ = _import_templates(monkeypatch) + + missing_path = tmp_path / "nonexistent" / "AIPASS_REGISTRY.json" + + with patch.object(templates, "REGISTRY_PATH", missing_path): + result = templates._load_branches_from_registry() + + assert result is None + + def test_returns_none_when_read_fails(self, tmp_path: Path, monkeypatch) -> None: + """Returns None when read_memory_file_data returns None.""" + templates, _ = _import_templates(monkeypatch) + + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text("{}", encoding="utf-8") + + with ( + patch.object(templates, "REGISTRY_PATH", registry_path), + patch.object(templates, "read_memory_file_data", return_value=None), + ): + result = templates._load_branches_from_registry() + + assert result is None + + def test_empty_branches_returns_empty_list(self, tmp_path: Path, monkeypatch) -> None: + """Returns empty list when branches key is empty.""" + templates, _ = _import_templates(monkeypatch) + + registry_data: dict = {"branches": []} + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry_data), encoding="utf-8") + + with ( + patch.object(templates, "REGISTRY_PATH", registry_path), + patch.object(templates, "_REPO_ROOT", tmp_path), + patch.object(templates, "read_memory_file_data", return_value=registry_data), + ): + result = templates._load_branches_from_registry() + + assert result is not None + assert result == [] + + +# =========================================================================== +# Tests: _SUBCOMMANDS dict +# =========================================================================== + +class TestSubcommands: + """Verify the _SUBCOMMANDS dict exists with expected keys.""" + + def test_subcommands_exists(self, monkeypatch) -> None: + templates, _ = _import_templates(monkeypatch) + assert hasattr(templates, "_SUBCOMMANDS") + + def test_subcommands_has_push_templates(self, monkeypatch) -> None: + templates, _ = _import_templates(monkeypatch) + assert "push-templates" in templates._SUBCOMMANDS + + def test_subcommands_has_diff_templates(self, monkeypatch) -> None: + templates, _ = _import_templates(monkeypatch) + assert "diff-templates" in templates._SUBCOMMANDS + + def test_subcommands_has_template_status(self, monkeypatch) -> None: + templates, _ = _import_templates(monkeypatch) + assert "template-status" in templates._SUBCOMMANDS + + def test_subcommands_values_are_strings(self, monkeypatch) -> None: + templates, _ = _import_templates(monkeypatch) + for key, value in templates._SUBCOMMANDS.items(): + assert isinstance(key, str), f"Key {key!r} is not a string" + assert isinstance(value, str), f"Value for {key!r} is not a string" diff --git a/src/aipass/memory/tests/test_verify.py b/src/aipass/memory/tests/test_verify.py new file mode 100644 index 00000000..5c434594 --- /dev/null +++ b/src/aipass/memory/tests/test_verify.py @@ -0,0 +1,563 @@ +# =================== META ==================== +# Name: test_verify.py +# Description: Tests for verify module (plan vectorization check) +# Version: 1.0.0 +# Created: 2026-03-24 +# Modified: 2026-03-24 +# ============================================= + +"""Tests for the verify module: _get_memory_python, _check_plan_subprocess, +is_plan_vectorized, handle_command, _discover_handlers. + +Covers: from aipass.memory.apps.modules.verify import handle_command +""" + +import json +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + + +# --------------------------------------------------------------------------- +# Helper: mock CLI modules that verify.py imports at module level +# --------------------------------------------------------------------------- + +def _mock_cli_modules(monkeypatch): + """Inject mocks for aipass.cli.apps.modules (console, error).""" + mock_console = MagicMock() + mock_error = MagicMock() + cli_modules = MagicMock() + cli_modules.console = mock_console + cli_modules.error = mock_error + monkeypatch.setitem(sys.modules, "aipass.cli", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules) + return mock_console, mock_error + + +def _import_verify(): + """Import verify module inside the test (after mocks are in place).""" + from aipass.memory.apps.modules import verify + return verify + + +# =========================================================================== +# 1. _get_memory_python() +# =========================================================================== + + +class TestGetMemoryPython: + """Tests for _get_memory_python() resolution logic.""" + + def test_env_override_takes_precedence(self, monkeypatch): + """AIPASS_MEMORY_PYTHON env var overrides all other paths.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + monkeypatch.setenv("AIPASS_MEMORY_PYTHON", "/custom/bin/python3") + result = verify._get_memory_python() + assert result == "/custom/bin/python3" + + def test_venv_python_used_when_exists(self, monkeypatch, tmp_path): + """When the memory .venv python exists, it should be returned.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + # Remove env override so it doesn't interfere + monkeypatch.delenv("AIPASS_MEMORY_PYTHON", raising=False) + + # Create a fake venv python + fake_python = tmp_path / ".venv" / "bin" / "python" + fake_python.parent.mkdir(parents=True) + fake_python.touch() + + # Patch the module-level constant + monkeypatch.setattr(verify, "_MEMORY_VENV_PYTHON", fake_python) + result = verify._get_memory_python() + assert result == str(fake_python) + + def test_fallback_to_sys_executable(self, monkeypatch): + """When no env var and no venv, fall back to sys.executable.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + monkeypatch.delenv("AIPASS_MEMORY_PYTHON", raising=False) + # Point to a path that does not exist + monkeypatch.setattr( + verify, "_MEMORY_VENV_PYTHON", Path("/nonexistent/.venv/bin/python") + ) + result = verify._get_memory_python() + assert result == sys.executable + + def test_env_override_empty_string_is_falsy(self, monkeypatch): + """An empty AIPASS_MEMORY_PYTHON env var should be treated as unset.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + monkeypatch.setenv("AIPASS_MEMORY_PYTHON", "") + # Empty string is falsy, so it should NOT be returned + monkeypatch.setattr( + verify, "_MEMORY_VENV_PYTHON", Path("/nonexistent/.venv/bin/python") + ) + result = verify._get_memory_python() + # Falls through to venv check (missing) then sys.executable + assert result == sys.executable + + +# =========================================================================== +# 2. _check_plan_subprocess() +# =========================================================================== + + +class TestCheckPlanSubprocess: + """Tests for _check_plan_subprocess() subprocess orchestration.""" + + def test_success_returns_parsed_json(self, monkeypatch): + """Successful subprocess returns parsed JSON dict.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + expected = {"success": True, "found": True, "count": 5, "source_files": ["a.md"]} + fake_result = MagicMock() + fake_result.returncode = 0 + fake_result.stdout = json.dumps(expected) + + monkeypatch.setattr(verify, "_get_memory_python", lambda: "/usr/bin/python3") + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: fake_result) + + result = verify._check_plan_subprocess("FPLAN-0126") + assert result["success"] is True + assert result["found"] is True + assert result["count"] == 5 + assert "a.md" in result["source_files"] + + def test_nonzero_returncode_returns_error(self, monkeypatch): + """Non-zero return code produces error dict with stderr message.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_result = MagicMock() + fake_result.returncode = 1 + fake_result.stderr = "collection not found" + + monkeypatch.setattr(verify, "_get_memory_python", lambda: "/usr/bin/python3") + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: fake_result) + + result = verify._check_plan_subprocess("FPLAN-9999") + assert result["success"] is False + assert "collection not found" in result["error"] + + def test_nonzero_returncode_empty_stderr(self, monkeypatch): + """Non-zero return with empty stderr gives generic failure message.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_result = MagicMock() + fake_result.returncode = 1 + fake_result.stderr = "" + + monkeypatch.setattr(verify, "_get_memory_python", lambda: "/usr/bin/python3") + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: fake_result) + + result = verify._check_plan_subprocess("FPLAN-0001") + assert result["success"] is False + assert result["error"] == "Subprocess failed" + + def test_timeout_returns_error(self, monkeypatch): + """Subprocess timeout produces a clear error message.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + def raise_timeout(*args, **kwargs): + raise subprocess.TimeoutExpired(cmd="python", timeout=60) + + monkeypatch.setattr(verify, "_get_memory_python", lambda: "/usr/bin/python3") + monkeypatch.setattr(subprocess, "run", raise_timeout) + + result = verify._check_plan_subprocess("FPLAN-0126") + assert result["success"] is False + assert "timed out" in result["error"] + + def test_invalid_json_returns_error(self, monkeypatch): + """Invalid JSON stdout produces a JSON decode error.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_result = MagicMock() + fake_result.returncode = 0 + fake_result.stdout = "not valid json {{" + + monkeypatch.setattr(verify, "_get_memory_python", lambda: "/usr/bin/python3") + monkeypatch.setattr(subprocess, "run", lambda *a, **kw: fake_result) + + result = verify._check_plan_subprocess("FPLAN-0126") + assert result["success"] is False + assert "Invalid JSON" in result["error"] + + def test_generic_exception_returns_error(self, monkeypatch): + """Any other exception is caught and returned as error.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + def raise_oserror(*args, **kwargs): + raise OSError("No such file or directory") + + monkeypatch.setattr(verify, "_get_memory_python", lambda: "/usr/bin/python3") + monkeypatch.setattr(subprocess, "run", raise_oserror) + + result = verify._check_plan_subprocess("FPLAN-0126") + assert result["success"] is False + assert "No such file or directory" in result["error"] + + def test_subprocess_receives_correct_input(self, monkeypatch): + """Verify the input JSON sent to subprocess contains correct fields.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + captured_kwargs = {} + + def capture_run(*args, **kwargs): + captured_kwargs.update(kwargs) + result = MagicMock() + result.returncode = 0 + result.stdout = json.dumps({"success": True, "found": False, "count": 0}) + return result + + monkeypatch.setattr(verify, "_get_memory_python", lambda: "/usr/bin/python3") + monkeypatch.setattr(subprocess, "run", capture_run) + + verify._check_plan_subprocess("HPLAN-0042") + input_data = json.loads(captured_kwargs["input"]) + assert input_data["operation"] == "check_plan" + assert input_data["plan_label"] == "HPLAN-0042" + + +# =========================================================================== +# 3. is_plan_vectorized() +# =========================================================================== + + +class TestIsPlanVectorized: + """Tests for the public is_plan_vectorized() API.""" + + def test_delegates_to_check_plan_subprocess(self, monkeypatch): + """is_plan_vectorized is a thin wrapper around _check_plan_subprocess.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + expected = {"success": True, "found": True, "count": 3, "source_files": []} + monkeypatch.setattr(verify, "_check_plan_subprocess", lambda label: expected) + + result = verify.is_plan_vectorized("FPLAN-0126") + assert result is expected + assert result["found"] is True + assert result["count"] == 3 + + def test_returns_failure_dict_on_error(self, monkeypatch): + """When subprocess fails, the error dict propagates through.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + error_result = {"success": False, "error": "something broke"} + monkeypatch.setattr(verify, "_check_plan_subprocess", lambda label: error_result) + + result = verify.is_plan_vectorized("FPLAN-0000") + assert result["success"] is False + assert result["error"] == "something broke" + + +# =========================================================================== +# 4. handle_command() +# =========================================================================== + + +class TestHandleCommand: + """Tests for handle_command() routing.""" + + def test_verify_no_args_calls_introspection(self, monkeypatch): + """'verify' with no args triggers print_introspection.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + called = {"introspection": False} + + def fake_introspection(): + called["introspection"] = True + + monkeypatch.setattr(verify, "print_introspection", fake_introspection) + + result = verify.handle_command("verify", []) + assert result is True + assert called["introspection"] is True + + def test_verify_help_flag(self, monkeypatch): + """'verify --help' triggers print_help.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + called = {"help": False} + + def fake_help(): + called["help"] = True + + monkeypatch.setattr(verify, "print_help", fake_help) + + result = verify.handle_command("verify", ["--help"]) + assert result is True + assert called["help"] is True + + def test_verify_h_flag(self, monkeypatch): + """'verify -h' triggers print_help.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + called = {"help": False} + monkeypatch.setattr(verify, "print_help", lambda: called.update(help=True)) + + result = verify.handle_command("verify", ["-h"]) + assert result is True + assert called["help"] is True + + def test_verify_help_word(self, monkeypatch): + """'verify help' triggers print_help.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + called = {"help": False} + monkeypatch.setattr(verify, "print_help", lambda: called.update(help=True)) + + result = verify.handle_command("verify", ["help"]) + assert result is True + assert called["help"] is True + + def test_verify_with_plan_label(self, monkeypatch): + """'verify FPLAN-0126' calls _verify_plan with the label.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + captured_label = {} + + def fake_verify_plan(label): + captured_label["label"] = label + + monkeypatch.setattr(verify, "_verify_plan", fake_verify_plan) + + result = verify.handle_command("verify", ["FPLAN-0126"]) + assert result is True + assert captured_label["label"] == "FPLAN-0126" + + def test_unknown_command_returns_false(self, monkeypatch): + """Unknown command returns False (not handled).""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + result = verify.handle_command("bogus", ["arg1"]) + assert result is False + + def test_toplevel_help_flag(self, monkeypatch): + """Top-level '--help' command triggers print_help.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + called = {"help": False} + monkeypatch.setattr(verify, "print_help", lambda: called.update(help=True)) + + result = verify.handle_command("--help", []) + assert result is True + assert called["help"] is True + + def test_toplevel_h_flag(self, monkeypatch): + """Top-level '-h' command triggers print_help.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + called = {"help": False} + monkeypatch.setattr(verify, "print_help", lambda: called.update(help=True)) + + result = verify.handle_command("-h", []) + assert result is True + assert called["help"] is True + + def test_toplevel_help_word(self, monkeypatch): + """Top-level 'help' command triggers print_help.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + called = {"help": False} + monkeypatch.setattr(verify, "print_help", lambda: called.update(help=True)) + + result = verify.handle_command("help", []) + assert result is True + assert called["help"] is True + + +# =========================================================================== +# 5. _discover_handlers() +# =========================================================================== + + +def _setup_fake_module_tree(tmp_path): + """Build tmp_path/apps/modules/verify.py and apps/handlers/ so that + patching verify.__file__ makes _discover_handlers() scan our fake tree. + + _discover_handlers does: + Path(__file__).resolve().parent.parent / "handlers" + so __file__ must sit at /apps/modules/verify.py + and handlers at /apps/handlers/ + """ + modules_dir = tmp_path / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_file = modules_dir / "verify.py" + fake_file.touch() + handlers_dir = tmp_path / "apps" / "handlers" + handlers_dir.mkdir(parents=True) + return str(fake_file), handlers_dir + + +class TestDiscoverHandlers: + """Tests for _discover_handlers() directory scanning.""" + + def test_discovers_handler_directories_with_py_files(self, monkeypatch, tmp_path): + """Finds handler dirs containing .py files (excluding __init__.py).""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_file, handlers_dir = _setup_fake_module_tree(tmp_path) + + storage_dir = handlers_dir / "storage" + storage_dir.mkdir() + (storage_dir / "__init__.py").write_text("", encoding="utf-8") + (storage_dir / "chroma_subprocess.py").write_text("", encoding="utf-8") + (storage_dir / "vector_store.py").write_text("", encoding="utf-8") + + search_dir = handlers_dir / "search" + search_dir.mkdir() + (search_dir / "__init__.py").write_text("", encoding="utf-8") + (search_dir / "query_engine.py").write_text("", encoding="utf-8") + + monkeypatch.setattr(verify, "__file__", fake_file) + + result = verify._discover_handlers() + assert "storage" in result + assert "search" in result + assert "chroma_subprocess.py" in result["storage"] + assert "vector_store.py" in result["storage"] + assert "query_engine.py" in result["search"] + # __init__.py should be excluded + assert "__init__.py" not in result["storage"] + + def test_skips_dunder_directories(self, monkeypatch, tmp_path): + """Directories starting with __ (like __pycache__) are skipped.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_file, handlers_dir = _setup_fake_module_tree(tmp_path) + + pycache = handlers_dir / "__pycache__" + pycache.mkdir() + (pycache / "verify.cpython-312.pyc").write_text("", encoding="utf-8") + + real_dir = handlers_dir / "json" + real_dir.mkdir() + (real_dir / "json_handler.py").write_text("", encoding="utf-8") + + monkeypatch.setattr(verify, "__file__", fake_file) + + result = verify._discover_handlers() + assert "__pycache__" not in result + assert "json" in result + + def test_empty_handlers_dir_returns_empty(self, monkeypatch, tmp_path): + """An empty handlers directory returns an empty dict.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_file, _handlers_dir = _setup_fake_module_tree(tmp_path) + # handlers_dir exists but is empty (no subdirs) + monkeypatch.setattr(verify, "__file__", fake_file) + + result = verify._discover_handlers() + assert result == {} + + def test_missing_handlers_dir_returns_empty(self, monkeypatch, tmp_path): + """A nonexistent handlers directory returns an empty dict.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + # Point __file__ at a tree with no handlers/ at all + modules_dir = tmp_path / "no_handlers" / "apps" / "modules" + modules_dir.mkdir(parents=True) + fake_file = modules_dir / "verify.py" + fake_file.touch() + + monkeypatch.setattr(verify, "__file__", str(fake_file)) + + result = verify._discover_handlers() + assert result == {} + + def test_dir_with_only_init_is_excluded(self, monkeypatch, tmp_path): + """A handler dir with only __init__.py (no real modules) is excluded.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_file, handlers_dir = _setup_fake_module_tree(tmp_path) + + empty_handler = handlers_dir / "empty_handler" + empty_handler.mkdir() + (empty_handler / "__init__.py").write_text("", encoding="utf-8") + + monkeypatch.setattr(verify, "__file__", fake_file) + + result = verify._discover_handlers() + assert "empty_handler" not in result + assert result == {} + + def test_non_py_files_are_ignored(self, monkeypatch, tmp_path): + """Non-.py files (README, .json, etc.) are not included.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_file, handlers_dir = _setup_fake_module_tree(tmp_path) + + config_dir = handlers_dir / "config" + config_dir.mkdir() + (config_dir / "settings.json").write_text("{}", encoding="utf-8") + (config_dir / "README.md").write_text("docs", encoding="utf-8") + (config_dir / "config_handler.py").write_text("", encoding="utf-8") + + monkeypatch.setattr(verify, "__file__", fake_file) + + result = verify._discover_handlers() + assert "config" in result + assert result["config"] == ["config_handler.py"] + # Confirm non-py files are absent + all_files = [f for files in result.values() for f in files] + assert "settings.json" not in all_files + assert "README.md" not in all_files + + def test_results_are_sorted(self, monkeypatch, tmp_path): + """Handler dirs and files within them are returned in sorted order.""" + _mock_cli_modules(monkeypatch) + verify = _import_verify() + + fake_file, handlers_dir = _setup_fake_module_tree(tmp_path) + + # Create dirs in reverse alphabetical order + for name in ["zebra", "alpha", "middle"]: + d = handlers_dir / name + d.mkdir() + (d / "handler.py").write_text("", encoding="utf-8") + + # Add multiple files to alpha in non-sorted order + (handlers_dir / "alpha" / "z_module.py").write_text("", encoding="utf-8") + (handlers_dir / "alpha" / "a_module.py").write_text("", encoding="utf-8") + + monkeypatch.setattr(verify, "__file__", fake_file) + + result = verify._discover_handlers() + dir_names = list(result.keys()) + assert dir_names == ["alpha", "middle", "zebra"] + assert result["alpha"] == ["a_module.py", "handler.py", "z_module.py"]