Compare commits
1 Commits
fix/duplic
...
fix/chatlo
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b46fb9d2bf |
137
tests/test_chatlog.py
Normal file
137
tests/test_chatlog.py
Normal 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"
|
||||
Reference in New Issue
Block a user