the-nexus/tests/test_bridge.py

#!/usr/bin/env python3
"""
Tests for multi_user_bridge.py

Validates:
  1. Session isolation (2 users, no cross-contamination)
  2. Crisis detection (low, moderate, high risk)
  3. Room awareness (Timmy knows which room user is in)
  4. Session timeout (inactive sessions cleaned up)
  5. Max sessions limit (eviction works)
"""

import json
import os
import sys
import time
import threading
import unittest
from unittest.mock import patch, MagicMock
from http.client import HTTPConnection
from datetime import datetime, timedelta

# Ensure bridge module is importable
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# Patch out AIAgent before importing the bridge module
_mock_agent = MagicMock()
_mock_agent.chat.return_value = "Test response from Timmy"

with patch.dict("sys.modules", {"run_agent": MagicMock(AIAgent=lambda **kw: _mock_agent)}):
    import multi_user_bridge as bridge


class TestPresenceManager(unittest.TestCase):
    """Test the PresenceManager class."""

    def setUp(self):
        self.pm = bridge.PresenceManager()

    def test_enter_and_leave_room(self):
        ev = self.pm.enter_room("u1", "Alice", "Lobby")
        self.assertEqual(ev["event"], "enter")
        self.assertEqual(ev["username"], "Alice")
        self.assertEqual(ev["room"], "Lobby")

        players = self.pm.get_players_in_room("Lobby")
        self.assertEqual(len(players), 1)
        self.assertEqual(players[0]["username"], "Alice")

        lev = self.pm.leave_room("u1", "Lobby")
        self.assertEqual(lev["event"], "leave")
        self.assertEqual(len(self.pm.get_players_in_room("Lobby")), 0)

    def test_say_event(self):
        self.pm.enter_room("u1", "Bob", "Tavern")
        ev = self.pm.say("u1", "Bob", "Tavern", "hello world")
        self.assertEqual(ev["type"], "say")
        self.assertEqual(ev["message"], "hello world")

        events = self.pm.get_room_events("Tavern")
        self.assertEqual(len(events), 2)  # enter + say

    def test_leave_nonexistent_room_returns_none(self):
        result = self.pm.leave_room("u1", "Nowhere")
        self.assertIsNone(result)

    def test_cleanup_user(self):
        self.pm.enter_room("u1", "Carol", "RoomA")
        self.pm.enter_room("u1", "Carol", "RoomB")
        events = self.pm.cleanup_user("u1")
        self.assertEqual(len(events), 2)
        self.assertEqual(len(self.pm.get_players_in_room("RoomA")), 0)
        self.assertEqual(len(self.pm.get_players_in_room("RoomB")), 0)


class TestSessionIsolation(unittest.TestCase):
    """Test 1: Session isolation — 2 users, verify no cross-contamination."""

    def setUp(self):
        # Fresh SessionManager with mocked AIAgent
        self.sm = bridge.SessionManager(max_sessions=10, session_timeout=3600)
        # Patch UserSession._init_agent to avoid real imports
        self.patcher = patch.object(bridge.UserSession, "_init_agent", lambda self: None)
        self.patcher.start()

    def tearDown(self):
        self.patcher.stop()

    def test_two_users_isolated_messages(self):
        s1 = self.sm.get_or_create("u1", "Alice", "Lobby")
        s2 = self.sm.get_or_create("u2", "Bob", "Lobby")

        # Simulate messages
        s1.messages.append({"role": "user", "content": "Alice secret"})
        s2.messages.append({"role": "user", "content": "Bob secret"})

        # Verify no cross-contamination
        s1_contents = [m["content"] for m in s1.messages]
        s2_contents = [m["content"] for m in s2.messages]

        self.assertIn("Alice secret", s1_contents)
        self.assertNotIn("Alice secret", s2_contents)
        self.assertIn("Bob secret", s2_contents)
        self.assertNotIn("Bob secret", s1_contents)

    def test_two_users_different_sessions(self):
        s1 = self.sm.get_or_create("u1", "Alice", "Lobby")
        s2 = self.sm.get_or_create("u2", "Bob", "Tavern")

        self.assertNotEqual(id(s1), id(s2))
        self.assertEqual(s1.username, "Alice")
        self.assertEqual(s2.username, "Bob")
        self.assertEqual(s1.room, "Lobby")
        self.assertEqual(s2.room, "Tavern")

    def test_get_or_create_returns_same_session(self):
        s1 = self.sm.get_or_create("u1", "Alice", "Lobby")
        s2 = self.sm.get_or_create("u1", "Alice", "Lobby")
        self.assertIs(s1, s2)


