83 lines
3.0 KiB
Python
83 lines
3.0 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
os.environ["DATABASE_URL"] = f"sqlite:///{tempfile.NamedTemporaryFile(delete=False).name}"
|
|
os.environ["JWT_SECRET_KEY"] = "test-secret"
|
|
|
|
from fastapi.testclient import TestClient # noqa: E402
|
|
|
|
from app.main import app # noqa: E402
|
|
|
|
|
|
class SmokeTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.client_context = TestClient(app)
|
|
self.client = self.client_context.__enter__()
|
|
|
|
def tearDown(self) -> None:
|
|
self.client_context.__exit__(None, None, None)
|
|
|
|
def auth_headers(self) -> dict[str, str]:
|
|
response = self.client.post(
|
|
"/api/auth/login",
|
|
json={"phone": "13800000000", "code": "123456"},
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
token = response.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
def test_health(self) -> None:
|
|
response = self.client.get("/health")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["status"], "ok")
|
|
|
|
def test_login_me_documents_stats_and_session(self) -> None:
|
|
headers = self.auth_headers()
|
|
|
|
me = self.client.get("/api/auth/me", headers=headers)
|
|
self.assertEqual(me.status_code, 200, me.text)
|
|
self.assertEqual(me.json()["phone"], "13800000000")
|
|
|
|
document = self.client.post(
|
|
"/api/documents",
|
|
headers=headers,
|
|
json={
|
|
"title": "差旅费报销管理办法.pdf",
|
|
"secrecy_level": "internal",
|
|
"owner_dept": "财务部",
|
|
"tags": ["差旅", "报销"],
|
|
},
|
|
)
|
|
self.assertEqual(document.status_code, 200, document.text)
|
|
self.assertEqual(document.json()["status"], "parsed")
|
|
|
|
session = self.client.post("/api/chat/sessions", headers=headers, json={"title": "差旅咨询"})
|
|
self.assertEqual(session.status_code, 200, session.text)
|
|
self.assertEqual(session.json()["title"], "差旅咨询")
|
|
session_id = session.json()["id"]
|
|
|
|
with self.client.stream(
|
|
"POST",
|
|
"/api/chat/stream",
|
|
headers=headers,
|
|
json={"conversation_id": session_id, "question": "差旅报销需要哪些材料？"},
|
|
) as stream:
|
|
body = "".join(stream.iter_text())
|
|
self.assertEqual(stream.status_code, 200, body)
|
|
self.assertIn("answer.delta", body)
|
|
|
|
messages = self.client.get(f"/api/chat/sessions/{session_id}/messages", headers=headers)
|
|
self.assertEqual(messages.status_code, 200, messages.text)
|
|
self.assertEqual(len(messages.json()), 1)
|
|
self.assertEqual(messages.json()[0]["status"], "done")
|
|
|
|
stats = self.client.get("/api/stats/overview", headers=headers)
|
|
self.assertEqual(stats.status_code, 200, stats.text)
|
|
self.assertGreaterEqual(stats.json()["document_count"], 1)
|
|
self.assertGreaterEqual(stats.json()["question_count"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|