jerryandClaude Fable 5
1e94de7464
- BFF 薄适配层（FastAPI，8090）：QA 代理/SSE 中继/Neo4j 直连/抽取代理/会话持久化 - Vue3 前端：五相位检索流水线、文献来源与证据链、知识图谱、提取任务四 tab - 四档响应式（桌面/笔记本/平板/手机）、深浅主题 - 81 个 FR 全部实现，测试追溯矩阵 81/81，后端 121 测/前端 156 测 - 真实凭据已脱敏，运行时配置在 backend/.env（不入 git） Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
"""app/test_db.py —— SQLite 层单测（aiosqlite + WAL，幂等迁移，JSON 编解码）。
|
|
|
|
覆盖：connect 建表/索引、重复 connect 幂等、外键级联删除、JSON 编解码容错。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.db import Database, dumps_json, loads_json
|
|
|
|
|
|
async def test_connect_creates_schema_and_wal(tmp_path):
|
|
db = Database(tmp_path / "t1" / "plantai.db")
|
|
await db.connect()
|
|
try:
|
|
# WAL 模式
|
|
row = await db.fetchone("PRAGMA journal_mode")
|
|
assert row[0].lower() == "wal"
|
|
# 表存在
|
|
tables = await db.fetchall("SELECT name FROM sqlite_master WHERE type='table'")
|
|
names = {t["name"] for t in tables}
|
|
assert {"sessions", "messages"} <= names
|
|
# 索引存在
|
|
indexes = await db.fetchall(
|
|
"SELECT name FROM sqlite_master WHERE type='index' AND name LIKE 'idx_%'"
|
|
)
|
|
inames = {i["name"] for i in indexes}
|
|
assert "idx_messages_session" in inames
|
|
assert "idx_sessions_updated" in inames
|
|
finally:
|
|
await db.close()
|
|
|
|
|
|
async def test_connect_idempotent(tmp_path):
|
|
db = Database(tmp_path / "t2" / "plantai.db")
|
|
await db.connect()
|
|
await db.close()
|
|
await db.connect() # 二次迁移不报错
|
|
await db.close()
|
|
|
|
|
|
async def test_cascade_delete_messages(tmp_path):
|
|
db = Database(tmp_path / "t3" / "plantai.db")
|
|
await db.connect()
|
|
try:
|
|
await db.execute(
|
|
"INSERT INTO sessions (id, title, status, meta, created_at, updated_at) "
|
|
"VALUES ('s1', 't', 'idle', '{}', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')"
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO messages (id, session_id, role, kind, status, blocks, raw, text, "
|
|
"created_at, updated_at) VALUES "
|
|
"('m1', 's1', 'user', 'qa', 'done', '[]', NULL, 'hi', "
|
|
"'2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')"
|
|
)
|
|
row = await db.fetchone("SELECT COUNT(*) AS n FROM messages WHERE session_id='s1'")
|
|
assert row["n"] == 1
|
|
await db.execute("DELETE FROM sessions WHERE id='s1'")
|
|
row = await db.fetchone("SELECT COUNT(*) AS n FROM messages WHERE session_id='s1'")
|
|
assert row["n"] == 0 # ON DELETE CASCADE
|
|
finally:
|
|
await db.close()
|
|
|
|
|
|
def test_dumps_loads_json_roundtrip():
|
|
payload = {"a": 1, "b": [1, 2], "c": {"d": "中文"}}
|
|
assert loads_json(dumps_json(payload)) == payload
|
|
assert loads_json(None, default=[]) == []
|
|
assert loads_json("not-json", default={}) == {}
|
|
|
|
|
|
def test_dumps_json_fallback_on_unserializable():
|
|
import threading
|
|
|
|
assert dumps_json({"lock": threading.Lock()}, default=None) == "null"
|