File size: 6,244 Bytes
55a76a3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | import io
from app.workflow.state import DocumentChunk
# 1. Test GET /health
def test_health_check(client):
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert data["components"]["vector_store"] == "ok"
assert data["components"]["document_registry"] == "ok"
# 2. Test POST /query
def test_query_assistant(client, test_vector_store, mock_embeddings):
# Add mock chunk to the vector store so retrieval succeeds
mock_chunk = DocumentChunk(
content="FastAPI is a modern web framework.",
source_file="fastapi.md",
document_id="doc_fastapi",
chunk_index=0
)
test_vector_store.add_chunks([mock_chunk], mock_embeddings.embed_documents([mock_chunk.content]))
payload = {
"question": "What is FastAPI?",
"session_id": "12345678-1234-5678-1234-567812345678"
}
response = client.post("/query", json=payload)
assert response.status_code == 200
data = response.json()
assert "answer" in data
assert "sources" in data
assert data["answer"] == "FastAPI is a modern web framework."
def test_query_validation_errors(client):
# Empty query
response = client.post("/query", json={"question": ""})
assert response.status_code == 422 # Pydantic min_length validation raises 422
assert "detail" in response.json() or "error" in response.json()
# Too long query
response = client.post("/query", json={"question": "a" * 2001})
assert response.status_code == 422
assert "detail" in response.json() or "error" in response.json()
# Invalid UUID session_id format
response = client.post("/query", json={"question": "What is FastAPI?", "session_id": "invalid-uuid"})
assert response.status_code == 422
assert "detail" in response.json() or "error" in response.json()
# 3. Test POST /ingest
def test_ingest_document_validation(client):
# Neither URL nor file
# We pass an empty files dict to force multipart encoding
response = client.post("/ingest", files={})
assert response.status_code == 422
assert response.json()["error"]["code"] == "VALIDATION_ERROR"
# Both URL and file
file_data = ("test.txt", io.BytesIO(b"Hello world"))
response = client.post(
"/ingest",
files={
"url": (None, "http://example.com"),
"file": file_data
}
)
assert response.status_code == 422
assert response.json()["error"]["code"] == "VALIDATION_ERROR"
# Invalid URL scheme
response = client.post("/ingest", files={"url": (None, "ftp://example.com")})
assert response.status_code == 422
assert response.json()["error"]["code"] == "VALIDATION_ERROR"
# Invalid file extension
file_data = ("test.exe", io.BytesIO(b"Hello world"))
response = client.post("/ingest", files={"file": file_data})
assert response.status_code == 422
assert response.json()["error"]["code"] == "VALIDATION_ERROR"
def test_ingest_url_success(client):
response = client.post("/ingest", files={"url": (None, "https://fastapi.tiangolo.com/")})
assert response.status_code == 201
data = response.json()
assert "document_id" in data
assert data["status"] == "indexed"
def test_ingest_file_success(client):
file_data = ("sample.md", io.BytesIO(b"# Sample Header\nTest file content."))
response = client.post("/ingest", files={"file": file_data})
assert response.status_code == 201
data = response.json()
assert "document_id" in data
assert data["status"] == "indexed"
# 4. Test GET /documents
def test_list_indexed_documents(client):
response = client.get("/documents")
assert response.status_code == 200
data = response.json()
assert "documents" in data
assert "total" in data
# 5. Test POST /feedback
def test_submit_feedback(client):
feedback_payload = {
"query": "What is FastAPI?",
"answer": "FastAPI is a modern web framework.",
"rating": "thumbs_up",
"comment": "Super helpful!",
"session_id": "12345678-1234-5678-1234-567812345678"
}
response = client.post("/feedback", json=feedback_payload)
assert response.status_code == 200
data = response.json()
assert "feedback_id" in data
assert data["status"] == "recorded"
def test_list_feedback(client):
# Submit one feedback first
feedback_payload = {
"query": "What is FastAPI?",
"answer": "FastAPI is a modern web framework.",
"rating": "thumbs_down",
"comment": "Not what I expected.",
"session_id": "12345678-1234-5678-1234-567812345678"
}
client.post("/feedback", json=feedback_payload)
# Get list
response = client.get("/feedback?rating=thumbs_down")
assert response.status_code == 200
data = response.json()
assert "feedback" in data
assert len(data["feedback"]) > 0
assert data["feedback"][0]["rating"] == "thumbs_down"
# 6. Test DELETE /documents/{id}
def test_delete_document(client):
# Ingest a document first to delete it
file_data = ("todelete.md", io.BytesIO(b"# To Delete\nContent to delete."))
ingest_res = client.post("/ingest", files={"file": file_data})
doc_id = ingest_res.json()["document_id"]
# Delete it
del_res = client.delete(f"/documents/{doc_id}")
assert del_res.status_code == 200
assert del_res.json()["status"] == "deleted"
# 7. Test conversational query
def test_conversational_query(client, mock_llm_client):
# Setup mock LLM responses specifically for this test
mock_llm_client.responses = {
"Classify the query type": '{"rewritten_query": "Hello", "query_type": "conversational"}',
"The user is greeting you": "Hello! I am your Technical Documentation Copilot. How can I help you today?"
}
payload = {
"question": "Hello!",
"session_id": "12345678-1234-5678-1234-567812345678"
}
response = client.post("/query", json=payload)
assert response.status_code == 200
data = response.json()
assert "answer" in data
assert "Hello! I am your Technical Documentation Copilot." in data["answer"]
assert len(data["sources"]) == 0
|