""" Tests for inbox endpoints: GET /api/v1/inbox/conversations GET /api/v1/inbox/conversations/{id} PATCH /api/v1/inbox/conversations/{id}/status POST /api/v1/inbox/conversations/{id}/reply DELETE /api/v1/inbox/conversations/{id} """ import pytest from unittest.mock import MagicMock, patch # ── Helpers ──────────────────────────────────────────────────────────────────── def make_user(uid="user-1"): u = MagicMock() u.id = uid u.email = "user@example.com" return u def make_supabase(plan="starter", company_id="company-1", conversations=None, messages=None, conversation=None): """ Build a Supabase mock wired for inbox tests. table() calls are routed by table name; every chain returns self so .select().eq()…execute() works. """ sb = MagicMock() def table_side(name): t = MagicMock() t.select = MagicMock(return_value=t) t.insert = MagicMock(return_value=t) t.update = MagicMock(return_value=t) t.delete = MagicMock(return_value=t) t.eq = MagicMock(return_value=t) t.neq = MagicMock(return_value=t) t.in_ = MagicMock(return_value=t) t.order = MagicMock(return_value=t) t.range = MagicMock(return_value=t) t.limit = MagicMock(return_value=t) if name == "subscriptions": t.execute = MagicMock(return_value=MagicMock(data=[{"plan": plan}])) elif name == "companies": t.execute = MagicMock(return_value=MagicMock(data=[{"id": company_id}])) elif name == "chatbots": t.execute = MagicMock(return_value=MagicMock(data=[ {"id": "chatbot-1", "name": "My Bot", "company_id": company_id} ])) elif name == "conversations": if conversation is not None: t.execute = MagicMock(return_value=MagicMock(data=[conversation])) else: rows = conversations or [ {"id": "conv-1", "chatbot_id": "chatbot-1", "session_id": "s1", "language": "en", "message_count": 3, "status": "open", "last_agent_reply_at": None, "created_at": "2024-01-01T00:00:00"}, ] t.execute = MagicMock(return_value=MagicMock(data=rows)) elif name == "messages": rows = messages or [ {"id": "msg-1", "role": "user", "content": "Hello", "sources": None, "confidence_score": None, "is_handoff": False, "created_at": "2024-01-01T00:00:00"}, ] t.execute = MagicMock(return_value=MagicMock(data=rows)) else: t.execute = MagicMock(return_value=MagicMock(data=[])) return t sb.table.side_effect = table_side return sb # ── Tests: list conversations ────────────────────────────────────────────────── class TestListConversations: def test_requires_auth(self, client): resp = client.get("/api/v1/inbox/conversations") assert resp.status_code == 401 def test_free_plan_returns_402(self, client): user = make_user() sb = make_supabase(plan="free") with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.get("/api/v1/inbox/conversations", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 402 def test_returns_conversation_list(self, client): user = make_user() sb = make_supabase(plan="starter") with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.get("/api/v1/inbox/conversations", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 data = resp.json() assert isinstance(data, list) assert len(data) == 1 assert data[0]["id"] == "conv-1" assert data[0]["status"] == "open" assert data[0]["chatbot_name"] == "My Bot" def test_empty_when_no_chatbots(self, client): user = make_user() sb = make_supabase(plan="starter") # Override chatbots table to return nothing original_side = sb.table.side_effect def patched(name): t = original_side(name) if name == "chatbots": t.execute = MagicMock(return_value=MagicMock(data=[])) return t sb.table.side_effect = patched with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.get("/api/v1/inbox/conversations", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 assert resp.json() == [] def test_pagination_params_accepted(self, client): user = make_user() sb = make_supabase(plan="starter") with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.get("/api/v1/inbox/conversations?page=2&limit=5", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 # ── Tests: get single conversation ──────────────────────────────────────────── class TestGetConversation: def _owned_conv(self, company_id="company-1"): return { "id": "conv-1", "chatbot_id": "chatbot-1", "session_id": "s1", "language": "en", "status": "open", "last_agent_reply_at": None, "created_at": "2024-01-01T00:00:00", "chatbots": {"company_id": company_id, "name": "My Bot"}, } def test_requires_auth(self, client): resp = client.get("/api/v1/inbox/conversations/conv-1") assert resp.status_code == 401 def test_returns_messages(self, client): user = make_user() conv = self._owned_conv() sb = make_supabase(plan="starter", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.get("/api/v1/inbox/conversations/conv-1", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 body = resp.json() assert body["conversation_id"] == "conv-1" assert len(body["messages"]) == 1 assert body["messages"][0]["role"] == "user" def test_returns_404_when_not_found(self, client): user = make_user() sb = make_supabase(plan="starter", conversation=None, conversations=[]) # conversations table returns empty original = sb.table.side_effect def patched(name): t = original(name) if name == "conversations": t.execute = MagicMock(return_value=MagicMock(data=[])) return t sb.table.side_effect = patched with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.get("/api/v1/inbox/conversations/nonexistent", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 404 def test_returns_403_for_other_companys_conversation(self, client): user = make_user() # Conversation belongs to company-OTHER, user belongs to company-1 conv = self._owned_conv(company_id="company-OTHER") sb = make_supabase(plan="starter", company_id="company-1", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.get("/api/v1/inbox/conversations/conv-1", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 403 # ── Tests: update conversation status ───────────────────────────────────────── class TestUpdateConversationStatus: def _owned_conv(self, company_id="company-1", status="open"): return { "id": "conv-1", "status": status, "chatbots": {"company_id": company_id}, } def test_requires_auth(self, client): resp = client.patch("/api/v1/inbox/conversations/conv-1/status", json={"status": "resolved"}) assert resp.status_code == 401 def test_valid_statuses_accepted(self, client): for s in ("open", "agent_handling", "resolved"): user = make_user() conv = self._owned_conv(status="open") sb = make_supabase(plan="starter", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.patch("/api/v1/inbox/conversations/conv-1/status", json={"status": s}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 assert resp.json()["status"] == s def test_invalid_status_returns_400(self, client): user = make_user() conv = self._owned_conv() sb = make_supabase(plan="starter", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.patch("/api/v1/inbox/conversations/conv-1/status", json={"status": "invalid_status"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 400 def test_403_for_other_companys_conversation(self, client): user = make_user() conv = self._owned_conv(company_id="company-OTHER") sb = make_supabase(plan="starter", company_id="company-1", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.patch("/api/v1/inbox/conversations/conv-1/status", json={"status": "resolved"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 403 def test_404_when_not_found(self, client): user = make_user() sb = make_supabase(plan="starter") original = sb.table.side_effect def patched(name): t = original(name) if name == "conversations": t.execute = MagicMock(return_value=MagicMock(data=[])) return t sb.table.side_effect = patched with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.patch("/api/v1/inbox/conversations/nonexistent/status", json={"status": "resolved"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 404 # ── Tests: agent reply ───────────────────────────────────────────────────────── class TestAgentReply: def _owned_conv(self, company_id="company-1", status="open"): return { "id": "conv-1", "status": status, "chatbots": {"company_id": company_id}, } def test_requires_auth(self, client): resp = client.post("/api/v1/inbox/conversations/conv-1/reply", json={"message": "Hello!"}) assert resp.status_code == 401 def test_sends_reply_and_returns_message_id(self, client): user = make_user() conv = self._owned_conv(status="open") sb = make_supabase(plan="starter", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.post("/api/v1/inbox/conversations/conv-1/reply", json={"message": "Thanks for contacting us!"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 body = resp.json() assert body["success"] is True assert "message_id" in body def test_empty_message_returns_422(self, client): user = make_user() conv = self._owned_conv() sb = make_supabase(plan="starter", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.post("/api/v1/inbox/conversations/conv-1/reply", json={"message": ""}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 422 def test_sets_status_to_agent_handling_when_open(self, client): user = make_user() conv = self._owned_conv(status="open") sb = make_supabase(plan="starter", conversation=conv) update_calls = [] original = sb.table.side_effect def patched(name): t = original(name) if name == "conversations": orig_update = t.update def track_update(data): update_calls.append(data) return t t.update = track_update return t sb.table.side_effect = patched with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.post("/api/v1/inbox/conversations/conv-1/reply", json={"message": "Hello"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 # Should have updated status to agent_handling statuses = [c.get("status") for c in update_calls if "status" in c] assert "agent_handling" in statuses def test_does_not_change_status_when_already_agent_handling(self, client): user = make_user() conv = self._owned_conv(status="agent_handling") sb = make_supabase(plan="starter", conversation=conv) update_calls = [] original = sb.table.side_effect def patched(name): t = original(name) if name == "conversations": orig_update = t.update def track_update(data): update_calls.append(data) return t t.update = track_update return t sb.table.side_effect = patched with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.post("/api/v1/inbox/conversations/conv-1/reply", json={"message": "Follow-up"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 # status should NOT appear in any update when already agent_handling new_statuses = [c.get("status") for c in update_calls if "status" in c] assert "agent_handling" not in new_statuses def test_403_for_other_companys_conversation(self, client): user = make_user() conv = self._owned_conv(company_id="company-OTHER") sb = make_supabase(plan="starter", company_id="company-1", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.post("/api/v1/inbox/conversations/conv-1/reply", json={"message": "Hi"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 403 def test_free_plan_returns_402(self, client): user = make_user() sb = make_supabase(plan="free") with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.post("/api/v1/inbox/conversations/conv-1/reply", json={"message": "Hi"}, headers={"Authorization": "Bearer tok"}) assert resp.status_code == 402 # ── Tests: delete conversation ───────────────────────────────────────────────── class TestDeleteConversation: def _owned_conv(self, company_id="company-1"): return { "id": "conv-1", "chatbots": {"company_id": company_id}, } def test_requires_auth(self, client): resp = client.delete("/api/v1/inbox/conversations/conv-1") assert resp.status_code == 401 def test_deletes_successfully(self, client): user = make_user() conv = self._owned_conv() sb = make_supabase(plan="starter", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.delete("/api/v1/inbox/conversations/conv-1", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 200 assert resp.json()["success"] is True def test_returns_404_when_not_found(self, client): user = make_user() sb = make_supabase(plan="starter") original = sb.table.side_effect def patched(name): t = original(name) if name == "conversations": t.execute = MagicMock(return_value=MagicMock(data=[])) return t sb.table.side_effect = patched with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.delete("/api/v1/inbox/conversations/nonexistent", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 404 def test_returns_403_for_other_companys_conversation(self, client): user = make_user() conv = self._owned_conv(company_id="company-OTHER") sb = make_supabase(plan="starter", company_id="company-1", conversation=conv) with patch("app.routers.inbox.get_current_user", return_value=user), \ patch("app.routers.inbox.get_supabase", return_value=sb): resp = client.delete("/api/v1/inbox/conversations/conv-1", headers={"Authorization": "Bearer tok"}) assert resp.status_code == 403