class TestCrisisDetection(unittest.TestCase):
    """Test 2: Crisis detection (low, moderate, high risk keywords)."""

    def setUp(self):
        self.pm = bridge.PresenceManager()

    def test_crisis_protocol_defined(self):
        """Crisis protocol must exist and have required steps."""
        protocol = bridge.CRISIS_PROTOCOL
        self.assertIsInstance(protocol, list)
        self.assertGreaterEqual(len(protocol), 3)
        # Must include 988 lifeline
        combined = " ".join(protocol).lower()
        self.assertIn("988", combined)
        self.assertIn("crisis", combined)

    def test_low_risk_normal_message(self):
        """Low risk: normal chat does not trigger crisis protocol."""
        ev = self.pm.say("u1", "Dan", "Lobby", "Hello, how are you?")
        self.assertEqual(ev["type"], "say")
        # No crisis flag on normal messages
        self.assertNotIn("crisis", ev)

    def test_moderate_risk_keyword_detection(self):
        """Moderate risk: crisis keywords present in message."""
        crisis_keywords = ["hopeless", "tired of living", "can't go on"]
        for kw in crisis_keywords:
            ev = self.pm.say("u1", "Dan", "Lobby", f"I feel {kw}")
            self.assertEqual(ev["type"], "say")
            # The message is recorded; detection happens at agent level
            self.assertIn(kw, ev["message"])

    def test_high_risk_message_recorded(self):
        """High risk: message with strong crisis indicators is captured."""
        msg = "I want to end my life tonight"
        ev = self.pm.say("u1", "Dan", "Lobby", msg)
        self.assertEqual(ev["message"], msg)
        events = self.pm.get_room_events("Lobby")
        self.assertEqual(len(events), 1)

    def test_crisis_protocol_has_grounding(self):
        """Crisis protocol must reference grounding exercise."""
        combined = " ".join(bridge.CRISIS_PROTOCOL).lower()
        self.assertIn("grounding", combined)


class TestRoomAwareness(unittest.TestCase):
    """Test 3: Room awareness — know which room each user is in."""

    def setUp(self):
        self.sm = bridge.SessionManager(max_sessions=10, session_timeout=3600)
        self.pm = bridge.PresenceManager()
        self.patcher = patch.object(bridge.UserSession, "_init_agent", lambda self: None)
        self.patcher.start()

    def tearDown(self):
        self.patcher.stop()

    def test_session_tracks_room(self):
        s = self.sm.get_or_create("u1", "Alice", "Dark Cave")
        self.assertEqual(s.room, "Dark Cave")

    def test_room_update_on_move(self):
        s = self.sm.get_or_create("u1", "Alice", "Lobby")
        self.assertEqual(s.room, "Lobby")
        s2 = self.sm.get_or_create("u1", "Alice", "Tower Top")
        self.assertIs(s, s2)
        self.assertEqual(s2.room, "Tower Top")

    def test_presence_tracks_multiple_rooms(self):
        self.pm.enter_room("u1", "Alice", "Lobby")
        self.pm.enter_room("u2", "Bob", "Tavern")
        self.pm.enter_room("u3", "Carol", "Lobby")

        lobby_players = self.pm.get_players_in_room("Lobby")
        tavern_players = self.pm.get_players_in_room("Tavern")

        lobby_names = {p["username"] for p in lobby_players}
        tavern_names = {p["username"] for p in tavern_players}

        self.assertEqual(lobby_names, {"Alice", "Carol"})
        self.assertEqual(tavern_names, {"Bob"})

    def test_context_summary_has_room(self):
        s = self.sm.get_or_create("u1", "Alice", "Library")
        summary = s.get_context_summary()
        self.assertEqual(summary["room"], "Library")
        self.assertEqual(summary["user"], "Alice")


class TestSessionTimeout(unittest.TestCase):
    """Test 4: Session timeout — inactive sessions cleaned up."""

    def setUp(self):
        # Very short timeout for testing (1 second)
        self.sm = bridge.SessionManager(max_sessions=10, session_timeout=1)
        self.patcher = patch.object(bridge.UserSession, "_init_agent", lambda self: None)
        self.patcher.start()

    def tearDown(self):
        self.patcher.stop()

    def test_stale_session_cleaned_up(self):
        s = self.sm.get_or_create("u1", "Alice", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 1)

        # Simulate session being old
        s.last_active = time.time() - 5  # 5 seconds ago

        # Next get_or_create triggers cleanup
        self.sm.get_or_create("u2", "Bob", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 1)
        self.assertNotIn("u1", self.sm.sessions)
        self.assertIn("u2", self.sm.sessions)

    def test_active_session_not_cleaned(self):
        s = self.sm.get_or_create("u1", "Alice", "Lobby")
        # Session is fresh, should not be cleaned
        self.sm.get_or_create("u2", "Bob", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 2)
        self.assertIn("u1", self.sm.sessions)

    def test_cleanup_stale_direct(self):
        s1 = self.sm.get_or_create("u1", "Alice", "Lobby")
        s2 = self.sm.get_or_create("u2", "Bob", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 2)

        # Age both sessions
        s1.last_active = time.time() - 10
        s2.last_active = time.time() - 10

        # Trigger cleanup
        self.sm.get_or_create("u3", "Carol", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 1)
        self.assertIn("u3", self.sm.sessions)


