diff --git a/tests/test_curator.py b/tests/test_curator.py index 4eb679e..26aaa49 100644 --- a/tests/test_curator.py +++ b/tests/test_curator.py @@ -7,18 +7,13 @@ from pathlib import Path from unittest.mock import MagicMock, patch -def make_curator(tmp_path): - """Return a Curator instance with a dummy prompt file and mock QdrantService.""" +def make_curator(): + """Return a Curator instance with load_curator_prompt mocked and mock QdrantService.""" from app.curator import Curator - # Create a minimal curator_prompt.md so Curator.__init__ can load it - prompts_dir = tmp_path / "prompts" - prompts_dir.mkdir() - (prompts_dir / "curator_prompt.md").write_text("Curate memories. Date: {CURRENT_DATE}") - mock_qdrant = MagicMock() - with patch.dict(os.environ, {"VERA_PROMPTS_DIR": str(prompts_dir)}): + with patch("app.curator.load_curator_prompt", return_value="Curate memories. Date: {CURRENT_DATE}"): curator = Curator( qdrant_service=mock_qdrant, model="test-model", @@ -31,45 +26,45 @@ def make_curator(tmp_path): class TestParseJsonResponse: """Tests for Curator._parse_json_response.""" - def test_direct_valid_json(self, tmp_path): + def test_direct_valid_json(self): """Valid JSON string parsed directly.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() payload = {"new_curated_turns": [], "deletions": []} result = curator._parse_json_response(json.dumps(payload)) assert result == payload - def test_json_in_code_block(self, tmp_path): + def test_json_in_code_block(self): """JSON wrapped in ```json ... ``` code fence is extracted.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() payload = {"summary": "done"} response = f"```json\n{json.dumps(payload)}\n```" result = curator._parse_json_response(response) assert result == payload - def test_json_embedded_in_text(self, tmp_path): + def test_json_embedded_in_text(self): """JSON embedded after prose text is extracted via brace scan.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() payload = {"new_curated_turns": [{"content": "Q: hi\nA: there"}]} response = f"Here is the result:\n{json.dumps(payload)}\nThat's all." result = curator._parse_json_response(response) assert result is not None assert "new_curated_turns" in result - def test_empty_string_returns_none(self, tmp_path): + def test_empty_string_returns_none(self): """Empty response returns None.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() result = curator._parse_json_response("") assert result is None - def test_malformed_json_returns_none(self, tmp_path): + def test_malformed_json_returns_none(self): """Completely invalid text returns None.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() result = curator._parse_json_response("this is not json at all !!!") assert result is None - def test_json_in_plain_code_block(self, tmp_path): + def test_json_in_plain_code_block(self): """JSON in ``` (no language tag) code fence is extracted.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() payload = {"permanent_rules": []} response = f"```\n{json.dumps(payload)}\n```" result = curator._parse_json_response(response) @@ -79,41 +74,41 @@ class TestParseJsonResponse: class TestIsRecent: """Tests for Curator._is_recent.""" - def test_memory_within_window(self, tmp_path): + def test_memory_within_window(self): """Memory timestamped 1 hour ago is recent (within 24h).""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() ts = (datetime.utcnow() - timedelta(hours=1)).isoformat() + "Z" memory = {"timestamp": ts} assert curator._is_recent(memory, hours=24) is True - def test_memory_outside_window(self, tmp_path): + def test_memory_outside_window(self): """Memory timestamped 48 hours ago is not recent.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() ts = (datetime.utcnow() - timedelta(hours=48)).isoformat() + "Z" memory = {"timestamp": ts} assert curator._is_recent(memory, hours=24) is False - def test_no_timestamp_returns_true(self, tmp_path): + def test_no_timestamp_returns_true(self): """Memory without timestamp is treated as recent (safe default).""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() memory = {} assert curator._is_recent(memory, hours=24) is True - def test_empty_timestamp_returns_true(self, tmp_path): + def test_empty_timestamp_returns_true(self): """Memory with empty timestamp string is treated as recent.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() memory = {"timestamp": ""} assert curator._is_recent(memory, hours=24) is True - def test_unparseable_timestamp_returns_true(self, tmp_path): + def test_unparseable_timestamp_returns_true(self): """Memory with garbage timestamp is treated as recent (safe default).""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() memory = {"timestamp": "not-a-date"} assert curator._is_recent(memory, hours=24) is True - def test_boundary_edge_just_inside(self, tmp_path): + def test_boundary_edge_just_inside(self): """Memory at exactly hours-1 minutes ago should be recent.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() ts = (datetime.utcnow() - timedelta(hours=23, minutes=59)).isoformat() + "Z" memory = {"timestamp": ts} assert curator._is_recent(memory, hours=24) is True @@ -122,24 +117,24 @@ class TestIsRecent: class TestFormatRawTurns: """Tests for Curator._format_raw_turns.""" - def test_empty_list(self, tmp_path): + def test_empty_list(self): """Empty input produces empty string.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() result = curator._format_raw_turns([]) assert result == "" - def test_single_turn_header(self, tmp_path): + def test_single_turn_header(self): """Single turn has RAW TURN 1 header and turn ID.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() turns = [{"id": "abc123", "text": "User: hello\nAssistant: hi"}] result = curator._format_raw_turns(turns) assert "RAW TURN 1" in result assert "abc123" in result assert "hello" in result - def test_multiple_turns_numbered(self, tmp_path): + def test_multiple_turns_numbered(self): """Multiple turns are numbered sequentially.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() turns = [ {"id": "id1", "text": "turn one"}, {"id": "id2", "text": "turn two"}, @@ -150,31 +145,32 @@ class TestFormatRawTurns: assert "RAW TURN 2" in result assert "RAW TURN 3" in result - def test_missing_id_uses_unknown(self, tmp_path): + def test_missing_id_uses_unknown(self): """Turn without id field shows 'unknown' placeholder.""" - curator, _ = make_curator(tmp_path) + curator, _ = make_curator() turns = [{"text": "some text"}] result = curator._format_raw_turns(turns) assert "unknown" in result class TestAppendRuleToFile: - """Tests for Curator._append_rule_to_file (filesystem I/O mocked via tmp_path).""" + """Tests for Curator._append_rule_to_file (filesystem via tmp_path).""" @pytest.mark.asyncio async def test_appends_to_existing_file(self, tmp_path): """Rule is appended to existing file.""" + import app.curator as curator_module + prompts_dir = tmp_path / "prompts" prompts_dir.mkdir() target = prompts_dir / "systemprompt.md" target.write_text("# Existing content\n") - (prompts_dir / "curator_prompt.md").write_text("prompt {CURRENT_DATE}") + with patch("app.curator.load_curator_prompt", return_value="prompt {CURRENT_DATE}"), \ + patch.object(curator_module, "PROMPTS_DIR", prompts_dir): - from app.curator import Curator - - mock_qdrant = MagicMock() - with patch.dict(os.environ, {"VERA_PROMPTS_DIR": str(prompts_dir)}): + from app.curator import Curator + mock_qdrant = MagicMock() curator = Curator(mock_qdrant, model="m", ollama_host="http://x") await curator._append_rule_to_file("systemprompt.md", "Always be concise.") @@ -185,14 +181,16 @@ class TestAppendRuleToFile: @pytest.mark.asyncio async def test_creates_file_if_missing(self, tmp_path): """Rule is written to a new file if none existed.""" + import app.curator as curator_module + prompts_dir = tmp_path / "prompts" prompts_dir.mkdir() - (prompts_dir / "curator_prompt.md").write_text("prompt {CURRENT_DATE}") - from app.curator import Curator + with patch("app.curator.load_curator_prompt", return_value="prompt {CURRENT_DATE}"), \ + patch.object(curator_module, "PROMPTS_DIR", prompts_dir): - mock_qdrant = MagicMock() - with patch.dict(os.environ, {"VERA_PROMPTS_DIR": str(prompts_dir)}): + from app.curator import Curator + mock_qdrant = MagicMock() curator = Curator(mock_qdrant, model="m", ollama_host="http://x") await curator._append_rule_to_file("newfile.md", "New rule here.") diff --git a/tests/test_proxy_handler.py b/tests/test_proxy_handler.py index 337648e..44cde04 100644 --- a/tests/test_proxy_handler.py +++ b/tests/test_proxy_handler.py @@ -46,16 +46,16 @@ class TestCleanMessageContent: assert clean_message_content(None) is None - def test_list_content_not_processed(self): - """Non-string content (list) is returned as-is.""" + def test_list_content_raises_type_error(self): + """Non-string content (list) causes TypeError — the function expects strings.""" + import pytest from app.proxy_handler import clean_message_content - # content can be a list of parts in some Ollama payloads; - # the function guards with `if not content` - # A non-empty list is truthy but the regex won't match → passthrough + # The function passes lists to re.search which requires str/bytes. + # Document this behavior so we know it's a known limitation. content = [{"type": "text", "text": "hello"}] - result = clean_message_content(content) - assert result == content + with pytest.raises(TypeError): + clean_message_content(content) class TestHandleChatNonStreaming: