From 7aef11adac35a2b04f53a34a8c668db7056fb76c Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sat, 25 Apr 2026 23:41:43 -0700 Subject: [PATCH] feat(memory): test(memory): cover 73 untested functions per TDPLAN-0003 Co-Authored-By: @memory --- .../memory/tests/test_rollover_pipeline.py | 1094 +++++++++++ src/aipass/memory/tests/test_search_extras.py | 596 ++++++ .../memory/tests/test_symbolic_extras.py | 1651 +++++++++++++++++ .../memory/tests/test_symbolic_module.py | 1243 +++++++++++++ src/aipass/memory/tests/test_watcher.py | 492 +++++ 5 files changed, 5076 insertions(+) create mode 100644 src/aipass/memory/tests/test_rollover_pipeline.py create mode 100644 src/aipass/memory/tests/test_search_extras.py create mode 100644 src/aipass/memory/tests/test_symbolic_extras.py create mode 100644 src/aipass/memory/tests/test_symbolic_module.py create mode 100644 src/aipass/memory/tests/test_watcher.py diff --git a/src/aipass/memory/tests/test_rollover_pipeline.py b/src/aipass/memory/tests/test_rollover_pipeline.py new file mode 100644 index 00000000..d446098d --- /dev/null +++ b/src/aipass/memory/tests/test_rollover_pipeline.py @@ -0,0 +1,1094 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_rollover_pipeline.py +# Date: 2026-04-25 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for untested public functions in the rollover pipeline. + +Covers: + from aipass.memory.apps.handlers.rollover.orchestrator import store_vectors_subprocess + from aipass.memory.apps.handlers.rollover.orchestrator import encode_batch_subprocess + from aipass.memory.apps.handlers.rollover.orchestrator import get_branch_local_chroma_path + from aipass.memory.apps.handlers.rollover.orchestrator import extract_text_from_memories + from aipass.memory.apps.handlers.rollover.extractor import extract_with_metadata + from aipass.memory.apps.modules.rollover import run_rollover + from aipass.memory.apps.modules.rollover import show_status + from aipass.memory.apps.modules.rollover import check_triggers + from aipass.memory.apps.handlers.schema.normalize import normalize_all_memory_files + from aipass.memory.apps.handlers.tracking.line_counter import update_all_memory_files + from aipass.memory.apps.handlers.learnings.manager import process_all_branches + +All tests use mocks or tmp_path -- no live filesystem or infrastructure access. +""" + +import json +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Import helpers -- each handler has module-level imports that need mocking +# --------------------------------------------------------------------------- + + +def _import_orchestrator(monkeypatch): + """Import orchestrator with mocked infrastructure dependencies.""" + mock_detector = MagicMock() + mock_detector._read_registry = MagicMock(return_value=[]) + mock_detector.check_all_branches = MagicMock(return_value={"success": True, "triggers": []}) + + mock_extractor = MagicMock() + mock_line_counter = MagicMock() + + monitor_pkg = MagicMock() + monitor_pkg.detector = mock_detector + + rollover_pkg = MagicMock() + rollover_pkg.extractor = mock_extractor + + tracking_pkg = MagicMock() + tracking_pkg.line_counter = mock_line_counter + + 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.rollover.extractor", mock_extractor) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.tracking", tracking_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.tracking.line_counter", mock_line_counter) + + sys.modules.pop("aipass.memory.apps.handlers.rollover.orchestrator", None) + parent = sys.modules.get("aipass.memory.apps.handlers.rollover") + if parent is not None and hasattr(parent, "orchestrator"): + delattr(parent, "orchestrator") + + from aipass.memory.apps.handlers.rollover import orchestrator + + return orchestrator, { + "detector": mock_detector, + "extractor": mock_extractor, + "line_counter": mock_line_counter, + } + + +def _import_extractor(monkeypatch): + """Import extractor with mocked infrastructure dependencies.""" + mock_json_handler = MagicMock() + mock_json_handler.log_operation = MagicMock(return_value=True) + mock_memory_files = MagicMock() + mock_memory_files.read_memory_file_data = MagicMock(return_value=None) + mock_memory_files.write_memory_file_simple = MagicMock() + + 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) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.memory_files", mock_memory_files) + + sys.modules.pop("aipass.memory.apps.handlers.rollover.extractor", None) + parent = sys.modules.get("aipass.memory.apps.handlers.rollover") + if parent is not None and hasattr(parent, "extractor"): + delattr(parent, "extractor") + + from aipass.memory.apps.handlers.rollover import extractor + + return extractor, { + "json_handler": mock_json_handler, + "memory_files": mock_memory_files, + } + + +def _import_rollover_module(monkeypatch): + """Import the rollover module with mocked infrastructure dependencies.""" + # 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}) + + monitor_pkg = MagicMock() + monitor_pkg.detector = mock_detector + + 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.rollover", rollover_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.rollover.orchestrator", mock_orchestrator) + + sys.modules.pop("aipass.memory.apps.modules.rollover", None) + 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, { + "console": mock_console, + "error": mock_error, + "warning": mock_warning, + "detector": mock_detector, + "orchestrator": mock_orchestrator, + } + + +def _import_normalize(monkeypatch): + """Import normalize with mocked infrastructure dependencies.""" + 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) + + sys.modules.pop("aipass.memory.apps.handlers.schema.normalize", None) + parent = sys.modules.get("aipass.memory.apps.handlers.schema") + if parent is not None and hasattr(parent, "normalize"): + delattr(parent, "normalize") + + from aipass.memory.apps.handlers.schema import normalize + + return normalize, { + "json_handler": mock_json_handler, + } + + +def _import_line_counter(monkeypatch): + """Import line_counter with mocked infrastructure dependencies.""" + mock_json_handler = MagicMock() + mock_json_handler.log_operation = MagicMock(return_value=True) + mock_memory_files = MagicMock() + mock_memory_files.update_metadata = MagicMock(return_value={"success": 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) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.memory_files", mock_memory_files) + + sys.modules.pop("aipass.memory.apps.handlers.tracking.line_counter", None) + parent = sys.modules.get("aipass.memory.apps.handlers.tracking") + if parent is not None and hasattr(parent, "line_counter"): + delattr(parent, "line_counter") + + from aipass.memory.apps.handlers.tracking import line_counter + + return line_counter, { + "json_handler": mock_json_handler, + "memory_files": mock_memory_files, + } + + +def _import_manager(monkeypatch): + """Import manager with mocked infrastructure dependencies.""" + mock_json_handler = MagicMock() + mock_json_handler.log_operation = MagicMock(return_value=True) + mock_memory_files = MagicMock() + mock_memory_files.read_memory_file_data = MagicMock(return_value=None) + mock_memory_files.write_memory_file_simple = MagicMock() + + 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) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.memory_files", mock_memory_files) + + sys.modules.pop("aipass.memory.apps.handlers.learnings.manager", None) + parent = sys.modules.get("aipass.memory.apps.handlers.learnings") + if parent is not None and hasattr(parent, "manager"): + delattr(parent, "manager") + + from aipass.memory.apps.handlers.learnings import manager + + return manager, { + "json_handler": mock_json_handler, + "memory_files": mock_memory_files, + } + + +# =========================================================================== +# Tests: orchestrator.store_vectors_subprocess +# =========================================================================== + + +class TestStoreVectorsSubprocess: + """Test store_vectors_subprocess calls subprocess and returns dict.""" + + def test_success_returns_parsed_json(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + expected = {"success": True, "collection": "test_col", "total_vectors": 5} + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = json.dumps(expected) + + with patch.object(subprocess, "run", return_value=mock_result) as mock_run: + result = orch.store_vectors_subprocess( + branch="TEST", + memory_type="sessions", + embeddings=[[0.1, 0.2]], + documents=["doc1"], + metadatas=[{"key": "val"}], + db_path="/tmp/test.chroma", + ) + + assert result["success"] is True + assert result["collection"] == "test_col" + mock_run.assert_called_once() + + def test_nonzero_returncode_returns_failure(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stderr = "some error" + + with patch.object(subprocess, "run", return_value=mock_result): + result = orch.store_vectors_subprocess( + branch="TEST", + memory_type="sessions", + embeddings=[[0.1]], + documents=["doc1"], + metadatas=[{}], + ) + + assert result["success"] is False + assert "some error" in result["error"] + + def test_timeout_returns_failure(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + + with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="test", timeout=60)): + result = orch.store_vectors_subprocess( + branch="TEST", + memory_type="sessions", + embeddings=[[0.1]], + documents=["doc1"], + metadatas=[{}], + ) + + assert result["success"] is False + assert "timed out" in result["error"] + + def test_invalid_json_response_returns_failure(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = "not valid json" + + with patch.object(subprocess, "run", return_value=mock_result): + result = orch.store_vectors_subprocess( + branch="TEST", + memory_type="sessions", + embeddings=[[0.1]], + documents=["doc1"], + metadatas=[{}], + ) + + assert result["success"] is False + assert "Invalid JSON" in result["error"] + + def test_numpy_array_tolist_conversion(self, monkeypatch): + """Embeddings with tolist() method get serialized correctly.""" + orch, _ = _import_orchestrator(monkeypatch) + expected = {"success": True} + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = json.dumps(expected) + + # Simulate a numpy array with tolist method + mock_embedding = MagicMock() + mock_embedding.tolist.return_value = [0.1, 0.2, 0.3] + + with patch.object(subprocess, "run", return_value=mock_result) as mock_run: + result = orch.store_vectors_subprocess( + branch="TEST", + memory_type="sessions", + embeddings=[mock_embedding], + documents=["doc1"], + metadatas=[{}], + ) + + assert result["success"] is True + mock_embedding.tolist.assert_called_once() + # Verify the serialized data includes the converted list + call_kwargs = mock_run.call_args + input_data = json.loads(call_kwargs.kwargs.get("input", call_kwargs[1].get("input", ""))) + assert input_data["embeddings"] == [[0.1, 0.2, 0.3]] + + +# =========================================================================== +# Tests: orchestrator.encode_batch_subprocess +# =========================================================================== + + +class TestEncodeBatchSubprocess: + """Test encode_batch_subprocess calls subprocess for embedding.""" + + def test_success_returns_embeddings(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + expected = {"success": True, "embeddings": [[0.1, 0.2]], "count": 1, "dimension": 2} + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = json.dumps(expected) + + with patch.object(subprocess, "run", return_value=mock_result): + result = orch.encode_batch_subprocess(["hello world"]) + + assert result["success"] is True + assert result["embeddings"] == [[0.1, 0.2]] + + def test_nonzero_returncode_returns_failure(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stderr = "embedding error" + + with patch.object(subprocess, "run", return_value=mock_result): + result = orch.encode_batch_subprocess(["text"]) + + assert result["success"] is False + assert "embedding error" in result["error"] + + def test_timeout_returns_failure(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + + with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="test", timeout=120)): + result = orch.encode_batch_subprocess(["text"]) + + assert result["success"] is False + assert "timed out" in result["error"] + + def test_invalid_json_returns_failure(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = "bad json" + + with patch.object(subprocess, "run", return_value=mock_result): + result = orch.encode_batch_subprocess(["text"]) + + assert result["success"] is False + assert "Invalid JSON" in result["error"] + + def test_generic_exception_returns_failure(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + + with patch.object(subprocess, "run", side_effect=OSError("no such file")): + result = orch.encode_batch_subprocess(["text"]) + + assert result["success"] is False + assert "no such file" in result["error"] + + +# =========================================================================== +# Tests: orchestrator.get_branch_local_chroma_path +# =========================================================================== + + +class TestGetBranchLocalChromaPath: + """Test get_branch_local_chroma_path looks up branch in registry.""" + + def test_returns_chroma_path_for_existing_branch(self, monkeypatch, tmp_path): + orch, mocks = _import_orchestrator(monkeypatch) + branch_dir = tmp_path / "my_branch" + branch_dir.mkdir() + + mocks["detector"]._read_registry.return_value = [ + {"name": "MY_BRANCH", "path": str(branch_dir)}, + ] + + result = orch.get_branch_local_chroma_path("MY_BRANCH") + + assert result is not None + assert result == branch_dir / ".chroma" + assert result.exists() # auto-created + + def test_returns_none_for_empty_name(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + assert orch.get_branch_local_chroma_path("") is None + + def test_returns_none_for_unknown_branch(self, monkeypatch): + orch, mocks = _import_orchestrator(monkeypatch) + mocks["detector"]._read_registry.return_value = [ + {"name": "OTHER", "path": "/nonexistent"}, + ] + result = orch.get_branch_local_chroma_path("MISSING_BRANCH") + assert result is None + + def test_case_insensitive_lookup(self, monkeypatch, tmp_path): + orch, mocks = _import_orchestrator(monkeypatch) + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + + mocks["detector"]._read_registry.return_value = [ + {"name": "My_Branch", "path": str(branch_dir)}, + ] + + result = orch.get_branch_local_chroma_path("my_branch") + assert result is not None + assert result == branch_dir / ".chroma" + + def test_returns_existing_chroma_dir(self, monkeypatch, tmp_path): + orch, mocks = _import_orchestrator(monkeypatch) + branch_dir = tmp_path / "branch" + chroma_dir = branch_dir / ".chroma" + chroma_dir.mkdir(parents=True) + + mocks["detector"]._read_registry.return_value = [ + {"name": "BRANCH", "path": str(branch_dir)}, + ] + + result = orch.get_branch_local_chroma_path("BRANCH") + assert result == chroma_dir + + +# =========================================================================== +# Tests: orchestrator.extract_text_from_memories +# =========================================================================== + + +class TestExtractTextFromMemories: + """Test extract_text_from_memories extracts text from memory items.""" + + def test_extracts_from_activities(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [{"activities": ["task 1", "task 2"]}] + texts = orch.extract_text_from_memories(memories) + assert len(texts) == 1 + assert "task 1" in texts[0] + assert "task 2" in texts[0] + + def test_extracts_from_summary(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [{"summary": "Session summary text"}] + texts = orch.extract_text_from_memories(memories) + assert texts == ["Session summary text"] + + def test_extracts_from_key_learning(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [{"_type": "key_learning", "key": "pattern", "value": "use pathlib"}] + texts = orch.extract_text_from_memories(memories) + assert len(texts) == 1 + assert "pattern" in texts[0] + assert "use pathlib" in texts[0] + + def test_extracts_from_content_field(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [{"content": "some content"}] + texts = orch.extract_text_from_memories(memories) + assert texts == ["some content"] + + def test_extracts_from_text_field(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [{"text": "raw text"}] + texts = orch.extract_text_from_memories(memories) + assert texts == ["raw text"] + + def test_extracts_from_message_field(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [{"message": "a message"}] + texts = orch.extract_text_from_memories(memories) + assert texts == ["a message"] + + def test_fallback_to_string_representation(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [{"unknown_field": 42}] + texts = orch.extract_text_from_memories(memories) + assert len(texts) == 1 + assert "unknown_field" in texts[0] + + def test_empty_list_returns_empty(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + assert orch.extract_text_from_memories([]) == [] + + def test_multiple_memory_types(self, monkeypatch): + orch, _ = _import_orchestrator(monkeypatch) + memories = [ + {"summary": "session 1"}, + {"content": "observation"}, + {"_type": "key_learning", "key": "k", "value": "v"}, + ] + texts = orch.extract_text_from_memories(memories) + assert len(texts) == 3 + + +# =========================================================================== +# Tests: extractor.extract_with_metadata +# =========================================================================== + + +class TestExtractWithMetadata: + """Test extract_with_metadata enriches extracted items.""" + + def test_returns_failure_for_nonexistent_file(self, monkeypatch, tmp_path): + ext, _ = _import_extractor(monkeypatch) + result = ext.extract_with_metadata(tmp_path / "nonexistent.json") + assert result["success"] is False + assert "not found" in result["error"].lower() + + def test_returns_failure_when_file_cannot_be_parsed(self, monkeypatch, tmp_path): + ext, mocks = _import_extractor(monkeypatch) + file_path = tmp_path / "bad.json" + file_path.write_text("{}", encoding="utf-8") + mocks["memory_files"].read_memory_file_data.return_value = None + + result = ext.extract_with_metadata(file_path) + assert result["success"] is False + + def test_v2_extraction_enriches_entries(self, monkeypatch, tmp_path): + """v2 schema extraction adds _metadata to each extracted entry.""" + ext, mocks = _import_extractor(monkeypatch) + + # Create a v2 file with sessions exceeding limits + data = { + "document_metadata": { + "schema_version": "2.0.0", + "limits": {"max_sessions": 2}, + "status": {}, + }, + "sessions": [ + {"session_number": 1, "summary": "newest"}, + {"session_number": 2, "summary": "middle"}, + {"session_number": 3, "summary": "oldest"}, + ], + } + file_path = tmp_path / ".trinity" / "local.json" + file_path.parent.mkdir(parents=True) + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + mocks["memory_files"].read_memory_file_data.return_value = data + mocks["memory_files"].write_memory_file_simple.return_value = None + + result = ext.extract_with_metadata(file_path) + + assert result["success"] is True + assert "entries" in result + assert result["branch"] is not None + assert result["type"] is not None + # Enriched entries should have _metadata + for entry in result.get("entries", []): + assert "_metadata" in entry + assert "branch" in entry["_metadata"] + assert "extracted_at" in entry["_metadata"] + + def test_skipped_result_passes_through(self, monkeypatch, tmp_path): + """When extract_items returns skipped (under limit), extract_with_metadata passes it.""" + ext, mocks = _import_extractor(monkeypatch) + + # v2 file under limits (no extraction needed) + data = { + "document_metadata": { + "schema_version": "2.0.0", + "limits": {"max_sessions": 10}, + "status": {}, + }, + "sessions": [{"session_number": 1, "summary": "only one"}], + } + file_path = tmp_path / ".trinity" / "local.json" + file_path.parent.mkdir(parents=True) + file_path.write_text(json.dumps(data, indent=2), encoding="utf-8") + + mocks["memory_files"].read_memory_file_data.return_value = data + + result = ext.extract_with_metadata(file_path) + # _extract_items_v2 returns skipped when under limit; extract_with_metadata + # passes the result dict through unchanged when nothing was extracted + assert result["success"] is True + # Either skipped=True (passthrough) or entries is empty (wrapped) + assert result.get("skipped") is True or result.get("count", 0) == 0 + + +# =========================================================================== +# Tests: modules.rollover.run_rollover +# =========================================================================== + + +class TestRunRollover: + """Test run_rollover delegates to handler and renders Rich output.""" + + def test_returns_true_when_no_triggers(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["orchestrator"].execute_rollover.return_value = { + "success": True, + "triggers_count": 0, + "success_count": 0, + "failed": [], + "results": [], + } + result = rollover.run_rollover() + assert result is True + + def test_returns_false_on_handler_exception(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["orchestrator"].execute_rollover.side_effect = RuntimeError("boom") + result = rollover.run_rollover() + assert result is False + mocks["error"].assert_called() + + def test_returns_false_on_error_result(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["orchestrator"].execute_rollover.return_value = { + "success": False, + "error": "Registry missing", + "triggers_count": 0, + } + result = rollover.run_rollover() + assert result is False + + def test_returns_true_with_successful_rollover(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["orchestrator"].execute_rollover.return_value = { + "success": True, + "triggers_count": 1, + "success_count": 1, + "failed": [], + "results": [ + { + "trigger": "TEST.local.json", + "memories_count": 5, + "old_lines": 600, + "new_lines": 400, + "global_collection": "test_col", + "global_total": 50, + "local_stored": True, + } + ], + } + result = rollover.run_rollover() + assert result is True + + def test_displays_failure_details(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["orchestrator"].execute_rollover.return_value = { + "success": False, + "triggers_count": 1, + "success_count": 0, + "failed": [{"trigger": "BAD.local.json", "stage": "embedding", "error": "model not found"}], + "results": [], + } + rollover.run_rollover() + mocks["error"].assert_called() + + +# =========================================================================== +# Tests: modules.rollover.show_status +# =========================================================================== + + +class TestShowStatus: + """Test show_status calls detector.get_rollover_stats and prints output.""" + + def test_displays_stats_on_success(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["detector"].get_rollover_stats.return_value = { + "success": True, + "total_branches": 2, + "files_checked": 4, + "files_ready": 1, + "branches": { + "TEST": { + "local": { + "current": 500, + "max": 600, + "ready": False, + "remaining": 100, + "schema_version": "1.0.0", + } + } + }, + } + rollover.show_status() + mocks["console"].print.assert_called() + + def test_displays_error_on_failure(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["detector"].get_rollover_stats.return_value = { + "success": False, + "error": "Registry not found", + } + rollover.show_status() + mocks["error"].assert_called() + + def test_displays_v2_branch_details(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["detector"].get_rollover_stats.return_value = { + "success": True, + "total_branches": 1, + "files_checked": 1, + "files_ready": 1, + "branches": { + "V2BRANCH": { + "local": { + "current": 25, + "max": 20, + "ready": True, + "remaining": 0, + "schema_version": "2.0.0", + "v2_reason": "sessions: 25/20", + } + } + }, + } + rollover.show_status() + # Should have printed without error + mocks["error"].assert_not_called() + + +# =========================================================================== +# Tests: modules.rollover.check_triggers +# =========================================================================== + + +class TestCheckTriggers: + """Test check_triggers calls detector.check_all_branches and prints output.""" + + def test_no_triggers_prints_clean(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["detector"].check_all_branches.return_value = {"success": True, "triggers": []} + rollover.check_triggers() + mocks["error"].assert_not_called() + + def test_displays_triggers_when_found(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mock_trigger = MagicMock() + mock_trigger.__str__ = MagicMock(return_value="TEST.local.json (650/600 lines)") + mocks["detector"].check_all_branches.return_value = { + "success": True, + "triggers": [mock_trigger], + } + rollover.check_triggers() + mocks["error"].assert_not_called() + + def test_displays_error_on_failure(self, monkeypatch): + rollover, mocks = _import_rollover_module(monkeypatch) + mocks["detector"].check_all_branches.return_value = { + "success": False, + "error": "Cannot read registry", + } + rollover.check_triggers() + mocks["error"].assert_called() + + +# =========================================================================== +# Tests: normalize.normalize_all_memory_files +# =========================================================================== + + +class TestNormalizeAllMemoryFiles: + """Test normalize_all_memory_files iterates registry branches.""" + + def test_returns_error_when_registry_not_found(self, monkeypatch): + norm, _ = _import_normalize(monkeypatch) + with patch.object(norm, "_find_repo_root", return_value=Path("/nonexistent")): + result = norm.normalize_all_memory_files() + assert result["success"] is False + assert "not found" in result["error"] + + def test_normalizes_files_for_existing_branches(self, monkeypatch, tmp_path): + norm, _ = _import_normalize(monkeypatch) + + # Create registry + branch_dir = tmp_path / "src" / "aipass" / "test_branch" + branch_dir.mkdir(parents=True) + + # Create memory file that needs normalization (root-level limits) + memory_data = { + "limits": {"max_lines": 600}, + "document_metadata": {"status": {}}, + "sessions": [], + } + file_path = branch_dir / "TEST_BRANCH.local.json" + file_path.write_text(json.dumps(memory_data, indent=2), encoding="utf-8") + + registry = { + "branches": [ + {"name": "TEST_BRANCH", "path": str(branch_dir)}, + ] + } + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry), encoding="utf-8") + + with patch.object(norm, "_find_repo_root", return_value=tmp_path): + result = norm.normalize_all_memory_files() + + assert result["success"] is True + assert result["files_checked"] >= 1 + + def test_skips_branches_with_missing_paths(self, monkeypatch, tmp_path): + norm, _ = _import_normalize(monkeypatch) + + registry = { + "branches": [ + {"name": "MISSING", "path": str(tmp_path / "nonexistent")}, + ] + } + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry), encoding="utf-8") + + with patch.object(norm, "_find_repo_root", return_value=tmp_path): + result = norm.normalize_all_memory_files() + + assert result["success"] is True + assert result["files_checked"] == 0 + + def test_dry_run_does_not_modify_files(self, monkeypatch, tmp_path): + norm, _ = _import_normalize(monkeypatch) + + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + + memory_data = { + "limits": {"max_lines": 600}, + "document_metadata": {"status": {}}, + "sessions": [], + } + file_path = branch_dir / "BRANCH.local.json" + original_content = json.dumps(memory_data, indent=2) + file_path.write_text(original_content, encoding="utf-8") + + registry = {"branches": [{"name": "BRANCH", "path": str(branch_dir)}]} + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry), encoding="utf-8") + + with patch.object(norm, "_find_repo_root", return_value=tmp_path): + result = norm.normalize_all_memory_files(dry_run=True) + + assert result["dry_run"] is True + # File content should not be changed in dry_run + assert file_path.read_text(encoding="utf-8") == original_content + + +# =========================================================================== +# Tests: line_counter.update_all_memory_files +# =========================================================================== + + +class TestUpdateAllMemoryFiles: + """Test update_all_memory_files iterates registry branches.""" + + def test_returns_empty_when_no_branches(self, monkeypatch): + lc, _ = _import_line_counter(monkeypatch) + + mock_read_registry = MagicMock(return_value=[]) + mock_get_path = MagicMock(return_value=None) + + with ( + patch( + "aipass.memory.apps.handlers.monitor.detector._read_registry", + mock_read_registry, + ), + patch( + "aipass.memory.apps.handlers.monitor.detector._get_memory_file_path", + mock_get_path, + ), + ): + result = lc.update_all_memory_files() + + assert result["success"] is True + assert result["updated"] == 0 + + def test_updates_existing_files(self, monkeypatch, tmp_path): + lc, mocks = _import_line_counter(monkeypatch) + + # Create a real memory file + file_path = tmp_path / "local.json" + file_path.write_text('{\n "test": true\n}\n', encoding="utf-8") + + branch = {"name": "TEST", "path": str(tmp_path)} + + mock_read_registry = MagicMock(return_value=[branch]) + + def mock_get_path(b, mem_type): + if mem_type == "local": + return file_path + return None + + with ( + patch( + "aipass.memory.apps.handlers.monitor.detector._read_registry", + mock_read_registry, + ), + patch( + "aipass.memory.apps.handlers.monitor.detector._get_memory_file_path", + mock_get_path, + ), + ): + result = lc.update_all_memory_files() + + assert result["success"] is True + assert result["updated"] >= 1 + + def test_tracks_failures(self, monkeypatch, tmp_path): + lc, mocks = _import_line_counter(monkeypatch) + + # Make update_metadata fail + mocks["memory_files"].update_metadata.return_value = {"success": False, "error": "write error"} + + file_path = tmp_path / "local.json" + file_path.write_text("{}\n", encoding="utf-8") + + branch = {"name": "TEST", "path": str(tmp_path)} + + mock_read_registry = MagicMock(return_value=[branch]) + + def mock_get_path(b, mem_type): + if mem_type == "local": + return file_path + return None + + with ( + patch( + "aipass.memory.apps.handlers.monitor.detector._read_registry", + mock_read_registry, + ), + patch( + "aipass.memory.apps.handlers.monitor.detector._get_memory_file_path", + mock_get_path, + ), + ): + result = lc.update_all_memory_files() + + assert result["success"] is True + assert result["failed"] >= 1 + + +# =========================================================================== +# Tests: manager.process_all_branches +# =========================================================================== + + +class TestProcessAllBranches: + """Test process_all_branches iterates registry branches.""" + + def test_returns_error_when_registry_not_found(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + with patch.object(mgr, "_find_repo_root", return_value=Path("/nonexistent")): + result = mgr.process_all_branches() + assert result["success"] is False + assert "not found" in result["error"] + + def test_processes_branches_with_local_files(self, monkeypatch, tmp_path): + mgr, mocks = _import_manager(monkeypatch) + + # Create branch with local file + branch_dir = tmp_path / "branch" + branch_dir.mkdir() + + local_data = { + "document_metadata": { + "limits": {"max_learnings": 100, "max_recently_completed": 20}, + "status": {}, + }, + "key_learnings": {"item1": "test learning [2026-01-01]"}, + "recently_completed": [], + } + local_file = branch_dir / "TEST.local.json" + local_file.write_text(json.dumps(local_data, indent=2), encoding="utf-8") + + # Mock read_memory_file_data to return the data + mocks["memory_files"].read_memory_file_data.return_value = local_data + + registry = {"branches": [{"name": "TEST", "path": str(branch_dir)}]} + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry), encoding="utf-8") + + with patch.object(mgr, "_find_repo_root", return_value=tmp_path): + result = mgr.process_all_branches() + + assert result["success"] is True + assert result["processed"] >= 1 + + def test_skips_branches_without_local_file(self, monkeypatch, tmp_path): + mgr, _ = _import_manager(monkeypatch) + + # Create branch dir without local file + branch_dir = tmp_path / "empty_branch" + branch_dir.mkdir() + + registry = {"branches": [{"name": "EMPTY", "path": str(branch_dir)}]} + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry), encoding="utf-8") + + with patch.object(mgr, "_find_repo_root", return_value=tmp_path): + result = mgr.process_all_branches() + + assert result["success"] is True + assert result["skipped"] >= 1 + assert result["processed"] == 0 + + def test_skips_branches_with_nonexistent_paths(self, monkeypatch, tmp_path): + mgr, _ = _import_manager(monkeypatch) + + registry = { + "branches": [ + {"name": "MISSING", "path": str(tmp_path / "no_such_dir")}, + ] + } + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text(json.dumps(registry), encoding="utf-8") + + with patch.object(mgr, "_find_repo_root", return_value=tmp_path): + result = mgr.process_all_branches() + + assert result["success"] is True + assert result["skipped"] >= 1 + assert result["processed"] == 0 + + def test_handles_read_registry_failure(self, monkeypatch, tmp_path): + mgr, _ = _import_manager(monkeypatch) + + # Create a malformed registry + registry_path = tmp_path / "AIPASS_REGISTRY.json" + registry_path.write_text("not json", encoding="utf-8") + + with patch.object(mgr, "_find_repo_root", return_value=tmp_path): + result = mgr.process_all_branches() + + assert result["success"] is False + assert "error" in result diff --git a/src/aipass/memory/tests/test_search_extras.py b/src/aipass/memory/tests/test_search_extras.py new file mode 100644 index 00000000..a105586c --- /dev/null +++ b/src/aipass/memory/tests/test_search_extras.py @@ -0,0 +1,596 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_search_extras.py +# Date: 2026-04-25 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for search handler internals (query_executor and vector_search). + +Covers: + from aipass.memory.apps.handlers.search.query_executor import encode_query_subprocess + from aipass.memory.apps.handlers.search.query_executor import search_vectors_subprocess + from aipass.memory.apps.handlers.search.vector_search import search_collection + from aipass.memory.apps.handlers.search.vector_search import encode_query + from aipass.memory.apps.handlers.search.vector_search import search_all_collections + +Tests subprocess-based encoding/search, ChromaDB collection queries, +query encoding via the singleton QueryEncoder, and multi-collection search. +All tests use mocks -- no live subprocess, ML model, or ChromaDB access. +""" + +import json +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Helpers: prepare the mock graph needed to import search handlers +# --------------------------------------------------------------------------- + + +def _prepare_query_executor_mocks(monkeypatch): + """Insert mocks for query_executor module-level imports. + + Returns a dict of key mock objects so tests can assert against them. + """ + # Mock the chroma_subprocess and embed_subprocess script paths + mock_chroma_script = MagicMock() + mock_embed_script = MagicMock() + + return { + "chroma_script": mock_chroma_script, + "embed_script": mock_embed_script, + } + + +def _import_query_executor(monkeypatch): + """Prepare mocks and import (or reimport) query_executor. + + Returns (module, mocks_dict). + """ + mocks = _prepare_query_executor_mocks(monkeypatch) + + # Remove cached module so it gets re-imported with our mocks + sys.modules.pop("aipass.memory.apps.handlers.search.query_executor", None) + + # Clear parent package attribute so Python re-executes module code + parent = sys.modules.get("aipass.memory.apps.handlers.search") + if parent is not None and hasattr(parent, "query_executor"): + delattr(parent, "query_executor") + + from aipass.memory.apps.handlers.search import query_executor # noqa: E402 + + return query_executor, mocks + + +def _prepare_vector_search_mocks(monkeypatch): + """Insert mocks for vector_search module-level imports. + + Returns a dict of key mock objects for assertions. + """ + # Mock chromadb client via the chroma module + mock_client = MagicMock() + mock_chroma = MagicMock() + mock_chroma.get_client = MagicMock(return_value=mock_client) + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.storage.chroma", + mock_chroma, + ) + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.storage", + MagicMock(), + ) + + # Mock sentence_transformers and torch for QueryEncoder + mock_model = MagicMock() + mock_model.encode.return_value = MagicMock(tolist=MagicMock(return_value=[0.1] * 384)) + mock_model.to.return_value = mock_model + + mock_st_cls = MagicMock(return_value=mock_model) + + mock_sentence_transformers = MagicMock() + mock_sentence_transformers.SentenceTransformer = mock_st_cls + monkeypatch.setitem(sys.modules, "sentence_transformers", mock_sentence_transformers) + + mock_torch = MagicMock() + mock_torch.cuda.is_available.return_value = False + monkeypatch.setitem(sys.modules, "torch", mock_torch) + + return { + "client": mock_client, + "model": mock_model, + "st_cls": mock_st_cls, + "torch": mock_torch, + } + + +def _import_vector_search(monkeypatch): + """Prepare mocks and import (or reimport) vector_search. + + Returns (module, mocks_dict). + """ + mocks = _prepare_vector_search_mocks(monkeypatch) + + # Remove cached modules so they get re-imported with our mocks + sys.modules.pop("aipass.memory.apps.handlers.search.vector_search", None) + + # Clear parent package attribute so Python re-executes module code + parent = sys.modules.get("aipass.memory.apps.handlers.search") + if parent is not None and hasattr(parent, "vector_search"): + delattr(parent, "vector_search") + + from aipass.memory.apps.handlers.search import vector_search # noqa: E402 + + # Reset singletons for a clean slate + setattr(vector_search, "_query_encoder", None) + setattr(vector_search, "_search_service", None) + setattr(vector_search, "_local_services", {}) + + return vector_search, mocks + + +# =========================================================================== +# Tests: encode_query_subprocess +# =========================================================================== + + +class TestEncodeQuerySubprocess: + """Verify encode_query_subprocess subprocess-based encoding.""" + + def test_encode_returns_embedding_on_success(self, monkeypatch): + """Successful subprocess returns embedding and dimension.""" + mod, mocks = _import_query_executor(monkeypatch) + + fake_output = json.dumps({"success": True, "embeddings": [[0.1, 0.2, 0.3]], "dimension": 3}) + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = fake_output + + with patch.object(subprocess, "run", return_value=mock_result) as mock_run: + result = mod.encode_query_subprocess("test query") + + assert result["success"] is True + assert result["embedding"] == [0.1, 0.2, 0.3] + assert result["dimension"] == 3 + mock_run.assert_called_once() + + def test_encode_returns_error_on_nonzero_exit(self, monkeypatch): + """Non-zero return code produces error dict.""" + mod, mocks = _import_query_executor(monkeypatch) + + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stderr = "Model not found" + + with patch.object(subprocess, "run", return_value=mock_result): + result = mod.encode_query_subprocess("test query") + + assert result["success"] is False + assert "Model not found" in result["error"] + + def test_encode_handles_timeout(self, monkeypatch): + """TimeoutExpired produces a timeout error.""" + mod, mocks = _import_query_executor(monkeypatch) + + with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="python", timeout=120)): + result = mod.encode_query_subprocess("slow query") + + assert result["success"] is False + assert "timed out" in result["error"].lower() + + def test_encode_handles_invalid_json(self, monkeypatch): + """Invalid JSON from subprocess produces error dict.""" + mod, mocks = _import_query_executor(monkeypatch) + + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = "not valid json" + + with patch.object(subprocess, "run", return_value=mock_result): + result = mod.encode_query_subprocess("test query") + + assert result["success"] is False + assert "json" in result["error"].lower() + + def test_encode_handles_empty_embeddings(self, monkeypatch): + """Response with empty embeddings list produces error.""" + mod, mocks = _import_query_executor(monkeypatch) + + fake_output = json.dumps({"success": True, "embeddings": [], "dimension": 384}) + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = fake_output + + with patch.object(subprocess, "run", return_value=mock_result): + result = mod.encode_query_subprocess("test query") + + assert result["success"] is False + assert "no embedding" in result["error"].lower() + + def test_encode_handles_generic_exception(self, monkeypatch): + """Unexpected exception produces error dict.""" + mod, mocks = _import_query_executor(monkeypatch) + + with patch.object(subprocess, "run", side_effect=OSError("Cannot execute")): + result = mod.encode_query_subprocess("test query") + + assert result["success"] is False + assert "Cannot execute" in result["error"] + + +# =========================================================================== +# Tests: search_vectors_subprocess +# =========================================================================== + + +class TestSearchVectorsSubprocess: + """Verify search_vectors_subprocess subprocess-based search.""" + + def test_search_returns_results_on_success(self, monkeypatch): + """Successful subprocess returns parsed results.""" + mod, mocks = _import_query_executor(monkeypatch) + + fake_output = json.dumps( + {"success": True, "results": [{"document": "hello", "distance": 0.1}], "total_results": 1} + ) + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = fake_output + + with patch.object(subprocess, "run", return_value=mock_result) as mock_run: + result = mod.search_vectors_subprocess( + query_embedding=[0.1, 0.2, 0.3], + branch="TEST", + n_results=5, + ) + + assert result["success"] is True + assert result["total_results"] == 1 + mock_run.assert_called_once() + + def test_search_returns_error_on_nonzero_exit(self, monkeypatch): + """Non-zero exit code produces error dict.""" + mod, mocks = _import_query_executor(monkeypatch) + + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stderr = "DB not found" + + with patch.object(subprocess, "run", return_value=mock_result): + result = mod.search_vectors_subprocess(query_embedding=[0.1]) + + assert result["success"] is False + assert "DB not found" in result["error"] + + def test_search_handles_timeout(self, monkeypatch): + """TimeoutExpired produces a timeout error.""" + mod, mocks = _import_query_executor(monkeypatch) + + with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="python", timeout=60)): + result = mod.search_vectors_subprocess(query_embedding=[0.1]) + + assert result["success"] is False + assert "timed out" in result["error"].lower() + + def test_search_handles_invalid_json(self, monkeypatch): + """Invalid JSON from subprocess produces error dict.""" + mod, mocks = _import_query_executor(monkeypatch) + + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = "{{bad json" + + with patch.object(subprocess, "run", return_value=mock_result): + result = mod.search_vectors_subprocess(query_embedding=[0.1]) + + assert result["success"] is False + assert "json" in result["error"].lower() + + def test_search_passes_db_path_as_string(self, monkeypatch): + """db_path is converted to string in the input data.""" + mod, mocks = _import_query_executor(monkeypatch) + + fake_output = json.dumps({"success": True, "results": []}) + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = fake_output + + with patch.object(subprocess, "run", return_value=mock_result) as mock_run: + mod.search_vectors_subprocess( + query_embedding=[0.1], + db_path=Path("/tmp/test_chroma"), + ) + + call_args = mock_run.call_args + input_data = json.loads(call_args.kwargs.get("input", call_args[1].get("input", ""))) + assert input_data["db_path"] == "/tmp/test_chroma" + + def test_search_handles_generic_exception(self, monkeypatch): + """Unexpected exception produces error dict.""" + mod, mocks = _import_query_executor(monkeypatch) + + with patch.object(subprocess, "run", side_effect=OSError("Cannot execute")): + result = mod.search_vectors_subprocess(query_embedding=[0.1]) + + assert result["success"] is False + assert "Cannot execute" in result["error"] + + +# =========================================================================== +# Tests: search_collection (vector_search) +# =========================================================================== + + +class TestSearchCollection: + """Verify search_collection wraps SearchService.query_collection.""" + + def test_search_returns_results_on_success(self, monkeypatch, tmp_path): + """Successful collection query returns success with results.""" + mod, mocks = _import_vector_search(monkeypatch) + + mock_collection = MagicMock() + mock_collection.query.return_value = { + "ids": [["id1", "id2"]], + "documents": [["doc1", "doc2"]], + "metadatas": [[{"branch": "TEST"}, {"branch": "TEST"}]], + "distances": [[0.1, 0.2]], + } + mocks["client"].get_collection.return_value = mock_collection + + result = mod.search_collection( + query_embedding=[0.1] * 384, + collection_name="test_collection", + n_results=5, + db_path=tmp_path / ".chroma", + ) + + assert result["success"] is True + assert result["count"] == 2 + assert "doc1" in result["documents"] + + def test_search_returns_error_for_missing_collection(self, monkeypatch, tmp_path): + """Missing collection returns success=False with error message.""" + mod, mocks = _import_vector_search(monkeypatch) + + mocks["client"].get_collection.side_effect = ValueError("Collection not found") + + result = mod.search_collection( + query_embedding=[0.1] * 384, + collection_name="nonexistent", + db_path=tmp_path / ".chroma", + ) + + assert result["success"] is False + assert "not found" in result["error"].lower() + + def test_search_returns_error_for_empty_embedding(self, monkeypatch): + """Empty query embedding returns error.""" + mod, mocks = _import_vector_search(monkeypatch) + + result = mod.search_collection( + query_embedding=[], + collection_name="test_collection", + ) + + assert result["success"] is False + assert "no query embedding" in result["error"].lower() + + def test_search_accepts_string_db_path(self, monkeypatch, tmp_path): + """String db_path is converted to Path internally.""" + mod, mocks = _import_vector_search(monkeypatch) + + mock_collection = MagicMock() + mock_collection.query.return_value = { + "ids": [["id1"]], + "documents": [["doc1"]], + "metadatas": [[{}]], + "distances": [[0.05]], + } + mocks["client"].get_collection.return_value = mock_collection + + result = mod.search_collection( + query_embedding=[0.1] * 384, + collection_name="test_collection", + db_path=str(tmp_path / ".chroma"), + ) + + assert result["success"] is True + + def test_search_passes_where_filter(self, monkeypatch, tmp_path): + """Metadata where filter is forwarded to collection.query.""" + mod, mocks = _import_vector_search(monkeypatch) + + mock_collection = MagicMock() + mock_collection.query.return_value = { + "ids": [[]], + "documents": [[]], + "metadatas": [[]], + "distances": [[]], + } + mocks["client"].get_collection.return_value = mock_collection + + mod.search_collection( + query_embedding=[0.1] * 384, + collection_name="test_collection", + where={"branch": "SEEDGO"}, + db_path=tmp_path / ".chroma", + ) + + call_kwargs = mock_collection.query.call_args.kwargs + assert call_kwargs["where"] == {"branch": "SEEDGO"} + + +# =========================================================================== +# Tests: encode_query (vector_search) +# =========================================================================== + + +class TestEncodeQuery: + """Verify encode_query uses QueryEncoder singleton.""" + + def test_encode_returns_embedding(self, monkeypatch): + """Successful encoding returns embedding with dimension.""" + mod, mocks = _import_vector_search(monkeypatch) + + result = mod.encode_query("test query") + + assert result["success"] is True + assert len(result["embedding"]) == 384 + assert result["dimension"] == 384 + assert result["model"] == "all-MiniLM-L6-v2" + + def test_encode_rejects_empty_query(self, monkeypatch): + """Empty query string returns error.""" + mod, mocks = _import_vector_search(monkeypatch) + + result = mod.encode_query("") + + assert result["success"] is False + assert "empty" in result["error"].lower() + + def test_encode_rejects_whitespace_only_query(self, monkeypatch): + """Whitespace-only query string returns error.""" + mod, mocks = _import_vector_search(monkeypatch) + + result = mod.encode_query(" ") + + assert result["success"] is False + assert "empty" in result["error"].lower() + + def test_encode_handles_model_error(self, monkeypatch): + """If model.encode raises, error is caught and returned.""" + mod, mocks = _import_vector_search(monkeypatch) + + mocks["model"].encode.side_effect = RuntimeError("CUDA out of memory") + + result = mod.encode_query("test query") + + assert result["success"] is False + assert "CUDA" in result["error"] + + def test_encode_uses_singleton(self, monkeypatch): + """Successive calls reuse the same QueryEncoder instance.""" + mod, mocks = _import_vector_search(monkeypatch) + + mod.encode_query("first query") + mod.encode_query("second query") + + # SentenceTransformer should only be constructed once + mocks["st_cls"].assert_called_once() + + +# =========================================================================== +# Tests: search_all_collections (vector_search) +# =========================================================================== + + +class TestSearchAllCollections: + """Verify search_all_collections aggregates results across collections.""" + + def test_search_all_returns_aggregated_results(self, monkeypatch, tmp_path): + """Searching all collections returns results from each.""" + mod, mocks = _import_vector_search(monkeypatch) + + mock_coll_a = MagicMock() + mock_coll_a.name = "coll_a" + mock_coll_b = MagicMock() + mock_coll_b.name = "coll_b" + mocks["client"].list_collections.return_value = [mock_coll_a, mock_coll_b] + + mock_collection = MagicMock() + mock_collection.query.return_value = { + "ids": [["id1"]], + "documents": [["doc1"]], + "metadatas": [[{"branch": "TEST"}]], + "distances": [[0.1]], + } + mocks["client"].get_collection.return_value = mock_collection + + result = mod.search_all_collections( + query_embedding=[0.1] * 384, + n_results=3, + db_path=tmp_path / ".chroma", + ) + + assert result["success"] is True + assert result["collections_searched"] == 2 + assert result["total_results"] == 2 + assert "coll_a" in result["results"] + assert "coll_b" in result["results"] + + def test_search_all_returns_empty_when_no_collections(self, monkeypatch, tmp_path): + """No collections returns success with empty results.""" + mod, mocks = _import_vector_search(monkeypatch) + + mocks["client"].list_collections.return_value = [] + + result = mod.search_all_collections( + query_embedding=[0.1] * 384, + db_path=tmp_path / ".chroma", + ) + + assert result["success"] is True + assert result["results"] == {} + assert "no collections" in result["message"].lower() + + def test_search_all_returns_error_for_empty_embedding(self, monkeypatch): + """Empty embedding returns error.""" + mod, mocks = _import_vector_search(monkeypatch) + + result = mod.search_all_collections(query_embedding=[]) + + assert result["success"] is False + assert "no query embedding" in result["error"].lower() + + def test_search_all_handles_exception(self, monkeypatch, tmp_path): + """Exception during search returns error dict.""" + mod, mocks = _import_vector_search(monkeypatch) + + mocks["client"].list_collections.side_effect = RuntimeError("DB corrupted") + + result = mod.search_all_collections( + query_embedding=[0.1] * 384, + db_path=tmp_path / ".chroma", + ) + + assert result["success"] is False + assert "DB corrupted" in result["error"] + + def test_search_all_accepts_string_db_path(self, monkeypatch, tmp_path): + """String db_path is accepted and converted.""" + mod, mocks = _import_vector_search(monkeypatch) + + mocks["client"].list_collections.return_value = [] + + result = mod.search_all_collections( + query_embedding=[0.1] * 384, + db_path=str(tmp_path / ".chroma"), + ) + + assert result["success"] is True + + def test_search_all_skips_nonexistent_collections(self, monkeypatch, tmp_path): + """Collections that fail get_collection are excluded from results.""" + mod, mocks = _import_vector_search(monkeypatch) + + mock_coll_a = MagicMock() + mock_coll_a.name = "coll_a" + mocks["client"].list_collections.return_value = [mock_coll_a] + + # get_collection raises => query_collection returns exists=False + mocks["client"].get_collection.side_effect = ValueError("Collection not found") + + result = mod.search_all_collections( + query_embedding=[0.1] * 384, + db_path=tmp_path / ".chroma", + ) + + assert result["success"] is True + assert result["collections_searched"] == 0 + assert result["total_results"] == 0 diff --git a/src/aipass/memory/tests/test_symbolic_extras.py b/src/aipass/memory/tests/test_symbolic_extras.py new file mode 100644 index 00000000..789d944c --- /dev/null +++ b/src/aipass/memory/tests/test_symbolic_extras.py @@ -0,0 +1,1651 @@ +# =================== AIPass ==================== +# Name: test_symbolic_extras.py +# Description: Tests for symbolic handler public functions +# Version: 1.0.0 +# Created: 2026-04-25 +# Modified: 2026-04-25 +# ============================================= + +"""Tests for 23 untested public functions in symbolic handler files. + +Covers imports required by the seedgo test scanner: + from aipass.memory.apps.handlers.symbolic.hook import save_config + from aipass.memory.apps.handlers.symbolic.hook import extract_conversation_context + from aipass.memory.apps.handlers.symbolic.hook import find_relevant_fragments + from aipass.memory.apps.handlers.symbolic.hook import format_fragment_recall + from aipass.memory.apps.handlers.symbolic.hook import format_multiple_recalls + from aipass.memory.apps.handlers.symbolic.hook import should_surface_fragment + from aipass.memory.apps.handlers.symbolic.hook import record_surface + from aipass.memory.apps.handlers.symbolic.hook import record_message + from aipass.memory.apps.handlers.symbolic.hook import reset_session + from aipass.memory.apps.handlers.symbolic.hook import get_session_state + from aipass.memory.apps.handlers.symbolic.hook import process_hook + from aipass.memory.apps.handlers.symbolic.storage import flatten_dimensions + from aipass.memory.apps.handlers.symbolic.storage import store_fragment + from aipass.memory.apps.handlers.symbolic.storage import store_fragments_batch + from aipass.memory.apps.handlers.symbolic.storage import store_llm_fragment + from aipass.memory.apps.handlers.symbolic.storage import store_llm_fragments_batch + from aipass.memory.apps.handlers.symbolic.storage import delete_fragment + from aipass.memory.apps.handlers.symbolic.deduplicator import deduplicate_fragment + from aipass.memory.apps.handlers.symbolic.chroma_client import get_chroma_client + from aipass.memory.apps.handlers.symbolic.retriever import search_by_vector + from aipass.memory.apps.handlers.symbolic.retriever import search_by_dimensions + from aipass.memory.apps.handlers.symbolic.retriever import search_by_triggers + from aipass.memory.apps.handlers.symbolic.retriever import retrieve_fragments +""" + +import json +import sys +import time +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _mock_handler_deps(monkeypatch): + """Mock heavy handler-level imports used directly by the handler files. + + Builds on the conftest autouse ``_mock_infrastructure`` which already mocks + prax logger, json_handler, and trigger at the sys.modules level. + + Here we additionally mock: + - memory_files (used by hook.py save_config/load_config) + - embedder (used by storage.py and retriever.py) + - chromadb (used by chroma_client.py) + - api keys (used by deduplicator.py) + """ + # -- memory_files (used by hook) ---------------------------------------- + mock_memory_files = MagicMock() + mock_memory_files.write_memory_file = MagicMock(return_value={"success": True}) + mock_memory_files.read_memory_file = MagicMock( + return_value={"success": True, "data": {"enabled": True, "threshold": 0.3}} + ) + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.json.memory_files", + mock_memory_files, + ) + + # -- embedder (used by storage + retriever) ----------------------------- + mock_embedder = MagicMock() + mock_embedder.encode_batch = MagicMock(return_value={"success": True, "embeddings": [[0.1, 0.2, 0.3]]}) + vector_pkg = MagicMock() + vector_pkg.embedder = mock_embedder + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector", vector_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector.embedder", mock_embedder) + + # -- chromadb (used by chroma_client) ----------------------------------- + mock_chromadb = MagicMock() + monkeypatch.setitem(sys.modules, "chromadb", mock_chromadb) + + # -- api keys (used by deduplicator) ------------------------------------ + mock_keys = MagicMock() + mock_keys.get_api_key = MagicMock(return_value="test-api-key") + monkeypatch.setitem(sys.modules, "aipass.api", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.api.apps", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.api.apps.handlers", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.api.apps.handlers.auth", MagicMock()) + monkeypatch.setitem(sys.modules, "aipass.api.apps.handlers.auth.keys", mock_keys) + + # Force-clear cached handler modules so they re-import with mocks + for mod_name in [ + "aipass.memory.apps.handlers.symbolic.hook", + "aipass.memory.apps.handlers.symbolic.storage", + "aipass.memory.apps.handlers.symbolic.retriever", + "aipass.memory.apps.handlers.symbolic.deduplicator", + "aipass.memory.apps.handlers.symbolic.chroma_client", + ]: + monkeypatch.delitem(sys.modules, mod_name, raising=False) + + +@pytest.fixture(autouse=True) +def _reset_hook_session(): + """Reset hook SESSION_STATE before each test.""" + yield + # Post-test cleanup: if the module was loaded, reset its state + hook_mod = sys.modules.get("aipass.memory.apps.handlers.symbolic.hook") + if hook_mod and hasattr(hook_mod, "reset_session"): + hook_mod.reset_session() + + +@pytest.fixture(autouse=True) +def _reset_chroma_clients(): + """Clear the chroma_client singleton cache between tests.""" + yield + cc_mod = sys.modules.get("aipass.memory.apps.handlers.symbolic.chroma_client") + if cc_mod and hasattr(cc_mod, "_clients"): + cc_mod._clients.clear() + + +# --------------------------------------------------------------------------- +# Import helpers (must be called inside tests, after mocks are installed) +# --------------------------------------------------------------------------- + + +def _import_hook(): # noqa: D103 + sys.modules.pop("aipass.memory.apps.handlers.symbolic.hook", None) + from aipass.memory.apps.handlers.symbolic.hook import ( # noqa: E402 + extract_conversation_context, + find_relevant_fragments, + format_fragment_recall, + format_multiple_recalls, + get_session_state, + process_hook, + record_message, + record_surface, + reset_session, + save_config, + should_surface_fragment, + ) + + return { + "save_config": save_config, + "extract_conversation_context": extract_conversation_context, + "find_relevant_fragments": find_relevant_fragments, + "format_fragment_recall": format_fragment_recall, + "format_multiple_recalls": format_multiple_recalls, + "should_surface_fragment": should_surface_fragment, + "record_surface": record_surface, + "record_message": record_message, + "reset_session": reset_session, + "get_session_state": get_session_state, + "process_hook": process_hook, + } + + +def _import_storage(): # noqa: D103 + sys.modules.pop("aipass.memory.apps.handlers.symbolic.storage", None) + from aipass.memory.apps.handlers.symbolic.storage import ( # noqa: E402 + delete_fragment, + flatten_dimensions, + store_fragment, + store_fragments_batch, + store_llm_fragment, + store_llm_fragments_batch, + ) + + return { + "flatten_dimensions": flatten_dimensions, + "store_fragment": store_fragment, + "store_fragments_batch": store_fragments_batch, + "store_llm_fragment": store_llm_fragment, + "store_llm_fragments_batch": store_llm_fragments_batch, + "delete_fragment": delete_fragment, + } + + +def _import_deduplicator(): # noqa: D103 + sys.modules.pop("aipass.memory.apps.handlers.symbolic.deduplicator", None) + from aipass.memory.apps.handlers.symbolic.deduplicator import deduplicate_fragment # noqa: E402 + + return deduplicate_fragment + + +def _import_chroma_client(): # noqa: D103 + sys.modules.pop("aipass.memory.apps.handlers.symbolic.chroma_client", None) + from aipass.memory.apps.handlers.symbolic.chroma_client import get_chroma_client # noqa: E402 + + return get_chroma_client + + +def _import_retriever(): # noqa: D103 + sys.modules.pop("aipass.memory.apps.handlers.symbolic.retriever", None) + from aipass.memory.apps.handlers.symbolic.retriever import ( # noqa: E402 + retrieve_fragments, + search_by_dimensions, + search_by_triggers, + search_by_vector, + ) + + return { + "search_by_vector": search_by_vector, + "search_by_dimensions": search_by_dimensions, + "search_by_triggers": search_by_triggers, + "retrieve_fragments": retrieve_fragments, + } + + +# --------------------------------------------------------------------------- +# Sample data helpers +# --------------------------------------------------------------------------- + + +def _sample_messages(): # noqa: D103 + return [ + {"role": "user", "content": "I found an error in the module"}, + {"role": "assistant", "content": "Let me debug that issue for you"}, + ] + + +def _v1_fragment(): # noqa: D103 + return { + "id": "frag_20260401_120000_abcd1234", + "content": "Technical debugging session", + "dimensions": { + "technical": ["debugging_session"], + "emotional": ["frustration_to_breakthrough"], + "collaboration": ["pair_debugging"], + "learnings": ["edge_case_handling"], + "triggers": ["parser", "bug", "fix"], + }, + "metadata": { + "timestamp": "2026-04-01T12:00:00", + "message_count": 10, + "depth": "deep", + "total_words": 500, + "source_branch": "test", + }, + } + + +def _v2_fragment(): # noqa: D103 + return { + "id": "frag_20260402_130000_efgh5678", + "content": "LLM extracted fragment content", + "metadata": { + "schema_version": "v2", + "summary": "Debugged a parser edge case", + "insight": "Always check boundary conditions", + "type": "episodic", + "emotional_tone": "determined", + "technical_domain": "parsing", + "triggers": "parser,bug,edge", + "timestamp": "2026-04-02T13:00:00", + }, + } + + +def _mock_collection(): + """Create a mock ChromaDB collection with standard methods.""" + coll = MagicMock() + coll.count.return_value = 5 + coll.upsert = MagicMock() + coll.delete = MagicMock() + coll.get = MagicMock(return_value={"ids": [], "documents": [], "metadatas": []}) + coll.query = MagicMock( + return_value={ + "ids": [["id1"]], + "documents": [["doc content"]], + "metadatas": [[{"triggers": "parser,bug"}]], + "distances": [[0.25]], + } + ) + return coll + + +def _mock_chroma_client(collection=None): + """Create a mock ChromaDB client.""" + client = MagicMock() + coll = collection or _mock_collection() + client.get_or_create_collection.return_value = coll + client.get_collection.return_value = coll + return client + + +# =========================================================================== +# 1. hook.save_config +# =========================================================================== + + +class TestSaveConfig: + """Tests for hook.save_config.""" + + def test_saves_via_memory_files(self, tmp_path): + """Verify save_config delegates to memory_files.write_memory_file.""" + h = _import_hook() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + mock_mf = MagicMock() + mock_mf.write_memory_file = MagicMock(return_value={"success": True}) + setattr(hook_mod, "memory_files", mock_mf) # noqa: B010 + + config = {"enabled": True, "threshold": 0.5} + path = tmp_path / "config.json" + + result = h["save_config"](config, config_path=path) + + assert result["success"] is True + mock_mf.write_memory_file.assert_called_once_with(path, config) + + def test_creates_parent_directories(self, tmp_path): + """Verify save_config creates missing parent directories.""" + h = _import_hook() + path = tmp_path / "deep" / "nested" / "config.json" + + h["save_config"]({"enabled": True}, config_path=path) + + assert path.parent.exists() + + +# =========================================================================== +# 2. hook.extract_conversation_context +# =========================================================================== + + +class TestExtractConversationContext: + """Tests for hook.extract_conversation_context.""" + + def test_extracts_keywords_from_messages(self): + """Verify keywords are extracted from message content.""" + h = _import_hook() + result = h["extract_conversation_context"](_sample_messages()) + + assert result["success"] is True + assert isinstance(result["keywords"], list) + assert "error" in result["keywords"] or "debug" in result["keywords"] + assert result["analyzed_messages"] == 2 + + def test_empty_messages_returns_neutral(self): + """Verify empty messages produce neutral defaults.""" + h = _import_hook() + result = h["extract_conversation_context"]([]) + + assert result["success"] is True + assert result["keywords"] == [] + assert result["mood"] == "neutral" + + def test_max_messages_limits_analysis(self): + """Verify max_messages parameter limits analyzed messages.""" + h = _import_hook() + msgs = [{"role": "user", "content": f"msg {i}"} for i in range(20)] + result = h["extract_conversation_context"](msgs, max_messages=3) + + assert result["analyzed_messages"] == 3 + + def test_mood_detection(self): + """Verify frustrated mood is detected from content.""" + h = _import_hook() + msgs = [{"role": "user", "content": "I'm frustrated and stuck on this ugh"}] + result = h["extract_conversation_context"](msgs) + + assert result["mood"] == "frustrated" + + def test_theme_extraction(self): + """Verify debugging theme is extracted from content.""" + h = _import_hook() + msgs = [{"role": "user", "content": "debug the error and fix the trace bug"}] + result = h["extract_conversation_context"](msgs) + + assert "debugging" in result["themes"] + + +# =========================================================================== +# 3. hook.find_relevant_fragments +# =========================================================================== + + +class TestFindRelevantFragments: + """Tests for hook.find_relevant_fragments.""" + + def test_returns_empty_when_no_context(self): + """Verify empty context returns no fragments.""" + h = _import_hook() + context = {"keywords": [], "mood": "neutral", "themes": []} + + result = h["find_relevant_fragments"](context) + + assert result["success"] is True + assert result["fragments"] == [] + + def test_calls_retriever_with_query(self): + """Verify retriever is called with extracted context.""" + h = _import_hook() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + mock_retriever = MagicMock() + mock_retriever.retrieve_fragments = MagicMock( + return_value={ + "success": True, + "results": [{"id": "frag1", "relevance_score": 0.8, "content": "test"}], + } + ) + setattr(hook_mod, "retriever", mock_retriever) # noqa: B010 + + context = { + "keywords": ["error", "debug"], + "mood": "frustrated", + "themes": ["debugging"], + } + result = h["find_relevant_fragments"](context) + + assert result["success"] is True + mock_retriever.retrieve_fragments.assert_called_once() + + def test_filters_below_threshold(self): + """Verify fragments below threshold are filtered out.""" + h = _import_hook() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + mock_retriever = MagicMock() + mock_retriever.retrieve_fragments = MagicMock( + return_value={ + "success": True, + "results": [ + {"id": "frag1", "relevance_score": 0.1, "content": "low score"}, + ], + } + ) + setattr(hook_mod, "retriever", mock_retriever) # noqa: B010 + + context = {"keywords": ["error"], "mood": "neutral", "themes": ["debugging"]} + result = h["find_relevant_fragments"](context) + + assert result["success"] is True + assert len(result["fragments"]) == 0 + + +# =========================================================================== +# 4. hook.format_fragment_recall +# =========================================================================== + + +class TestFormatFragmentRecall: + """Tests for hook.format_fragment_recall.""" + + def test_v1_fragment_basic_format(self): + """Verify v1 fragments format with dimension metadata.""" + h = _import_hook() + frag = { + "content": "Short content", + "metadata": { + "emotional_0": "frustration_to_breakthrough", + "technical_0": "debugging_session", + "learnings_0": "edge_case_handling", + }, + } + result = h["format_fragment_recall"](frag) + + assert "frustration" in result + assert "debugging session" in result + assert "edge case handling" in result + + def test_v2_fragment_episodic_format(self): + """Verify v2 episodic fragments use correct prefix.""" + h = _import_hook() + frag = { + "content": "test", + "metadata": { + "schema_version": "v2", + "summary": "Debugged parser", + "insight": "Check boundaries", + "type": "episodic", + }, + } + result = h["format_fragment_recall"](frag) + + assert "During a session" in result + assert "Debugged parser" in result + assert "Check boundaries" in result + + def test_v2_procedural_type(self): + """Verify v2 procedural fragments use learned-how-to prefix.""" + h = _import_hook() + frag = { + "content": "", + "metadata": { + "schema_version": "v2", + "summary": "Use pathlib for paths", + "insight": "", + "type": "procedural", + }, + } + result = h["format_fragment_recall"](frag) + + assert "We learned how to:" in result + + def test_v2_no_insight_ends_with_period(self): + """Verify v2 recall without insight ends with period.""" + h = _import_hook() + frag = { + "content": "", + "metadata": { + "schema_version": "v2", + "summary": "Something happened", + "insight": "", + "type": "semantic", + }, + } + result = h["format_fragment_recall"](frag) + + assert result.endswith(".") + + def test_v1_fragment_no_metadata(self): + """Verify v1 fragments with empty metadata use default text.""" + h = _import_hook() + frag = {"content": "Short", "metadata": {}} + result = h["format_fragment_recall"](frag) + + assert "This reminds me of a past conversation" in result + + +# =========================================================================== +# 5. hook.format_multiple_recalls +# =========================================================================== + + +class TestFormatMultipleRecalls: + """Tests for hook.format_multiple_recalls.""" + + def test_empty_list_returns_empty_string(self): + """Verify empty list produces empty string.""" + h = _import_hook() + assert h["format_multiple_recalls"]([]) == "" + + def test_single_fragment(self): + """Verify single fragment includes schema tag.""" + h = _import_hook() + frag = { + "content": "Test", + "metadata": { + "schema_version": "v2", + "summary": "Hello", + "insight": "", + "type": "", + }, + } + result = h["format_multiple_recalls"]([frag]) + + assert "[v2]" in result + + def test_multiple_fragments_separated_by_dividers(self): + """Verify multiple fragments are separated by dividers.""" + h = _import_hook() + frags = [ + {"content": "A", "metadata": {}}, + { + "content": "B", + "metadata": { + "schema_version": "v2", + "summary": "B frag", + "insight": "", + "type": "", + }, + }, + ] + result = h["format_multiple_recalls"](frags) + + assert "---" in result + assert "[v1]" in result + assert "[v2]" in result + + +# =========================================================================== +# 6. hook.should_surface_fragment +# =========================================================================== + + +class TestShouldSurfaceFragment: + """Tests for hook.should_surface_fragment.""" + + def test_disabled_hook(self): + """Verify disabled hook returns False.""" + h = _import_hook() + h["reset_session"]() + can, reason = h["should_surface_fragment"](config={"enabled": False}) + + assert can is False + assert "disabled" in reason + + def test_max_fragments_reached(self): + """Verify max fragments per session blocks surfacing.""" + h = _import_hook() + h["reset_session"]() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + hook_mod.SESSION_STATE["fragments_surfaced"] = 5 + + can, reason = h["should_surface_fragment"]( + config={ + "enabled": True, + "max_fragments_per_session": 5, + "min_messages_between": 0, + "cooldown_seconds": 0, + } + ) + assert can is False + assert "Max fragments" in reason + + def test_not_enough_messages(self): + """Verify insufficient messages blocks surfacing.""" + h = _import_hook() + h["reset_session"]() + can, reason = h["should_surface_fragment"]( + config={ + "enabled": True, + "max_fragments_per_session": 10, + "min_messages_between": 5, + "cooldown_seconds": 0, + } + ) + assert can is False + assert "messages since last" in reason + + def test_cooldown_active(self): + """Verify active cooldown blocks surfacing.""" + h = _import_hook() + h["reset_session"]() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + hook_mod.SESSION_STATE["messages_since_last"] = 100 + hook_mod.SESSION_STATE["last_surface_time"] = time.time() + + can, reason = h["should_surface_fragment"]( + config={ + "enabled": True, + "max_fragments_per_session": 10, + "min_messages_between": 0, + "cooldown_seconds": 600, + } + ) + assert can is False + assert "Cooldown" in reason + + def test_already_surfaced_fragment(self): + """Verify duplicate fragment is blocked.""" + h = _import_hook() + h["reset_session"]() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + hook_mod.SESSION_STATE["messages_since_last"] = 100 + hook_mod.SESSION_STATE["surfaced_ids"].add("frag-001") + + can, reason = h["should_surface_fragment"]( + fragment={"id": "frag-001"}, + config={ + "enabled": True, + "max_fragments_per_session": 10, + "min_messages_between": 0, + "cooldown_seconds": 0, + }, + ) + assert can is False + assert "already surfaced" in reason + + def test_ready_to_surface(self): + """Verify all conditions met returns True.""" + h = _import_hook() + h["reset_session"]() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + hook_mod.SESSION_STATE["messages_since_last"] = 100 + + can, reason = h["should_surface_fragment"]( + config={ + "enabled": True, + "max_fragments_per_session": 10, + "min_messages_between": 0, + "cooldown_seconds": 0, + } + ) + assert can is True + assert "Ready" in reason + + +# =========================================================================== +# 7. hook.record_surface +# =========================================================================== + + +class TestRecordSurface: + """Tests for hook.record_surface.""" + + def test_increments_counters(self): + """Verify surface recording updates all counters.""" + h = _import_hook() + h["reset_session"]() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + hook_mod.SESSION_STATE["messages_since_last"] = 15 + + h["record_surface"]({"id": "frag-001"}) + + state = h["get_session_state"]() + assert state["fragments_surfaced"] == 1 + assert state["messages_since_last"] == 0 + assert state["surfaced_count"] == 1 + + def test_records_fragment_without_id(self): + """Verify fragment without id does not raise.""" + h = _import_hook() + h["reset_session"]() + h["record_surface"]({"content": "no id here"}) + + state = h["get_session_state"]() + assert state["fragments_surfaced"] == 1 + assert state["surfaced_count"] == 0 + + +# =========================================================================== +# 8. hook.record_message +# =========================================================================== + + +class TestRecordMessage: + """Tests for hook.record_message.""" + + def test_increments_counter(self): + """Verify message counter increments correctly.""" + h = _import_hook() + h["reset_session"]() + + h["record_message"]() + h["record_message"]() + h["record_message"]() + + state = h["get_session_state"]() + assert state["messages_since_last"] == 3 + + +# =========================================================================== +# 9. hook.reset_session +# =========================================================================== + + +class TestResetSession: + """Tests for hook.reset_session.""" + + def test_clears_all_state(self): + """Verify reset clears all session state fields.""" + h = _import_hook() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + + hook_mod.SESSION_STATE["fragments_surfaced"] = 99 + hook_mod.SESSION_STATE["messages_since_last"] = 99 + hook_mod.SESSION_STATE["last_surface_time"] = 99999 + hook_mod.SESSION_STATE["surfaced_ids"] = {"a", "b", "c"} + + h["reset_session"]() + + state = h["get_session_state"]() + assert state["fragments_surfaced"] == 0 + assert state["messages_since_last"] == 0 + assert state["last_surface_time"] == 0 + assert state["surfaced_count"] == 0 + + +# =========================================================================== +# 10. hook.get_session_state +# =========================================================================== + + +class TestGetSessionState: + """Tests for hook.get_session_state.""" + + def test_returns_dict_with_expected_keys(self): + """Verify returned dict contains all expected keys.""" + h = _import_hook() + h["reset_session"]() + + state = h["get_session_state"]() + + assert "fragments_surfaced" in state + assert "messages_since_last" in state + assert "last_surface_time" in state + assert "surfaced_count" in state + + +# =========================================================================== +# 11. hook.process_hook +# =========================================================================== + + +class TestProcessHook: + """Tests for hook.process_hook.""" + + def test_returns_not_surfaced_when_blocked(self): + """Verify blocked state returns surfaced=False.""" + h = _import_hook() + h["reset_session"]() + result = h["process_hook"](_sample_messages()) + + assert result["success"] is True + assert result["surfaced"] is False + + def test_returns_not_surfaced_when_disabled(self): + """Verify disabled config returns surfaced=False.""" + h = _import_hook() + h["reset_session"]() + result = h["process_hook"](_sample_messages(), config={"enabled": False}) + + assert result["success"] is True + assert result["surfaced"] is False + assert "disabled" in result["reason"] + + def test_surfaces_when_conditions_met(self): + """Verify full pipeline surfaces a fragment when conditions are met.""" + h = _import_hook() + h["reset_session"]() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + hook_mod.SESSION_STATE["messages_since_last"] = 100 + + mock_retriever = MagicMock() + mock_retriever.retrieve_fragments = MagicMock( + return_value={ + "success": True, + "results": [ + { + "id": "frag-surface-1", + "content": "Past debugging session", + "metadata": {}, + "relevance_score": 0.9, + } + ], + } + ) + setattr(hook_mod, "retriever", mock_retriever) # noqa: B010 + + config = { + "enabled": True, + "threshold": 0.3, + "max_fragments_per_session": 10, + "min_messages_between": 0, + "cooldown_seconds": 0, + } + + result = h["process_hook"](_sample_messages(), config=config) + + assert result["success"] is True + assert result["surfaced"] is True + assert "recall" in result + + def test_no_fragments_above_threshold(self): + """Verify low-scoring fragments are not surfaced.""" + h = _import_hook() + h["reset_session"]() + hook_mod = sys.modules["aipass.memory.apps.handlers.symbolic.hook"] + hook_mod.SESSION_STATE["messages_since_last"] = 100 + + mock_retriever = MagicMock() + mock_retriever.retrieve_fragments = MagicMock( + return_value={ + "success": True, + "results": [ + { + "id": "frag-low", + "content": "low", + "metadata": {}, + "relevance_score": 0.05, + } + ], + } + ) + setattr(hook_mod, "retriever", mock_retriever) # noqa: B010 + + config = { + "enabled": True, + "threshold": 0.3, + "max_fragments_per_session": 10, + "min_messages_between": 0, + "cooldown_seconds": 0, + } + + result = h["process_hook"](_sample_messages(), config=config) + + assert result["success"] is True + assert result["surfaced"] is False + + +# =========================================================================== +# 12. storage.flatten_dimensions +# =========================================================================== + + +class TestFlattenDimensions: + """Tests for storage.flatten_dimensions.""" + + def test_flattens_dimension_lists(self): + """Verify nested dimensions are flattened to indexed keys.""" + s = _import_storage() + frag = _v1_fragment() + + result = s["flatten_dimensions"](frag) + + assert result["success"] is True + meta = result["metadata"] + assert meta["technical_0"] == "debugging_session" + assert meta["emotional_0"] == "frustration_to_breakthrough" + assert "parser,bug,fix" in meta["triggers"] + + def test_empty_fragment_fails(self): + """Verify None fragment returns failure.""" + s = _import_storage() + result = s["flatten_dimensions"](None) + + assert result["success"] is False + + def test_includes_metadata_fields(self): + """Verify metadata fields are preserved in flat output.""" + s = _import_storage() + frag = _v1_fragment() + + result = s["flatten_dimensions"](frag) + meta = result["metadata"] + + assert meta["timestamp"] == "2026-04-01T12:00:00" + assert meta["message_count"] == 10 + assert meta["depth"] == "deep" + assert meta["source_branch"] == "test" + + def test_limits_to_5_per_dimension(self): + """Verify dimensions are limited to 5 values.""" + s = _import_storage() + frag = { + "dimensions": { + "technical": [f"tech_{i}" for i in range(10)], + "triggers": [], + }, + "metadata": {}, + } + result = s["flatten_dimensions"](frag) + meta = result["metadata"] + + assert "technical_4" in meta + assert "technical_5" not in meta + + +# =========================================================================== +# 13. storage.store_fragment +# =========================================================================== + + +class TestStoreFragment: + """Tests for storage.store_fragment.""" + + def test_stores_valid_fragment(self): + """Verify valid fragment is stored successfully.""" + s = _import_storage() + client = _mock_chroma_client() + + with patch( + "aipass.memory.apps.handlers.symbolic.chroma_client.get_client", + return_value=client, + ): + result = s["store_fragment"](_v1_fragment()) + + assert result["success"] is True + assert "fragment_id" in result + + def test_fails_on_empty_fragment(self): + """Verify None fragment returns failure.""" + s = _import_storage() + result = s["store_fragment"](None) + + assert result["success"] is False + + def test_fails_on_missing_content(self): + """Verify fragment without content returns failure.""" + s = _import_storage() + result = s["store_fragment"]({"id": "frag-1", "content": ""}) + + assert result["success"] is False + + def test_handles_embedding_failure(self): + """Verify embedding failure is reported correctly.""" + s = _import_storage() + embedder_mod = sys.modules["aipass.memory.apps.handlers.vector.embedder"] + setattr( + embedder_mod, + "encode_batch", + MagicMock(return_value={"success": False, "error": "model not loaded"}), + ) + + result = s["store_fragment"](_v1_fragment()) + + assert result["success"] is False + assert "Embedding failed" in result["error"] + + +# =========================================================================== +# 14. storage.store_fragments_batch +# =========================================================================== + + +class TestStoreFragmentsBatch: + """Tests for storage.store_fragments_batch.""" + + def test_stores_multiple_fragments(self): + """Verify batch storage stores all valid fragments.""" + s = _import_storage() + client = _mock_chroma_client() + embedder_mod = sys.modules["aipass.memory.apps.handlers.vector.embedder"] + setattr( + embedder_mod, + "encode_batch", + MagicMock( + return_value={ + "success": True, + "embeddings": [[0.1, 0.2], [0.3, 0.4]], + } + ), + ) + + frags = [_v1_fragment(), _v1_fragment()] + frags[1]["id"] = "frag_20260401_120001_wxyz9012" + + with patch( + "aipass.memory.apps.handlers.symbolic.chroma_client.get_client", + return_value=client, + ): + result = s["store_fragments_batch"](frags) + + assert result["success"] is True + assert result["stored"] == 2 + + def test_empty_list_returns_zero(self): + """Verify empty list returns zero stored.""" + s = _import_storage() + result = s["store_fragments_batch"]([]) + + assert result["success"] is True + assert result["stored"] == 0 + + def test_fails_on_embedding_count_mismatch(self): + """Verify embedding count mismatch is caught.""" + s = _import_storage() + embedder_mod = sys.modules["aipass.memory.apps.handlers.vector.embedder"] + setattr( + embedder_mod, + "encode_batch", + MagicMock(return_value={"success": True, "embeddings": [[0.1]]}), + ) + + frags = [_v1_fragment(), _v1_fragment()] + frags[1]["id"] = "frag_20260401_120001_wxyz9012" + + result = s["store_fragments_batch"](frags) + + assert result["success"] is False + assert "mismatch" in result["error"] + + +# =========================================================================== +# 15. storage.store_llm_fragment +# =========================================================================== + + +class TestStoreLlmFragment: + """Tests for storage.store_llm_fragment.""" + + def test_stores_v2_fragment(self): + """Verify v2 LLM fragment is stored successfully.""" + s = _import_storage() + client = _mock_chroma_client() + + llm_frag = { + "summary": "Debugged parser edge case", + "insight": "Check boundary conditions", + "type": "episodic", + "emotional_tone": "determined", + "technical_domain": "parsing", + "triggers": ["parser", "boundary"], + } + + with patch( + "aipass.memory.apps.handlers.symbolic.chroma_client.get_client", + return_value=client, + ): + result = s["store_llm_fragment"](llm_frag, source_branch="memory") + + assert result["success"] is True + assert "fragment_id" in result + + def test_fails_on_empty_fragment(self): + """Verify None fragment returns failure.""" + s = _import_storage() + result = s["store_llm_fragment"](None) + + assert result["success"] is False + + def test_fails_on_missing_summary(self): + """Verify fragment without summary returns failure.""" + s = _import_storage() + result = s["store_llm_fragment"]({"insight": "no summary"}) + + assert result["success"] is False + assert "missing summary" in result["error"] + + +# =========================================================================== +# 16. storage.store_llm_fragments_batch +# =========================================================================== + + +class TestStoreLlmFragmentsBatch: + """Tests for storage.store_llm_fragments_batch.""" + + def test_stores_multiple_v2_fragments(self): + """Verify batch v2 storage stores all valid fragments.""" + s = _import_storage() + client = _mock_chroma_client() + embedder_mod = sys.modules["aipass.memory.apps.handlers.vector.embedder"] + setattr( + embedder_mod, + "encode_batch", + MagicMock(return_value={"success": True, "embeddings": [[0.1], [0.2]]}), + ) + + frags = [ + { + "summary": "First memory", + "insight": "insight A", + "type": "episodic", + "triggers": ["test"], + }, + { + "summary": "Second memory", + "insight": "", + "type": "semantic", + "triggers": [], + }, + ] + + with patch( + "aipass.memory.apps.handlers.symbolic.chroma_client.get_client", + return_value=client, + ): + result = s["store_llm_fragments_batch"](frags, source_branch="test") + + assert result["success"] is True + assert result["stored"] == 2 + + def test_empty_list_returns_zero(self): + """Verify empty list returns zero stored.""" + s = _import_storage() + result = s["store_llm_fragments_batch"]([]) + + assert result["success"] is True + assert result["stored"] == 0 + + def test_skips_fragments_without_summary(self): + """Verify fragments without summary are rejected.""" + s = _import_storage() + frags = [{"insight": "no summary here"}, {"insight": "also no summary"}] + result = s["store_llm_fragments_batch"](frags) + + assert result["success"] is False + assert "No valid" in result["error"] + + +# =========================================================================== +# 17. storage.delete_fragment +# =========================================================================== + + +class TestDeleteFragment: + """Tests for storage.delete_fragment.""" + + def test_deletes_by_id(self): + """Verify fragment is deleted by ID.""" + s = _import_storage() + client = _mock_chroma_client() + + with patch( + "aipass.memory.apps.handlers.symbolic.chroma_client.get_client", + return_value=client, + ): + result = s["delete_fragment"]("frag-to-delete") + + assert result["success"] is True + assert result["deleted_id"] == "frag-to-delete" + client.get_or_create_collection.return_value.delete.assert_called_once_with(ids=["frag-to-delete"]) + + def test_handles_chroma_exception(self): + """Verify ChromaDB exception is caught gracefully.""" + s = _import_storage() + client = _mock_chroma_client() + client.get_or_create_collection.side_effect = RuntimeError("db locked") + + with patch( + "aipass.memory.apps.handlers.symbolic.chroma_client.get_client", + return_value=client, + ): + result = s["delete_fragment"]("frag-missing") + + assert result["success"] is False + assert "failed" in result["error"].lower() + + +# =========================================================================== +# 18. deduplicator.deduplicate_fragment +# =========================================================================== + + +class TestDeduplicateFragment: + """Tests for deduplicator.deduplicate_fragment.""" + + def test_add_when_no_existing(self): + """Verify ADD action when no existing fragments.""" + deduplicate = _import_deduplicator() + new_frag = { + "summary": "New insight", + "insight": "Fresh", + "type": "episodic", + "triggers": [], + } + + result = deduplicate(new_frag, []) + + assert result["success"] is True + assert result["action"] == "ADD" + + def test_noop_on_empty_new_fragment(self): + """Verify NOOP action when new fragment is empty.""" + deduplicate = _import_deduplicator() + empty_frag: dict = {} # type: ignore[assignment] + result = deduplicate(empty_frag, []) + + assert result["action"] == "NOOP" + + def test_llm_returns_update_action(self): + """Verify UPDATE action merges fragment content.""" + deduplicate = _import_deduplicator() + + llm_response = json.dumps( + { + "action": "UPDATE", + "merged_summary": "Combined summary", + "merged_insight": "Combined insight", + "delete_id": "", + "reason": "Overlapping content", + } + ) + + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps({"choices": [{"message": {"content": llm_response}}]}).encode("utf-8") + mock_resp.__enter__ = MagicMock(return_value=mock_resp) + mock_resp.__exit__ = MagicMock(return_value=False) + + new_frag = { + "summary": "New", + "insight": "Fresh", + "type": "episodic", + "triggers": [], + } + existing = [ + { + "id": "old-1", + "content": "Old content", + "metadata": {"summary": "Old"}, + } + ] + + with patch("urllib.request.urlopen", return_value=mock_resp): + result = deduplicate(new_frag, existing) + + assert result["success"] is True + assert result["action"] == "UPDATE" + assert result["fragment"]["summary"] == "Combined summary" + + def test_falls_back_to_add_on_api_error(self): + """Verify ADD fallback on API error.""" + deduplicate = _import_deduplicator() + + new_frag = { + "summary": "New", + "insight": "", + "type": "episodic", + "triggers": [], + } + existing = [{"id": "old-1", "content": "Old content", "metadata": {}}] + + import urllib.error + + with patch( + "urllib.request.urlopen", + side_effect=urllib.error.URLError("connection refused"), + ): + result = deduplicate(new_frag, existing) + + assert result["success"] is True + assert result["action"] == "ADD" + assert "failed" in result["reason"].lower() + + def test_falls_back_to_add_when_no_api_key(self): + """Verify ADD fallback when API key is missing.""" + deduplicate = _import_deduplicator() + + keys_mod = sys.modules["aipass.api.apps.handlers.auth.keys"] + setattr(keys_mod, "get_api_key", MagicMock(return_value=None)) # noqa: B010 + + new_frag = { + "summary": "New", + "insight": "", + "type": "episodic", + "triggers": [], + } + existing = [{"id": "old-1", "content": "Old", "metadata": {}}] + + result = deduplicate(new_frag, existing) + + assert result["success"] is True + assert result["action"] == "ADD" + assert "key" in result["reason"].lower() or "unavailable" in result["reason"].lower() + + +# =========================================================================== +# 19. chroma_client.get_chroma_client +# =========================================================================== + + +class TestGetChromaClient: + """Tests for chroma_client.get_chroma_client.""" + + def test_returns_persistent_client(self, tmp_path): + """Verify PersistentClient is created and returned.""" + get_chroma_client = _import_chroma_client() + chromadb_mod = sys.modules["chromadb"] + mock_client = MagicMock() + setattr(chromadb_mod, "PersistentClient", MagicMock(return_value=mock_client)) # noqa: B010 + + db_path = tmp_path / "test_chroma" + result = get_chroma_client(db_path) + + assert result == mock_client + chromadb_mod.PersistentClient.assert_called_once() + + def test_caches_client_by_path(self, tmp_path): + """Verify same path returns cached client.""" + get_chroma_client = _import_chroma_client() + chromadb_mod = sys.modules["chromadb"] + mock_client = MagicMock() + setattr(chromadb_mod, "PersistentClient", MagicMock(return_value=mock_client)) # noqa: B010 + + db_path = tmp_path / "cached_chroma" + client1 = get_chroma_client(db_path) + client2 = get_chroma_client(db_path) + + assert client1 is client2 + assert chromadb_mod.PersistentClient.call_count == 1 + + def test_accepts_string_path(self, tmp_path): + """Verify string path is accepted and converted.""" + get_chroma_client = _import_chroma_client() + chromadb_mod = sys.modules["chromadb"] + mock_client = MagicMock() + setattr(chromadb_mod, "PersistentClient", MagicMock(return_value=mock_client)) # noqa: B010 + + result = get_chroma_client(str(tmp_path / "str_chroma")) + + assert result == mock_client + + +# =========================================================================== +# 20. retriever.search_by_vector +# =========================================================================== + + +class TestSearchByVector: + """Tests for retriever.search_by_vector.""" + + def test_returns_results(self): + """Verify vector search returns formatted results.""" + r = _import_retriever() + client = _mock_chroma_client() + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["search_by_vector"]("debugging session") + + assert result["success"] is True + assert result["search_type"] == "vector" + assert len(result["results"]) > 0 + + def test_empty_query_fails(self): + """Verify empty query returns failure.""" + r = _import_retriever() + result = r["search_by_vector"]("") + + assert result["success"] is False + assert "Query" in result["error"] or "required" in result["error"] + + def test_handles_missing_collection(self): + """Verify missing collection returns empty results.""" + r = _import_retriever() + client = MagicMock() + client.get_collection.side_effect = Exception("collection not found") + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["search_by_vector"]("test query") + + assert result["success"] is True + assert result["results"] == [] + + +# =========================================================================== +# 21. retriever.search_by_dimensions +# =========================================================================== + + +class TestSearchByDimensions: + """Tests for retriever.search_by_dimensions.""" + + def test_filters_by_dimension(self): + """Verify dimension filter returns matching fragments.""" + r = _import_retriever() + coll = _mock_collection() + coll.get.return_value = { + "ids": ["frag-1"], + "documents": ["content here"], + "metadatas": [{"emotional_0": "frustration"}], + } + client = _mock_chroma_client(coll) + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["search_by_dimensions"]({"emotional_0": "frustration"}) + + assert result["success"] is True + assert result["search_type"] == "dimension_filter" + assert len(result["results"]) == 1 + + def test_empty_filters_fails(self): + """Verify empty filters returns failure.""" + r = _import_retriever() + result = r["search_by_dimensions"]({}) + + assert result["success"] is False + + def test_handles_missing_collection(self): + """Verify missing collection returns empty results.""" + r = _import_retriever() + client = MagicMock() + client.get_collection.side_effect = Exception("not found") + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["search_by_dimensions"]({"emotional_0": "curiosity"}) + + assert result["success"] is True + assert result["results"] == [] + + +# =========================================================================== +# 22. retriever.search_by_triggers +# =========================================================================== + + +class TestSearchByTriggers: + """Tests for retriever.search_by_triggers.""" + + def test_finds_matching_triggers(self): + """Verify trigger search finds matching fragments.""" + r = _import_retriever() + coll = _mock_collection() + coll.get.return_value = { + "ids": ["frag-1", "frag-2"], + "documents": ["doc1", "doc2"], + "metadatas": [ + {"triggers": "parser,bug,fix"}, + {"triggers": "deploy,release"}, + ], + } + client = _mock_chroma_client(coll) + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["search_by_triggers"](["parser"]) + + assert result["success"] is True + assert result["search_type"] == "trigger_keywords" + assert len(result["results"]) == 1 + + def test_empty_keywords_fails(self): + """Verify empty keywords returns failure.""" + r = _import_retriever() + result = r["search_by_triggers"]([]) + + assert result["success"] is False + + def test_case_insensitive_matching(self): + """Verify trigger matching is case-insensitive.""" + r = _import_retriever() + coll = _mock_collection() + coll.get.return_value = { + "ids": ["frag-1"], + "documents": ["doc1"], + "metadatas": [{"triggers": "Parser,BUG"}], + } + client = _mock_chroma_client(coll) + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["search_by_triggers"](["PARSER"]) + + assert result["success"] is True + assert len(result["results"]) == 1 + + +# =========================================================================== +# 23. retriever.retrieve_fragments +# =========================================================================== + + +class TestRetrieveFragments: + """Tests for retriever.retrieve_fragments.""" + + def test_combined_search(self): + """Verify combined search uses multiple methods.""" + r = _import_retriever() + client = _mock_chroma_client() + coll = client.get_collection.return_value + coll.get.return_value = { + "ids": ["frag-1"], + "documents": ["doc1"], + "metadatas": [{"triggers": "parser,bug"}], + } + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["retrieve_fragments"](query="debugging", trigger_keywords=["parser"]) + + assert result["success"] is True + assert len(result["search_methods"]) >= 1 + + def test_no_search_params_fails(self): + """Verify no search params returns failure.""" + r = _import_retriever() + result = r["retrieve_fragments"]() + + assert result["success"] is False + assert "At least one" in result["error"] + + def test_query_only_uses_vector(self): + """Verify query-only search uses vector method.""" + r = _import_retriever() + client = _mock_chroma_client() + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["retrieve_fragments"](query="test query") + + assert result["success"] is True + assert "vector" in result["search_methods"] + + def test_deduplicates_across_methods(self): + """Verify results are deduplicated across search methods.""" + r = _import_retriever() + coll = _mock_collection() + coll.query.return_value = { + "ids": [["frag-1"]], + "documents": [["doc1"]], + "metadatas": [[{"triggers": "parser"}]], + "distances": [[0.2]], + } + coll.get.return_value = { + "ids": ["frag-1"], + "documents": ["doc1"], + "metadatas": [{"triggers": "parser"}], + } + client = _mock_chroma_client(coll) + + with patch( + "aipass.memory.apps.handlers.symbolic.retriever.get_chroma_client", + return_value=client, + ): + result = r["retrieve_fragments"](query="parser", trigger_keywords=["parser"]) + + assert result["success"] is True + frag_ids = [res["id"] for res in result["results"]] + assert frag_ids.count("frag-1") == 1 + if result["results"]: + assert result["results"][0].get("relevance_score", 0) > 0 + + +# ============================================================================= +# analyze_conversation_llm (extractor.py) +# ============================================================================= + + +class TestAnalyzeConversationLlm: + """Tests for extractor.analyze_conversation_llm().""" + + def test_empty_history_returns_empty(self): + from aipass.memory.apps.handlers.symbolic.extractor import analyze_conversation_llm + + result = analyze_conversation_llm([]) + assert result["success"] is True + assert result["fragments"] == [] + assert result["message_count"] == 0 + + def test_merges_llm_and_regex_results(self, monkeypatch): + from aipass.memory.apps.handlers.symbolic import extractor + + monkeypatch.setattr( + extractor, + "extract_fragments_llm", + lambda history: { + "success": True, + "fragments": [{"summary": "test frag"}], + "chunk_count": 1, + }, + ) + monkeypatch.setattr( + extractor, + "analyze_conversation", + lambda history: { + "metadata": {"timestamp": "2026-01-01", "total_chars": 100, "total_words": 20, "depth": "deep"}, + "dimensions": {"technical": ["coding"]}, + "message_count": 2, + }, + ) + + result = extractor.analyze_conversation_llm([{"role": "user", "content": "hello"}]) + assert result["success"] is True + assert len(result["fragments"]) == 1 + assert result["metadata"]["depth"] == "deep" + assert result["message_count"] == 2 diff --git a/src/aipass/memory/tests/test_symbolic_module.py b/src/aipass/memory/tests/test_symbolic_module.py new file mode 100644 index 00000000..3e9c101d --- /dev/null +++ b/src/aipass/memory/tests/test_symbolic_module.py @@ -0,0 +1,1243 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_symbolic_module.py +# Date: 2026-04-25 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for untested public functions in apps/modules/symbolic.py. + +Covers the 25 public functions listed below. The module under test is a thin +delegation layer: most functions forward to handler sub-modules. We mock those +handlers so tests stay lightweight with no live ChromaDB / filesystem / API. + +Thin wrappers: + from aipass.memory.apps.modules.symbolic import analyze_conversation + from aipass.memory.apps.modules.symbolic import store_fragment + from aipass.memory.apps.modules.symbolic import store_fragments_batch + from aipass.memory.apps.modules.symbolic import flatten_dimensions + from aipass.memory.apps.modules.symbolic import store_llm_fragment + from aipass.memory.apps.modules.symbolic import store_llm_fragments_batch + from aipass.memory.apps.modules.symbolic import deduplicate_fragment + from aipass.memory.apps.modules.symbolic import extract_and_store_llm + from aipass.memory.apps.modules.symbolic import retrieve_fragments + from aipass.memory.apps.modules.symbolic import search_fragments_by_vector + from aipass.memory.apps.modules.symbolic import search_fragments_by_dimensions + from aipass.memory.apps.modules.symbolic import search_fragments_by_triggers + from aipass.memory.apps.modules.symbolic import extract_conversation_context + from aipass.memory.apps.modules.symbolic import find_relevant_fragments + from aipass.memory.apps.modules.symbolic import format_fragment_recall + from aipass.memory.apps.modules.symbolic import should_surface_fragment + from aipass.memory.apps.modules.symbolic import process_hook + from aipass.memory.apps.modules.symbolic import load_hook_config + from aipass.memory.apps.modules.symbolic import reset_hook_session + from aipass.memory.apps.modules.symbolic import get_hook_session_state + +CLI/display: + from aipass.memory.apps.modules.symbolic import run_demo + from aipass.memory.apps.modules.symbolic import search_fragments_cli + from aipass.memory.apps.modules.symbolic import run_hook_test + from aipass.memory.apps.modules.symbolic import analyze_file + from aipass.memory.apps.modules.symbolic import bootstrap_from_jsonl +""" + +import sys +import types +import pytest +from pathlib import Path +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(), + hook=MagicMock(), + deduplicator=MagicMock(), + trigger=MagicMock(), + console=MagicMock(), + header=MagicMock(), + error_fn=MagicMock(), + warning_fn=MagicMock(), + json_handler=MagicMock(), + memory_files=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 ------------------------------------------------ + mock_console = MagicMock() + mock_header = MagicMock() + mock_error = MagicMock() + mock_warning = MagicMock() + cli_modules = MagicMock() + cli_modules.console = mock_console + cli_modules.header = mock_header + cli_modules.error = mock_error + cli_modules.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) + + # -- 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 + mock_memory_files = MagicMock() + json_pkg.memory_files = mock_memory_files + 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) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.memory_files", mock_memory_files) + + # -- symbolic handler sub-modules (the delegation targets) -------------- + mock_extractor = MagicMock() + mock_storage = MagicMock() + mock_retriever = MagicMock() + mock_hook = MagicMock() + mock_deduplicator = MagicMock() + + # Give hook a SESSION_STATE dict for run_hook_test + mock_hook.SESSION_STATE = {"messages_since_last": 0, "last_surface_time": 0} + + 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 / store_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) + + # -- trigger error report (lazy import inside extract_and_store_llm) --- + mock_errors_mod = MagicMock() + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules.errors", mock_errors_mod) + + # -- rich Panel (lazy import in search_fragments_cli / run_hook_test) --- + monkeypatch.setitem(sys.modules, "rich", MagicMock()) + monkeypatch.setitem(sys.modules, "rich.panel", MagicMock()) + + # -- chromadb (used in bootstrap_from_jsonl summary) -------------------- + monkeypatch.setitem(sys.modules, "chromadb", MagicMock()) + + # 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.hook = mock_hook + _handler_mocks.deduplicator = mock_deduplicator + _handler_mocks.trigger = mock_trigger + _handler_mocks.console = mock_console + _handler_mocks.header = mock_header + _handler_mocks.error_fn = mock_error + _handler_mocks.warning_fn = mock_warning + _handler_mocks.json_handler = mock_json_handler + _handler_mocks.memory_files = mock_memory_files + + +# --------------------------------------------------------------------------- +# 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.""" + sys.modules.pop("aipass.memory.apps.modules.symbolic", None) + 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 + + +# =========================================================================== +# 1. analyze_conversation +# =========================================================================== + + +class TestAnalyzeConversation: + """analyze_conversation delegates to extractor.analyze_conversation.""" + + def test_delegates_to_extractor(self): + symbolic = _import_symbolic() + expected = { + "success": True, + "dimensions": {"technical": ["debug"]}, + "metadata": {"total_words": 20}, + "message_count": 2, + } + _handler_mocks.extractor.analyze_conversation.return_value = expected + + result = symbolic.analyze_conversation(_sample_chat()) + + _handler_mocks.extractor.analyze_conversation.assert_called_once_with(_sample_chat()) + assert result == expected + + def test_returns_handler_result_unchanged(self): + symbolic = _import_symbolic() + handler_result = {"success": False, "error": "bad input"} + _handler_mocks.extractor.analyze_conversation.return_value = handler_result + + assert symbolic.analyze_conversation([]) == handler_result + + +# =========================================================================== +# 2. store_fragment +# =========================================================================== + + +class TestStoreFragment: + """store_fragment delegates to storage.store_fragment and fires trigger.""" + + def test_delegates_to_storage(self): + symbolic = _import_symbolic() + expected = {"success": True, "fragment_id": "f-001"} + _handler_mocks.storage.store_fragment.return_value = expected + + frag = {"id": "f-001", "content": "test data"} + result = symbolic.store_fragment(frag) + + _handler_mocks.storage.store_fragment.assert_called_once_with(frag, None) + assert result == expected + + def test_passes_db_path(self): + symbolic = _import_symbolic() + db = Path("/tmp/test.chroma") + _handler_mocks.storage.store_fragment.return_value = { + "success": True, + "fragment_id": "f-002", + } + + symbolic.store_fragment({"id": "f-002"}, db_path=db) + + _handler_mocks.storage.store_fragment.assert_called_once_with({"id": "f-002"}, db) + + def test_fires_trigger_on_success(self): + symbolic = _import_symbolic() + _handler_mocks.storage.store_fragment.return_value = { + "success": True, + "fragment_id": "f-003", + } + _handler_mocks.trigger.reset_mock() + + symbolic.store_fragment({"id": "f-003"}) + + _handler_mocks.trigger.fire.assert_called_once_with("fragment_stored", fragment_id="f-003") + + def test_no_trigger_on_failure(self): + symbolic = _import_symbolic() + _handler_mocks.storage.store_fragment.return_value = { + "success": False, + "error": "db error", + } + _handler_mocks.trigger.reset_mock() + + symbolic.store_fragment({"id": "f-bad"}) + + _handler_mocks.trigger.fire.assert_not_called() + + +# =========================================================================== +# 3. store_fragments_batch +# =========================================================================== + + +class TestStoreFragmentsBatch: + """store_fragments_batch delegates to storage.store_fragments_batch.""" + + def test_delegates_to_storage(self): + symbolic = _import_symbolic() + expected = {"success": True, "stored": 3} + _handler_mocks.storage.store_fragments_batch.return_value = expected + + frags = [{"id": "a"}, {"id": "b"}, {"id": "c"}] + result = symbolic.store_fragments_batch(frags) + + _handler_mocks.storage.store_fragments_batch.assert_called_once_with(frags, None) + assert result == expected + + def test_passes_db_path(self): + symbolic = _import_symbolic() + db = Path("/tmp/batch.chroma") + _handler_mocks.storage.store_fragments_batch.return_value = { + "success": True, + "stored": 1, + } + + symbolic.store_fragments_batch([{"id": "x"}], db_path=db) + + _handler_mocks.storage.store_fragments_batch.assert_called_once_with([{"id": "x"}], db) + + +# =========================================================================== +# 4. flatten_dimensions +# =========================================================================== + + +class TestFlattenDimensions: + """flatten_dimensions delegates to storage.flatten_dimensions.""" + + def test_delegates_to_storage(self): + symbolic = _import_symbolic() + expected = {"success": True, "metadata": {"technical_0": "debug"}} + _handler_mocks.storage.flatten_dimensions.return_value = expected + + frag = {"dimensions": {"technical": ["debug"]}} + result = symbolic.flatten_dimensions(frag) + + _handler_mocks.storage.flatten_dimensions.assert_called_once_with(frag) + assert result == expected + + +# =========================================================================== +# 5. store_llm_fragment +# =========================================================================== + + +class TestStoreLlmFragment: + """store_llm_fragment delegates to storage.store_llm_fragment.""" + + def test_delegates_to_storage(self): + symbolic = _import_symbolic() + expected = { + "success": True, + "fragment_id": "llm-001", + "collection": "symbolic_fragments", + } + _handler_mocks.storage.store_llm_fragment.return_value = expected + + frag = {"summary": "test insight", "type": "episodic"} + result = symbolic.store_llm_fragment(frag, source_branch="memory") + + _handler_mocks.storage.store_llm_fragment.assert_called_once_with(frag, "memory", None) + assert result == expected + + def test_passes_all_args(self): + symbolic = _import_symbolic() + db = Path("/tmp/llm.chroma") + _handler_mocks.storage.store_llm_fragment.return_value = {"success": True} + + symbolic.store_llm_fragment({"summary": "x"}, source_branch="drone", db_path=db) + + _handler_mocks.storage.store_llm_fragment.assert_called_once_with({"summary": "x"}, "drone", db) + + +# =========================================================================== +# 6. store_llm_fragments_batch +# =========================================================================== + + +class TestStoreLlmFragmentsBatch: + """store_llm_fragments_batch delegates to storage.store_llm_fragments_batch.""" + + def test_delegates_to_storage(self): + symbolic = _import_symbolic() + expected = {"success": True, "stored": 2} + _handler_mocks.storage.store_llm_fragments_batch.return_value = expected + + frags = [{"summary": "a"}, {"summary": "b"}] + result = symbolic.store_llm_fragments_batch(frags, source_branch="api") + + _handler_mocks.storage.store_llm_fragments_batch.assert_called_once_with(frags, "api", None) + assert result == expected + + +# =========================================================================== +# 7. deduplicate_fragment +# =========================================================================== + + +class TestDeduplicateFragment: + """deduplicate_fragment delegates to deduplicator.deduplicate_fragment.""" + + def test_delegates_to_deduplicator(self): + symbolic = _import_symbolic() + expected = {"success": True, "action": "ADD", "fragment": {"summary": "new"}} + _handler_mocks.deduplicator.deduplicate_fragment.return_value = expected + + new_frag = {"summary": "new insight"} + existing = [{"summary": "old insight"}] + result = symbolic.deduplicate_fragment(new_frag, existing) + + _handler_mocks.deduplicator.deduplicate_fragment.assert_called_once_with(new_frag, existing) + assert result == expected + + def test_noop_action(self): + symbolic = _import_symbolic() + expected = { + "success": True, + "action": "NOOP", + "reason": "duplicate", + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = expected + + result = symbolic.deduplicate_fragment({"summary": "dup"}, [{"summary": "dup"}]) + + assert result["action"] == "NOOP" + + +# =========================================================================== +# 8. extract_and_store_llm +# =========================================================================== + + +class TestExtractAndStoreLlm: + """extract_and_store_llm runs the end-to-end pipeline.""" + + def test_success_with_add_action(self): + symbolic = _import_symbolic() + # Step 1: extraction returns fragments + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "insight A"}], + } + # Step 2: vector search returns no similar + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [], + } + # Step 3: dedup says ADD + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "success": True, + "action": "ADD", + "fragment": {"summary": "insight A"}, + } + # Step 4: store succeeds + _handler_mocks.storage.store_llm_fragment.return_value = { + "success": True, + "fragment_id": "llm-new", + } + + result = symbolic.extract_and_store_llm(_sample_chat(), source_branch="test") + + assert result["success"] is True + assert result["added"] == 1 + assert result["updated"] == 0 + assert result["skipped"] == 0 + + def test_extraction_failure_returns_error(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": False, + "error": "API unavailable", + } + + result = symbolic.extract_and_store_llm(_sample_chat()) + + assert result["success"] is False + assert result["processed"] == 0 + assert "API unavailable" in result["errors"] + + def test_no_fragments_extracted(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [], + } + + result = symbolic.extract_and_store_llm(_sample_chat()) + + assert result["success"] is True + assert result["processed"] == 0 + assert result["added"] == 0 + + def test_noop_action_skips(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "duplicate"}], + } + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [{"summary": "duplicate"}], + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "success": True, + "action": "NOOP", + "reason": "already exists", + } + + result = symbolic.extract_and_store_llm(_sample_chat()) + + assert result["success"] is True + assert result["skipped"] == 1 + assert result["added"] == 0 + + def test_update_action(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "updated insight"}], + } + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [{"summary": "old insight"}], + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "success": True, + "action": "UPDATE", + "fragment": {"summary": "updated insight"}, + } + _handler_mocks.storage.store_llm_fragment.return_value = { + "success": True, + "fragment_id": "llm-upd", + } + + result = symbolic.extract_and_store_llm(_sample_chat()) + + assert result["success"] is True + assert result["updated"] == 1 + + def test_delete_action(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "obsolete"}], + } + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [], + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "success": True, + "action": "DELETE", + "delete_id": "old-id-123", + "reason": "superseded", + } + _handler_mocks.storage.delete_fragment.return_value = {"success": True} + + result = symbolic.extract_and_store_llm(_sample_chat()) + + assert result["success"] is True + assert result["skipped"] == 1 + _handler_mocks.storage.delete_fragment.assert_called_once_with("old-id-123", None) + + +# =========================================================================== +# 9. retrieve_fragments +# =========================================================================== + + +class TestRetrieveFragments: + """retrieve_fragments delegates to retriever.retrieve_fragments.""" + + def test_delegates_to_retriever(self): + symbolic = _import_symbolic() + expected = {"success": True, "results": [{"content": "frag1"}]} + _handler_mocks.retriever.retrieve_fragments.return_value = expected + + result = symbolic.retrieve_fragments(query="debug error") + + _handler_mocks.retriever.retrieve_fragments.assert_called_once_with("debug error", None, None, 5, None) + assert result == expected + + def test_passes_all_filters(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + } + db = Path("/tmp/ret.chroma") + + symbolic.retrieve_fragments( + query="test", + dimension_filters={"emotional_0": "frustrated"}, + trigger_keywords=["error"], + n_results=10, + db_path=db, + ) + + _handler_mocks.retriever.retrieve_fragments.assert_called_once_with( + "test", {"emotional_0": "frustrated"}, ["error"], 10, db + ) + + +# =========================================================================== +# 10. search_fragments_by_vector +# =========================================================================== + + +class TestSearchFragmentsByVector: + """search_fragments_by_vector delegates to retriever.search_by_vector.""" + + def test_delegates_to_retriever(self): + symbolic = _import_symbolic() + expected = {"success": True, "results": []} + _handler_mocks.retriever.search_by_vector.return_value = expected + + result = symbolic.search_fragments_by_vector("semantic query") + + _handler_mocks.retriever.search_by_vector.assert_called_once_with("semantic query", 5, None) + assert result == expected + + +# =========================================================================== +# 11. search_fragments_by_dimensions +# =========================================================================== + + +class TestSearchFragmentsByDimensions: + """search_fragments_by_dimensions delegates to retriever.search_by_dimensions.""" + + def test_delegates_to_retriever(self): + symbolic = _import_symbolic() + expected = {"success": True, "results": [{"content": "matched"}]} + _handler_mocks.retriever.search_by_dimensions.return_value = expected + + filters = {"technical_0": "debugging_session"} + result = symbolic.search_fragments_by_dimensions(filters, n_results=3) + + _handler_mocks.retriever.search_by_dimensions.assert_called_once_with(filters, 3, None) + assert result == expected + + +# =========================================================================== +# 12. search_fragments_by_triggers +# =========================================================================== + + +class TestSearchFragmentsByTriggers: + """search_fragments_by_triggers delegates to retriever.search_by_triggers.""" + + def test_delegates_to_retriever(self): + symbolic = _import_symbolic() + expected = {"success": True, "results": []} + _handler_mocks.retriever.search_by_triggers.return_value = expected + + result = symbolic.search_fragments_by_triggers(["error", "debug"], n_results=10) + + _handler_mocks.retriever.search_by_triggers.assert_called_once_with(["error", "debug"], 10, None) + assert result == expected + + +# =========================================================================== +# 13. extract_conversation_context +# =========================================================================== + + +class TestExtractConversationContext: + """extract_conversation_context delegates to hook.extract_conversation_context.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + expected = { + "success": True, + "keywords": ["error", "debug"], + "mood": "frustrated", + "themes": ["troubleshooting"], + } + _handler_mocks.hook.extract_conversation_context.return_value = expected + + msgs = [{"role": "user", "content": "I have an error"}] + result = symbolic.extract_conversation_context(msgs) + + _handler_mocks.hook.extract_conversation_context.assert_called_once_with(msgs, 5) + assert result == expected + + def test_custom_max_messages(self): + symbolic = _import_symbolic() + _handler_mocks.hook.extract_conversation_context.return_value = {"success": True} + + symbolic.extract_conversation_context([], max_messages=10) + + _handler_mocks.hook.extract_conversation_context.assert_called_once_with([], 10) + + +# =========================================================================== +# 14. find_relevant_fragments +# =========================================================================== + + +class TestFindRelevantFragments: + """find_relevant_fragments delegates to hook.find_relevant_fragments.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + expected = {"success": True, "fragments": [{"content": "match"}]} + _handler_mocks.hook.find_relevant_fragments.return_value = expected + + context = {"keywords": ["debug"], "mood": "neutral"} + result = symbolic.find_relevant_fragments(context, n_results=5) + + _handler_mocks.hook.find_relevant_fragments.assert_called_once_with(context, 5, None) + assert result == expected + + +# =========================================================================== +# 15. format_fragment_recall +# =========================================================================== + + +class TestFormatFragmentRecall: + """format_fragment_recall delegates to hook.format_fragment_recall.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + expected = "This reminds me of a debugging session..." + _handler_mocks.hook.format_fragment_recall.return_value = expected + + frag = {"content": "debug session", "metadata": {"type": "episodic"}} + result = symbolic.format_fragment_recall(frag) + + _handler_mocks.hook.format_fragment_recall.assert_called_once_with(frag) + assert result == expected + + +# =========================================================================== +# 16. should_surface_fragment +# =========================================================================== + + +class TestShouldSurfaceFragment: + """should_surface_fragment delegates to hook.should_surface_fragment.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + expected = (True, "relevance threshold met") + _handler_mocks.hook.should_surface_fragment.return_value = expected + + frag = {"content": "test", "metadata": {}} + result = symbolic.should_surface_fragment(frag) + + _handler_mocks.hook.should_surface_fragment.assert_called_once_with(frag, None) + assert result == expected + + def test_with_config(self): + symbolic = _import_symbolic() + _handler_mocks.hook.should_surface_fragment.return_value = ( + False, + "cooldown active", + ) + config = {"min_relevance": 0.8} + + result = symbolic.should_surface_fragment(config=config) + + _handler_mocks.hook.should_surface_fragment.assert_called_once_with(None, config) + assert result[0] is False + + +# =========================================================================== +# 17. process_hook +# =========================================================================== + + +class TestProcessHook: + """process_hook delegates to hook.process_hook.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + expected = {"success": True, "surfaced": True, "recall": "I remember..."} + _handler_mocks.hook.process_hook.return_value = expected + + msgs = [{"role": "user", "content": "debugging"}] + result = symbolic.process_hook(msgs) + + _handler_mocks.hook.process_hook.assert_called_once_with(msgs, None, None) + assert result == expected + + def test_passes_config_and_db_path(self): + symbolic = _import_symbolic() + _handler_mocks.hook.process_hook.return_value = {"success": True} + config = {"enabled": True} + db = Path("/tmp/hook.chroma") + + symbolic.process_hook([], config=config, db_path=db) + + _handler_mocks.hook.process_hook.assert_called_once_with([], config, db) + + +# =========================================================================== +# 18. load_hook_config +# =========================================================================== + + +class TestLoadHookConfig: + """load_hook_config delegates to hook.load_config.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + expected = {"enabled": True, "min_relevance": 0.5} + _handler_mocks.hook.load_config.return_value = expected + + result = symbolic.load_hook_config() + + _handler_mocks.hook.load_config.assert_called_once_with(None) + assert result == expected + + def test_passes_config_path(self): + symbolic = _import_symbolic() + _handler_mocks.hook.load_config.return_value = {} + p = Path("/tmp/hook_config.json") + + symbolic.load_hook_config(config_path=p) + + _handler_mocks.hook.load_config.assert_called_once_with(p) + + +# =========================================================================== +# 19. reset_hook_session +# =========================================================================== + + +class TestResetHookSession: + """reset_hook_session delegates to hook.reset_session.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + _handler_mocks.hook.reset_session.return_value = None + + result = symbolic.reset_hook_session() + + _handler_mocks.hook.reset_session.assert_called_once() + assert result is None + + +# =========================================================================== +# 20. get_hook_session_state +# =========================================================================== + + +class TestGetHookSessionState: + """get_hook_session_state delegates to hook.get_session_state.""" + + def test_delegates_to_hook(self): + symbolic = _import_symbolic() + expected = {"fragments_surfaced": 3, "messages_since_last": 12} + _handler_mocks.hook.get_session_state.return_value = expected + + result = symbolic.get_hook_session_state() + + _handler_mocks.hook.get_session_state.assert_called_once() + assert result == expected + + +# =========================================================================== +# 21. run_demo +# =========================================================================== + + +class TestRunDemo: + """run_demo runs demo analysis with Rich output.""" + + def test_does_not_raise(self): + symbolic = _import_symbolic() + # Mock the functions run_demo calls internally + _handler_mocks.extractor.analyze_conversation.return_value = { + "success": True, + "dimensions": { + "technical": ["debug"], + "emotional": ["frustrated"], + "collaboration": ["balanced"], + "learnings": ["fix"], + "triggers": ["error"], + }, + "metadata": {"total_words": 50, "depth": "shallow"}, + "message_count": 5, + } + _handler_mocks.hook.format_fragment_recall.return_value = "This reminds me of..." + + # Should not raise + symbolic.run_demo() + + # Verify console was used + assert _handler_mocks.console.print.called + + def test_handles_analysis_failure(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.analyze_conversation.return_value = { + "success": False, + "error": "test error", + } + _handler_mocks.hook.format_fragment_recall.return_value = "recall" + + # Should not raise even on failure + symbolic.run_demo() + + +# =========================================================================== +# 22. search_fragments_cli +# =========================================================================== + + +class TestSearchFragmentsCli: + """search_fragments_cli executes CLI fragment search.""" + + def test_basic_query_search(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [ + { + "content": "debug session", + "metadata": {"timestamp": "2026-01-01"}, + "relevance_score": 0.85, + "_sources": ["vector"], + } + ], + "search_methods": ["vector"], + } + + symbolic.search_fragments_cli(["debug", "error"]) + + _handler_mocks.retriever.retrieve_fragments.assert_called_once() + assert _handler_mocks.console.print.called + + def test_no_args_shows_error(self): + symbolic = _import_symbolic() + + symbolic.search_fragments_cli([]) + + # Should print error about missing query + assert _handler_mocks.console.print.called + + def test_dimension_filter_parsing(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["dimension"], + } + + symbolic.search_fragments_cli(["query", "--dimension", "emotional_0=frustrated"]) + + call_args = _handler_mocks.retriever.retrieve_fragments.call_args + assert call_args is not None + # dimension_filters is the second positional arg + assert call_args[0][1] == {"emotional_0": "frustrated"} + + def test_trigger_keyword_parsing(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["trigger"], + } + + symbolic.search_fragments_cli(["--trigger", "error", "--trigger", "debug"]) + + call_args = _handler_mocks.retriever.retrieve_fragments.call_args + assert call_args is not None + # trigger_keywords is the third positional arg + assert call_args[0][2] == ["error", "debug"] + + def test_search_failure(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": False, + "error": "ChromaDB unavailable", + } + + symbolic.search_fragments_cli(["test query"]) + + assert _handler_mocks.console.print.called + + +# =========================================================================== +# 23. run_hook_test +# =========================================================================== + + +class TestRunHookTest: + """run_hook_test tests hook with sample text.""" + + def test_does_not_raise(self): + symbolic = _import_symbolic() + _handler_mocks.hook.reset_session.return_value = None + _handler_mocks.hook.extract_conversation_context.return_value = { + "success": True, + "keywords": ["error"], + "mood": "frustrated", + "themes": ["debugging"], + } + _handler_mocks.hook.find_relevant_fragments.return_value = { + "success": True, + "fragments": [], + "query_used": "error debugging", + "threshold_applied": 0.3, + } + _handler_mocks.hook.should_surface_fragment.return_value = ( + False, + "no fragments", + ) + _handler_mocks.hook.process_hook.return_value = { + "success": True, + "surfaced": False, + "reason": "no matching fragments", + } + _handler_mocks.hook.get_session_state.return_value = { + "fragments_surfaced": 0, + "messages_since_last": 0, + } + + symbolic.run_hook_test(["test", "text"]) + + assert _handler_mocks.console.print.called + + def test_bypass_flag(self): + symbolic = _import_symbolic() + _handler_mocks.hook.reset_session.return_value = None + _handler_mocks.hook.extract_conversation_context.return_value = { + "success": True, + "keywords": [], + "mood": "neutral", + "themes": [], + } + _handler_mocks.hook.find_relevant_fragments.return_value = { + "success": True, + "fragments": [], + "query_used": "", + "threshold_applied": 0.3, + } + _handler_mocks.hook.should_surface_fragment.return_value = ( + False, + "no fragments", + ) + _handler_mocks.hook.process_hook.return_value = { + "success": True, + "surfaced": False, + "reason": "nothing to surface", + } + _handler_mocks.hook.get_session_state.return_value = { + "fragments_surfaced": 0, + "messages_since_last": 0, + } + + # Should not raise with --bypass flag + symbolic.run_hook_test(["--bypass", "test"]) + + assert _handler_mocks.hook.reset_session.called + + def test_context_extraction_failure(self): + symbolic = _import_symbolic() + _handler_mocks.hook.reset_session.return_value = None + _handler_mocks.hook.extract_conversation_context.return_value = { + "success": False, + "error": "extraction failed", + } + + # Should not raise even when extraction fails (returns early) + symbolic.run_hook_test(["broken input"]) + + assert _handler_mocks.console.print.called + + +# =========================================================================== +# 24. analyze_file +# =========================================================================== + + +class TestAnalyzeFile: + """analyze_file analyzes a conversation JSON file.""" + + def test_file_not_found(self, tmp_path): + symbolic = _import_symbolic() + nonexistent = str(tmp_path / "does_not_exist.json") + + symbolic.analyze_file(nonexistent) + + # Should print error about missing file + assert _handler_mocks.console.print.called + + def test_successful_analysis(self, tmp_path): + symbolic = _import_symbolic() + + # Create a real file so Path(file_path).exists() is True + chat_file = tmp_path / "chat.json" + chat_file.write_text("[]", encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "hi"}, + ], + } + _handler_mocks.extractor.analyze_conversation.return_value = { + "success": True, + "dimensions": { + "technical": [], + "emotional": [], + "collaboration": [], + "learnings": [], + "triggers": [], + }, + "metadata": {"total_words": 2, "depth": "shallow"}, + "message_count": 2, + } + + symbolic.analyze_file(str(chat_file)) + + assert _handler_mocks.console.print.called + + def test_invalid_json_data(self, tmp_path): + symbolic = _import_symbolic() + + chat_file = tmp_path / "bad.json" + chat_file.write_text("{}", encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": {"not": "a list"}, + } + + symbolic.analyze_file(str(chat_file)) + + # Should print error about expected array + assert _handler_mocks.console.print.called + + def test_read_failure(self, tmp_path): + symbolic = _import_symbolic() + + chat_file = tmp_path / "unreadable.json" + chat_file.write_text("[]", encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": False, + "error": "permission denied", + } + + symbolic.analyze_file(str(chat_file)) + + assert _handler_mocks.console.print.called + + +# =========================================================================== +# 25. bootstrap_from_jsonl +# =========================================================================== + + +class TestBootstrapFromJsonl: + """bootstrap_from_jsonl bootstraps from session JONLs.""" + + def test_no_sessions_found(self, monkeypatch): + symbolic = _import_symbolic() + + # Mock _find_bootstrap_sessions to return empty + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: []) + + symbolic.bootstrap_from_jsonl(max_sessions=5) + + # Should print error about no files found + assert _handler_mocks.error_fn.called or _handler_mocks.console.print.called + + def test_processes_sessions(self, monkeypatch, tmp_path): + symbolic = _import_symbolic() + + # Create a fake JSONL file + jsonl_file = tmp_path / "session.jsonl" + jsonl_file.write_text("", encoding="utf-8") + + # Mock _find_bootstrap_sessions + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: [jsonl_file]) + + # Mock _parse_jsonl_to_chat_history to return enough messages + monkeypatch.setattr( + symbolic, + "_parse_jsonl_to_chat_history", + lambda path: [ + {"role": "user", "content": "hello"}, + {"role": "assistant", "content": "hi"}, + {"role": "user", "content": "test"}, + {"role": "assistant", "content": "response"}, + ], + ) + + # Mock extract_and_store_llm (called internally) + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "test"}], + } + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [], + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "success": True, + "action": "ADD", + "fragment": {"summary": "test"}, + } + _handler_mocks.storage.store_llm_fragment.return_value = { + "success": True, + "fragment_id": "boot-001", + } + + symbolic.bootstrap_from_jsonl(max_sessions=1) + + assert _handler_mocks.console.print.called + + def test_skips_sessions_with_few_messages(self, monkeypatch, tmp_path): + symbolic = _import_symbolic() + + jsonl_file = tmp_path / "tiny.jsonl" + jsonl_file.write_text("", encoding="utf-8") + + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: [jsonl_file]) + # Only 2 messages -- below the 4-message threshold + monkeypatch.setattr( + symbolic, + "_parse_jsonl_to_chat_history", + lambda path: [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "hello"}, + ], + ) + + symbolic.bootstrap_from_jsonl(max_sessions=1) + + # extract_and_store_llm should NOT have been called + _handler_mocks.extractor.extract_fragments_llm.assert_not_called() diff --git a/src/aipass/memory/tests/test_watcher.py b/src/aipass/memory/tests/test_watcher.py new file mode 100644 index 00000000..0f5ce064 --- /dev/null +++ b/src/aipass/memory/tests/test_watcher.py @@ -0,0 +1,492 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_watcher.py +# Date: 2026-04-25 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for the memory watcher handler. + +Covers: + from aipass.memory.apps.handlers.monitor.memory_watcher import start_memory_watcher + from aipass.memory.apps.handlers.monitor.memory_watcher import stop_memory_watcher + from aipass.memory.apps.handlers.monitor.memory_watcher import is_memory_watcher_active + from aipass.memory.apps.handlers.monitor.memory_watcher import get_watcher_status + from aipass.memory.apps.handlers.monitor.memory_watcher import MemoryFileWatcher + +Tests watcher lifecycle (start/stop/status), the MemoryFileWatcher.on_modified +callback, and edge cases like missing watchdog or already-running observers. +All tests use mocks -- no live filesystem watchers or infrastructure access. +""" + +import sys +from unittest.mock import MagicMock + + +# --------------------------------------------------------------------------- +# Helpers: prepare the mock graph needed to import memory_watcher +# --------------------------------------------------------------------------- + + +def _prepare_watcher_mocks(monkeypatch): + """Insert mocks for every module-level import memory_watcher.py touches. + + Returns a dict of key mock objects so tests can assert against them. + """ + # Mock line_counter + mock_update_line_count = MagicMock(return_value={"success": True, "lines": 150}) + mock_line_counter = MagicMock() + mock_line_counter.update_line_count = mock_update_line_count + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.tracking.line_counter", + mock_line_counter, + ) + + # Mock detector + mock_check_single_file = MagicMock(return_value={"success": True, "should_rollover": False}) + mock_detector = MagicMock() + mock_detector.check_single_file = mock_check_single_file + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.monitor.detector", + mock_detector, + ) + + # Mock watchdog + mock_observer_instance = MagicMock() + mock_observer_instance.is_alive.return_value = True + mock_observer_cls = MagicMock(return_value=mock_observer_instance) + + mock_watchdog_observers = MagicMock() + mock_watchdog_observers.Observer = mock_observer_cls + + mock_fse_handler = type("FileSystemEventHandler", (), {"__init__": lambda self: None}) + + mock_watchdog_events = MagicMock() + mock_watchdog_events.FileSystemEventHandler = mock_fse_handler + + monkeypatch.setitem(sys.modules, "watchdog", MagicMock()) + monkeypatch.setitem(sys.modules, "watchdog.observers", mock_watchdog_observers) + monkeypatch.setitem(sys.modules, "watchdog.events", mock_watchdog_events) + + # Mock rollover orchestrator (lazy import inside on_modified) + mock_execute_rollover = MagicMock(return_value={"success": True}) + mock_orchestrator = MagicMock() + mock_orchestrator.execute_rollover = mock_execute_rollover + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.rollover", + MagicMock(), + ) + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.rollover.orchestrator", + mock_orchestrator, + ) + + return { + "update_line_count": mock_update_line_count, + "check_single_file": mock_check_single_file, + "observer_instance": mock_observer_instance, + "observer_cls": mock_observer_cls, + "execute_rollover": mock_execute_rollover, + } + + +def _import_watcher(monkeypatch): + """Prepare mocks and import (or reimport) memory_watcher. + + Returns (module, mocks_dict). + """ + mocks = _prepare_watcher_mocks(monkeypatch) + + # Remove cached module so it gets re-imported with our mocks + sys.modules.pop("aipass.memory.apps.handlers.monitor.memory_watcher", None) + + # Also clear parent package's cached attribute so Python re-executes + # the module code with fresh mocks instead of returning a stale ref. + parent = sys.modules.get("aipass.memory.apps.handlers.monitor") + if parent is not None and hasattr(parent, "memory_watcher"): + delattr(parent, "memory_watcher") + + from aipass.memory.apps.handlers.monitor import memory_watcher # noqa: E402 + + # Reset the global _observer to None for a clean slate each test + setattr(memory_watcher, "_observer", None) + + return memory_watcher, mocks + + +# =========================================================================== +# Tests: start_memory_watcher +# =========================================================================== + + +class TestStartMemoryWatcher: + """Verify start_memory_watcher lifecycle.""" + + def test_start_returns_success_with_paths(self, monkeypatch, tmp_path): + """Starting watcher with valid branch paths returns success.""" + mod, mocks = _import_watcher(monkeypatch) + + branch_path = tmp_path / "src" / "aipass" / "test_branch" + branch_path.mkdir(parents=True) + monkeypatch.setattr(mod, "_get_branch_paths", lambda: [branch_path]) + + result = mod.start_memory_watcher() + + assert result["success"] is True + assert result["count"] == 1 + assert str(branch_path) in result["watched_paths"] + mocks["observer_cls"].assert_called_once() + mocks["observer_instance"].start.assert_called_once() + + def test_start_fails_when_already_running(self, monkeypatch, tmp_path): + """Starting when observer is already alive returns error.""" + mod, mocks = _import_watcher(monkeypatch) + + # Simulate already running observer + mock_existing = MagicMock() + mock_existing.is_alive.return_value = True + setattr(mod, "_observer", mock_existing) + + result = mod.start_memory_watcher() + + assert result["success"] is False + assert "already running" in result["error"].lower() + + def test_start_fails_when_no_branches(self, monkeypatch): + """Starting with no branch paths returns error.""" + mod, mocks = _import_watcher(monkeypatch) + monkeypatch.setattr(mod, "_get_branch_paths", lambda: []) + + result = mod.start_memory_watcher() + + assert result["success"] is False + assert "no branch" in result["error"].lower() + + def test_start_handles_schedule_error_gracefully(self, monkeypatch, tmp_path): + """If scheduling a path raises an exception, other paths still work.""" + mod, mocks = _import_watcher(monkeypatch) + + good_path = tmp_path / "good_branch" + good_path.mkdir(parents=True) + bad_path = tmp_path / "bad_branch" + bad_path.mkdir(parents=True) + + call_count = 0 + + def _mock_schedule(handler, path, recursive=False): + """Mock observer.schedule that fails on the first call.""" + nonlocal call_count + call_count += 1 + if call_count == 1: + raise OSError("Permission denied") + + mocks["observer_instance"].schedule = _mock_schedule + monkeypatch.setattr(mod, "_get_branch_paths", lambda: [bad_path, good_path]) + + result = mod.start_memory_watcher() + + assert result["success"] is True + # Only the second path should succeed + assert result["count"] == 1 + + +# =========================================================================== +# Tests: stop_memory_watcher +# =========================================================================== + + +class TestStopMemoryWatcher: + """Verify stop_memory_watcher lifecycle.""" + + def test_stop_returns_success(self, monkeypatch): + """Stopping a running watcher returns success.""" + mod, mocks = _import_watcher(monkeypatch) + + # Set up a running observer + mock_obs = MagicMock() + mock_obs.is_alive.return_value = True + setattr(mod, "_observer", mock_obs) + + result = mod.stop_memory_watcher() + + assert result["success"] is True + mock_obs.stop.assert_called_once() + mock_obs.join.assert_called_once() + assert mod._observer is None # type: ignore[union-attr] + + def test_stop_fails_when_not_running(self, monkeypatch): + """Stopping when no watcher is running returns error.""" + mod, mocks = _import_watcher(monkeypatch) + + result = mod.stop_memory_watcher() + + assert result["success"] is False + assert "not running" in result["error"].lower() + + def test_stop_fails_when_observer_not_alive(self, monkeypatch): + """Stopping when observer exists but is not alive returns error.""" + mod, mocks = _import_watcher(monkeypatch) + + mock_obs = MagicMock() + mock_obs.is_alive.return_value = False + setattr(mod, "_observer", mock_obs) + + result = mod.stop_memory_watcher() + + assert result["success"] is False + + +# =========================================================================== +# Tests: is_memory_watcher_active +# =========================================================================== + + +class TestIsMemoryWatcherActive: + """Verify is_memory_watcher_active boolean checks.""" + + def test_active_when_observer_alive(self, monkeypatch): + """Returns True when observer is alive.""" + mod, mocks = _import_watcher(monkeypatch) + + mock_obs = MagicMock() + mock_obs.is_alive.return_value = True + setattr(mod, "_observer", mock_obs) + + assert mod.is_memory_watcher_active() is True + + def test_inactive_when_no_observer(self, monkeypatch): + """Returns False when _observer is None.""" + mod, mocks = _import_watcher(monkeypatch) + + assert mod.is_memory_watcher_active() is False + + def test_inactive_when_observer_not_alive(self, monkeypatch): + """Returns False when observer exists but is_alive returns False.""" + mod, mocks = _import_watcher(monkeypatch) + + mock_obs = MagicMock() + mock_obs.is_alive.return_value = False + setattr(mod, "_observer", mock_obs) + + assert mod.is_memory_watcher_active() is False + + +# =========================================================================== +# Tests: get_watcher_status +# =========================================================================== + + +class TestGetWatcherStatus: + """Verify get_watcher_status returns correct status info.""" + + def test_status_when_inactive(self, monkeypatch): + """Returns inactive status when watcher is not running.""" + mod, mocks = _import_watcher(monkeypatch) + + result = mod.get_watcher_status() + + assert result["active"] is False + assert "not running" in result["message"].lower() + + def test_status_when_active(self, monkeypatch, tmp_path): + """Returns active status with directory count when watcher is running.""" + mod, mocks = _import_watcher(monkeypatch) + + mock_obs = MagicMock() + mock_obs.is_alive.return_value = True + setattr(mod, "_observer", mock_obs) + + branch_path = tmp_path / "branch1" + branch_path.mkdir() + monkeypatch.setattr(mod, "_get_branch_paths", lambda: [branch_path]) + + result = mod.get_watcher_status() + + assert result["active"] is True + assert result["watched_directories"] == 1 + assert str(branch_path) in result["paths"] + + def test_status_returns_multiple_paths(self, monkeypatch, tmp_path): + """Returns all watched directory paths.""" + mod, mocks = _import_watcher(monkeypatch) + + mock_obs = MagicMock() + mock_obs.is_alive.return_value = True + setattr(mod, "_observer", mock_obs) + + paths = [] + for name in ["branch_a", "branch_b", "branch_c"]: + p = tmp_path / name + p.mkdir() + paths.append(p) + + monkeypatch.setattr(mod, "_get_branch_paths", lambda: paths) + + result = mod.get_watcher_status() + + assert result["watched_directories"] == 3 + assert len(result["paths"]) == 3 + + +# =========================================================================== +# Tests: MemoryFileWatcher.on_modified +# =========================================================================== + + +class TestMemoryFileWatcherOnModified: + """Verify MemoryFileWatcher.on_modified callback behavior.""" + + def test_ignores_directory_events(self, monkeypatch): + """Directory modification events are ignored.""" + mod, mocks = _import_watcher(monkeypatch) + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = True + event.src_path = "/some/.trinity/local.json" + + watcher.on_modified(event) + + mocks["update_line_count"].assert_not_called() + + def test_ignores_non_memory_files(self, monkeypatch): + """Non-memory files (not in .trinity/) are ignored.""" + mod, mocks = _import_watcher(monkeypatch) + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = False + event.src_path = "/some/dir/config.json" + + watcher.on_modified(event) + + mocks["update_line_count"].assert_not_called() + + def test_processes_memory_file_modification(self, monkeypatch): + """Valid memory file modification triggers line count update and check.""" + mod, mocks = _import_watcher(monkeypatch) + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = False + event.src_path = "/some/branch/.trinity/local.json" + + watcher.on_modified(event) + + mocks["update_line_count"].assert_called_once() + mocks["check_single_file"].assert_called_once() + + def test_rollover_triggered_when_threshold_exceeded(self, monkeypatch): + """When check_single_file says should_rollover, execute_rollover is called.""" + mod, mocks = _import_watcher(monkeypatch) + + mocks["check_single_file"].return_value = { + "success": True, + "should_rollover": True, + "trigger": "lines exceeded 600", + } + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = False + event.src_path = "/some/branch/.trinity/local.json" + + watcher.on_modified(event) + + mocks["execute_rollover"].assert_called_once() + + def test_skips_recently_modified_file(self, monkeypatch): + """Files already in the recent modifications set are skipped.""" + mod, mocks = _import_watcher(monkeypatch) + + watcher = mod.MemoryFileWatcher() + file_path = "/some/branch/.trinity/local.json" + watcher._recent_modifications.add(file_path) + + event = MagicMock() + event.is_directory = False + event.src_path = file_path + + watcher.on_modified(event) + + # Should skip and not call update_line_count + mocks["update_line_count"].assert_not_called() + # The file key should be removed from recent modifications after skip + assert file_path not in watcher._recent_modifications + + def test_handles_line_count_update_failure(self, monkeypatch): + """When update_line_count fails, check_single_file is not called.""" + mod, mocks = _import_watcher(monkeypatch) + + mocks["update_line_count"].return_value = { + "success": False, + "error": "File not found", + } + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = False + event.src_path = "/some/branch/.trinity/local.json" + + watcher.on_modified(event) + + mocks["update_line_count"].assert_called_once() + mocks["check_single_file"].assert_not_called() + + def test_handles_check_failure(self, monkeypatch): + """When check_single_file fails, rollover is not triggered.""" + mod, mocks = _import_watcher(monkeypatch) + + mocks["check_single_file"].return_value = { + "success": False, + "error": "Read error", + } + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = False + event.src_path = "/some/branch/.trinity/local.json" + + watcher.on_modified(event) + + mocks["check_single_file"].assert_called_once() + mocks["execute_rollover"].assert_not_called() + + def test_processes_observations_json(self, monkeypatch): + """observations.json files in .trinity/ are also processed.""" + mod, mocks = _import_watcher(monkeypatch) + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = False + event.src_path = "/some/branch/.trinity/observations.json" + + watcher.on_modified(event) + + mocks["update_line_count"].assert_called_once() + + def test_rollover_exception_does_not_propagate(self, monkeypatch): + """If execute_rollover raises, the exception is caught.""" + mod, mocks = _import_watcher(monkeypatch) + + mocks["check_single_file"].return_value = { + "success": True, + "should_rollover": True, + "trigger": "lines exceeded", + } + mocks["execute_rollover"].side_effect = RuntimeError("Rollover crashed") + + watcher = mod.MemoryFileWatcher() + event = MagicMock() + event.is_directory = False + event.src_path = "/some/branch/.trinity/local.json" + + # Should not raise + watcher.on_modified(event) + + mocks["execute_rollover"].assert_called_once()