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

feat(memory): test(memory): cover 73 untested functions per TDPLAN-0003

Browse Source 
Co-Authored-By: @memory <memory@aipass>
This commit is contained in:
AIOSAI
2026-04-25 23:41:43 -07:00
co-authored by @memory
parent 63a76f6bd6
commit 7aef11adac
5 changed files with 5076 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/memory/tests/test_rollover_pipeline.py
+1094
View File
File diff suppressed because it is too large Load Diff
src/aipass/memory/tests/test_search_extras.py
+596
View File
@@ -0,0 +1,596 @@
# ===================AIPASS====================
# META DATA HEADER
# Name: tests/test_search_extras.py
# Date: 2026-04-25
# Version: 1.0.0
# Category: memory/tests
# =============================================
"""Tests for search handler internals (query_executor and vector_search).
Covers:
from aipass.memory.apps.handlers.search.query_executor import encode_query_subprocess
from aipass.memory.apps.handlers.search.query_executor import search_vectors_subprocess
from aipass.memory.apps.handlers.search.vector_search import search_collection
from aipass.memory.apps.handlers.search.vector_search import encode_query
from aipass.memory.apps.handlers.search.vector_search import search_all_collections
Tests subprocess-based encoding/search, ChromaDB collection queries,
query encoding via the singleton QueryEncoder, and multi-collection search.
All tests use mocks -- no live subprocess, ML model, or ChromaDB access.
"""
import json
import subprocess
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
# ---------------------------------------------------------------------------
# Helpers: prepare the mock graph needed to import search handlers
# ---------------------------------------------------------------------------
def _prepare_query_executor_mocks(monkeypatch):
"""Insert mocks for query_executor module-level imports.
Returns a dict of key mock objects so tests can assert against them.
"""
# Mock the chroma_subprocess and embed_subprocess script paths
mock_chroma_script = MagicMock()
mock_embed_script = MagicMock()
return {
"chroma_script": mock_chroma_script,
"embed_script": mock_embed_script,
}
def _import_query_executor(monkeypatch):
"""Prepare mocks and import (or reimport) query_executor.
Returns (module, mocks_dict).
"""
mocks = _prepare_query_executor_mocks(monkeypatch)
# Remove cached module so it gets re-imported with our mocks
sys.modules.pop("aipass.memory.apps.handlers.search.query_executor", None)
# Clear parent package attribute so Python re-executes module code
parent = sys.modules.get("aipass.memory.apps.handlers.search")
if parent is not None and hasattr(parent, "query_executor"):
delattr(parent, "query_executor")
from aipass.memory.apps.handlers.search import query_executor # noqa: E402
return query_executor, mocks
def _prepare_vector_search_mocks(monkeypatch):
"""Insert mocks for vector_search module-level imports.
Returns a dict of key mock objects for assertions.
"""
# Mock chromadb client via the chroma module
mock_client = MagicMock()
mock_chroma = MagicMock()
mock_chroma.get_client = MagicMock(return_value=mock_client)
monkeypatch.setitem(
sys.modules,
"aipass.memory.apps.handlers.storage.chroma",
mock_chroma,
)
monkeypatch.setitem(
sys.modules,
"aipass.memory.apps.handlers.storage",
MagicMock(),
)
# Mock sentence_transformers and torch for QueryEncoder
mock_model = MagicMock()
mock_model.encode.return_value = MagicMock(tolist=MagicMock(return_value=[0.1] * 384))
mock_model.to.return_value = mock_model
mock_st_cls = MagicMock(return_value=mock_model)
mock_sentence_transformers = MagicMock()
mock_sentence_transformers.SentenceTransformer = mock_st_cls
monkeypatch.setitem(sys.modules, "sentence_transformers", mock_sentence_transformers)
mock_torch = MagicMock()
mock_torch.cuda.is_available.return_value = False
monkeypatch.setitem(sys.modules, "torch", mock_torch)
return {
"client": mock_client,
"model": mock_model,
"st_cls": mock_st_cls,
"torch": mock_torch,
}
def _import_vector_search(monkeypatch):
"""Prepare mocks and import (or reimport) vector_search.
Returns (module, mocks_dict).
"""
mocks = _prepare_vector_search_mocks(monkeypatch)
# Remove cached modules so they get re-imported with our mocks
sys.modules.pop("aipass.memory.apps.handlers.search.vector_search", None)
# Clear parent package attribute so Python re-executes module code
parent = sys.modules.get("aipass.memory.apps.handlers.search")
if parent is not None and hasattr(parent, "vector_search"):
delattr(parent, "vector_search")
from aipass.memory.apps.handlers.search import vector_search # noqa: E402
# Reset singletons for a clean slate
setattr(vector_search, "_query_encoder", None)
setattr(vector_search, "_search_service", None)
setattr(vector_search, "_local_services", {})
return vector_search, mocks
# ===========================================================================
# Tests: encode_query_subprocess
# ===========================================================================
class TestEncodeQuerySubprocess:
"""Verify encode_query_subprocess subprocess-based encoding."""
def test_encode_returns_embedding_on_success(self, monkeypatch):
"""Successful subprocess returns embedding and dimension."""
mod, mocks = _import_query_executor(monkeypatch)
fake_output = json.dumps({"success": True, "embeddings": [[0.1, 0.2, 0.3]], "dimension": 3})
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = fake_output
with patch.object(subprocess, "run", return_value=mock_result) as mock_run:
result = mod.encode_query_subprocess("test query")
assert result["success"] is True
assert result["embedding"] == [0.1, 0.2, 0.3]
assert result["dimension"] == 3
mock_run.assert_called_once()
def test_encode_returns_error_on_nonzero_exit(self, monkeypatch):
"""Non-zero return code produces error dict."""
mod, mocks = _import_query_executor(monkeypatch)
mock_result = MagicMock()
mock_result.returncode = 1
mock_result.stderr = "Model not found"
with patch.object(subprocess, "run", return_value=mock_result):
result = mod.encode_query_subprocess("test query")
assert result["success"] is False
assert "Model not found" in result["error"]
def test_encode_handles_timeout(self, monkeypatch):
"""TimeoutExpired produces a timeout error."""
mod, mocks = _import_query_executor(monkeypatch)
with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="python", timeout=120)):
result = mod.encode_query_subprocess("slow query")
assert result["success"] is False
assert "timed out" in result["error"].lower()
def test_encode_handles_invalid_json(self, monkeypatch):
"""Invalid JSON from subprocess produces error dict."""
mod, mocks = _import_query_executor(monkeypatch)
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = "not valid json"
with patch.object(subprocess, "run", return_value=mock_result):
result = mod.encode_query_subprocess("test query")
assert result["success"] is False
assert "json" in result["error"].lower()
def test_encode_handles_empty_embeddings(self, monkeypatch):
"""Response with empty embeddings list produces error."""
mod, mocks = _import_query_executor(monkeypatch)
fake_output = json.dumps({"success": True, "embeddings": [], "dimension": 384})
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = fake_output
with patch.object(subprocess, "run", return_value=mock_result):
result = mod.encode_query_subprocess("test query")
assert result["success"] is False
assert "no embedding" in result["error"].lower()
def test_encode_handles_generic_exception(self, monkeypatch):
"""Unexpected exception produces error dict."""
mod, mocks = _import_query_executor(monkeypatch)
with patch.object(subprocess, "run", side_effect=OSError("Cannot execute")):
result = mod.encode_query_subprocess("test query")
assert result["success"] is False
assert "Cannot execute" in result["error"]
# ===========================================================================
# Tests: search_vectors_subprocess
# ===========================================================================
class TestSearchVectorsSubprocess:
"""Verify search_vectors_subprocess subprocess-based search."""
def test_search_returns_results_on_success(self, monkeypatch):
"""Successful subprocess returns parsed results."""
mod, mocks = _import_query_executor(monkeypatch)
fake_output = json.dumps(
{"success": True, "results": [{"document": "hello", "distance": 0.1}], "total_results": 1}
)
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = fake_output
with patch.object(subprocess, "run", return_value=mock_result) as mock_run:
result = mod.search_vectors_subprocess(
query_embedding=[0.1, 0.2, 0.3],
branch="TEST",
n_results=5,
)
assert result["success"] is True
assert result["total_results"] == 1
mock_run.assert_called_once()
def test_search_returns_error_on_nonzero_exit(self, monkeypatch):
"""Non-zero exit code produces error dict."""
mod, mocks = _import_query_executor(monkeypatch)
mock_result = MagicMock()
mock_result.returncode = 1
mock_result.stderr = "DB not found"
with patch.object(subprocess, "run", return_value=mock_result):
result = mod.search_vectors_subprocess(query_embedding=[0.1])
assert result["success"] is False
assert "DB not found" in result["error"]
def test_search_handles_timeout(self, monkeypatch):
"""TimeoutExpired produces a timeout error."""
mod, mocks = _import_query_executor(monkeypatch)
with patch.object(subprocess, "run", side_effect=subprocess.TimeoutExpired(cmd="python", timeout=60)):
result = mod.search_vectors_subprocess(query_embedding=[0.1])
assert result["success"] is False
assert "timed out" in result["error"].lower()
def test_search_handles_invalid_json(self, monkeypatch):
"""Invalid JSON from subprocess produces error dict."""
mod, mocks = _import_query_executor(monkeypatch)
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = "{{bad json"
with patch.object(subprocess, "run", return_value=mock_result):
result = mod.search_vectors_subprocess(query_embedding=[0.1])
assert result["success"] is False
assert "json" in result["error"].lower()
def test_search_passes_db_path_as_string(self, monkeypatch):
"""db_path is converted to string in the input data."""
mod, mocks = _import_query_executor(monkeypatch)
fake_output = json.dumps({"success": True, "results": []})
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = fake_output
with patch.object(subprocess, "run", return_value=mock_result) as mock_run:
mod.search_vectors_subprocess(
query_embedding=[0.1],
db_path=Path("/tmp/test_chroma"),
)
call_args = mock_run.call_args
input_data = json.loads(call_args.kwargs.get("input", call_args[1].get("input", "")))
assert input_data["db_path"] == "/tmp/test_chroma"
def test_search_handles_generic_exception(self, monkeypatch):
"""Unexpected exception produces error dict."""
mod, mocks = _import_query_executor(monkeypatch)
with patch.object(subprocess, "run", side_effect=OSError("Cannot execute")):
result = mod.search_vectors_subprocess(query_embedding=[0.1])
assert result["success"] is False
assert "Cannot execute" in result["error"]
# ===========================================================================
# Tests: search_collection (vector_search)
# ===========================================================================
class TestSearchCollection:
"""Verify search_collection wraps SearchService.query_collection."""
def test_search_returns_results_on_success(self, monkeypatch, tmp_path):
"""Successful collection query returns success with results."""
mod, mocks = _import_vector_search(monkeypatch)
mock_collection = MagicMock()
mock_collection.query.return_value = {
"ids": [["id1", "id2"]],
"documents": [["doc1", "doc2"]],
"metadatas": [[{"branch": "TEST"}, {"branch": "TEST"}]],
"distances": [[0.1, 0.2]],
}
mocks["client"].get_collection.return_value = mock_collection
result = mod.search_collection(
query_embedding=[0.1] * 384,
collection_name="test_collection",
n_results=5,
db_path=tmp_path / ".chroma",
)
assert result["success"] is True
assert result["count"] == 2
assert "doc1" in result["documents"]
def test_search_returns_error_for_missing_collection(self, monkeypatch, tmp_path):
"""Missing collection returns success=False with error message."""
mod, mocks = _import_vector_search(monkeypatch)
mocks["client"].get_collection.side_effect = ValueError("Collection not found")
result = mod.search_collection(
query_embedding=[0.1] * 384,
collection_name="nonexistent",
db_path=tmp_path / ".chroma",
)
assert result["success"] is False
assert "not found" in result["error"].lower()
def test_search_returns_error_for_empty_embedding(self, monkeypatch):
"""Empty query embedding returns error."""
mod, mocks = _import_vector_search(monkeypatch)
result = mod.search_collection(
query_embedding=[],
collection_name="test_collection",
)
assert result["success"] is False
assert "no query embedding" in result["error"].lower()
def test_search_accepts_string_db_path(self, monkeypatch, tmp_path):
"""String db_path is converted to Path internally."""
mod, mocks = _import_vector_search(monkeypatch)
mock_collection = MagicMock()
mock_collection.query.return_value = {
"ids": [["id1"]],
"documents": [["doc1"]],
"metadatas": [[{}]],
"distances": [[0.05]],
}
mocks["client"].get_collection.return_value = mock_collection
result = mod.search_collection(
query_embedding=[0.1] * 384,
collection_name="test_collection",
db_path=str(tmp_path / ".chroma"),
)
assert result["success"] is True
def test_search_passes_where_filter(self, monkeypatch, tmp_path):
"""Metadata where filter is forwarded to collection.query."""
mod, mocks = _import_vector_search(monkeypatch)
mock_collection = MagicMock()
mock_collection.query.return_value = {
"ids": [[]],
"documents": [[]],
"metadatas": [[]],
"distances": [[]],
}
mocks["client"].get_collection.return_value = mock_collection
mod.search_collection(
query_embedding=[0.1] * 384,
collection_name="test_collection",
where={"branch": "SEEDGO"},
db_path=tmp_path / ".chroma",
)
call_kwargs = mock_collection.query.call_args.kwargs
assert call_kwargs["where"] == {"branch": "SEEDGO"}
# ===========================================================================
# Tests: encode_query (vector_search)
# ===========================================================================
class TestEncodeQuery:
"""Verify encode_query uses QueryEncoder singleton."""
def test_encode_returns_embedding(self, monkeypatch):
"""Successful encoding returns embedding with dimension."""
mod, mocks = _import_vector_search(monkeypatch)
result = mod.encode_query("test query")
assert result["success"] is True
assert len(result["embedding"]) == 384
assert result["dimension"] == 384
assert result["model"] == "all-MiniLM-L6-v2"
def test_encode_rejects_empty_query(self, monkeypatch):
"""Empty query string returns error."""
mod, mocks = _import_vector_search(monkeypatch)
result = mod.encode_query("")
assert result["success"] is False
assert "empty" in result["error"].lower()
def test_encode_rejects_whitespace_only_query(self, monkeypatch):
"""Whitespace-only query string returns error."""
mod, mocks = _import_vector_search(monkeypatch)
result = mod.encode_query(" ")
assert result["success"] is False
assert "empty" in result["error"].lower()
def test_encode_handles_model_error(self, monkeypatch):
"""If model.encode raises, error is caught and returned."""
mod, mocks = _import_vector_search(monkeypatch)
mocks["model"].encode.side_effect = RuntimeError("CUDA out of memory")
result = mod.encode_query("test query")
assert result["success"] is False
assert "CUDA" in result["error"]
def test_encode_uses_singleton(self, monkeypatch):
"""Successive calls reuse the same QueryEncoder instance."""
mod, mocks = _import_vector_search(monkeypatch)
mod.encode_query("first query")
mod.encode_query("second query")
# SentenceTransformer should only be constructed once
mocks["st_cls"].assert_called_once()
# ===========================================================================
# Tests: search_all_collections (vector_search)
# ===========================================================================
class TestSearchAllCollections:
"""Verify search_all_collections aggregates results across collections."""
def test_search_all_returns_aggregated_results(self, monkeypatch, tmp_path):
"""Searching all collections returns results from each."""
mod, mocks = _import_vector_search(monkeypatch)
mock_coll_a = MagicMock()
mock_coll_a.name = "coll_a"
mock_coll_b = MagicMock()
mock_coll_b.name = "coll_b"
mocks["client"].list_collections.return_value = [mock_coll_a, mock_coll_b]
mock_collection = MagicMock()
mock_collection.query.return_value = {
"ids": [["id1"]],
"documents": [["doc1"]],
"metadatas": [[{"branch": "TEST"}]],
"distances": [[0.1]],
}
mocks["client"].get_collection.return_value = mock_collection
result = mod.search_all_collections(
query_embedding=[0.1] * 384,
n_results=3,
db_path=tmp_path / ".chroma",
)
assert result["success"] is True
assert result["collections_searched"] == 2
assert result["total_results"] == 2
assert "coll_a" in result["results"]
assert "coll_b" in result["results"]
def test_search_all_returns_empty_when_no_collections(self, monkeypatch, tmp_path):
"""No collections returns success with empty results."""
mod, mocks = _import_vector_search(monkeypatch)
mocks["client"].list_collections.return_value = []
result = mod.search_all_collections(
query_embedding=[0.1] * 384,
db_path=tmp_path / ".chroma",
)
assert result["success"] is True
assert result["results"] == {}
assert "no collections" in result["message"].lower()
def test_search_all_returns_error_for_empty_embedding(self, monkeypatch):
"""Empty embedding returns error."""
mod, mocks = _import_vector_search(monkeypatch)
result = mod.search_all_collections(query_embedding=[])
assert result["success"] is False
assert "no query embedding" in result["error"].lower()
def test_search_all_handles_exception(self, monkeypatch, tmp_path):
"""Exception during search returns error dict."""
mod, mocks = _import_vector_search(monkeypatch)
mocks["client"].list_collections.side_effect = RuntimeError("DB corrupted")
result = mod.search_all_collections(
query_embedding=[0.1] * 384,
db_path=tmp_path / ".chroma",
)
assert result["success"] is False
assert "DB corrupted" in result["error"]
def test_search_all_accepts_string_db_path(self, monkeypatch, tmp_path):
"""String db_path is accepted and converted."""
mod, mocks = _import_vector_search(monkeypatch)
mocks["client"].list_collections.return_value = []
result = mod.search_all_collections(
query_embedding=[0.1] * 384,
db_path=str(tmp_path / ".chroma"),
)
assert result["success"] is True
def test_search_all_skips_nonexistent_collections(self, monkeypatch, tmp_path):
"""Collections that fail get_collection are excluded from results."""
mod, mocks = _import_vector_search(monkeypatch)
mock_coll_a = MagicMock()
mock_coll_a.name = "coll_a"
mocks["client"].list_collections.return_value = [mock_coll_a]
# get_collection raises => query_collection returns exists=False
mocks["client"].get_collection.side_effect = ValueError("Collection not found")
result = mod.search_all_collections(
query_embedding=[0.1] * 384,
db_path=tmp_path / ".chroma",
)
assert result["success"] is True
assert result["collections_searched"] == 0
assert result["total_results"] == 0
src/aipass/memory/tests/test_symbolic_extras.py
+1651
View File
File diff suppressed because it is too large Load Diff
src/aipass/memory/tests/test_symbolic_module.py
+1243
View File
File diff suppressed because it is too large Load Diff
src/aipass/memory/tests/test_watcher.py
+492
View File
@@ -0,0 +1,492 @@
# ===================AIPASS====================
# META DATA HEADER
# Name: tests/test_watcher.py
# Date: 2026-04-25
# Version: 1.0.0
# Category: memory/tests
# =============================================
"""Tests for the memory watcher handler.
Covers:
from aipass.memory.apps.handlers.monitor.memory_watcher import start_memory_watcher
from aipass.memory.apps.handlers.monitor.memory_watcher import stop_memory_watcher
from aipass.memory.apps.handlers.monitor.memory_watcher import is_memory_watcher_active
from aipass.memory.apps.handlers.monitor.memory_watcher import get_watcher_status
from aipass.memory.apps.handlers.monitor.memory_watcher import MemoryFileWatcher
Tests watcher lifecycle (start/stop/status), the MemoryFileWatcher.on_modified
callback, and edge cases like missing watchdog or already-running observers.
All tests use mocks -- no live filesystem watchers or infrastructure access.
"""
import sys
from unittest.mock import MagicMock
# ---------------------------------------------------------------------------
# Helpers: prepare the mock graph needed to import memory_watcher
# ---------------------------------------------------------------------------
def _prepare_watcher_mocks(monkeypatch):
"""Insert mocks for every module-level import memory_watcher.py touches.
Returns a dict of key mock objects so tests can assert against them.
"""
# Mock line_counter
mock_update_line_count = MagicMock(return_value={"success": True, "lines": 150})
mock_line_counter = MagicMock()
mock_line_counter.update_line_count = mock_update_line_count
monkeypatch.setitem(
sys.modules,
"aipass.memory.apps.handlers.tracking.line_counter",
mock_line_counter,
)
# Mock detector
mock_check_single_file = MagicMock(return_value={"success": True, "should_rollover": False})
mock_detector = MagicMock()
mock_detector.check_single_file = mock_check_single_file
monkeypatch.setitem(
sys.modules,
"aipass.memory.apps.handlers.monitor.detector",
mock_detector,
)
# Mock watchdog
mock_observer_instance = MagicMock()
mock_observer_instance.is_alive.return_value = True
mock_observer_cls = MagicMock(return_value=mock_observer_instance)
mock_watchdog_observers = MagicMock()
mock_watchdog_observers.Observer = mock_observer_cls
mock_fse_handler = type("FileSystemEventHandler", (), {"__init__": lambda self: None})
mock_watchdog_events = MagicMock()
mock_watchdog_events.FileSystemEventHandler = mock_fse_handler
monkeypatch.setitem(sys.modules, "watchdog", MagicMock())
monkeypatch.setitem(sys.modules, "watchdog.observers", mock_watchdog_observers)
monkeypatch.setitem(sys.modules, "watchdog.events", mock_watchdog_events)
# Mock rollover orchestrator (lazy import inside on_modified)
mock_execute_rollover = MagicMock(return_value={"success": True})
mock_orchestrator = MagicMock()
mock_orchestrator.execute_rollover = mock_execute_rollover
monkeypatch.setitem(
sys.modules,
"aipass.memory.apps.handlers.rollover",
MagicMock(),
)
monkeypatch.setitem(
sys.modules,
"aipass.memory.apps.handlers.rollover.orchestrator",
mock_orchestrator,
)
return {
"update_line_count": mock_update_line_count,
"check_single_file": mock_check_single_file,
"observer_instance": mock_observer_instance,
"observer_cls": mock_observer_cls,
"execute_rollover": mock_execute_rollover,
}
def _import_watcher(monkeypatch):
"""Prepare mocks and import (or reimport) memory_watcher.
Returns (module, mocks_dict).
"""
mocks = _prepare_watcher_mocks(monkeypatch)
# Remove cached module so it gets re-imported with our mocks
sys.modules.pop("aipass.memory.apps.handlers.monitor.memory_watcher", None)
# Also clear parent package's cached attribute so Python re-executes
# the module code with fresh mocks instead of returning a stale ref.
parent = sys.modules.get("aipass.memory.apps.handlers.monitor")
if parent is not None and hasattr(parent, "memory_watcher"):
delattr(parent, "memory_watcher")
from aipass.memory.apps.handlers.monitor import memory_watcher # noqa: E402
# Reset the global _observer to None for a clean slate each test
setattr(memory_watcher, "_observer", None)
return memory_watcher, mocks
# ===========================================================================
# Tests: start_memory_watcher
# ===========================================================================
class TestStartMemoryWatcher:
"""Verify start_memory_watcher lifecycle."""
def test_start_returns_success_with_paths(self, monkeypatch, tmp_path):
"""Starting watcher with valid branch paths returns success."""
mod, mocks = _import_watcher(monkeypatch)
branch_path = tmp_path / "src" / "aipass" / "test_branch"
branch_path.mkdir(parents=True)
monkeypatch.setattr(mod, "_get_branch_paths", lambda: [branch_path])
result = mod.start_memory_watcher()
assert result["success"] is True
assert result["count"] == 1
assert str(branch_path) in result["watched_paths"]
mocks["observer_cls"].assert_called_once()
mocks["observer_instance"].start.assert_called_once()
def test_start_fails_when_already_running(self, monkeypatch, tmp_path):
"""Starting when observer is already alive returns error."""
mod, mocks = _import_watcher(monkeypatch)
# Simulate already running observer
mock_existing = MagicMock()
mock_existing.is_alive.return_value = True
setattr(mod, "_observer", mock_existing)
result = mod.start_memory_watcher()
assert result["success"] is False
assert "already running" in result["error"].lower()
def test_start_fails_when_no_branches(self, monkeypatch):
"""Starting with no branch paths returns error."""
mod, mocks = _import_watcher(monkeypatch)
monkeypatch.setattr(mod, "_get_branch_paths", lambda: [])
result = mod.start_memory_watcher()
assert result["success"] is False
assert "no branch" in result["error"].lower()
def test_start_handles_schedule_error_gracefully(self, monkeypatch, tmp_path):
"""If scheduling a path raises an exception, other paths still work."""
mod, mocks = _import_watcher(monkeypatch)
good_path = tmp_path / "good_branch"
good_path.mkdir(parents=True)
bad_path = tmp_path / "bad_branch"
bad_path.mkdir(parents=True)
call_count = 0
def _mock_schedule(handler, path, recursive=False):
"""Mock observer.schedule that fails on the first call."""
nonlocal call_count
call_count += 1
if call_count == 1:
raise OSError("Permission denied")
mocks["observer_instance"].schedule = _mock_schedule
monkeypatch.setattr(mod, "_get_branch_paths", lambda: [bad_path, good_path])
result = mod.start_memory_watcher()
assert result["success"] is True
# Only the second path should succeed
assert result["count"] == 1
# ===========================================================================
# Tests: stop_memory_watcher
# ===========================================================================
class TestStopMemoryWatcher:
"""Verify stop_memory_watcher lifecycle."""
def test_stop_returns_success(self, monkeypatch):
"""Stopping a running watcher returns success."""
mod, mocks = _import_watcher(monkeypatch)
# Set up a running observer
mock_obs = MagicMock()
mock_obs.is_alive.return_value = True
setattr(mod, "_observer", mock_obs)
result = mod.stop_memory_watcher()
assert result["success"] is True
mock_obs.stop.assert_called_once()
mock_obs.join.assert_called_once()
assert mod._observer is None # type: ignore[union-attr]
def test_stop_fails_when_not_running(self, monkeypatch):
"""Stopping when no watcher is running returns error."""
mod, mocks = _import_watcher(monkeypatch)
result = mod.stop_memory_watcher()
assert result["success"] is False
assert "not running" in result["error"].lower()
def test_stop_fails_when_observer_not_alive(self, monkeypatch):
"""Stopping when observer exists but is not alive returns error."""
mod, mocks = _import_watcher(monkeypatch)
mock_obs = MagicMock()
mock_obs.is_alive.return_value = False
setattr(mod, "_observer", mock_obs)
result = mod.stop_memory_watcher()
assert result["success"] is False
# ===========================================================================
# Tests: is_memory_watcher_active
# ===========================================================================
class TestIsMemoryWatcherActive:
"""Verify is_memory_watcher_active boolean checks."""
def test_active_when_observer_alive(self, monkeypatch):
"""Returns True when observer is alive."""
mod, mocks = _import_watcher(monkeypatch)
mock_obs = MagicMock()
mock_obs.is_alive.return_value = True
setattr(mod, "_observer", mock_obs)
assert mod.is_memory_watcher_active() is True
def test_inactive_when_no_observer(self, monkeypatch):
"""Returns False when _observer is None."""
mod, mocks = _import_watcher(monkeypatch)
assert mod.is_memory_watcher_active() is False
def test_inactive_when_observer_not_alive(self, monkeypatch):
"""Returns False when observer exists but is_alive returns False."""
mod, mocks = _import_watcher(monkeypatch)
mock_obs = MagicMock()
mock_obs.is_alive.return_value = False
setattr(mod, "_observer", mock_obs)
assert mod.is_memory_watcher_active() is False
# ===========================================================================
# Tests: get_watcher_status
# ===========================================================================
class TestGetWatcherStatus:
"""Verify get_watcher_status returns correct status info."""
def test_status_when_inactive(self, monkeypatch):
"""Returns inactive status when watcher is not running."""
mod, mocks = _import_watcher(monkeypatch)
result = mod.get_watcher_status()
assert result["active"] is False
assert "not running" in result["message"].lower()
def test_status_when_active(self, monkeypatch, tmp_path):
"""Returns active status with directory count when watcher is running."""
mod, mocks = _import_watcher(monkeypatch)
mock_obs = MagicMock()
mock_obs.is_alive.return_value = True
setattr(mod, "_observer", mock_obs)
branch_path = tmp_path / "branch1"
branch_path.mkdir()
monkeypatch.setattr(mod, "_get_branch_paths", lambda: [branch_path])
result = mod.get_watcher_status()
assert result["active"] is True
assert result["watched_directories"] == 1
assert str(branch_path) in result["paths"]
def test_status_returns_multiple_paths(self, monkeypatch, tmp_path):
"""Returns all watched directory paths."""
mod, mocks = _import_watcher(monkeypatch)
mock_obs = MagicMock()
mock_obs.is_alive.return_value = True
setattr(mod, "_observer", mock_obs)
paths = []
for name in ["branch_a", "branch_b", "branch_c"]:
p = tmp_path / name
p.mkdir()
paths.append(p)
monkeypatch.setattr(mod, "_get_branch_paths", lambda: paths)
result = mod.get_watcher_status()
assert result["watched_directories"] == 3
assert len(result["paths"]) == 3
# ===========================================================================
# Tests: MemoryFileWatcher.on_modified
# ===========================================================================
class TestMemoryFileWatcherOnModified:
"""Verify MemoryFileWatcher.on_modified callback behavior."""
def test_ignores_directory_events(self, monkeypatch):
"""Directory modification events are ignored."""
mod, mocks = _import_watcher(monkeypatch)
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = True
event.src_path = "/some/.trinity/local.json"
watcher.on_modified(event)
mocks["update_line_count"].assert_not_called()
def test_ignores_non_memory_files(self, monkeypatch):
"""Non-memory files (not in .trinity/) are ignored."""
mod, mocks = _import_watcher(monkeypatch)
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = False
event.src_path = "/some/dir/config.json"
watcher.on_modified(event)
mocks["update_line_count"].assert_not_called()
def test_processes_memory_file_modification(self, monkeypatch):
"""Valid memory file modification triggers line count update and check."""
mod, mocks = _import_watcher(monkeypatch)
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = False
event.src_path = "/some/branch/.trinity/local.json"
watcher.on_modified(event)
mocks["update_line_count"].assert_called_once()
mocks["check_single_file"].assert_called_once()
def test_rollover_triggered_when_threshold_exceeded(self, monkeypatch):
"""When check_single_file says should_rollover, execute_rollover is called."""
mod, mocks = _import_watcher(monkeypatch)
mocks["check_single_file"].return_value = {
"success": True,
"should_rollover": True,
"trigger": "lines exceeded 600",
}
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = False
event.src_path = "/some/branch/.trinity/local.json"
watcher.on_modified(event)
mocks["execute_rollover"].assert_called_once()
def test_skips_recently_modified_file(self, monkeypatch):
"""Files already in the recent modifications set are skipped."""
mod, mocks = _import_watcher(monkeypatch)
watcher = mod.MemoryFileWatcher()
file_path = "/some/branch/.trinity/local.json"
watcher._recent_modifications.add(file_path)
event = MagicMock()
event.is_directory = False
event.src_path = file_path
watcher.on_modified(event)
# Should skip and not call update_line_count
mocks["update_line_count"].assert_not_called()
# The file key should be removed from recent modifications after skip
assert file_path not in watcher._recent_modifications
def test_handles_line_count_update_failure(self, monkeypatch):
"""When update_line_count fails, check_single_file is not called."""
mod, mocks = _import_watcher(monkeypatch)
mocks["update_line_count"].return_value = {
"success": False,
"error": "File not found",
}
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = False
event.src_path = "/some/branch/.trinity/local.json"
watcher.on_modified(event)
mocks["update_line_count"].assert_called_once()
mocks["check_single_file"].assert_not_called()
def test_handles_check_failure(self, monkeypatch):
"""When check_single_file fails, rollover is not triggered."""
mod, mocks = _import_watcher(monkeypatch)
mocks["check_single_file"].return_value = {
"success": False,
"error": "Read error",
}
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = False
event.src_path = "/some/branch/.trinity/local.json"
watcher.on_modified(event)
mocks["check_single_file"].assert_called_once()
mocks["execute_rollover"].assert_not_called()
def test_processes_observations_json(self, monkeypatch):
"""observations.json files in .trinity/ are also processed."""
mod, mocks = _import_watcher(monkeypatch)
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = False
event.src_path = "/some/branch/.trinity/observations.json"
watcher.on_modified(event)
mocks["update_line_count"].assert_called_once()
def test_rollover_exception_does_not_propagate(self, monkeypatch):
"""If execute_rollover raises, the exception is caught."""
mod, mocks = _import_watcher(monkeypatch)
mocks["check_single_file"].return_value = {
"success": True,
"should_rollover": True,
"trigger": "lines exceeded",
}
mocks["execute_rollover"].side_effect = RuntimeError("Rollover crashed")
watcher = mod.MemoryFileWatcher()
event = MagicMock()
event.is_directory = False
event.src_path = "/some/branch/.trinity/local.json"
# Should not raise
watcher.on_modified(event)
mocks["execute_rollover"].assert_called_once()