mirror of
http://88.130.71.182:3000/BlitTech/contexta_be.git
synced 2026-06-12 23:23:21 +00:00
116 lines
4.2 KiB
Python
116 lines
4.2 KiB
Python
"""Tests for URL source refresh endpoint."""
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
AUTH = {"Authorization": "Bearer test-token"}
|
|
|
|
|
|
def _make_sb(source=None, chatbot_company="company-1"):
|
|
sb = MagicMock()
|
|
default_source = {
|
|
"id": "src-1",
|
|
"chatbot_id": "cb-1",
|
|
"url": "https://example.com/faq",
|
|
"status": "completed",
|
|
"page_title": "FAQ",
|
|
"chunk_count": 10,
|
|
"error_message": None,
|
|
}
|
|
|
|
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.returning.return_value = m
|
|
m.limit.return_value = m
|
|
m.order.return_value = m
|
|
|
|
if name == "chatbots":
|
|
m.execute.return_value = MagicMock(data=[{
|
|
"id": "cb-1",
|
|
"company_id": chatbot_company,
|
|
"qdrant_collection_name": "col-1",
|
|
}])
|
|
elif name == "companies":
|
|
m.execute.return_value = MagicMock(data=[{"id": chatbot_company, "owner_id": "user-1"}])
|
|
elif name == "url_sources":
|
|
m.execute.return_value = MagicMock(data=[source or default_source])
|
|
else:
|
|
m.execute.return_value = MagicMock(data=[], count=0)
|
|
return m
|
|
|
|
sb.table.side_effect = table_side
|
|
sb.auth = MagicMock()
|
|
return sb
|
|
|
|
|
|
class TestUrlRefresh:
|
|
def _refresh(self, client, source=None):
|
|
with patch("app.routers.documents.get_supabase") as mock_sb, \
|
|
patch("app.routers.documents.vector_store") as mock_vs, \
|
|
patch("app.routers.documents.response_cache") as mock_cache, \
|
|
patch("app.dependencies.get_current_user") as mock_user, \
|
|
patch("app.routers.documents._process_url_source", new_callable=AsyncMock):
|
|
mock_sb.return_value = _make_sb(source=source)
|
|
mock_vs.delete_by_document_id = MagicMock()
|
|
mock_vs.collection_exists = MagicMock(return_value=True)
|
|
mock_cache.invalidate = MagicMock()
|
|
user = MagicMock()
|
|
user.id = "user-1"
|
|
mock_user.return_value = user
|
|
return client.post(
|
|
"/api/v1/chatbots/cb-1/url-sources/src-1/refresh",
|
|
headers=AUTH,
|
|
)
|
|
|
|
def test_returns_200(self, client):
|
|
resp = self._refresh(client)
|
|
assert resp.status_code == 200
|
|
|
|
def test_source_reset_to_pending(self, client):
|
|
resp = self._refresh(client)
|
|
body = resp.json()
|
|
assert body["status"] == "pending"
|
|
assert body["chunk_count"] == 0
|
|
|
|
def test_returns_404_for_unknown_source(self, client):
|
|
with patch("app.routers.documents.get_supabase") as mock_sb, \
|
|
patch("app.dependencies.get_current_user") as mock_user:
|
|
sb = _make_sb()
|
|
# Override url_sources to return empty
|
|
def table_side(name):
|
|
m = MagicMock()
|
|
m.select.return_value = m
|
|
m.update.return_value = m
|
|
m.eq.return_value = m
|
|
if name == "chatbots":
|
|
m.execute.return_value = MagicMock(data=[{
|
|
"id": "cb-1", "company_id": "company-1",
|
|
"qdrant_collection_name": "col-1",
|
|
}])
|
|
elif name == "companies":
|
|
m.execute.return_value = MagicMock(data=[{"id": "company-1", "owner_id": "user-1"}])
|
|
else:
|
|
m.execute.return_value = MagicMock(data=[])
|
|
return m
|
|
sb2 = MagicMock()
|
|
sb2.table.side_effect = table_side
|
|
sb2.auth = MagicMock()
|
|
mock_sb.return_value = sb2
|
|
user = MagicMock()
|
|
user.id = "user-1"
|
|
mock_user.return_value = user
|
|
resp = client.post(
|
|
"/api/v1/chatbots/cb-1/url-sources/no-such-src/refresh",
|
|
headers=AUTH,
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_requires_authentication(self, client):
|
|
resp = client.post("/api/v1/chatbots/cb-1/url-sources/src-1/refresh")
|
|
assert resp.status_code == 401
|