test: add unit tests for multi_user_bridge.py (#1541)

tests/test_multi_user_bridge.py

@@ -0,0 +1,313 @@
+#!/usr/bin/env python3
+"""
+Unit tests for multi_user_bridge.py
+
+23 tests across 5 test classes:
+- TestPluginRegistry: Register, unregister, list, fire hooks, thread-safety
+- TestChatLogIsolation: Room isolation, rolling buffer, history, thread-safety
+- TestPresenceManager: Enter/leave, room queries, isolation, concurrency
+- TestConcurrentUsers: Multi-user concurrent chat with isolation verification
+"""
+
+import json
+import os
+import sys
+import threading
+import time
+import pytest
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+# Add repo root to path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+# Mock WORLD_DIR before importing to prevent file writes
+with patch.dict(os.environ, {"TIMMY_BRIDGE_PORT": "0"}):
+    import multi_user_bridge as mub
+
+
+# ─── TestPluginRegistry ──────────────────────────────────────────────
+
+class TestPluginRegistry:
+    """Test the plugin registry system."""
+
+    def setup_method(self):
+        self.registry = mub.PluginRegistry()
+
+    def test_register_plugin(self):
+        plugin = mub.Plugin("test", "A test plugin")
+        self.registry.register(plugin)
+        assert self.registry.get("test") is plugin
+
+    def test_unregister_plugin(self):
+        plugin = mub.Plugin("test", "A test plugin")
+        self.registry.register(plugin)
+        result = self.registry.unregister("test")
+        assert result is True
+        assert self.registry.get("test") is None
+
+    def test_unregister_nonexistent(self):
+        result = self.registry.unregister("nonexistent")
+        assert result is False
+
+    def test_list_plugins(self):
+        p1 = mub.Plugin("alpha", "First")
+        p2 = mub.Plugin("beta", "Second")
+        self.registry.register(p1)
+        self.registry.register(p2)
+        names = [p["name"] for p in self.registry.list_plugins()]
+        assert "alpha" in names
+        assert "beta" in names
+
+    def test_fire_on_message(self):
+        class EchoPlugin(mub.Plugin):
+            def on_message(self, user_id, message, room):
+                return f"echo: {message}"
+
+        self.registry.register(EchoPlugin("echo", "Echoes"))
+        result = self.registry.fire_on_message("u1", "hello", "garden")
+        assert result == "echo: hello"
+
+    def test_fire_on_message_no_handler(self):
+        plugin = mub.Plugin("noop", "Does nothing")
+        self.registry.register(plugin)
+        result = self.registry.fire_on_message("u1", "hello", "garden")
+        assert result is None
+
+    def test_thread_safety(self):
+        errors = []
+
+        def register_many(prefix):
+            try:
+                for i in range(50):
+                    p = mub.Plugin(f"{prefix}_{i}", f"Plugin {i}")
+                    self.registry.register(p)
+            except Exception as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=register_many, args=(f"t{t}",)) for t in range(4)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        assert len(errors) == 0
+        assert len(self.registry.list_plugins()) == 200
+
+
+# ─── TestChatLogIsolation ────────────────────────────────────────────
+
+class TestChatLogIsolation:
+    """Test chat log room isolation and rolling buffer."""
+
+    def setup_method(self):
+        # Patch CHATLOG_FILE to prevent disk writes
+        self._patcher = patch.object(mub, "CHATLOG_FILE", Path("/dev/null"))
+        self._patcher.start()
+        self.chat_log = mub.ChatLog(max_per_room=5)
+
+    def teardown_method(self):
+        self._patcher.stop()
+
+    def test_room_isolation(self):
+        self.chat_log.log("garden", "say", "Hello garden", user_id="u1")
+        self.chat_log.log("tower", "say", "Hello tower", user_id="u2")
+
+        garden_msgs = self.chat_log.get_history("garden")
+        tower_msgs = self.chat_log.get_history("tower")
+
+        assert len(garden_msgs) == 1
+        assert garden_msgs[0]["message"] == "Hello garden"
+        assert len(tower_msgs) == 1
+        assert tower_msgs[0]["message"] == "Hello tower"
+
+    def test_rolling_buffer(self):
+        for i in range(10):
+            self.chat_log.log("garden", "say", f"msg_{i}", user_id="u1")
+
+        history = self.chat_log.get_history("garden")
+        assert len(history) == 5  # max_per_room
+        assert history[0]["message"] == "msg_5"  # oldest kept
+        assert history[-1]["message"] == "msg_9"  # newest
+
+    def test_get_all_rooms(self):
+        self.chat_log.log("garden", "say", "hi", user_id="u1")
+        self.chat_log.log("tower", "say", "hi", user_id="u1")
+        rooms = self.chat_log.get_all_rooms()
+        assert set(rooms) == {"garden", "tower"}
+
+    def test_history_with_limit(self):
+        for i in range(10):
+            self.chat_log.log("garden", "say", f"msg_{i}", user_id="u1")
+
+        limited = self.chat_log.get_history("garden", limit=3)
+        assert len(limited) == 3
+        assert limited[0]["message"] == "msg_7"
+
+    def test_empty_room_history(self):
+        history = self.chat_log.get_history("nonexistent")
+        assert history == []
+
+    def test_message_types(self):
+        self.chat_log.log("garden", "say", "said", user_id="u1")
+        self.chat_log.log("garden", "ask", "asked", user_id="u1")
+        self.chat_log.log("garden", "system", "notice", user_id=None)
+
+        history = self.chat_log.get_history("garden")
+        types = [m["type"] for m in history]
+        assert "say" in types
+        assert "ask" in types
+        assert "system" in types
+
+    def test_since_filter(self):
+        self.chat_log.log("garden", "say", "old", user_id="u1")
+        time.sleep(0.05)
+        cutoff = time.strftime("%Y-%m-%dT%H:%M:%S")
+        time.sleep(0.05)
+        self.chat_log.log("garden", "say", "new", user_id="u1")
+
+        recent = self.chat_log.get_history("garden", since=cutoff)
+        assert len(recent) == 1
+        assert recent[0]["message"] == "new"
+
+    def test_thread_safety(self):
+        errors = []
+
+        def log_many(room, n):
+            try:
+                for i in range(n):
+                    self.chat_log.log(room, "say", f"{room}_{i}", user_id="u1")
+            except Exception as e:
+                errors.append(e)
+
+        threads = [
+            threading.Thread(target=log_many, args=("garden", 50)),
+            threading.Thread(target=log_many, args=("tower", 50)),
+        ]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert len(errors) == 0
+        # Each room should have exactly max_per_room (5) messages
+        assert len(self.chat_log.get_history("garden")) == 5
+        assert len(self.chat_log.get_history("tower")) == 5
+
+
+# ─── TestPresenceManager ─────────────────────────────────────────────
+
+class TestPresenceManager:
+    """Test presence tracking (enter/leave/room queries)."""
+
+    def setup_method(self):
+        self.pm = mub.PresenceManager()
+
+    def test_enter_room(self):
+        event = self.pm.enter_room("u1", "Alice", "garden")
+        assert event["event"] == "enter"
+        assert event["user_id"] == "u1"
+        assert event["room"] == "garden"
+
+    def test_leave_room(self):
+        self.pm.enter_room("u1", "Alice", "garden")
+        event = self.pm.leave_room("u1", "garden")
+        assert event is not None
+        assert event["event"] == "leave"
+
+    def test_leave_nonexistent(self):
+        event = self.pm.leave_room("u1", "garden")
+        assert event is None
+
+    def test_get_players_in_room(self):
+        self.pm.enter_room("u1", "Alice", "garden")
+        self.pm.enter_room("u2", "Bob", "garden")
+        self.pm.enter_room("u3", "Charlie", "tower")
+
+        players = self.pm.get_players_in_room("garden")
+        names = {p["username"] for p in players}
+        assert names == {"Alice", "Bob"}
+
+    def test_room_isolation(self):
+        self.pm.enter_room("u1", "Alice", "garden")
+        self.pm.enter_room("u2", "Bob", "tower")
+
+        garden_players = {p["user_id"] for p in self.pm.get_players_in_room("garden")}
+        tower_players = {p["user_id"] for p in self.pm.get_players_in_room("tower")}
+
+        assert garden_players == {"u1"}
+        assert tower_players == {"u2"}
+
+    def test_cleanup_user(self):
+        self.pm.enter_room("u1", "Alice", "garden")
+        self.pm.enter_room("u1", "Alice", "tower")
+
+        events = self.pm.cleanup_user("u1")
+        assert len(events) == 2
+
+        assert len(self.pm.get_players_in_room("garden")) == 0
+        assert len(self.pm.get_players_in_room("tower")) == 0
+
+    def test_say_event(self):
+        self.pm.enter_room("u1", "Alice", "garden")
+        event = self.pm.say("u1", "Alice", "garden", "Hello!")
+        assert event["type"] == "say"
+        assert event["message"] == "Hello!"
+
+    def test_get_room_events(self):
+        self.pm.enter_room("u1", "Alice", "garden")
+        self.pm.say("u1", "Alice", "garden", "Hi!")
+        self.pm.leave_room("u1", "garden")
+
+        events = self.pm.get_room_events("garden")
+        types = [e["event"] for e in events]
+        assert "enter" in types
+        assert "message" in types
+        assert "leave" in types
+
+
+# ─── TestConcurrentUsers ─────────────────────────────────────────────
+
+class TestConcurrentUsers:
+    """Test multi-user concurrent operations."""
+
+    def test_concurrent_chat(self):
+        pm = mub.PresenceManager()
+        _patcher = patch.object(mub, "CHATLOG_FILE", Path("/dev/null"))
+        _patcher.start()
+        cl = mub.ChatLog(max_per_room=100)
+        _patcher.stop()
+
+        errors = []
+        rooms = ["garden", "tower", "library"]
+
+        def user_session(user_id, username):
+            try:
+                for room in rooms:
+                    pm.enter_room(user_id, username, room)
+                    for i in range(5):
+                        pm.say(user_id, username, room, f"{username}: msg {i}")
+                        cl.log(room, "say", f"{username}: msg {i}", user_id=user_id, username=username)
+                    pm.leave_room(user_id, room)
+            except Exception as e:
+                errors.append(e)
+
+        threads = [
+            threading.Thread(target=user_session, args=(f"u{i}", f"user_{i}"))
+            for i in range(10)
+        ]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert len(errors) == 0
+
+        # Verify no cross-contamination: each room should have messages
+        for room in rooms:
+            players = pm.get_players_in_room(room)
+            # All users should have left
+            assert len(players) == 0
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])