feat: implement backend knowledge management

backend/tests/test_runtime.py

@@ -0,0 +1,78 @@
+"""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"]