From 093fde79bdd7f4a242c25f85f1c7a7bb21be4aed Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sun, 26 Apr 2026 02:11:03 -0700 Subject: [PATCH] feat(memory): test(memory): improve line coverage on handler files Co-Authored-By: @memory --- .../memory/tests/test_manager_vectorize.py | 425 ++++++ .../memory/tests/test_orchestrator_exec.py | 752 ++++++++++ .../memory/tests/test_plans_processor.py | 689 +++++++++ src/aipass/memory/tests/test_symbolic_cli.py | 1301 +++++++++++++++++ .../memory/tests/test_templates_display.py | 675 +++++++++ 5 files changed, 3842 insertions(+) create mode 100644 src/aipass/memory/tests/test_manager_vectorize.py create mode 100644 src/aipass/memory/tests/test_orchestrator_exec.py create mode 100644 src/aipass/memory/tests/test_plans_processor.py create mode 100644 src/aipass/memory/tests/test_symbolic_cli.py create mode 100644 src/aipass/memory/tests/test_templates_display.py diff --git a/src/aipass/memory/tests/test_manager_vectorize.py b/src/aipass/memory/tests/test_manager_vectorize.py new file mode 100644 index 00000000..d750268b --- /dev/null +++ b/src/aipass/memory/tests/test_manager_vectorize.py @@ -0,0 +1,425 @@ +# ===================AIPASS==================== +# Name: tests/test_manager_vectorize.py +# Date: 2026-04-26 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= +"""Tests for manager vectorization and location helpers -- line coverage. + +Covers: from aipass.memory.apps.handlers.learnings.manager import process_all_branches +""" + +import sys +import json +import subprocess +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Helper: import learnings manager with mocked dependencies +# --------------------------------------------------------------------------- + + +def _import_manager(monkeypatch): + """Import manager with mocked memory_files dependency.""" + mock_memory_files = MagicMock() + 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, mock_memory_files + + +# =========================================================================== +# _find_learnings_location +# =========================================================================== + + +class TestFindLearningsLocation: + """Tests for _find_learnings_location helper.""" + + def test_find_learnings_at_root(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"key_learnings": {"a": "val"}} + parent, loc = mgr._find_learnings_location(data) + assert parent is data + assert loc == "root" + + def test_find_learnings_in_active_tasks(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"active_tasks": {"key_learnings": {"b": "val2"}}} + parent, loc = mgr._find_learnings_location(data) + assert parent is data["active_tasks"] + assert loc == "active_tasks" + + def test_find_learnings_not_found(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"other_key": 1} + parent, loc = mgr._find_learnings_location(data) + assert parent is None + assert loc == "" + + +# =========================================================================== +# _get_learnings / _set_learnings +# =========================================================================== + + +class TestGetSetLearnings: + """Tests for _get_learnings and _set_learnings helpers.""" + + def test_get_learnings_returns_dict(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"key_learnings": {"x": "y [2026-01-01]"}} + result = mgr._get_learnings(data) + assert result == {"x": "y [2026-01-01]"} + + def test_get_learnings_empty(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"something_else": True} + result = mgr._get_learnings(data) + assert result == {} + + def test_set_learnings_existing(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"key_learnings": {"old": "val"}} + ok = mgr._set_learnings(data, {"new": "val2"}) + assert ok is True + assert data["key_learnings"] == {"new": "val2"} + + def test_set_learnings_creates_at_root(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"no_learnings_here": True} + ok = mgr._set_learnings(data, {"fresh": "entry"}) + assert ok is True + assert data["key_learnings"] == {"fresh": "entry"} + + +# =========================================================================== +# _find_recently_completed_location +# =========================================================================== + + +class TestFindRecentlyCompletedLocation: + """Tests for _find_recently_completed_location helper.""" + + def test_find_recently_completed_at_root(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"recently_completed": ["task1"]} + parent, loc = mgr._find_recently_completed_location(data) + assert parent is data + assert loc == "root" + + def test_find_recently_completed_in_active_tasks(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"active_tasks": {"recently_completed": ["task2"]}} + parent, loc = mgr._find_recently_completed_location(data) + assert parent is data["active_tasks"] + assert loc == "active_tasks" + + def test_find_recently_completed_not_found(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"nothing": 0} + parent, loc = mgr._find_recently_completed_location(data) + assert parent is None + assert loc == "" + + +# =========================================================================== +# _get_recently_completed / _set_recently_completed +# =========================================================================== + + +class TestGetSetRecentlyCompleted: + """Tests for _get_recently_completed and _set_recently_completed helpers.""" + + def test_get_recently_completed_returns_list(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"recently_completed": ["a", "b"]} + result = mgr._get_recently_completed(data) + assert result == ["a", "b"] + + def test_get_recently_completed_empty(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {} + result = mgr._get_recently_completed(data) + assert result == [] + + def test_set_recently_completed_existing(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"recently_completed": ["old"]} + ok = mgr._set_recently_completed(data, ["new1", "new2"]) + assert ok is True + assert data["recently_completed"] == ["new1", "new2"] + + def test_set_recently_completed_creates_at_root(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + data = {"other": True} + ok = mgr._set_recently_completed(data, ["fresh"]) + assert ok is True + assert data["recently_completed"] == ["fresh"] + + +# =========================================================================== +# _vectorize_learnings +# =========================================================================== + + +def _mock_embedder(monkeypatch, encode_return): + """Install a mock embedder in sys.modules so the in-function import succeeds.""" + mock_emb = MagicMock() + mock_emb.encode_batch = MagicMock(return_value=encode_return) + vector_pkg = MagicMock() + vector_pkg.embedder = mock_emb + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector", vector_pkg) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector.embedder", mock_emb) + return mock_emb + + +class TestVectorizeLearnings: + """Tests for _vectorize_learnings.""" + + def test_empty_learnings(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + result = mgr._vectorize_learnings("BRANCH", []) + assert result["success"] is True + assert "No learnings" in result["message"] + + def test_success(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.1, 0.2]]}) + + mock_completed = MagicMock() + mock_completed.returncode = 0 + mock_completed.stdout = json.dumps({"success": True, "stored": 1}) + with patch.object(subprocess, "run", return_value=mock_completed) as mock_run: + result = mgr._vectorize_learnings("TEST", [("key1", "value1 [2026-01-01]")]) + assert result["success"] is True + assert mock_run.called + + def test_embedding_fails(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": False, "error": "model not found"}) + + result = mgr._vectorize_learnings("BRANCH", [("k", "v")]) + assert result["success"] is False + assert "Embedding failed" in result["error"] + + def test_no_embeddings(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": []}) + + result = mgr._vectorize_learnings("BRANCH", [("k", "v")]) + assert result["success"] is False + assert "No embeddings" in result["error"] + + def test_embedding_import_exception(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + # Make the import itself raise by inserting a broken module + broken = MagicMock() + broken.embedder = MagicMock() + broken.embedder.encode_batch = MagicMock(side_effect=RuntimeError("import boom")) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector", broken) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector.embedder", broken.embedder) + + result = mgr._vectorize_learnings("BRANCH", [("k", "v")]) + assert result["success"] is False + assert "import boom" in result["error"] + + def test_subprocess_timeout(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.1]]}) + + with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="x", timeout=60)): + result = mgr._vectorize_learnings("BR", [("k", "v")]) + assert result["success"] is False + assert "timed out" in result["error"] + + def test_subprocess_bad_json(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.1]]}) + + mock_proc = MagicMock() + mock_proc.returncode = 0 + mock_proc.stdout = "NOT JSON {" + with patch.object(subprocess, "run", return_value=mock_proc): + result = mgr._vectorize_learnings("BR", [("k", "v")]) + assert result["success"] is False + assert "Invalid JSON" in result["error"] + + def test_subprocess_nonzero_return(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.1]]}) + + mock_proc = MagicMock() + mock_proc.returncode = 1 + mock_proc.stderr = "segfault" + with patch.object(subprocess, "run", return_value=mock_proc): + result = mgr._vectorize_learnings("BR", [("k", "v")]) + assert result["success"] is False + assert "segfault" in result["error"] + + def test_subprocess_generic_exception(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.1]]}) + + with patch.object(subprocess, "run", side_effect=OSError("disk full")): + result = mgr._vectorize_learnings("BR", [("k", "v")]) + assert result["success"] is False + assert "disk full" in result["error"] + + def test_numpy_tolist_conversion(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + + # Simulate numpy-like objects with a tolist() method + class FakeNdarray: + def __init__(self, data): + self._data = data + + def tolist(self): + return self._data + + _mock_embedder( + monkeypatch, + {"success": True, "embeddings": [FakeNdarray([0.3, 0.4])]}, + ) + + mock_proc = MagicMock() + mock_proc.returncode = 0 + mock_proc.stdout = json.dumps({"success": True}) + with patch.object(subprocess, "run", return_value=mock_proc) as mock_run: + result = mgr._vectorize_learnings("BR", [("k", "v [2026-01-01]")]) + assert result["success"] is True + # Verify the input sent to subprocess had plain lists, not FakeNdarray + call_kwargs = mock_run.call_args + sent_input = json.loads(call_kwargs.kwargs.get("input") or call_kwargs[1].get("input")) + assert sent_input["embeddings"] == [[0.3, 0.4]] + + +# =========================================================================== +# _vectorize_completed_tasks +# =========================================================================== + + +class TestVectorizeCompletedTasks: + """Tests for _vectorize_completed_tasks.""" + + def test_empty_tasks(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + result = mgr._vectorize_completed_tasks("BRANCH", []) + assert result["success"] is True + assert "No tasks" in result["message"] + + def test_success(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.5, 0.6]]}) + + mock_proc = MagicMock() + mock_proc.returncode = 0 + mock_proc.stdout = json.dumps({"success": True, "stored": 1}) + with patch.object(subprocess, "run", return_value=mock_proc): + result = mgr._vectorize_completed_tasks("TEST", ["did thing [2026-01-01]"]) + assert result["success"] is True + + def test_embedding_fails(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": False, "error": "no model"}) + + result = mgr._vectorize_completed_tasks("BR", ["task1"]) + assert result["success"] is False + assert "Embedding failed" in result["error"] + + def test_no_embeddings(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": []}) + + result = mgr._vectorize_completed_tasks("BR", ["task1"]) + assert result["success"] is False + assert "No embeddings" in result["error"] + + def test_embedding_import_exception(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + broken = MagicMock() + broken.embedder = MagicMock() + broken.embedder.encode_batch = MagicMock(side_effect=ValueError("bad weights")) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector", broken) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.vector.embedder", broken.embedder) + + result = mgr._vectorize_completed_tasks("BR", ["task1"]) + assert result["success"] is False + assert "bad weights" in result["error"] + + def test_subprocess_timeout(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.1]]}) + + with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="x", timeout=60)): + result = mgr._vectorize_completed_tasks("BR", ["task1"]) + assert result["success"] is False + assert "timed out" in result["error"] + + def test_subprocess_bad_json(self, monkeypatch): + mgr, _ = _import_manager(monkeypatch) + _mock_embedder(monkeypatch, {"success": True, "embeddings": [[0.1]]}) + + mock_proc = MagicMock() + mock_proc.returncode = 0 + mock_proc.stdout = "<< 0 triggers it) + mock_trigger_mod = MagicMock() + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules.core", mock_trigger_mod) + mock_central = MagicMock() + mock_central.update_central = MagicMock(return_value={"success": True}) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.central_writer", mock_central) + mock_dash = MagicMock() + mock_dash.push_memory_bank_dashboard = MagicMock(return_value=True) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.dashboard_push", mock_dash) + mock_pool = MagicMock() + mock_pool.process_memory_pool = MagicMock(return_value={"files_processed": 0}) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake.pool_processor", mock_pool) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake", MagicMock(pool_processor=mock_pool)) + + result = orch.execute_rollover() + assert result["success"] is True + assert result["success_count"] == 1 + assert result["triggers_count"] == 2 + assert len(result["failed"]) == 1 + + def test_line_count_update_fails(self, monkeypatch, tmp_path): + """Line count update failure after successful storage is non-fatal.""" + orch, mocks = _import_orchestrator(monkeypatch) + trigger = _make_trigger(tmp_path) + mocks["detector"].check_all_branches.return_value = { + "success": True, + "triggers": [trigger], + } + mocks["extractor"].create_rollover_backup.return_value = { + "success": True, + "message": "ok", + } + mocks["extractor"].extract_with_metadata.return_value = { + "success": True, + "entries": [{"text": "m", "_metadata": {}, "timestamp": "2026-01-01"}], + "branch": "TEST", + "type": "sessions", + "old_lines": 100, + "new_lines": 50, + } + monkeypatch.setattr( + orch, + "encode_batch_subprocess", + lambda texts: {"success": True, "embeddings": [[0.1]]}, + ) + monkeypatch.setattr(orch, "get_branch_local_chroma_path", lambda b: None) + monkeypatch.setattr( + orch, + "store_vectors_subprocess", + lambda **kw: {"success": True, "collection": "c", "total_vectors": 1}, + ) + mocks["line_counter"].update_line_count.return_value = { + "success": False, + "error": "permission denied", + } + + # Mock post-rollover chain + mock_trigger_mod = MagicMock() + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules.core", mock_trigger_mod) + mock_central = MagicMock() + mock_central.update_central = MagicMock(return_value={"success": True}) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.central_writer", mock_central) + mock_dash = MagicMock() + mock_dash.push_memory_bank_dashboard = MagicMock(return_value=True) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.dashboard_push", mock_dash) + mock_pool = MagicMock() + mock_pool.process_memory_pool = MagicMock(return_value={"files_processed": 0}) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake.pool_processor", mock_pool) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake", MagicMock(pool_processor=mock_pool)) + + result = orch.execute_rollover() + # Still succeeds -- line count is non-fatal + assert result["success"] is True + assert result["success_count"] == 1 + + def test_local_storage_with_path(self, monkeypatch, tmp_path): + """Local chroma path exists and local storage succeeds.""" + orch, mocks = _import_orchestrator(monkeypatch) + trigger = _make_trigger(tmp_path) + mocks["detector"].check_all_branches.return_value = { + "success": True, + "triggers": [trigger], + } + mocks["extractor"].create_rollover_backup.return_value = { + "success": True, + "message": "ok", + } + mocks["extractor"].extract_with_metadata.return_value = { + "success": True, + "entries": [{"text": "m", "_metadata": {}, "timestamp": "2026-01-01"}], + "branch": "TEST", + "type": "sessions", + "old_lines": 100, + "new_lines": 50, + } + monkeypatch.setattr( + orch, + "encode_batch_subprocess", + lambda texts: {"success": True, "embeddings": [[0.1]]}, + ) + # Return a real path for local chroma + local_chroma = tmp_path / ".chroma" + monkeypatch.setattr(orch, "get_branch_local_chroma_path", lambda b: local_chroma) + monkeypatch.setattr( + orch, + "store_vectors_subprocess", + lambda **kw: {"success": True, "collection": "c", "total_vectors": 1}, + ) + mocks["line_counter"].update_line_count.return_value = {"success": True} + + # Mock post-rollover chain + mock_trigger_mod = MagicMock() + monkeypatch.setitem(sys.modules, "aipass.trigger.apps.modules.core", mock_trigger_mod) + mock_central = MagicMock() + mock_central.update_central = MagicMock(return_value={"success": True}) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.central_writer", mock_central) + mock_dash = MagicMock() + mock_dash.push_memory_bank_dashboard = MagicMock(return_value=True) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.dashboard_push", mock_dash) + mock_pool = MagicMock() + mock_pool.process_memory_pool = MagicMock(return_value={"files_processed": 0}) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake.pool_processor", mock_pool) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.intake", MagicMock(pool_processor=mock_pool)) + + result = orch.execute_rollover() + assert result["success"] is True + assert result["results"][0]["local_stored"] is True + + +# =========================================================================== +# sync_line_counts +# =========================================================================== + + +class TestSyncLineCounts: + """Tests for sync_line_counts.""" + + def test_sync_success(self, monkeypatch): + """line_counter returns success.""" + orch, mocks = _import_orchestrator(monkeypatch) + mocks["line_counter"].update_all_memory_files.return_value = { + "success": True, + "updated": 5, + "failed": 0, + } + result = orch.sync_line_counts() + assert result["success"] is True + assert result["updated"] == 5 + + def test_sync_failure(self, monkeypatch): + """line_counter returns failure.""" + orch, mocks = _import_orchestrator(monkeypatch) + mocks["line_counter"].update_all_memory_files.return_value = { + "success": False, + "updated": 0, + "failed": 3, + } + result = orch.sync_line_counts() + assert result["success"] is False diff --git a/src/aipass/memory/tests/test_plans_processor.py b/src/aipass/memory/tests/test_plans_processor.py new file mode 100644 index 00000000..47c766e7 --- /dev/null +++ b/src/aipass/memory/tests/test_plans_processor.py @@ -0,0 +1,689 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_plans_processor.py +# Date: 2026-04-26 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for plans_processor handler -- line coverage for all functions. + +Covers: from aipass.memory.apps.handlers.intake.plans_processor import process_plans +""" + +import json +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- +# Import helper +# --------------------------------------------------------------------------- + + +def _import_plans_processor(monkeypatch): + """Import plans_processor with mocked dependencies.""" + mock_memory_files = MagicMock() + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.json.memory_files", + mock_memory_files, + ) + + sys.modules.pop("aipass.memory.apps.handlers.intake.plans_processor", None) + parent = sys.modules.get("aipass.memory.apps.handlers.intake") + if parent is not None and hasattr(parent, "plans_processor"): + delattr(parent, "plans_processor") + + from aipass.memory.apps.handlers.intake import plans_processor + + return plans_processor + + +# =========================================================================== +# Tests: _chunk_plan_text +# =========================================================================== + + +class TestChunkPlanText: + """Test _chunk_plan_text function.""" + + def test_chunks_by_markdown_headers(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + text = ( + "## Introduction\n" + "This is the introduction section with enough text to pass the 30-char threshold easily.\n" + "## Details\n" + "Here are the details of the plan with plenty of content to exceed thirty characters.\n" + ) + + result = mod._chunk_plan_text(text, "plan.md") + + assert len(result) == 2 + assert result[0]["section"] == "Introduction" + assert result[1]["section"] == "Details" + assert "Introduction" in result[0]["text"] or "introduction" in result[0]["text"] + + def test_flushes_last_section(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + text = ( + "## Header One\n" + "Content for header one, long enough to pass thirty characters.\n" + "Trailing content without a following header, also long enough to be a real section." + ) + + result = mod._chunk_plan_text(text, "plan.md") + + assert len(result) == 1 + # The last section should be flushed since there is only one header + assert result[0]["section"] == "Header One" + assert "Trailing content" in result[0]["text"] + + def test_skips_short_sections(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + text = ( + "## Short\n" + "Tiny.\n" + "## Long Section\n" + "This section has enough content to pass the thirty-character minimum requirement.\n" + ) + + result = mod._chunk_plan_text(text, "plan.md") + + # The "Short" section has text "## Short\nTiny." stripped -> "## Short\nTiny." + # which is short, so it should be skipped + assert len(result) == 1 + assert result[0]["section"] == "Long Section" + + def test_fallback_to_size_chunking(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + # Headers present but every section body is tiny (< 30 chars), so the + # header-based pass produces zero chunks and the size-based fallback + # triggers on the full text which exceeds MAX_CHUNK_CHARS. + # Each pair "## Hxxx\nx\n" is ~12 chars; need > 1500 total. + num_sections = (mod.MAX_CHUNK_CHARS // 8) + 50 + header_lines = [] + for i in range(num_sections): + header_lines.append(f"## H{i:04d}") + header_lines.append("x") + text = "\n".join(header_lines) + # Confirm text is actually long enough for the size-based fallback + assert len(text) > mod.MAX_CHUNK_CHARS + + result = mod._chunk_plan_text(text, "plan.md") + + assert len(result) >= 2 + assert result[0]["section"].startswith("plan.md_part") + + def test_small_text_no_headers(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + text = "This is a plain text plan without any markdown headers at all." + + result = mod._chunk_plan_text(text, "plan.md") + + assert len(result) == 1 + assert result[0]["section"] == "plan.md" + assert result[0]["text"] == text + + def test_tiny_text_skipped(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + text = "Short." + + result = mod._chunk_plan_text(text, "plan.md") + + assert result == [] + + def test_splits_oversized_chunks(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + # Create a markdown section that is > MAX_CHUNK_CHARS * 2 + big_body = "X" * (mod.MAX_CHUNK_CHARS * 3) + text = f"## Big Section\n{big_body}\n" + + result = mod._chunk_plan_text(text, "plan.md") + + # The single chunk was > MAX_CHUNK_CHARS * 2, so it gets split + assert len(result) >= 2 + for chunk in result: + assert "_part" in chunk["section"] or chunk["section"] == "Big Section" + + def test_empty_text(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + + result = mod._chunk_plan_text("", "plan.md") + + assert result == [] + + +# =========================================================================== +# Tests: _load_manifest / _save_manifest +# =========================================================================== + + +class TestManifest: + """Test _load_manifest and _save_manifest.""" + + def test_load_manifest_file_exists(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.parent.mkdir(parents=True) + manifest_data = {"plan1.md": "2026-01-01T00:00:00", "plan2.md": "2026-01-02T00:00:00"} + manifest_path.write_text(json.dumps(manifest_data), encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + result = mod._load_manifest() + + assert result == manifest_data + + def test_load_manifest_file_missing(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + manifest_path = tmp_path / "config" / ".plans_processed.json" + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + result = mod._load_manifest() + + assert result == {} + + def test_load_manifest_bad_json(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.parent.mkdir(parents=True) + manifest_path.write_text("not valid json {{{{", encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + result = mod._load_manifest() + + assert result == {} + + def test_save_manifest_creates_parent_dirs(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + manifest_path = tmp_path / "deep" / "nested" / "config" / ".plans_processed.json" + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + data = {"file.md": "2026-04-26T12:00:00"} + + mod._save_manifest(data) + + assert manifest_path.exists() + loaded = json.loads(manifest_path.read_text(encoding="utf-8")) + assert loaded == data + + +# =========================================================================== +# Tests: _embed_texts +# =========================================================================== + + +class TestEmbedTexts: + """Test _embed_texts subprocess wrapper.""" + + def test_embed_texts_success(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + expected = {"success": True, "embeddings": [[0.1, 0.2], [0.3, 0.4]]} + 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 = mod._embed_texts(["hello", "world"]) + + assert result["success"] is True + assert result["embeddings"] == [[0.1, 0.2], [0.3, 0.4]] + mock_run.assert_called_once() + + def test_embed_texts_nonzero_return(self, monkeypatch): + mod = _import_plans_processor(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._embed_texts(["hello"]) + + assert result["success"] is False + assert "model not found" in result["error"] + + def test_embed_texts_exception(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + + with patch.object(subprocess, "run", side_effect=OSError("no such binary")): + result = mod._embed_texts(["hello"]) + + assert result["success"] is False + assert "no such binary" in result["error"] + + +# =========================================================================== +# Tests: _store_vectors +# =========================================================================== + + +class TestStoreVectors: + """Test _store_vectors subprocess wrapper.""" + + def test_store_vectors_success(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + expected = {"success": True, "stored": 5} + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stdout = json.dumps(expected) + + with patch.object(subprocess, "run", return_value=mock_result): + result = mod._store_vectors( + embeddings=[[0.1, 0.2]], + documents=["doc1"], + metadatas=[{"key": "val"}], + collection_name="test_col", + ) + + assert result["success"] is True + assert result["stored"] == 5 + + def test_store_vectors_nonzero_return(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stderr = "db locked" + + with patch.object(subprocess, "run", return_value=mock_result): + result = mod._store_vectors( + embeddings=[[0.1]], + documents=["doc"], + metadatas=[{}], + ) + + assert result["success"] is False + assert "db locked" in result["error"] + + def test_store_vectors_exception(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + + with patch.object(subprocess, "run", side_effect=TimeoutError("timed out")): + result = mod._store_vectors( + embeddings=[[0.1]], + documents=["doc"], + metadatas=[{}], + ) + + assert result["success"] is False + assert "timed out" in result["error"] + + +# =========================================================================== +# Tests: _find_repo_root / _get_memory_python (module-level) +# =========================================================================== + + +class TestFindRepoRoot: + """Test _find_repo_root function.""" + + def test_find_repo_root_with_registry(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + # Create a fake registry file + (tmp_path / "AIPASS_REGISTRY.json").write_text("{}", encoding="utf-8") + sub = tmp_path / "a" / "b" / "c" + sub.mkdir(parents=True) + fake_file = sub / "plans_processor.py" + fake_file.write_text("", encoding="utf-8") + + # Patch __file__ to be inside tmp_path tree + monkeypatch.setattr(mod, "__file__", str(fake_file)) + + # Re-call _find_repo_root which reads __file__ at module level; + # but the function uses Path(__file__) inside, so we need to patch the + # function's reference to __file__. We do this by calling it after + # monkeypatching the module's __file__. + result = mod._find_repo_root() + + assert result == tmp_path + + def test_find_repo_root_falls_back_to_cwd(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + # Point __file__ to a location with no registry + nowhere = tmp_path / "nowhere" / "file.py" + nowhere.parent.mkdir(parents=True) + nowhere.write_text("", encoding="utf-8") + monkeypatch.setattr(mod, "__file__", str(nowhere)) + monkeypatch.chdir(tmp_path) + + result = mod._find_repo_root() + + assert result == Path.cwd() + + +class TestGetMemoryPython: + """Test _get_memory_python function.""" + + def test_env_override(self, monkeypatch): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setenv("AIPASS_MEMORY_PYTHON", "/custom/python") + + result = mod._get_memory_python() + + assert result == "/custom/python" + + def test_venv_python_exists(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.delenv("AIPASS_MEMORY_PYTHON", raising=False) + venv_python = tmp_path / ".venv" / "bin" / "python" + venv_python.parent.mkdir(parents=True) + venv_python.write_text("#!/usr/bin/env python", encoding="utf-8") + monkeypatch.setattr(mod, "_MEMORY_VENV_PYTHON", venv_python) + + result = mod._get_memory_python() + + assert result == str(venv_python) + + def test_falls_back_to_sys_executable(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.delenv("AIPASS_MEMORY_PYTHON", raising=False) + # Point to a non-existent venv + monkeypatch.setattr(mod, "_MEMORY_VENV_PYTHON", tmp_path / "nonexistent" / "python") + + result = mod._get_memory_python() + + assert result == sys.executable + + +# =========================================================================== +# Tests: process_plans +# =========================================================================== + + +class TestProcessPlans: + """Test process_plans main entry point.""" + + def _setup_config(self, tmp_path, config_data): + """Write a memory_bank.config.json and return its path.""" + config_dir = tmp_path / "config" + config_dir.mkdir(parents=True, exist_ok=True) + config_path = config_dir / "memory_bank.config.json" + config_path.write_text(json.dumps(config_data), encoding="utf-8") + return config_path + + def test_process_plans_config_load_fails(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + # Point _MEMORY_ROOT to tmp_path -- no config file exists + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + + result = mod.process_plans() + + assert result["success"] is False + assert "Config load failed" in result["error"] + + def test_process_plans_disabled(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + self._setup_config(tmp_path, {"plans": {"enabled": False}}) + + result = mod.process_plans() + + assert result["success"] is True + assert result["skipped"] is True + assert "disabled" in result["reason"] + + def test_process_plans_dir_not_found(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + self._setup_config( + tmp_path, + {"plans": {"enabled": True, "path": "nonexistent/plans"}}, + ) + # _find_repo_root will return tmp_path + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + + result = mod.process_plans() + + assert result["success"] is True + assert result["files_processed"] == 0 + assert "not found" in result.get("reason", "") + + def test_process_plans_no_files(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + plans_dir = tmp_path / "plans" + plans_dir.mkdir() + self._setup_config( + tmp_path, + {"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}}, + ) + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + + result = mod.process_plans() + + assert result["success"] is True + assert result["files_processed"] == 0 + + def test_process_plans_all_already_processed(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + plans_dir = tmp_path / "plans" + plans_dir.mkdir() + plan_file = plans_dir / "done.md" + plan_file.write_text("Already processed plan content that is long enough.", encoding="utf-8") + self._setup_config( + tmp_path, + {"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}}, + ) + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + # Pre-populate the manifest + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.write_text(json.dumps({"done.md": "2026-01-01T00:00:00"}), encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + result = mod.process_plans() + + assert result["success"] is True + assert result["files_processed"] == 0 + assert "already processed" in result.get("reason", "") + + def test_process_plans_success(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + + # Create plans directory with a file + plans_dir = tmp_path / "plans" + plans_dir.mkdir() + plan_file = plans_dir / "new_plan.md" + plan_file.write_text( + "## Objective\nThis is the objective section with enough content to exceed thirty characters.\n" + "## Steps\nThese are the steps of the plan with sufficient length for chunking.\n", + encoding="utf-8", + ) + + self._setup_config( + tmp_path, + { + "plans": { + "enabled": True, + "path": str(plans_dir), + "supported_extensions": [".md"], + "collection_name": "test_plans", + } + }, + ) + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + + # Empty manifest + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.write_text("{}", encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + # Mock _embed_texts to return success + monkeypatch.setattr( + mod, + "_embed_texts", + lambda texts: {"success": True, "embeddings": [[0.1, 0.2]] * len(texts)}, + ) + # Mock _store_vectors to return success + monkeypatch.setattr( + mod, + "_store_vectors", + lambda emb, docs, metas, collection_name="flow_plans": {"success": True, "stored": len(docs)}, + ) + + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + + result = mod.process_plans() + + assert result["success"] is True + assert result["files_processed"] == 1 + assert result["total_chunks"] >= 2 + mock_jh.log_operation.assert_called_once() + + # Manifest should be updated + updated_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert "new_plan.md" in updated_manifest + + def test_process_plans_embed_fails(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + + plans_dir = tmp_path / "plans" + plans_dir.mkdir() + plan_file = plans_dir / "plan.md" + plan_file.write_text( + "## Section\nThis section has enough content to pass the minimum threshold for chunking.\n", + encoding="utf-8", + ) + + self._setup_config( + tmp_path, + {"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}}, + ) + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.write_text("{}", encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + # Mock _embed_texts to return failure + monkeypatch.setattr( + mod, + "_embed_texts", + lambda texts: {"success": False, "error": "GPU out of memory"}, + ) + + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + + result = mod.process_plans() + + assert result["success"] is False + assert "errors" in result + assert any("embed" in e for e in result["errors"]) + + def test_process_plans_no_embeddings(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + + plans_dir = tmp_path / "plans" + plans_dir.mkdir() + plan_file = plans_dir / "plan.md" + plan_file.write_text( + "## Section\nThis section has enough content to pass the minimum threshold for chunking.\n", + encoding="utf-8", + ) + + self._setup_config( + tmp_path, + {"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}}, + ) + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.write_text("{}", encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + # Embed succeeds but returns empty embeddings + monkeypatch.setattr( + mod, + "_embed_texts", + lambda texts: {"success": True, "embeddings": []}, + ) + + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + + result = mod.process_plans() + + assert "errors" in result + assert any("no embeddings" in e for e in result["errors"]) + + def test_process_plans_store_fails(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + + plans_dir = tmp_path / "plans" + plans_dir.mkdir() + plan_file = plans_dir / "plan.md" + plan_file.write_text( + "## Section\nThis section has enough content to pass the minimum threshold for chunking.\n", + encoding="utf-8", + ) + + self._setup_config( + tmp_path, + {"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}}, + ) + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.write_text("{}", encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + # Embed succeeds + monkeypatch.setattr( + mod, + "_embed_texts", + lambda texts: {"success": True, "embeddings": [[0.1, 0.2]] * len(texts)}, + ) + # Store fails + monkeypatch.setattr( + mod, + "_store_vectors", + lambda emb, docs, metas, collection_name="flow_plans": {"success": False, "error": "disk full"}, + ) + + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + + result = mod.process_plans() + + assert result["success"] is False + assert "errors" in result + assert any("store" in e for e in result["errors"]) + + def test_process_plans_empty_chunks(self, monkeypatch, tmp_path): + mod = _import_plans_processor(monkeypatch) + monkeypatch.setattr(mod, "_MEMORY_ROOT", tmp_path) + + plans_dir = tmp_path / "plans" + plans_dir.mkdir() + plan_file = plans_dir / "tiny.md" + # Content that will produce zero chunks (under 30 chars, no headers) + plan_file.write_text("Hi.", encoding="utf-8") + + self._setup_config( + tmp_path, + {"plans": {"enabled": True, "path": str(plans_dir), "supported_extensions": [".md"]}}, + ) + monkeypatch.setattr(mod, "_find_repo_root", lambda: tmp_path) + + manifest_path = tmp_path / "config" / ".plans_processed.json" + manifest_path.write_text("{}", encoding="utf-8") + monkeypatch.setattr(mod, "_PROCESSED_MANIFEST", manifest_path) + + mock_jh = MagicMock() + monkeypatch.setattr(mod, "json_handler", mock_jh) + + result = mod.process_plans() + + # No chunks produced, but no errors either -- files_without_chunks path + assert result["success"] is True + assert result["files_processed"] == 0 + + # File should still be marked in manifest (files_without_chunks) + updated_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert "tiny.md" in updated_manifest diff --git a/src/aipass/memory/tests/test_symbolic_cli.py b/src/aipass/memory/tests/test_symbolic_cli.py new file mode 100644 index 00000000..7df4c75f --- /dev/null +++ b/src/aipass/memory/tests/test_symbolic_cli.py @@ -0,0 +1,1301 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_symbolic_cli.py +# Date: 2026-04-26 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for symbolic module CLI/display functions -- line coverage. + +Covers: from aipass.memory.apps.modules.symbolic import handle_command +""" + +import json +import sys +import time +import types +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +# --------------------------------------------------------------------------- +# 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 _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 + + +def _default_analysis_result() -> dict: + """Standard successful analysis result.""" + return { + "success": True, + "dimensions": { + "technical": [], + "emotional": [], + "collaboration": [], + "learnings": [], + "triggers": [], + }, + "metadata": {"total_words": 50, "depth": "shallow", "timestamp": "2026-01-01", "total_chars": 200}, + "message_count": 3, + } + + +def _default_extract_store_result() -> dict: + """Standard successful extract_and_store_llm result.""" + return { + "success": True, + "processed": 2, + "added": 2, + "updated": 0, + "skipped": 0, + "errors": [], + } + + +# =========================================================================== +# handle_command routing (lines 637-751) +# =========================================================================== + + +class TestHandleCommand: + """Tests for handle_command routing.""" + + def test_help_flag(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("--help", []) + assert result is True + _handler_mocks.console.print.assert_called() + + def test_symbolic_no_args(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("symbolic", []) + assert result is True + # print_introspection calls console.print multiple times + assert _handler_mocks.console.print.call_count > 0 + + def test_symbolic_help(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("symbolic", ["--help"]) + assert result is True + _handler_mocks.header.assert_called() + + def test_symbolic_demo(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.analyze_conversation.return_value = _default_analysis_result() + _handler_mocks.hook.format_fragment_recall.return_value = "This reminds me of..." + result = symbolic.handle_command("symbolic", ["demo"]) + assert result is True + _handler_mocks.extractor.analyze_conversation.assert_called_once() + + def test_symbolic_analyze_no_file(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("symbolic", ["analyze"]) + assert result is True + # Should print error about missing file path + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("File path required" in c for c in calls) + + def test_symbolic_analyze_with_file(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = {"success": True, "data": chat_data} + _handler_mocks.extractor.analyze_conversation.return_value = _default_analysis_result() + + result = symbolic.handle_command("symbolic", ["analyze", str(chat_file)]) + assert result is True + + def test_symbolic_extract_no_file(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("symbolic", ["extract"]) + assert result is True + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("File path required" in c for c in calls) + + def test_symbolic_extract_with_branch(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = {"success": True, "data": chat_data} + # Mock the full pipeline that extract_file calls internally + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [], + "chunk_count": 0, + } + + result = symbolic.handle_command("symbolic", ["extract", str(chat_file), "memory"]) + assert result is True + + def test_symbolic_bootstrap(self, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: []) + result = symbolic.handle_command("symbolic", ["bootstrap"]) + assert result is True + + def test_symbolic_bootstrap_max(self, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: []) + result = symbolic.handle_command("symbolic", ["bootstrap", "--max=3"]) + assert result is True + + def test_symbolic_fragments(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["vector"], + } + result = symbolic.handle_command("symbolic", ["fragments", "query"]) + assert result is True + + def test_symbolic_hook_test(self): + symbolic = _import_symbolic() + _handler_mocks.hook.extract_conversation_context.return_value = { + "success": True, + "keywords": ["test"], + "mood": "neutral", + "themes": ["coding"], + } + _handler_mocks.hook.find_relevant_fragments.return_value = { + "success": True, + "fragments": [], + "query_used": "test", + "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 fragments", + } + _handler_mocks.hook.get_session_state.return_value = { + "fragments_surfaced": 0, + "messages_since_last": 0, + } + result = symbolic.handle_command("symbolic", ["hook-test", "test text"]) + assert result is True + + def test_symbolic_unknown_subcommand_returns_false(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("symbolic", ["nonexistent_sub"]) + assert result is False + + # Backward-compat routing + + def test_backward_compat_demo(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.analyze_conversation.return_value = _default_analysis_result() + _handler_mocks.hook.format_fragment_recall.return_value = "This reminds me of..." + result = symbolic.handle_command("demo", []) + assert result is True + + def test_backward_compat_analyze(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = {"success": True, "data": chat_data} + _handler_mocks.extractor.analyze_conversation.return_value = _default_analysis_result() + + result = symbolic.handle_command("analyze", [str(chat_file)]) + assert result is True + + def test_backward_compat_analyze_no_file(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("analyze", []) + assert result is True + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("File path required" in c for c in calls) + + def test_backward_compat_extract(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = {"success": True, "data": chat_data} + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [], + "chunk_count": 0, + } + result = symbolic.handle_command("extract", [str(chat_file)]) + assert result is True + + def test_backward_compat_extract_no_file(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("extract", []) + assert result is True + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("File path required" in c for c in calls) + + def test_backward_compat_bootstrap(self, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: []) + result = symbolic.handle_command("bootstrap", []) + assert result is True + + def test_backward_compat_fragments(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["vector"], + } + result = symbolic.handle_command("fragments", ["query"]) + assert result is True + + def test_backward_compat_hook_test(self): + symbolic = _import_symbolic() + _handler_mocks.hook.extract_conversation_context.return_value = { + "success": True, + "keywords": ["test"], + "mood": "neutral", + "themes": ["coding"], + } + _handler_mocks.hook.find_relevant_fragments.return_value = { + "success": True, + "fragments": [], + "query_used": "test", + "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 fragments", + } + _handler_mocks.hook.get_session_state.return_value = { + "fragments_surfaced": 0, + "messages_since_last": 0, + } + result = symbolic.handle_command("hook-test", ["text"]) + assert result is True + + def test_unknown_returns_false(self): + symbolic = _import_symbolic() + result = symbolic.handle_command("unknown", []) + assert result is False + + +# =========================================================================== +# print_help, print_introspection +# =========================================================================== + + +class TestPrintHelp: + """Tests for print_help (lines 754-815).""" + + def test_print_help(self): + symbolic = _import_symbolic() + symbolic.print_help() + _handler_mocks.header.assert_called() + assert _handler_mocks.console.print.call_count > 10 + + +class TestPrintIntrospection: + """Tests for print_introspection (lines 604-629).""" + + def test_introspection_with_symbolic_handlers(self, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr( + symbolic, + "_discover_handlers", + lambda: {"symbolic": ["extractor.py", "storage.py"]}, + ) + symbolic.print_introspection() + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Connected Handlers" in c for c in calls) + + def test_introspection_without_symbolic(self, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(symbolic, "_discover_handlers", lambda: {}) + symbolic.print_introspection() + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Subcommands" in c for c in calls) + # Should NOT contain "Connected Handlers" + assert not any("Connected Handlers" in c for c in calls) + + +# =========================================================================== +# run_demo (lines 818-916) +# =========================================================================== + + +class TestRunDemo: + """Tests for run_demo.""" + + def test_run_demo_success(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.analyze_conversation.return_value = _default_analysis_result() + _handler_mocks.hook.format_fragment_recall.return_value = "This reminds me of..." + + symbolic.run_demo() + + _handler_mocks.extractor.analyze_conversation.assert_called_once() + _handler_mocks.hook.format_fragment_recall.assert_called() + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Analysis complete" in c for c in calls) + + def test_run_demo_analysis_failure(self): + symbolic = _import_symbolic() + _handler_mocks.extractor.analyze_conversation.return_value = { + "success": False, + "error": "test failure", + } + _handler_mocks.hook.format_fragment_recall.return_value = "This reminds me of..." + + symbolic.run_demo() + + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Analysis failed" in c for c in calls) + + +# =========================================================================== +# search_fragments_cli (lines 919-1077) +# =========================================================================== + + +class TestSearchFragmentsCli: + """Tests for search_fragments_cli.""" + + def test_search_no_args(self): + symbolic = _import_symbolic() + symbolic.search_fragments_cli([]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("query, dimension filter, or trigger required" in c for c in calls) + + def test_search_query_only(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["vector"], + } + symbolic.search_fragments_cli(["test", "query"]) + _handler_mocks.retriever.retrieve_fragments.assert_called_once() + call_args = _handler_mocks.retriever.retrieve_fragments.call_args + # Called positionally: retriever.retrieve_fragments(query, dim, trig, n, db) + assert call_args[0][0] == "test query" + + def test_search_with_dimension_filter(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["vector"], + } + symbolic.search_fragments_cli(["query", "--dimension", "emotional_0=frustrated"]) + call_kwargs = _handler_mocks.retriever.retrieve_fragments.call_args + assert call_kwargs[1].get("dimension_filters") == {"emotional_0": "frustrated"} or ( + len(call_kwargs[0]) > 1 and call_kwargs[0][1] == {"emotional_0": "frustrated"} + ) + + def test_search_with_trigger(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["vector"], + } + symbolic.search_fragments_cli(["query", "--trigger", "error"]) + _handler_mocks.retriever.retrieve_fragments.assert_called_once() + + def test_search_with_n_results(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["vector"], + } + symbolic.search_fragments_cli(["query", "--n", "10"]) + call_kwargs = _handler_mocks.retriever.retrieve_fragments.call_args + assert call_kwargs[1].get("n_results") == 10 or (len(call_kwargs[0]) > 3 and call_kwargs[0][3] == 10) + + def test_search_invalid_n(self): + symbolic = _import_symbolic() + symbolic.search_fragments_cli(["query", "--n", "abc"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Invalid number" in c for c in calls) + + def test_search_invalid_dimension(self): + symbolic = _import_symbolic() + symbolic.search_fragments_cli(["query", "--dimension", "bad_format_no_equals"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Invalid dimension format" in c for c in calls) + + def test_search_no_results(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [], + "search_methods": ["vector"], + } + symbolic.search_fragments_cli(["test"]) + _handler_mocks.warning_fn.assert_called() + + def test_search_failure(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": False, + "error": "DB unavailable", + } + symbolic.search_fragments_cli(["test"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("DB unavailable" in c for c in calls) + + def test_search_v1_results(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [ + { + "content": "debugging pattern found", + "metadata": { + "timestamp": "2026-01-01", + "source_branch": "memory", + "depth": "deep", + "technical_0": "debug", + "emotional_0": "frustration", + }, + "relevance_score": 0.85, + "_sources": ["vector"], + "relevance_tier": "high", + }, + ], + "search_methods": ["vector"], + } + symbolic.search_fragments_cli(["debug"]) + # Should display a Panel for this result + assert _handler_mocks.console.print.call_count > 5 + + def test_search_v2_results(self): + symbolic = _import_symbolic() + _handler_mocks.retriever.retrieve_fragments.return_value = { + "success": True, + "results": [ + { + "content": "step-by-step debugging", + "metadata": { + "schema_version": "v2", + "summary": "Debugging session breakthrough", + "insight": "Step-by-step approach works best", + "type": "episodic", + "emotional_tone": "excited", + "technical_domain": "debugging", + "timestamp": "2026-01-01", + "source_branch": "memory", + }, + "relevance_score": 0.92, + "_sources": ["vector"], + "relevance_tier": "high", + }, + ], + "search_methods": ["vector"], + } + symbolic.search_fragments_cli(["debug"]) + assert _handler_mocks.console.print.call_count > 5 + + +# =========================================================================== +# run_hook_test (lines 1080-1204) +# =========================================================================== + + +class TestRunHookTest: + """Tests for run_hook_test.""" + + def _setup_hook_mocks( + self, + context_success: bool = True, + fragments: list | None = None, + surfaced: bool = False, + hook_success: bool = True, + ): + """Configure hook mocks for common test scenarios.""" + _handler_mocks.hook.extract_conversation_context.return_value = { + "success": context_success, + "keywords": ["test"], + "mood": "neutral", + "themes": ["coding"], + **({"error": "context extraction failed"} if not context_success else {}), + } + _handler_mocks.hook.find_relevant_fragments.return_value = { + "success": True, + "fragments": fragments or [], + "query_used": "test", + "threshold_applied": 0.3, + } + _handler_mocks.hook.should_surface_fragment.return_value = (surfaced, "test reason") + _handler_mocks.hook.process_hook.return_value = { + "success": hook_success, + "surfaced": surfaced, + "reason": "test reason", + **( + {"recall": "I remember debugging...", "fragment_id": "frag-1", "relevance_score": 0.8} + if surfaced + else {} + ), + **({"error": "hook process failed"} if not hook_success else {}), + } + _handler_mocks.hook.format_fragment_recall.return_value = "This reminds me of..." + _handler_mocks.hook.get_session_state.return_value = { + "fragments_surfaced": 1 if surfaced else 0, + "messages_since_last": 0, + } + _handler_mocks.hook.reset_session.return_value = None + + def test_hook_test_default_text(self): + symbolic = _import_symbolic() + self._setup_hook_mocks() + symbolic.run_hook_test([]) + # Should use default text + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("stuck on this error" in c for c in calls) + + def test_hook_test_custom_text(self): + symbolic = _import_symbolic() + self._setup_hook_mocks() + symbolic.run_hook_test(["my", "test", "text"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("my test text" in c for c in calls) + + def test_hook_test_bypass(self): + symbolic = _import_symbolic() + self._setup_hook_mocks() + symbolic.run_hook_test(["text", "--bypass"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("True" in c and "Bypass" in c for c in calls) + + def test_hook_test_context_extraction_fails(self): + symbolic = _import_symbolic() + self._setup_hook_mocks(context_success=False) + symbolic.run_hook_test(["text"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Failed" in c for c in calls) + + def test_hook_test_surfaced(self): + symbolic = _import_symbolic() + self._setup_hook_mocks(surfaced=True) + symbolic.run_hook_test(["text"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Fragment surfaced" in c for c in calls) + + def test_hook_test_not_surfaced(self): + symbolic = _import_symbolic() + self._setup_hook_mocks(surfaced=False) + symbolic.run_hook_test(["text"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Not surfaced" in c for c in calls) + + def test_hook_test_hook_fails(self): + symbolic = _import_symbolic() + self._setup_hook_mocks(hook_success=False) + symbolic.run_hook_test(["text"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Hook failed" in c for c in calls) + + def test_hook_test_v2_fragments(self): + symbolic = _import_symbolic() + v2_frag = { + "content": "debugging pattern", + "metadata": { + "schema_version": "v2", + "summary": "Debug breakthrough", + "insight": "Step by step", + "type": "episodic", + }, + "relevance_score": 0.85, + } + _handler_mocks.hook.extract_conversation_context.return_value = { + "success": True, + "keywords": ["test"], + "mood": "neutral", + "themes": ["coding"], + } + _handler_mocks.hook.find_relevant_fragments.return_value = { + "success": True, + "fragments": [v2_frag], + "query_used": "test", + "threshold_applied": 0.3, + } + _handler_mocks.hook.should_surface_fragment.return_value = (False, "test") + _handler_mocks.hook.process_hook.return_value = { + "success": True, + "surfaced": False, + "reason": "test", + } + _handler_mocks.hook.format_fragment_recall.return_value = "This reminds me of..." + _handler_mocks.hook.get_session_state.return_value = { + "fragments_surfaced": 0, + "messages_since_last": 0, + } + _handler_mocks.hook.reset_session.return_value = None + + symbolic.run_hook_test(["text"]) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + # v2 fragment preview path: "Fragment {i} (v2):" + assert any("(v2)" in c for c in calls) + + +# =========================================================================== +# analyze_file (lines 1207-1255) +# =========================================================================== + + +class TestAnalyzeFile: + """Tests for analyze_file.""" + + def test_analyze_file_not_found(self): + symbolic = _import_symbolic() + symbolic.analyze_file("/nonexistent/path/to/file.json") + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("File not found" in c for c in calls) + + def test_analyze_file_read_fails(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_file.write_text("[]", encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": False, + "error": "Read error", + } + symbolic.analyze_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Read error" in c for c in calls) + + def test_analyze_file_not_list(self, tmp_path): + symbolic = _import_symbolic() + 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": {"not": "a list"}, + } + symbolic.analyze_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Expected JSON array" in c for c in calls) + + def test_analyze_file_success(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": chat_data, + } + _handler_mocks.extractor.analyze_conversation.return_value = _default_analysis_result() + + symbolic.analyze_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Analysis complete" in c for c in calls) + + def test_analyze_file_failure(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": chat_data, + } + _handler_mocks.extractor.analyze_conversation.return_value = { + "success": False, + "error": "Analysis error", + } + + symbolic.analyze_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Analysis failed" in c for c in calls) + + +# =========================================================================== +# extract_file (lines 1258-1323) +# =========================================================================== + + +class TestExtractFile: + """Tests for extract_file.""" + + def test_extract_file_not_found(self): + symbolic = _import_symbolic() + symbolic.extract_file("/nonexistent/path/to/file.json") + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("File not found" in c for c in calls) + + def test_extract_file_read_fails(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_file.write_text("[]", encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": False, + "error": "Read error", + } + symbolic.extract_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Read error" in c for c in calls) + + def test_extract_file_not_list(self, tmp_path): + symbolic = _import_symbolic() + 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": {"not": "a list"}, + } + symbolic.extract_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Expected JSON array" in c for c in calls) + + def test_extract_file_success(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": chat_data, + } + # extract_and_store_llm is called internally by extract_file + # It calls extractor.extract_fragments_llm, then dedup + store loop + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "test frag"}], + "chunk_count": 1, + } + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [], + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "action": "ADD", + "fragment": {"summary": "test frag"}, + "reason": "new", + } + _handler_mocks.storage.store_llm_fragment.return_value = { + "success": True, + "fragment_id": "abc123", + } + + symbolic.extract_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Pipeline complete" in c for c in calls) + + def test_extract_file_with_branch(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": chat_data, + } + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [], + "chunk_count": 0, + } + + symbolic.extract_file(str(chat_file), source_branch="memory") + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("memory" in c for c in calls) + + def test_extract_file_pipeline_fails(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": chat_data, + } + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": False, + "error": "LLM unavailable", + } + + symbolic.extract_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Pipeline failed" in c for c in calls) + + def test_extract_file_with_errors(self, tmp_path): + symbolic = _import_symbolic() + chat_file = tmp_path / "chat.json" + chat_data = [{"role": "user", "content": "hello"}] + chat_file.write_text(json.dumps(chat_data), encoding="utf-8") + + _handler_mocks.memory_files.read_memory_file.return_value = { + "success": True, + "data": chat_data, + } + # Pipeline succeeds but has errors + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "frag1"}, {"summary": "frag2"}], + "chunk_count": 1, + } + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [], + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "action": "ADD", + "fragment": {"summary": "frag"}, + "reason": "new", + } + # First store succeeds, second fails + _handler_mocks.storage.store_llm_fragment.side_effect = [ + {"success": True, "fragment_id": "abc123"}, + {"success": False, "error": "store failed"}, + ] + + symbolic.extract_file(str(chat_file)) + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + # Pipeline should still complete with errors shown + assert any("Pipeline complete" in c for c in calls) + assert any("Errors" in c for c in calls) + + # Reset side_effect + _handler_mocks.storage.store_llm_fragment.side_effect = None + + +# =========================================================================== +# _parse_jsonl_to_chat_history (lines 1331-1390) +# =========================================================================== + + +class TestParseJsonlToChatHistory: + """Tests for _parse_jsonl_to_chat_history.""" + + def test_parse_user_text_message(self, tmp_path): + symbolic = _import_symbolic() + jsonl_file = tmp_path / "session.jsonl" + entry = {"type": "user", "message": {"role": "user", "content": "Hello world"}} + jsonl_file.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + result = symbolic._parse_jsonl_to_chat_history(jsonl_file) + assert len(result) == 1 + assert result[0]["role"] == "user" + assert result[0]["content"] == "Hello world" + + def test_parse_assistant_text_message(self, tmp_path): + symbolic = _import_symbolic() + jsonl_file = tmp_path / "session.jsonl" + entry = {"type": "assistant", "message": {"role": "assistant", "content": "I can help"}} + jsonl_file.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + result = symbolic._parse_jsonl_to_chat_history(jsonl_file) + assert len(result) == 1 + assert result[0]["role"] == "assistant" + assert result[0]["content"] == "I can help" + + def test_parse_user_list_content(self, tmp_path): + symbolic = _import_symbolic() + jsonl_file = tmp_path / "session.jsonl" + entry = { + "type": "user", + "message": { + "role": "user", + "content": [ + {"type": "text", "text": "First part"}, + {"type": "text", "text": "Second part"}, + ], + }, + } + jsonl_file.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + result = symbolic._parse_jsonl_to_chat_history(jsonl_file) + assert len(result) == 1 + assert result[0]["content"] == "First part Second part" + + def test_parse_assistant_list_content(self, tmp_path): + symbolic = _import_symbolic() + jsonl_file = tmp_path / "session.jsonl" + entry = { + "type": "assistant", + "message": { + "role": "assistant", + "content": [ + {"type": "text", "text": "Response part 1"}, + {"type": "text", "text": "Response part 2"}, + ], + }, + } + jsonl_file.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + result = symbolic._parse_jsonl_to_chat_history(jsonl_file) + assert len(result) == 1 + assert result[0]["content"] == "Response part 1 Response part 2" + + def test_parse_skips_malformed_lines(self, tmp_path): + symbolic = _import_symbolic() + jsonl_file = tmp_path / "session.jsonl" + good_entry = {"type": "user", "message": {"role": "user", "content": "valid"}} + content = "not valid json\n" + json.dumps(good_entry) + "\n" + "{{bad\n" + jsonl_file.write_text(content, encoding="utf-8") + + result = symbolic._parse_jsonl_to_chat_history(jsonl_file) + assert len(result) == 1 + assert result[0]["content"] == "valid" + + def test_parse_empty_file(self, tmp_path): + symbolic = _import_symbolic() + jsonl_file = tmp_path / "empty.jsonl" + jsonl_file.write_text("", encoding="utf-8") + + result = symbolic._parse_jsonl_to_chat_history(jsonl_file) + assert result == [] + + def test_parse_os_error(self, tmp_path): + symbolic = _import_symbolic() + # Path that does not exist + bad_path = tmp_path / "nonexistent.jsonl" + + result = symbolic._parse_jsonl_to_chat_history(bad_path) + assert result == [] + + +# =========================================================================== +# _find_bootstrap_sessions (lines 1393-1466) +# =========================================================================== + + +class TestFindBootstrapSessions: + """Tests for _find_bootstrap_sessions.""" + + def test_no_projects_dir(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(Path, "home", lambda: tmp_path) + # No .claude/projects directory exists + result = symbolic._find_bootstrap_sessions() + assert result == [] + + def test_finds_priority_branch_files(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(Path, "home", lambda: tmp_path) + + projects_dir = tmp_path / ".claude" / "projects" + branch_dir = projects_dir / "-home-patrick-Projects-AIPass-src-aipass-memory" + branch_dir.mkdir(parents=True) + + # Create a file in the valid size range (100KB-3MB) + jsonl_file = branch_dir / "session1.jsonl" + jsonl_file.write_text("x" * 200_000, encoding="utf-8") + + result = symbolic._find_bootstrap_sessions() + assert len(result) == 1 + assert result[0] == jsonl_file + + def test_skips_agent_files(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(Path, "home", lambda: tmp_path) + + projects_dir = tmp_path / ".claude" / "projects" + branch_dir = projects_dir / "-home-patrick-Projects-AIPass-src-aipass-memory" + branch_dir.mkdir(parents=True) + + # Agent file -- should be skipped + agent_file = branch_dir / "agent-build.jsonl" + agent_file.write_text("x" * 200_000, encoding="utf-8") + + result = symbolic._find_bootstrap_sessions() + assert len(result) == 0 + + def test_size_filtering(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(Path, "home", lambda: tmp_path) + + projects_dir = tmp_path / ".claude" / "projects" + branch_dir = projects_dir / "-home-patrick-Projects-AIPass-src-aipass-memory" + branch_dir.mkdir(parents=True) + + # Too small (< 100KB) + small_file = branch_dir / "small.jsonl" + small_file.write_text("x" * 50_000, encoding="utf-8") + + # Too large (> 3MB) + large_file = branch_dir / "large.jsonl" + large_file.write_text("x" * 4_000_000, encoding="utf-8") + + result = symbolic._find_bootstrap_sessions() + assert len(result) == 0 + + def test_max_sessions_limit(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(Path, "home", lambda: tmp_path) + + projects_dir = tmp_path / ".claude" / "projects" + + # Create multiple priority branch dirs with valid files + branches = [ + "-home-patrick-Projects-AIPass-src-aipass-memory", + "-home-patrick-Projects-AIPass-src-aipass-devpulse", + "-home-patrick-Projects-AIPass-src-aipass-seedgo", + "-home-patrick-Projects-AIPass-src-aipass-drone", + ] + for branch_name in branches: + branch_dir = projects_dir / branch_name + branch_dir.mkdir(parents=True) + jsonl_file = branch_dir / "session.jsonl" + jsonl_file.write_text("x" * 200_000, encoding="utf-8") + + result = symbolic._find_bootstrap_sessions(max_sessions=2) + assert len(result) == 2 + + +# =========================================================================== +# bootstrap_from_jsonl (lines 1469-1586) +# =========================================================================== + + +class TestBootstrapFromJsonl: + """Tests for bootstrap_from_jsonl.""" + + def test_bootstrap_no_sessions(self, monkeypatch): + symbolic = _import_symbolic() + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: []) + + symbolic.bootstrap_from_jsonl() + _handler_mocks.error_fn.assert_called() + + def test_bootstrap_success(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + + # Create a JSONL file with enough messages + jsonl_file = tmp_path / "-home-patrick-Projects-AIPass-src-aipass-memory" / "session.jsonl" + jsonl_file.parent.mkdir(parents=True) + lines = [] + for i in range(6): + role = "user" if i % 2 == 0 else "assistant" + entry = {"type": role, "message": {"role": role, "content": f"Message {i}"}} + lines.append(json.dumps(entry)) + jsonl_file.write_text("\n".join(lines), encoding="utf-8") + + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: [jsonl_file]) + monkeypatch.setattr(time, "sleep", lambda _: None) + + # Mock the pipeline + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": True, + "fragments": [{"summary": "test frag"}], + "chunk_count": 1, + } + _handler_mocks.retriever.search_by_vector.return_value = { + "success": True, + "results": [], + } + _handler_mocks.deduplicator.deduplicate_fragment.return_value = { + "action": "ADD", + "fragment": {"summary": "test frag"}, + "reason": "new", + } + _handler_mocks.storage.store_llm_fragment.return_value = { + "success": True, + "fragment_id": "abc123", + } + + symbolic.bootstrap_from_jsonl() + assert any("Bootstrap Summary" in str(c) for c in _handler_mocks.header.call_args_list) + + def test_bootstrap_few_messages_skipped(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + + # Create a JSONL with too few messages (< 4) + jsonl_file = tmp_path / "-home-patrick-Projects-AIPass-src-aipass-memory" / "session.jsonl" + jsonl_file.parent.mkdir(parents=True) + entry = {"type": "user", "message": {"role": "user", "content": "Just one msg"}} + jsonl_file.write_text(json.dumps(entry) + "\n", encoding="utf-8") + + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: [jsonl_file]) + monkeypatch.setattr(time, "sleep", lambda _: None) + + symbolic.bootstrap_from_jsonl() + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Too few messages" in c for c in calls) + + def test_bootstrap_pipeline_failure(self, tmp_path, monkeypatch): + symbolic = _import_symbolic() + + # Create a JSONL file with enough messages + jsonl_file = tmp_path / "-home-patrick-Projects-AIPass-src-aipass-memory" / "session.jsonl" + jsonl_file.parent.mkdir(parents=True) + lines = [] + for i in range(6): + role = "user" if i % 2 == 0 else "assistant" + entry = {"type": role, "message": {"role": role, "content": f"Message {i}"}} + lines.append(json.dumps(entry)) + jsonl_file.write_text("\n".join(lines), encoding="utf-8") + + monkeypatch.setattr(symbolic, "_find_bootstrap_sessions", lambda max_sessions: [jsonl_file]) + monkeypatch.setattr(time, "sleep", lambda _: None) + + # Pipeline fails + _handler_mocks.extractor.extract_fragments_llm.return_value = { + "success": False, + "error": "LLM unavailable", + } + + symbolic.bootstrap_from_jsonl() + calls = [str(c) for c in _handler_mocks.console.print.call_args_list] + assert any("Failed" in c for c in calls) diff --git a/src/aipass/memory/tests/test_templates_display.py b/src/aipass/memory/tests/test_templates_display.py new file mode 100644 index 00000000..6d385396 --- /dev/null +++ b/src/aipass/memory/tests/test_templates_display.py @@ -0,0 +1,675 @@ +# ===================AIPASS==================== +# META DATA HEADER +# Name: tests/test_templates_display.py +# Date: 2026-04-26 +# Version: 1.0.0 +# Category: memory/tests +# ============================================= + +"""Tests for templates module display functions -- line coverage. + +Covers: from aipass.memory.apps.modules.templates import handle_command +""" + +import sys +from pathlib import Path +from unittest.mock import MagicMock + + +# --------------------------------------------------------------------------- +# Helpers: build the full mock graph that templates.py needs at import time +# --------------------------------------------------------------------------- + + +def _prepare_templates_mocks(monkeypatch): + mock_panel = MagicMock() + rich_panel_mod = MagicMock() + rich_panel_mod.Panel = mock_panel + monkeypatch.setitem(sys.modules, "rich.panel", rich_panel_mod) + monkeypatch.setitem(sys.modules, "rich", MagicMock()) + + 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) + + mock_memory_files = MagicMock() + mock_memory_files.read_memory_file_data = MagicMock(return_value=None) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.json.memory_files", mock_memory_files) + + mock_pusher = MagicMock() + mock_pusher.push_templates = MagicMock( + return_value={ + "success": True, + "branches_scanned": 5, + "branches_updated": 2, + "files_modified": 3, + "changes": [], + "errors": [], + } + ) + mock_pusher.get_template_status = MagicMock( + return_value={ + "version": "2.0.0", + "last_push": "2026-03-20", + "local_template_exists": True, + "observations_template_exists": True, + "templates_dir": "/tmp/templates", + "last_push_branches": [], + } + ) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.templates.pusher", mock_pusher) + + mock_differ = MagicMock() + mock_differ.diff_template_vs_branch = MagicMock( + return_value={ + "local": [], + "observations": [], + "errors": [], + } + ) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.templates.differ", mock_differ) + + mock_spawn_pusher = MagicMock() + mock_spawn_pusher.push_to_spawn_templates = MagicMock( + return_value={ + "success": True, + "template_sets_found": [], + "template_sets_updated": 0, + "files_modified": 0, + "changes": [], + } + ) + monkeypatch.setitem(sys.modules, "aipass.memory.apps.handlers.templates.spawn_pusher", mock_spawn_pusher) + + monkeypatch.setitem( + sys.modules, + "aipass.memory.apps.handlers.templates", + MagicMock(pusher=mock_pusher, differ=mock_differ, spawn_pusher=mock_spawn_pusher), + ) + + return { + "console": mock_console, + "error": mock_error, + "warning": mock_warning, + "pusher": mock_pusher, + "differ": mock_differ, + "spawn_pusher": mock_spawn_pusher, + "memory_files": mock_memory_files, + } + + +def _import_templates(monkeypatch): + mocks = _prepare_templates_mocks(monkeypatch) + sys.modules.pop("aipass.memory.apps.modules.templates", None) + parent = sys.modules.get("aipass.memory.apps.modules") + if parent is not None and hasattr(parent, "templates"): + delattr(parent, "templates") + from aipass.memory.apps.modules import templates + + return templates, mocks + + +# =========================================================================== +# Tests: print_help +# =========================================================================== + + +class TestPrintHelp: + """Tests for print_help -- outputs usage panel and command list.""" + + def test_print_help_outputs_usage(self, monkeypatch) -> None: + """print_help calls console.print multiple times for usage info.""" + templates, mocks = _import_templates(monkeypatch) + + templates.print_help() + + # Lines 192-209: 7+ console.print calls (blank lines + panel + usage lines) + assert mocks["console"].print.call_count >= 7 + + +# =========================================================================== +# Tests: print_introspection +# =========================================================================== + + +class TestPrintIntrospection: + """Tests for print_introspection -- displays module identity, handlers, subcommands.""" + + def test_introspection_with_handlers(self, monkeypatch) -> None: + """With discovered handlers, prints handler listing and subcommands.""" + templates, mocks = _import_templates(monkeypatch) + + fake_handlers = { + "json": ["json_handler.py", "memory_files.py"], + "templates": ["pusher.py", "differ.py"], + } + monkeypatch.setattr(templates, "_discover_handlers", lambda: fake_handlers) + + templates.print_introspection() + + # Should print handler dirs and subcommand list + assert mocks["console"].print.call_count >= 10 + + def test_introspection_no_handlers(self, monkeypatch) -> None: + """With no handlers found, prints 'No handlers found' message.""" + templates, mocks = _import_templates(monkeypatch) + + monkeypatch.setattr(templates, "_discover_handlers", lambda: {}) + + templates.print_introspection() + + # Should still print subcommands and hints even with no handlers + assert mocks["console"].print.call_count >= 8 + + +# =========================================================================== +# Tests: _display_push_results +# =========================================================================== + + +class TestDisplayPushResults: + """Tests for _display_push_results -- formats push_templates() handler result.""" + + def test_push_results_success_with_changes(self, monkeypatch) -> None: + """Success with changes list, not dry_run -- displays change details.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": True, + "branches_scanned": 5, + "branches_updated": 2, + "files_modified": 3, + "changes": [ + { + "branch": "CLI", + "file": "local.json", + "changes": ["added field_x", "added field_y"], + }, + ], + "errors": [], + } + + templates._display_push_results(result, dry_run=False) + + # Verify console.print was called (panel, summary, changes, final status) + assert mocks["console"].print.call_count >= 8 + + def test_push_results_success_no_changes(self, monkeypatch) -> None: + """Success with no changes -- displays 'up to date' message.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": True, + "branches_scanned": 5, + "branches_updated": 0, + "files_modified": 0, + "changes": [], + "errors": [], + } + + templates._display_push_results(result, dry_run=False) + + # Should hit the 'All branches are up to date' and 'No updates needed' paths + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "up to date" in joined or "No updates needed" in joined + + def test_push_results_dry_run_with_changes(self, monkeypatch) -> None: + """dry_run=True with changes -- shows DRY RUN label and 'would be updated'.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": True, + "branches_scanned": 5, + "branches_updated": 2, + "files_modified": 3, + "changes": [ + {"branch": "CLI", "file": "local.json", "changes": ["added field"]}, + ], + "errors": [], + } + + templates._display_push_results(result, dry_run=True) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "DRY RUN" in joined + + def test_push_results_failure(self, monkeypatch) -> None: + """success=False with errors -- displays failure message and errors.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": False, + "branches_scanned": 0, + "branches_updated": 0, + "files_modified": 0, + "changes": [], + "errors": ["Registry not found", "Permission denied"], + } + + templates._display_push_results(result, dry_run=False) + + # error() should be called for each error plus the failure header + assert mocks["error"].call_count >= 2 + + def test_push_results_with_errors(self, monkeypatch) -> None: + """success=True but errors list non-empty -- shows both success summary and errors.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": True, + "branches_scanned": 5, + "branches_updated": 2, + "files_modified": 3, + "changes": [ + {"branch": "CLI", "file": "local.json", "changes": ["added field"]}, + ], + "errors": ["Branch ALPHA: permission denied"], + } + + templates._display_push_results(result, dry_run=False) + + # error() called for the error entry + assert mocks["error"].call_count >= 1 + # console.print still used for summary + assert mocks["console"].print.call_count >= 8 + + +# =========================================================================== +# Tests: _display_spawn_push_results +# =========================================================================== + + +class TestDisplaySpawnPushResults: + """Tests for _display_spawn_push_results -- formats spawn template push results.""" + + def test_spawn_push_failure(self, monkeypatch) -> None: + """success=False with errors -- calls error() for each.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": False, + "errors": ["Spawn dir not found", "Template missing"], + } + + templates._display_spawn_push_results(result, dry_run=False) + + assert mocks["error"].call_count >= 2 + + def test_spawn_push_up_to_date(self, monkeypatch) -> None: + """success=True, files_modified=0, no changes -- prints 'up to date'.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": True, + "template_sets_found": ["set_a"], + "template_sets_updated": 0, + "files_modified": 0, + "changes": [], + } + + templates._display_spawn_push_results(result, dry_run=False) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "up to date" in joined + + def test_spawn_push_with_changes(self, monkeypatch) -> None: + """success=True, files_modified>0, changes present -- prints update details.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": True, + "template_sets_found": ["set_a", "set_b"], + "template_sets_updated": 1, + "files_modified": 2, + "changes": [ + {"template_set": "set_a", "file": "local.json", "action": "updated"}, + {"template_set": "set_a", "file": "obs.json", "action": "created"}, + ], + } + + templates._display_spawn_push_results(result, dry_run=False) + + # Should print summary line + change detail lines + assert mocks["console"].print.call_count >= 3 + + def test_spawn_push_dry_run(self, monkeypatch) -> None: + """dry_run=True with changes -- uses 'would update' phrasing.""" + templates, mocks = _import_templates(monkeypatch) + + result = { + "success": True, + "template_sets_found": ["set_a"], + "template_sets_updated": 1, + "files_modified": 1, + "changes": [ + {"template_set": "set_a", "file": "local.json", "action": "updated"}, + ], + } + + templates._display_spawn_push_results(result, dry_run=True) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "would update" in joined + + +# =========================================================================== +# Tests: _display_diff_results +# =========================================================================== + + +class TestDisplayDiffResults: + """Tests for _display_diff_results -- calls differ handler, displays per-branch results.""" + + def _make_branches(self, tmp_path: Path, names: list[str]) -> list[dict]: + """Create branch dicts with real temp directories.""" + branches = [] + for name in names: + branch_dir = tmp_path / name.lower() + branch_dir.mkdir(parents=True, exist_ok=True) + branches.append( + { + "name": name, + "path": str(branch_dir), + "status": "active", + } + ) + return branches + + def test_diff_all_branches_no_diffs(self, tmp_path: Path, monkeypatch) -> None: + """All branches up to date -- shows 'up to date' for each and summary.""" + templates, mocks = _import_templates(monkeypatch) + + branches = self._make_branches(tmp_path, ["CLI", "MEMORY"]) + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: branches) + monkeypatch.setattr( + templates, + "diff_template_vs_branch", + lambda path: {"local": [], "observations": [], "errors": []}, + ) + + templates._display_diff_results(None) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "up to date" in joined.lower() + + def test_diff_with_diffs(self, tmp_path: Path, monkeypatch) -> None: + """Branches have local/obs diffs -- displays diff entries and warning.""" + templates, mocks = _import_templates(monkeypatch) + + branches = self._make_branches(tmp_path, ["CLI", "MEMORY"]) + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: branches) + + def fake_diff(path: str) -> dict: + """Return fake diff result with local and observations diffs.""" + return { + "local": [{"file": "local.json", "additions": ["field_x"]}], + "observations": [{"file": "observations.json", "removals": ["old_field"]}], + "errors": [], + } + + monkeypatch.setattr(templates, "diff_template_vs_branch", fake_diff) + + templates._display_diff_results(None) + + # warning() called because branches have diffs + assert mocks["warning"].call_count >= 1 + + def test_diff_specific_branch(self, tmp_path: Path, monkeypatch) -> None: + """branch_name filter applied -- only diffs the named branch.""" + templates, mocks = _import_templates(monkeypatch) + + branches = self._make_branches(tmp_path, ["CLI", "MEMORY", "DRONE"]) + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: branches) + monkeypatch.setattr( + templates, + "diff_template_vs_branch", + lambda path: {"local": [], "observations": [], "errors": []}, + ) + + templates._display_diff_results("CLI") + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "CLI" in joined + + def test_diff_branch_not_found(self, tmp_path: Path, monkeypatch) -> None: + """Filtered branch not in registry -- displays error.""" + templates, mocks = _import_templates(monkeypatch) + + branches = self._make_branches(tmp_path, ["CLI", "MEMORY"]) + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: branches) + + templates._display_diff_results("NONEXISTENT") + + assert mocks["error"].call_count >= 1 + + def test_diff_registry_load_fails(self, monkeypatch) -> None: + """_load_branches_from_registry returns None -- displays error.""" + templates, mocks = _import_templates(monkeypatch) + + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: None) + + templates._display_diff_results(None) + + assert mocks["error"].call_count >= 1 + + def test_diff_branch_path_missing(self, tmp_path: Path, monkeypatch) -> None: + """Branch path doesn't exist on disk -- displays error and increments error count.""" + templates, mocks = _import_templates(monkeypatch) + + branches = [ + {"name": "GHOST", "path": str(tmp_path / "nonexistent_dir"), "status": "active"}, + ] + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: branches) + + templates._display_diff_results(None) + + # error() called for missing path + assert mocks["error"].call_count >= 1 + + def test_diff_handler_exception(self, tmp_path: Path, monkeypatch) -> None: + """diff_template_vs_branch raises exception -- caught and error displayed.""" + templates, mocks = _import_templates(monkeypatch) + + branches = self._make_branches(tmp_path, ["CLI"]) + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: branches) + monkeypatch.setattr( + templates, + "diff_template_vs_branch", + MagicMock(side_effect=RuntimeError("handler exploded")), + ) + + templates._display_diff_results(None) + + assert mocks["error"].call_count >= 1 + + def test_diff_handler_returns_errors(self, tmp_path: Path, monkeypatch) -> None: + """Diff result has errors list -- errors displayed per branch.""" + templates, mocks = _import_templates(monkeypatch) + + branches = self._make_branches(tmp_path, ["CLI"]) + monkeypatch.setattr(templates, "_load_branches_from_registry", lambda: branches) + monkeypatch.setattr( + templates, + "diff_template_vs_branch", + lambda path: {"local": [], "observations": [], "errors": ["file not readable"]}, + ) + + templates._display_diff_results(None) + + # error() called for the error entry + assert mocks["error"].call_count >= 1 + + +# =========================================================================== +# Tests: _display_file_diffs +# =========================================================================== + + +class TestDisplayFileDiffs: + """Tests for _display_file_diffs -- displays individual file diff entries.""" + + def test_file_diffs_additions(self, monkeypatch) -> None: + """Entries with additions -- prints green + lines.""" + templates, mocks = _import_templates(monkeypatch) + + file_diffs = [ + {"file": "local.json", "additions": ["field_a", "field_b"]}, + ] + + templates._display_file_diffs(file_diffs) + + # One file header line + two addition lines + assert mocks["console"].print.call_count >= 3 + + def test_file_diffs_removals(self, monkeypatch) -> None: + """Entries with removals -- prints red - lines.""" + templates, mocks = _import_templates(monkeypatch) + + file_diffs = [ + {"file": "local.json", "removals": ["old_field"]}, + ] + + templates._display_file_diffs(file_diffs) + + assert mocks["console"].print.call_count >= 2 + + def test_file_diffs_modifications(self, monkeypatch) -> None: + """Entries with modifications -- prints yellow ~ lines.""" + templates, mocks = _import_templates(monkeypatch) + + file_diffs = [ + {"file": "local.json", "modifications": ["changed_field"]}, + ] + + templates._display_file_diffs(file_diffs) + + assert mocks["console"].print.call_count >= 2 + + def test_file_diffs_empty(self, monkeypatch) -> None: + """Empty list -- no console.print calls.""" + templates, mocks = _import_templates(monkeypatch) + + templates._display_file_diffs([]) + + assert mocks["console"].print.call_count == 0 + + +# =========================================================================== +# Tests: _display_status +# =========================================================================== + + +class TestDisplayStatus: + """Tests for _display_status -- displays template status info.""" + + def test_status_templates_exist(self, monkeypatch) -> None: + """local + obs templates found -- shows 'found' labels.""" + templates, mocks = _import_templates(monkeypatch) + + status = { + "version": "2.0.0", + "last_push": "2026-03-20", + "local_template_exists": True, + "observations_template_exists": True, + "templates_dir": "/tmp/templates", + "last_push_branches": [], + } + + templates._display_status(status) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "found" in joined + assert "2.0.0" in joined + + def test_status_templates_missing(self, monkeypatch) -> None: + """Templates not found -- shows 'MISSING' labels.""" + templates, mocks = _import_templates(monkeypatch) + + status = { + "version": None, + "last_push": None, + "local_template_exists": False, + "observations_template_exists": False, + "templates_dir": "/tmp/templates", + "last_push_branches": [], + } + + templates._display_status(status) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "MISSING" in joined + + def test_status_with_pushed_branches(self, monkeypatch) -> None: + """last_push_branches has entries -- shows branch count and names.""" + templates, mocks = _import_templates(monkeypatch) + + status = { + "version": "2.0.0", + "last_push": "2026-03-20", + "local_template_exists": True, + "observations_template_exists": True, + "templates_dir": "/tmp/templates", + "last_push_branches": ["CLI", "MEMORY", "DRONE"], + } + + templates._display_status(status) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "3" in joined + assert "CLI" in joined + + def test_status_many_branches(self, monkeypatch) -> None: + """>8 pushed branches -- display is truncated with '... (+N more)'.""" + templates, mocks = _import_templates(monkeypatch) + + branch_names = [f"BRANCH_{i}" for i in range(12)] + status = { + "version": "2.0.0", + "last_push": "2026-03-20", + "local_template_exists": True, + "observations_template_exists": True, + "templates_dir": "/tmp/templates", + "last_push_branches": branch_names, + } + + templates._display_status(status) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "+4 more" in joined + assert "12" in joined + + def test_status_no_pushed_branches(self, monkeypatch) -> None: + """Empty last_push_branches -- shows 'none'.""" + templates, mocks = _import_templates(monkeypatch) + + status = { + "version": "2.0.0", + "last_push": "2026-03-20", + "local_template_exists": True, + "observations_template_exists": True, + "templates_dir": "/tmp/templates", + "last_push_branches": [], + } + + templates._display_status(status) + + call_args_list = [str(c) for c in mocks["console"].print.call_args_list] + joined = " ".join(call_args_list) + assert "none" in joined