import pytest from fastapi.testclient import TestClient from httpx import ASGITransport, AsyncClient from app.main import app from app.database import get_database from app.repositories.session_repository import create_session, claim_session_atomically, get_session_by_id from app.repositories.tenant_repository import create_tenant from app.repositories.conversation_repository import get_or_create_conversation, get_conversation_by_session_id from app.repositories.message_repository import create_message from app.services.presence import ( active_sessions, add_active_session, remove_active_session, handle_agent_disconnect, get_active_sessions ) @pytest.fixture(autouse=True) def clear_presence(): active_sessions.clear() yield active_sessions.clear() @pytest.mark.anyio async def test_agent_disconnect_requeues_in_chat_session(): db = get_database() # 1. Create a session sess = await create_session(db, site_id="site_test_requeue") sid = sess["sessionId"] # 2. Claim session for Agent A claimed = await claim_session_atomically(db, sid, "agent_A") assert claimed is not None assert claimed["status"] == "IN_CHAT" assert claimed["assignedAgentId"] == "agent_A" # Register in memory presence add_active_session( session_id=sid, site_id="site_test_requeue", websocket=None, status_val="IN_CHAT", assigned_agent_id="agent_A" ) assert active_sessions[sid]["assignedAgentId"] == "agent_A" # 3. Simulate Agent A WebSocket disconnect await handle_agent_disconnect(db, "agent_A") # 4. Verify MongoDB state is requeued to RINGING updated_sess = await get_session_by_id(db, sid) assert updated_sess["status"] == "RINGING" assert updated_sess.get("assignedAgentId") is None # 5. Verify in-memory presence state is updated to RINGING assert active_sessions[sid]["status"] == "RINGING" assert active_sessions[sid]["assignedAgentId"] is None @pytest.mark.anyio async def test_agent_disconnect_does_not_requeue_already_reassigned_session(): db = get_database() # 1. Create a session and claim for Agent B sess = await create_session(db, site_id="site_test_reassign") sid = sess["sessionId"] claimed = await claim_session_atomically(db, sid, "agent_B") assert claimed["assignedAgentId"] == "agent_B" add_active_session( session_id=sid, site_id="site_test_reassign", websocket=None, status_val="IN_CHAT", assigned_agent_id="agent_B" ) # 2. Late disconnect event for Agent A arrives await handle_agent_disconnect(db, "agent_A") # 3. Verify Agent B's session remains untouched IN_CHAT updated_sess = await get_session_by_id(db, sid) assert updated_sess["status"] == "IN_CHAT" assert updated_sess["assignedAgentId"] == "agent_B" assert active_sessions[sid]["status"] == "IN_CHAT" assert active_sessions[sid]["assignedAgentId"] == "agent_B" @pytest.mark.anyio async def test_agent_disconnect_requeues_multiple_sessions(): db = get_database() sess1 = await create_session(db, site_id="site_multi") sess2 = await create_session(db, site_id="site_multi") await claim_session_atomically(db, sess1["sessionId"], "agent_multi") await claim_session_atomically(db, sess2["sessionId"], "agent_multi") add_active_session(sess1["sessionId"], "site_multi", None, status_val="IN_CHAT", assigned_agent_id="agent_multi") add_active_session(sess2["sessionId"], "site_multi", None, status_val="IN_CHAT", assigned_agent_id="agent_multi") await handle_agent_disconnect(db, "agent_multi") s1_doc = await get_session_by_id(db, sess1["sessionId"]) s2_doc = await get_session_by_id(db, sess2["sessionId"]) assert s1_doc["status"] == "RINGING" assert s1_doc.get("assignedAgentId") is None assert s2_doc["status"] == "RINGING" assert s2_doc.get("assignedAgentId") is None @pytest.mark.anyio async def test_visitor_temporary_disconnect_preserves_session_and_messages(): db = get_database() # 1. Setup tenant & session await db["tenants"].insert_one({"siteId": "site_temp_disc", "domain": "temp.com", "name": "Temp Store"}) sess = await create_session(db, site_id="site_temp_disc") sid = sess["sessionId"] await claim_session_atomically(db, sid, "agent_1") conv = await get_or_create_conversation(db, sid, "site_temp_disc") await create_message(db, conv["conversationId"], sid, "user", "Hello agent") await create_message(db, conv["conversationId"], sid, "agent", "Hi visitor") # 2. Visitor WebSocket disconnects temporarily (memory removed) remove_active_session(sid) # 3. Verify session & conversation data in MongoDB remains IN_CHAT and intact s_doc = await get_session_by_id(db, sid) assert s_doc["status"] == "IN_CHAT" assert s_doc["assignedAgentId"] == "agent_1" # 4. Visitor reconnects: history API is called async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: resp = await client.get(f"/api/v1/session/{sid}/history?siteId=site_temp_disc") assert resp.status_code == 200 body = resp.json() assert body["sessionId"] == sid assert body["status"] == "IN_CHAT" assert body["assignedAgentId"] == "agent_1" assert len(body["messages"]) == 2 assert body["messages"][0]["content"] == "Hello agent" assert body["messages"][1]["content"] == "Hi visitor" @pytest.mark.anyio async def test_explicit_end_chat_ends_session_permanently(): db = get_database() await db["tenants"].insert_one({"siteId": "site_end_test", "domain": "endtest.com", "name": "End Store"}) sess = await create_session(db, site_id="site_end_test") sid = sess["sessionId"] add_active_session(sid, "site_end_test", None, status_val="IN_CHAT", assigned_agent_id="agent_1") async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: # Call explicit end chat endpoint end_resp = await client.post("/api/v1/handoff/end", json={"sessionId": sid, "endedBy": "visitor"}) assert end_resp.status_code == 200 assert end_resp.json()["status"] == "ENDED" # Verify MongoDB status is ENDED s_doc = await get_session_by_id(db, sid) assert s_doc["status"] == "ENDED" # Verify active presence is purged assert sid not in get_active_sessions() @pytest.mark.anyio async def test_in_chat_visitor_disconnect_sets_grace_period(): db = get_database() sess = await create_session(db, site_id="site_grace_1") sid = sess["sessionId"] await claim_session_atomically(db, sid, "agent_1") with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_1") as agent_ws: with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws: # Consume visitor_connected agent_ws.receive_json() # Visitor disconnects while IN_CHAT event = agent_ws.receive_json() assert event["type"] == "visitor_disconnected" assert event["sessionId"] == sid assert event["connectionStatus"] == "disconnected" assert "disconnectedAt" in event # Session remains in active_sessions during grace period assert sid in active_sessions assert active_sessions[sid]["connectionStatus"] == "disconnected" # DB also reflects disconnected s_doc = await get_session_by_id(db, sid) assert s_doc["connectionStatus"] == "disconnected" @pytest.mark.anyio async def test_in_chat_visitor_reconnect_cancels_timer_and_restores_online(): db = get_database() sess = await create_session(db, site_id="site_grace_2") sid = sess["sessionId"] await claim_session_atomically(db, sid, "agent_1") with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_1") as agent_ws: with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws: agent_ws.receive_json() # Disconnect 1 agent_ws.receive_json() # Reconnect within grace period with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws2: reconnect_event = agent_ws.receive_json() assert reconnect_event["type"] == "visitor_reconnected" assert reconnect_event["sessionId"] == sid assert reconnect_event["connectionStatus"] == "online" assert active_sessions[sid]["connectionStatus"] == "online" @pytest.mark.anyio async def test_in_chat_visitor_timeout_removes_session(): import asyncio from app.services.presence import start_grace_period_timer db = get_database() sess = await create_session(db, site_id="site_grace_3") sid = sess["sessionId"] await claim_session_atomically(db, sid, "agent_1") add_active_session(sid, "site_grace_3", None, status_val="IN_CHAT", assigned_agent_id="agent_1") with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_1") as agent_ws: # Trigger grace period timer with fast 0.05s expiry await start_grace_period_timer(sid, "site_grace_3", grace_seconds=0.05) await asyncio.sleep(0.15) event = agent_ws.receive_json() assert event["type"] == "visitor_timeout" assert event["sessionId"] == sid assert event["connectionStatus"] == "offline" assert sid not in active_sessions @pytest.mark.anyio async def test_ringing_visitor_disconnect_removes_immediately(): db = get_database() sess = await create_session(db, site_id="site_grace_4") sid = sess["sessionId"] with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_1") as agent_ws: with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws: agent_ws.receive_json() # RINGING disconnect event = agent_ws.receive_json() assert event["type"] == "visitor_disconnected" assert "connectionStatus" not in event # Immediately removed assert sid not in active_sessions @pytest.mark.anyio async def test_rapid_disconnect_reconnect_cleans_up_timers(): from app.services.presence import grace_period_timers db = get_database() sess = await create_session(db, site_id="site_grace_5") sid = sess["sessionId"] await claim_session_atomically(db, sid, "agent_1") with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_1") as agent_ws: for _ in range(3): with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws: # While reconnected, timer is cancelled and cleaned up assert sid not in grace_period_timers or grace_period_timers[sid].done() @pytest.mark.anyio async def test_reconnect_after_abandoned_timeout_rering_and_history_intact(): import asyncio from app.services.presence import start_grace_period_timer db = get_database() await db["tenants"].insert_one({"siteId": "site_ab_1", "domain": "ab1.com", "name": "AB1"}) sess = await create_session(db, site_id="site_ab_1") sid = sess["sessionId"] await claim_session_atomically(db, sid, "agent_1") conv = await get_or_create_conversation(db, sid, "site_ab_1") await create_message(db, conv["conversationId"], sid, "user", "Prior message before timeout") add_active_session(sid, "site_ab_1", None, status_val="IN_CHAT", assigned_agent_id="agent_1") # 1. Trigger timeout expiry await start_grace_period_timer(sid, "site_ab_1", grace_seconds=0.05) await asyncio.sleep(0.15) # Verify session is ABANDONED in DB and unassigned s_doc = await get_session_by_id(db, sid) assert s_doc["status"] == "ABANDONED" assert s_doc.get("assignedAgentId") is None c_doc = await get_conversation_by_session_id(db, sid) assert c_doc["status"] == "ABANDONED" # 2. Reconnect visitor after timeout with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_2") as agent2_ws: with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws: # Ring event received on agent_2 event = agent2_ws.receive_json() assert event["type"] == "visitor_connected" assert event["sessionId"] == sid assert event["status"] == "RINGING" assert event["assignedAgentId"] is None # Status changed notice sent to visitor vis_event = visitor_ws.receive_json() assert vis_event["type"] == "status_changed" assert vis_event["status"] == "RINGING" # 3. Agent 2 (different agent) claims the re-rung chat accept_res = client.post("/api/v1/handoff/accept", json={"sessionId": sid, "agentId": "agent_2"}) assert accept_res.status_code == 200 assert accept_res.json()["status"] == "IN_CHAT" assert accept_res.json()["assignedAgentId"] == "agent_2" # 4. History contains prior message hist_res = client.get(f"/api/v1/session/{sid}/history?siteId=site_ab_1") assert hist_res.status_code == 200 body = hist_res.json() assert len(body["messages"]) == 1 assert body["messages"][0]["content"] == "Prior message before timeout" @pytest.mark.anyio async def test_reconnect_after_abandoned_original_agent_can_claim(): import asyncio from app.services.presence import start_grace_period_timer db = get_database() await db["tenants"].insert_one({"siteId": "site_ab_2", "domain": "ab2.com", "name": "AB2"}) sess = await create_session(db, site_id="site_ab_2") sid = sess["sessionId"] await claim_session_atomically(db, sid, "agent_1") conv = await get_or_create_conversation(db, sid, "site_ab_2") await create_message(db, conv["conversationId"], sid, "user", "Hello agent 1") add_active_session(sid, "site_ab_2", None, status_val="IN_CHAT", assigned_agent_id="agent_1") # Trigger timeout expiry await start_grace_period_timer(sid, "site_ab_2", grace_seconds=0.05) await asyncio.sleep(0.15) # Reconnect visitor with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_1") as agent1_ws: with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws: agent1_ws.receive_json() # Original Agent 1 claims first accept_res = client.post("/api/v1/handoff/accept", json={"sessionId": sid, "agentId": "agent_1"}) assert accept_res.status_code == 200 assert accept_res.json()["assignedAgentId"] == "agent_1" @pytest.mark.anyio async def test_reconnect_to_ended_session_does_not_trigger_rering(): from app.repositories.session_repository import update_session_status db = get_database() await db["tenants"].insert_one({"siteId": "site_ab_3", "domain": "ab3.com", "name": "AB3"}) sess = await create_session(db, site_id="site_ab_3") sid = sess["sessionId"] await update_session_status(db, sid, "ENDED") with TestClient(app) as client: with client.websocket_connect("/ws/agent?agentId=agent_1") as agent_ws: with client.websocket_connect(f"/ws/session/{sid}") as visitor_ws: # visitor_connected sent with status ENDED, not RINGING event = agent_ws.receive_json() assert event["type"] == "visitor_connected" assert event["status"] == "ENDED" @pytest.mark.anyio async def test_start_new_conversation_after_ended(): from app.repositories.session_repository import update_session_status db = get_database() await db["tenants"].insert_one({"siteId": "site_nc_1", "domain": "nc1.com", "name": "NC1"}) sess = await create_session(db, site_id="site_nc_1") sid = sess["sessionId"] # Active conversation 1 conv1 = await get_or_create_conversation(db, sid, "site_nc_1") c1_id = conv1["conversationId"] await claim_session_atomically(db, sid, "agent_1") await create_message(db, c1_id, sid, "user", "Old conversation message 1") await create_message(db, c1_id, sid, "agent", "Old conversation reply 1") # End conversation 1 await update_session_status(db, sid, "ENDED") await db["conversations"].update_one({"conversationId": c1_id}, {"$set": {"status": "ended", "endedBy": "agent"}}) with TestClient(app) as client: # 1. Post new-conversation after ENDED res = client.post(f"/api/v1/session/{sid}/new-conversation") assert res.status_code == 200 data = res.json() assert data["sessionId"] == sid assert data["status"] == "RINGING" c2_id = data["conversationId"] assert c2_id != c1_id # 2. Check old conversation history via GET /api/v1/conversation/{c1_id} c1_res = client.get(f"/api/v1/conversation/{c1_id}") assert c1_res.status_code == 200 c1_body = c1_res.json() assert c1_body["conversationId"] == c1_id assert len(c1_body["messages"]) == 2 assert c1_body["messages"][0]["content"] == "Old conversation message 1" # 3. Check active session history via GET /api/v1/session/{sid}/history -> returns empty messages for new conversation! hist_res = client.get(f"/api/v1/session/{sid}/history?siteId=site_nc_1") assert hist_res.status_code == 200 hist_body = hist_res.json() assert hist_body["conversationId"] == c2_id assert hist_body["status"] == "RINGING" assert len(hist_body["messages"]) == 0 # 4. Any agent (e.g. agent_2, different from agent_1) accepts the new conversation accept_res = client.post("/api/v1/handoff/accept", json={"sessionId": sid, "agentId": "agent_2"}) assert accept_res.status_code == 200 assert accept_res.json()["assignedAgentId"] == "agent_2" assert accept_res.json()["status"] == "IN_CHAT" @pytest.mark.anyio async def test_start_new_conversation_rejected_if_already_active(): db = get_database() await db["tenants"].insert_one({"siteId": "site_nc_2", "domain": "nc2.com", "name": "NC2"}) sess = await create_session(db, site_id="site_nc_2") sid = sess["sessionId"] await get_or_create_conversation(db, sid, "site_nc_2") with TestClient(app) as client: # Attempt new-conversation while status is still active (RINGING) res = client.post(f"/api/v1/session/{sid}/new-conversation") assert res.status_code == 400 assert "Cannot start a new conversation" in res.json()["detail"] @pytest.mark.anyio async def test_ended_session_without_click_stays_passive(): from app.repositories.session_repository import update_session_status db = get_database() await db["tenants"].insert_one({"siteId": "site_nc_3", "domain": "nc3.com", "name": "NC3"}) sess = await create_session(db, site_id="site_nc_3") sid = sess["sessionId"] await update_session_status(db, sid, "ENDED") with TestClient(app) as client: # GET history returns status ENDED res = client.get(f"/api/v1/session/{sid}/history?siteId=site_nc_3") assert res.status_code == 200 assert res.json()["status"] == "ENDED"