110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""Run the default application lifecycle over HTTP and verify persistence."""
|
|
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
|
|
@contextmanager
|
|
def server(engine):
|
|
with socket.socket() as socket_:
|
|
socket_.bind(("127.0.0.1", 0))
|
|
port = socket_.getsockname()[1]
|
|
environment = {**os.environ, "DATABASE_URL": engine.url.render_as_string(hide_password=False)}
|
|
backend = Path(__file__).resolve().parents[1]
|
|
process = subprocess.Popen(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"uvicorn",
|
|
"app.main:app",
|
|
"--host",
|
|
"127.0.0.1",
|
|
"--port",
|
|
str(port),
|
|
"--log-level",
|
|
"error",
|
|
],
|
|
cwd=backend,
|
|
env=environment,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
try:
|
|
with httpx.Client(base_url=f"http://127.0.0.1:{port}", timeout=2) as client:
|
|
deadline = time.monotonic() + 15
|
|
while time.monotonic() < deadline:
|
|
if process.poll() is not None:
|
|
raise AssertionError("默认后端进程启动失败")
|
|
try:
|
|
if client.get("/api/ready").status_code == 200:
|
|
break
|
|
except httpx.TransportError:
|
|
pass
|
|
time.sleep(0.05)
|
|
else:
|
|
raise AssertionError("默认后端未在时限内就绪")
|
|
yield client
|
|
finally:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
|
|
|
|
def test_default_startup_and_persistence_across_restart(migrated_database):
|
|
with server(migrated_database) as client:
|
|
assert client.get("/docs").status_code == 200
|
|
response = client.post(
|
|
"/api/knowledge",
|
|
json={
|
|
"title": "持久化验收占位",
|
|
"content": "仅为自动化测试数据",
|
|
"source_url": "",
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
item = response.json()
|
|
with server(migrated_database) as client:
|
|
assert client.get(f"/api/knowledge/{item['id']}").json()["content"] == "仅为自动化测试数据"
|
|
assert client.get("/api/knowledge").json()[0]["id"] == item["id"]
|
|
|
|
|
|
def test_retrieval_run_persists_across_restart(migrated_database):
|
|
with server(migrated_database) as client:
|
|
item = client.post(
|
|
"/api/knowledge",
|
|
json={
|
|
"title": "HTTP检索测试",
|
|
"content": "仅用于自动化测试的那拉提关键词。",
|
|
"source_url": "https://example.com/",
|
|
"collected_at": "2026-10-15T09:00:00+08:00",
|
|
},
|
|
).json()
|
|
assert (
|
|
client.post(
|
|
f"/api/knowledge/{item['id']}/review",
|
|
json={
|
|
"expected_version": item["version"],
|
|
"verification_status": "verified",
|
|
"reviewer": "test-reviewer",
|
|
"verification_notes": "仅验证检索流程",
|
|
},
|
|
).status_code
|
|
== 200
|
|
)
|
|
response = client.post("/api/retrieval", json={"query": "那拉提"})
|
|
assert response.status_code == 200
|
|
run = response.json()
|
|
assert run["results"][0]["knowledge_id"] == item["id"]
|
|
with server(migrated_database) as client:
|
|
assert client.get(f"/api/retrieval/runs/{run['run_id']}").json() == run
|