79 lines
2.4 KiB
Python
79 lines
2.4 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"]
|