jerryandClaude Fable 5
0875bb036c
- git mv 保留历史：ui/(Vue3+Vite)、sdk/(BFF 中间层)
- 路径语义修正：db_path/figures_cache 改相对 sdk/(BFF cwd)，
历史库 sdk/backend/data 迁移至 sdk/data/(会话/图片缓存无损)
- 引用修复 ~15 处：config.py、deploy/*.sh、vitest/playwright 配置、
单测导入路径、docs、README、gitignore、.env.example
- 骨架：agent/{qa,methodology,legacy-rag,vendor}、knowledge/manifest.yaml、
tooling/{compile,kb,crypto,assemble,audit,tools,scripts} + README
- 门禁验证：sdk pytest 125 通过、ui vitest 158 通过、
生产 BFF 脚本重启四上游全绿 + 静态托管 + 会话库无损
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"
|