class TestMaxSessions(unittest.TestCase):
    """Test 5: Max sessions limit — eviction works."""

    def setUp(self):
        self.sm = bridge.SessionManager(max_sessions=3, session_timeout=3600)
        self.patcher = patch.object(bridge.UserSession, "_init_agent", lambda self: None)
        self.patcher.start()

    def tearDown(self):
        self.patcher.stop()

    def test_eviction_at_max(self):
        s1 = self.sm.get_or_create("u1", "Alice", "Lobby")
        s2 = self.sm.get_or_create("u2", "Bob", "Lobby")
        s3 = self.sm.get_or_create("u3", "Carol", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 3)

        # Adding 4th should evict oldest (u1, least recently active)
        s4 = self.sm.get_or_create("u4", "Dan", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 3)
        self.assertNotIn("u1", self.sm.sessions)
        self.assertIn("u2", self.sm.sessions)
        self.assertIn("u3", self.sm.sessions)
        self.assertIn("u4", self.sm.sessions)

    def test_eviction_oldest_first(self):
        s1 = self.sm.get_or_create("u1", "Alice", "Lobby")
        time.sleep(0.01)
        s2 = self.sm.get_or_create("u2", "Bob", "Lobby")
        time.sleep(0.01)
        s3 = self.sm.get_or_create("u3", "Carol", "Lobby")

        # u1 is oldest (least recently active)
        s4 = self.sm.get_or_create("u4", "Dan", "Lobby")
        self.assertNotIn("u1", self.sm.sessions)

    def test_no_eviction_under_limit(self):
        self.sm.get_or_create("u1", "Alice", "Lobby")
        self.sm.get_or_create("u2", "Bob", "Lobby")
        self.assertEqual(self.sm.get_session_count(), 2)

    def test_list_sessions(self):
        self.sm.get_or_create("u1", "Alice", "Lobby")
        self.sm.get_or_create("u2", "Bob", "Tavern")
        sessions = self.sm.list_sessions()
        self.assertEqual(len(sessions), 2)
        users = {s["user"] for s in sessions}
        self.assertEqual(users, {"Alice", "Bob"})


class TestBridgeHTTPAPI(unittest.TestCase):
    """Integration tests: start the bridge server and hit endpoints."""

    @classmethod
    def setUpClass(cls):
        # Use a random port to avoid conflicts
        cls.port = 14000 + (os.getpid() % 1000)
        bridge.BRIDGE_PORT = cls.port
        bridge.BRIDGE_HOST = "127.0.0.1"

        # Reset global managers
        bridge.session_manager = bridge.SessionManager(max_sessions=5, session_timeout=3600)
        bridge.presence_manager = bridge.PresenceManager()

        # Patch UserSession to avoid real AIAgent
        cls.patcher = patch.object(bridge.UserSession, "_init_agent", lambda self: None)
        cls.patcher.start()

        # Start server in background thread
        cls.server = bridge.HTTPServer(("127.0.0.1", cls.port), bridge.BridgeHandler)
        cls.server_thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.server_thread.start()
        time.sleep(0.3)  # Let server start

    @classmethod
    def tearDownClass(cls):
        cls.server.shutdown()
        cls.patcher.stop()

    def _post(self, path, data):
        conn = HTTPConnection("127.0.0.1", self.port)
        body = json.dumps(data)
        conn.request("POST", path, body=body, headers={"Content-Type": "application/json"})
        resp = conn.getresponse()
        return resp.status, json.loads(resp.read())

    def _get(self, path):
        conn = HTTPConnection("127.0.0.1", self.port)
        conn.request("GET", path)
        resp = conn.getresponse()
        return resp.status, json.loads(resp.read())

    def test_health(self):
        status, data = self._get("/bridge/health")
        self.assertEqual(status, 200)
        self.assertEqual(data["status"], "ok")

    def test_say_endpoint(self):
        status, data = self._post("/bridge/say", {
            "user_id": "test_u1",
            "username": "Tester",
            "message": "Hello room!",
            "room": "TestRoom",
        })
        self.assertEqual(status, 200)
        self.assertTrue(data["ok"])
        self.assertEqual(data["event"]["message"], "Hello room!")

    def test_room_players_endpoint(self):
        # Ensure a player is present
        bridge.presence_manager.enter_room("http_u1", "HttpAlice", "HttpRoom")
        status, data = self._get("/bridge/room/HttpRoom/players")
        self.assertEqual(status, 200)
        names = [p["username"] for p in data["players"]]
        self.assertIn("HttpAlice", names)

    def test_room_events_endpoint(self):
        bridge.presence_manager.enter_room("ev_u1", "EvAlice", "EvRoom")
        bridge.presence_manager.say("ev_u1", "EvAlice", "EvRoom", "test message")
        status, data = self._get("/bridge/room/EvRoom/events")
        self.assertEqual(status, 200)
        self.assertGreater(len(data["events"]), 0)

    def test_sessions_endpoint(self):
        status, data = self._get("/bridge/sessions")
        self.assertEqual(status, 200)
        self.assertIn("sessions", data)

    def test_not_found(self):
        status, data = self._get("/bridge/nonexistent")
        self.assertEqual(status, 404)


if __name__ == "__main__":
    unittest.main()