mirror of
http://88.130.71.182:3000/BlitTech/contexta_be.git
synced 2026-06-13 08:45:24 +00:00
92d4c2fc5e
- Add new routers: admin, appointments, campaigns - Add storage service and logging config - Add migrations directory and test suite with pytest config - Add supabase_migration_features.sql - Update models, dependencies, config, and existing routers - Remove whatsapp_service (deleted) - Update pyproject.toml and uv.lock dependencies Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
429 lines
19 KiB
Python
429 lines
19 KiB
Python
"""
|
|
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
|