"""Tests for document upload, list, delete, and URL source endpoints.""" import pytest from unittest.mock import MagicMock, patch, AsyncMock AUTH = {"Authorization": "Bearer test-token"} def _make_doc_sb(company=True, chatbot=True, doc=None, url_source=None): """Build a supabase mock for document endpoint tests.""" sb = MagicMock() def table_side(name): m = MagicMock() m.select.return_value = m m.insert.return_value = m m.update.return_value = m m.delete.return_value = m m.eq.return_value = m m.in_.return_value = m m.limit.return_value = m m.order.return_value = m m.range.return_value = m m.gte.return_value = m m.lt.return_value = m if name == "companies": m.execute.return_value = MagicMock( data=[{"id": "company-1"}] if company else [], count=1 if company else 0, ) elif name == "chatbots": m.execute.return_value = MagicMock( data=[{"id": "cb-1", "company_id": "company-1", "qdrant_collection_name": "col-1"}] if chatbot else [], count=1 if chatbot else 0, ) elif name == "documents": if doc is not None: m.execute.return_value = MagicMock(data=[doc], count=1) else: m.execute.return_value = MagicMock( data=[{ "id": "doc-1", "chatbot_id": "cb-1", "file_name": "test.pdf", "file_type": ".pdf", "file_size": 1024, "chunk_count": 0, "status": "processing", "created_at": "2024-01-01T00:00:00", "updated_at": "2024-01-01T00:00:00", "error_message": None, "file_url": None, }], count=1, ) elif name == "subscriptions": m.execute.return_value = MagicMock(data=[{"plan": "starter"}], count=1) elif name == "url_sources": if url_source is not None: m.execute.return_value = MagicMock(data=[url_source], count=1) else: m.execute.return_value = MagicMock(data=[], count=0) else: m.execute.return_value = MagicMock(data=[], count=0) return m sb.table.side_effect = table_side sb.auth = MagicMock() return sb class TestDocumentAuth: def test_upload_requires_auth(self, client): resp = client.post( "/api/v1/chatbots/cb-1/documents", files={"file": ("test.pdf", b"PDF content", "application/pdf")}, ) assert resp.status_code == 401 def test_list_requires_auth(self, client): resp = client.get("/api/v1/chatbots/cb-1/documents") assert resp.status_code == 401 def test_delete_requires_auth(self, client): resp = client.delete("/api/v1/chatbots/cb-1/documents/doc-1") assert resp.status_code == 401 class TestDocumentUpload: def test_upload_unsupported_type_returns_400(self, client): with patch("app.routers.documents.get_supabase") as mock_sb: mock_sb.return_value = _make_doc_sb() resp = client.post( "/api/v1/chatbots/cb-1/documents", files={"file": ("image.png", b"image data", "image/png")}, headers=AUTH, ) assert resp.status_code == 400 assert "not supported" in resp.json()["detail"] @pytest.mark.parametrize("filename,mime", [ ("report.pdf", "application/pdf"), ("data.csv", "text/csv"), ("doc.txt", "text/plain"), ("notes.md", "text/markdown"), ]) def test_upload_accepted_types(self, client, filename, mime): with patch("app.routers.documents.get_supabase") as mock_sb, \ patch("app.routers.documents._process_document_bg", new_callable=AsyncMock): mock_sb.return_value = _make_doc_sb() resp = client.post( "/api/v1/chatbots/cb-1/documents", files={"file": (filename, b"content", mime)}, headers=AUTH, ) assert resp.status_code == 201 def test_upload_company_not_found_returns_404(self, client): with patch("app.routers.documents.get_supabase") as mock_sb: mock_sb.return_value = _make_doc_sb(company=False) resp = client.post( "/api/v1/chatbots/cb-1/documents", files={"file": ("test.pdf", b"data", "application/pdf")}, headers=AUTH, ) assert resp.status_code == 404 def test_upload_chatbot_not_found_returns_404(self, client): with patch("app.routers.documents.get_supabase") as mock_sb: mock_sb.return_value = _make_doc_sb(chatbot=False) resp = client.post( "/api/v1/chatbots/cb-1/documents", files={"file": ("test.pdf", b"data", "application/pdf")}, headers=AUTH, ) assert resp.status_code == 404 def test_upload_response_has_processing_status(self, client): with patch("app.routers.documents.get_supabase") as mock_sb, \ patch("app.routers.documents._process_document_bg", new_callable=AsyncMock): mock_sb.return_value = _make_doc_sb() resp = client.post( "/api/v1/chatbots/cb-1/documents", files={"file": ("report.pdf", b"pdf content", "application/pdf")}, headers=AUTH, ) assert resp.status_code == 201 assert resp.json()["status"] == "processing" class TestDocumentList: def test_list_returns_documents(self, client): with patch("app.routers.documents.get_supabase") as mock_sb: mock_sb.return_value = _make_doc_sb() resp = client.get("/api/v1/chatbots/cb-1/documents", headers=AUTH) assert resp.status_code == 200 data = resp.json() assert isinstance(data, list) assert len(data) == 1 assert data[0]["file_name"] == "test.pdf" def test_list_chatbot_not_found_returns_404(self, client): with patch("app.routers.documents.get_supabase") as mock_sb: mock_sb.return_value = _make_doc_sb(chatbot=False) resp = client.get("/api/v1/chatbots/cb-1/documents", headers=AUTH) assert resp.status_code == 404 class TestDocumentDelete: def test_delete_document_not_found_returns_404(self, client): # Override the documents table to return empty for doc lookup def table_side(name): m = MagicMock() m.select.return_value = m m.insert.return_value = m m.update.return_value = m m.delete.return_value = m m.eq.return_value = m m.in_.return_value = m m.limit.return_value = m m.order.return_value = m if name == "companies": m.execute.return_value = MagicMock(data=[{"id": "company-1"}]) elif name == "chatbots": m.execute.return_value = MagicMock( data=[{"id": "cb-1", "company_id": "company-1", "qdrant_collection_name": "col-1"}] ) elif name == "documents": m.execute.return_value = MagicMock(data=[]) else: m.execute.return_value = MagicMock(data=[]) return m sb = MagicMock() sb.table.side_effect = table_side sb.auth = MagicMock() with patch("app.routers.documents.get_supabase", return_value=sb): resp = client.delete("/api/v1/chatbots/cb-1/documents/no-such-doc", headers=AUTH) assert resp.status_code == 404 def test_delete_success(self, client): doc = { "id": "doc-1", "chatbot_id": "cb-1", "file_name": "report.pdf", "file_type": ".pdf", "file_size": 1024, "chunk_count": 5, "status": "completed", "created_at": "2024-01-01T00:00:00", "updated_at": "2024-01-01T00:00:00", "error_message": None, "file_url": None, } with patch("app.routers.documents.get_supabase") as mock_sb, \ patch("app.routers.documents.vector_store") as mock_vs: mock_vs.delete_by_document_id = MagicMock() mock_sb.return_value = _make_doc_sb(doc=doc) resp = client.delete("/api/v1/chatbots/cb-1/documents/doc-1", headers=AUTH) assert resp.status_code == 200 assert resp.json()["success"] is True class TestUrlSources: def test_list_url_sources_requires_auth(self, client): resp = client.get("/api/v1/chatbots/cb-1/url-sources") assert resp.status_code == 401 def test_add_url_source_requires_auth(self, client): resp = client.post("/api/v1/chatbots/cb-1/url-sources", json={"url": "https://example.com"}) assert resp.status_code == 401 def test_add_url_source_free_plan_blocked(self, client): def table_side(name): m = MagicMock() m.select.return_value = m m.eq.return_value = m m.execute.return_value = MagicMock(data=[], count=0) if name == "companies": m.execute.return_value = MagicMock(data=[{"id": "comp-1"}]) elif name == "chatbots": m.execute.return_value = MagicMock(data=[{"id": "cb-1", "company_id": "comp-1"}]) elif name == "subscriptions": m.execute.return_value = MagicMock(data=[{"plan": "free"}]) return m sb = MagicMock() sb.table.side_effect = table_side sb.auth = MagicMock() with patch("app.routers.documents.get_supabase", return_value=sb): resp = client.post( "/api/v1/chatbots/cb-1/url-sources", json={"url": "https://example.com"}, headers=AUTH, ) assert resp.status_code == 402 def test_list_url_sources_returns_empty(self, client): with patch("app.routers.documents.get_supabase") as mock_sb: mock_sb.return_value = _make_doc_sb() resp = client.get("/api/v1/chatbots/cb-1/url-sources", headers=AUTH) assert resp.status_code == 200 assert resp.json() == []