test: add comprehensive test coverage (35% → 58%, threshold 50%)
Add 180+ new tests across 10 test files covering previously untested modules: - instructions_cli (0% → 100%), hooks_cli (73% → 96%), spellcheck (28% → 84%) - palace_graph (9% → 91%), general_extractor (0% → 92%), entity_detector (0% → 69%) - entity_registry (0% → 70%), room_detector_local (0% → 55%), layers (0% → 28%) - onboarding (0% → 36%) Also fixes Windows encoding bug in onboarding.py (write_text without encoding="utf-8"). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Tal Muskal
@@ -1,17 +1,24 @@
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from mempalace.hooks_cli import (
|
||||
SAVE_INTERVAL,
|
||||
STOP_BLOCK_REASON,
|
||||
PRECOMPACT_BLOCK_REASON,
|
||||
_count_human_messages,
|
||||
_log,
|
||||
_maybe_auto_ingest,
|
||||
_parse_harness_input,
|
||||
_sanitize_session_id,
|
||||
hook_stop,
|
||||
hook_session_start,
|
||||
hook_precompact,
|
||||
run_hook,
|
||||
)
|
||||
|
||||
|
||||
@@ -190,3 +197,204 @@ def test_precompact_always_blocks(tmp_path):
|
||||
)
|
||||
assert result["decision"] == "block"
|
||||
assert result["reason"] == PRECOMPACT_BLOCK_REASON
|
||||
|
||||
|
||||
# --- _log ---
|
||||
|
||||
|
||||
def test_log_writes_to_hook_log(tmp_path):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
_log("test message")
|
||||
log_path = tmp_path / "hook.log"
|
||||
assert log_path.is_file()
|
||||
content = log_path.read_text()
|
||||
assert "test message" in content
|
||||
|
||||
|
||||
def test_log_oserror_is_silenced(tmp_path):
|
||||
"""_log should not raise if the directory cannot be created."""
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", Path("/nonexistent/deeply/nested/dir")):
|
||||
# Should not raise
|
||||
_log("this will fail silently")
|
||||
|
||||
|
||||
# --- _maybe_auto_ingest ---
|
||||
|
||||
|
||||
def test_maybe_auto_ingest_no_env(tmp_path):
|
||||
"""Without MEMPAL_DIR set, does nothing."""
|
||||
with patch.dict("os.environ", {}, clear=True):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
_maybe_auto_ingest() # should not raise
|
||||
|
||||
|
||||
def test_maybe_auto_ingest_with_env(tmp_path):
|
||||
"""With MEMPAL_DIR set to a valid directory, spawns subprocess."""
|
||||
mempal_dir = tmp_path / "project"
|
||||
mempal_dir.mkdir()
|
||||
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli.subprocess.Popen") as mock_popen:
|
||||
_maybe_auto_ingest()
|
||||
mock_popen.assert_called_once()
|
||||
|
||||
|
||||
def test_maybe_auto_ingest_oserror(tmp_path):
|
||||
"""OSError during subprocess spawn is silenced."""
|
||||
mempal_dir = tmp_path / "project"
|
||||
mempal_dir.mkdir()
|
||||
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli.subprocess.Popen", side_effect=OSError("fail")):
|
||||
_maybe_auto_ingest() # should not raise
|
||||
|
||||
|
||||
# --- _parse_harness_input ---
|
||||
|
||||
|
||||
def test_parse_harness_input_unknown():
|
||||
"""Unknown harness should sys.exit(1)."""
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_parse_harness_input({"session_id": "test"}, "unknown-harness")
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
def test_parse_harness_input_valid():
|
||||
result = _parse_harness_input(
|
||||
{"session_id": "abc-123", "stop_hook_active": True, "transcript_path": "/tmp/t.jsonl"},
|
||||
"claude-code",
|
||||
)
|
||||
assert result["session_id"] == "abc-123"
|
||||
assert result["stop_hook_active"] is True
|
||||
|
||||
|
||||
# --- hook_stop with OSError on write ---
|
||||
|
||||
|
||||
def test_stop_hook_oserror_on_last_save_read(tmp_path):
|
||||
"""When last_save_file has invalid content, falls back to 0."""
|
||||
transcript = tmp_path / "t.jsonl"
|
||||
_write_transcript(transcript, [
|
||||
{"message": {"role": "user", "content": f"msg {i}"}}
|
||||
for i in range(SAVE_INTERVAL)
|
||||
])
|
||||
# Write invalid content to last save file
|
||||
(tmp_path / "test_last_save").write_text("not_a_number")
|
||||
result = _capture_hook_output(
|
||||
hook_stop,
|
||||
{"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)},
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
assert result["decision"] == "block"
|
||||
|
||||
|
||||
def test_stop_hook_oserror_on_write(tmp_path):
|
||||
"""When write to last_save_file fails, hook still outputs correctly."""
|
||||
transcript = tmp_path / "t.jsonl"
|
||||
_write_transcript(transcript, [
|
||||
{"message": {"role": "user", "content": f"msg {i}"}}
|
||||
for i in range(SAVE_INTERVAL)
|
||||
])
|
||||
|
||||
def bad_write_text(*args, **kwargs):
|
||||
raise OSError("disk full")
|
||||
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch.object(Path, "write_text", bad_write_text):
|
||||
result = _capture_hook_output(
|
||||
hook_stop,
|
||||
{"session_id": "test", "stop_hook_active": False, "transcript_path": str(transcript)},
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
assert result["decision"] == "block"
|
||||
|
||||
|
||||
# --- hook_precompact with MEMPAL_DIR ---
|
||||
|
||||
|
||||
def test_precompact_with_mempal_dir(tmp_path):
|
||||
"""Precompact runs subprocess.run when MEMPAL_DIR is set."""
|
||||
mempal_dir = tmp_path / "project"
|
||||
mempal_dir.mkdir()
|
||||
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
||||
with patch("mempalace.hooks_cli.subprocess.run") as mock_run:
|
||||
result = _capture_hook_output(
|
||||
hook_precompact,
|
||||
{"session_id": "test"},
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
assert result["decision"] == "block"
|
||||
mock_run.assert_called_once()
|
||||
|
||||
|
||||
def test_precompact_with_mempal_dir_oserror(tmp_path):
|
||||
"""Precompact handles OSError from subprocess gracefully."""
|
||||
mempal_dir = tmp_path / "project"
|
||||
mempal_dir.mkdir()
|
||||
with patch.dict("os.environ", {"MEMPAL_DIR": str(mempal_dir)}):
|
||||
with patch("mempalace.hooks_cli.subprocess.run", side_effect=OSError("fail")):
|
||||
result = _capture_hook_output(
|
||||
hook_precompact,
|
||||
{"session_id": "test"},
|
||||
state_dir=tmp_path,
|
||||
)
|
||||
assert result["decision"] == "block"
|
||||
|
||||
|
||||
# --- run_hook ---
|
||||
|
||||
|
||||
def test_run_hook_dispatches_session_start(tmp_path):
|
||||
"""run_hook reads stdin JSON and dispatches to correct handler."""
|
||||
stdin_data = json.dumps({"session_id": "run-test"})
|
||||
with patch("sys.stdin", io.StringIO(stdin_data)):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli._output") as mock_output:
|
||||
run_hook("session-start", "claude-code")
|
||||
mock_output.assert_called_once_with({})
|
||||
|
||||
|
||||
def test_run_hook_dispatches_stop(tmp_path):
|
||||
transcript = tmp_path / "t.jsonl"
|
||||
_write_transcript(transcript, [
|
||||
{"message": {"role": "user", "content": f"msg {i}"}}
|
||||
for i in range(3)
|
||||
])
|
||||
stdin_data = json.dumps({
|
||||
"session_id": "run-test",
|
||||
"stop_hook_active": False,
|
||||
"transcript_path": str(transcript),
|
||||
})
|
||||
with patch("sys.stdin", io.StringIO(stdin_data)):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli._output") as mock_output:
|
||||
run_hook("stop", "claude-code")
|
||||
mock_output.assert_called_once_with({})
|
||||
|
||||
|
||||
def test_run_hook_dispatches_precompact(tmp_path):
|
||||
stdin_data = json.dumps({"session_id": "run-test"})
|
||||
with patch("sys.stdin", io.StringIO(stdin_data)):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli._output") as mock_output:
|
||||
run_hook("precompact", "claude-code")
|
||||
mock_output.assert_called_once()
|
||||
call_args = mock_output.call_args[0][0]
|
||||
assert call_args["decision"] == "block"
|
||||
|
||||
|
||||
def test_run_hook_unknown_hook():
|
||||
stdin_data = json.dumps({"session_id": "test"})
|
||||
with patch("sys.stdin", io.StringIO(stdin_data)):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
run_hook("nonexistent", "claude-code")
|
||||
assert exc_info.value.code == 1
|
||||
|
||||
|
||||
def test_run_hook_invalid_json(tmp_path):
|
||||
"""Invalid stdin JSON should not crash — falls back to empty dict."""
|
||||
with patch("sys.stdin", io.StringIO("not valid json")):
|
||||
with patch("mempalace.hooks_cli.STATE_DIR", tmp_path):
|
||||
with patch("mempalace.hooks_cli._output") as mock_output:
|
||||
run_hook("session-start", "claude-code")
|
||||
mock_output.assert_called_once_with({})
|
||||
|
||||
Reference in New Issue
Block a user