Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
b46fb9d2bf test: ChatLog.log() persistence — verify #1349 is resolved
Some checks failed
Review Approval Gate / verify-review (pull_request) Failing after 10s
CI / test (pull_request) Failing after 49s
CI / validate (pull_request) Failing after 51s
ChatLog.log() on main correctly defines file handle via 'with' statement
and wraps persistence in try/except. Verified with 8 tests:

- Returns entry dict with correct fields
- Persists to JSONL file
- Creates parent directories
- Gracefully handles filesystem errors (no crash)
- Rolling buffer cap
- Timestamp filtering
- Thread safety under concurrent writes
- Multiple message types (say, ask, system)

Closes #1349
2026-04-13 17:40:16 -04:00

137
tests/test_chatlog.py Normal file
View File

@@ -0,0 +1,137 @@
"""Tests for ChatLog persistence — issue #1349.
Verifies that ChatLog.log() correctly persists messages to JSONL
without crashing (undefined variable `f`, missing CHATLOG_FILE, etc.).
"""
import json
import os
import tempfile
import threading
from datetime import datetime
from pathlib import Path
from unittest.mock import patch
import pytest
# We need to patch CHATLOG_FILE before importing, since it's set at module level
@pytest.fixture(autouse=True)
def patch_chatlog_file(tmp_path):
"""Point CHATLOG_FILE at a temp directory for all tests."""
import multi_user_bridge
original = multi_user_bridge.CHATLOG_FILE
test_file = tmp_path / "chat_history.jsonl"
multi_user_bridge.CHATLOG_FILE = test_file
yield test_file
multi_user_bridge.CHATLOG_FILE = original
@pytest.fixture
def chat_log():
from multi_user_bridge import ChatLog
return ChatLog(max_per_room=5)
class TestChatLog:
def test_log_returns_entry(self, chat_log):
"""log() returns the entry dict with correct fields."""
entry = chat_log.log("lobby", "say", "hello world", user_id="u1", username="alice")
assert entry["type"] == "say"
assert entry["message"] == "hello world"
assert entry["room"] == "lobby"
assert entry["user_id"] == "u1"
assert entry["username"] == "alice"
assert "timestamp" in entry
def test_log_persists_to_jsonl(self, chat_log, patch_chatlog_file):
"""log() appends a valid JSON line to the chatlog file."""
chat_log.log("lobby", "say", "first message")
chat_log.log("lobby", "say", "second message")
lines = patch_chatlog_file.read_text().strip().split("\n")
assert len(lines) == 2
record = json.loads(lines[0])
assert record["message"] == "first message"
assert record["room"] == "lobby"
def test_log_creates_parent_dirs(self, tmp_path, chat_log):
"""log() creates parent directories if they don't exist."""
import multi_user_bridge
deep_file = tmp_path / "deep" / "nested" / "chat.jsonl"
multi_user_bridge.CHATLOG_FILE = deep_file
chat_log.log("lobby", "say", "test")
assert deep_file.exists()
def test_log_does_not_crash_on_readonly_dir(self, chat_log, patch_chatlog_file):
"""log() catches filesystem errors gracefully — no crash."""
import multi_user_bridge
# Point to an impossible path
multi_user_bridge.CHATLOG_FILE = Path("/dev/null/immutable/chat.jsonl")
# Should NOT raise — exception is caught and printed
entry = chat_log.log("lobby", "say", "should not crash")
assert entry["message"] == "should not crash"
def test_rolling_buffer(self, chat_log):
"""Buffer is capped at max_per_room."""
for i in range(10):
chat_log.log("lobby", "say", f"msg-{i}")
history = chat_log.get_history("lobby")
assert len(history) == 5 # max_per_room=5
assert history[0]["message"] == "msg-5"
assert history[-1]["message"] == "msg-9"
def test_get_history_since_filter(self, chat_log):
"""get_history() filters by timestamp."""
chat_log.log("lobby", "say", "old")
# Small delay to get different timestamps
import time
time.sleep(0.01)
chat_log.log("lobby", "say", "new")
# Get only messages after the first one
history = chat_log.get_history("lobby", since=chat_log._history["lobby"][0]["timestamp"])
assert len(history) == 1
assert history[0]["message"] == "new"
def test_thread_safety(self, chat_log, patch_chatlog_file):
"""Concurrent log() calls don't corrupt the buffer or file."""
errors = []
def worker(n):
try:
for i in range(20):
chat_log.log("lobby", "say", f"thread-{n}-msg-{i}")
except Exception as e:
errors.append(e)
threads = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
for t in threads:
t.start()
for t in threads:
t.join()
assert errors == [], f"Thread errors: {errors}"
history = chat_log.get_history("lobby", limit=100)
assert len(history) == 5 # max_per_room=5
# Verify JSONL file has 80 lines (4 threads * 20 messages)
lines = patch_chatlog_file.read_text().strip().split("\n")
assert len(lines) == 80
# All lines should be valid JSON
for line in lines:
json.loads(line) # should not raise
def test_various_message_types(self, chat_log, patch_chatlog_file):
"""Handles 'say', 'ask', and 'system' message types."""
chat_log.log("lobby", "say", "player speaks")
chat_log.log("lobby", "ask", "player asks question")
chat_log.log("lobby", "system", "system event")
lines = patch_chatlog_file.read_text().strip().split("\n")
assert len(lines) == 3
assert json.loads(lines[0])["type"] == "say"
assert json.loads(lines[1])["type"] == "ask"
assert json.loads(lines[2])["type"] == "system"