101 lines
4.1 KiB
Python
101 lines
4.1 KiB
Python
"""tests/test_db.py —— 三表迁移幂等 + JSON 编解码。"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.db import Database, dumps_json, loads_json
|
|
|
|
pytestmark = pytest.mark.bioflow_unit
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.new_event_loop().run_until_complete(coro)
|
|
|
|
|
|
def _tables(db):
|
|
return {r[0] for r in _run(db.fetchall("SELECT name FROM sqlite_master WHERE type='table'"))}
|
|
|
|
|
|
def test_schema_created(db):
|
|
tables = _tables(db)
|
|
assert {"bio_jobs", "bio_stages", "bio_archives"} <= tables
|
|
|
|
|
|
def test_migration_idempotent(tmp_path):
|
|
"""TC-BIO-DB：重复 connect 幂等（不报错不重复建表）。"""
|
|
path = tmp_path / "a" / "bioflow.db"
|
|
db1 = Database(path)
|
|
_run(db1.connect())
|
|
_run(db1.close())
|
|
db2 = Database(path)
|
|
_run(db2.connect())
|
|
tables = _tables(db2)
|
|
assert {"bio_jobs", "bio_stages", "bio_archives"} <= tables
|
|
_run(db2.close())
|
|
|
|
|
|
def test_incremental_column_migration(tmp_path):
|
|
"""老库（无增量列）connect 后补列（bio_jobs.error / bio_archives.report_artifact）。"""
|
|
import sqlite3
|
|
|
|
path = tmp_path / "old" / "bioflow.db"
|
|
path.parent.mkdir(parents=True)
|
|
conn = sqlite3.connect(path)
|
|
conn.execute("""CREATE TABLE bio_jobs (
|
|
id TEXT PRIMARY KEY, session_id TEXT NOT NULL DEFAULT '', title TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT 'created', stage_chain TEXT NOT NULL DEFAULT '[]',
|
|
params TEXT NOT NULL DEFAULT '{}', auto_continue INTEGER NOT NULL DEFAULT 0,
|
|
progress INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)""")
|
|
conn.execute("""CREATE TABLE bio_stages (
|
|
id TEXT PRIMARY KEY, job_id TEXT NOT NULL REFERENCES bio_jobs(id) ON DELETE CASCADE,
|
|
stage TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'waiting', params TEXT NOT NULL DEFAULT '{}',
|
|
manifest_path TEXT, log_tail TEXT NOT NULL DEFAULT '', attempt INTEGER NOT NULL DEFAULT 0,
|
|
started_at TEXT, finished_at TEXT, UNIQUE(job_id, stage))""")
|
|
conn.execute("""CREATE TABLE bio_archives (
|
|
id TEXT PRIMARY KEY, job_id TEXT NOT NULL, name TEXT NOT NULL DEFAULT '',
|
|
description TEXT NOT NULL DEFAULT '', stage_chain TEXT NOT NULL DEFAULT '[]',
|
|
params TEXT NOT NULL DEFAULT '{}', disk_path TEXT, size_bytes INTEGER NOT NULL DEFAULT 0,
|
|
reused_count INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL)""")
|
|
conn.commit()
|
|
conn.close()
|
|
db = Database(path)
|
|
_run(db.connect())
|
|
cols = {r[1] for r in _run(db.fetchall("PRAGMA table_info(bio_jobs)"))}
|
|
assert "error" in cols and "archive_id" in cols
|
|
cols2 = {r[1] for r in _run(db.fetchall("PRAGMA table_info(bio_archives)"))}
|
|
assert "report_artifact" in cols2
|
|
_run(db.close())
|
|
|
|
|
|
def test_job_crud(db):
|
|
_run(db.execute(
|
|
"INSERT INTO bio_jobs (id, session_id, title, status, stage_chain, params, created_at, updated_at) "
|
|
"VALUES ('j1', 's1', 't', 'created', ?, ?, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')",
|
|
(dumps_json(["identify"]), dumps_json({}))))
|
|
row = _run(db.fetchone("SELECT * FROM bio_jobs WHERE id='j1'"))
|
|
assert row["title"] == "t"
|
|
assert loads_json(row["stage_chain"]) == ["identify"]
|
|
assert loads_json(row["params"]) == {}
|
|
|
|
|
|
def test_stage_cascade_delete(db):
|
|
_run(db.execute(
|
|
"INSERT INTO bio_jobs (id, session_id, title, status, stage_chain, params, created_at, updated_at) "
|
|
"VALUES ('j1', 's1', 't', 'created', '[]', '{}', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')"))
|
|
_run(db.execute(
|
|
"INSERT INTO bio_stages (id, job_id, stage, status, params, log_tail, attempt) "
|
|
"VALUES ('j1:identify', 'j1', 'identify', 'waiting', '{}', '', 0)"))
|
|
_run(db.execute("DELETE FROM bio_jobs WHERE id='j1'"))
|
|
rows = _run(db.fetchall("SELECT * FROM bio_stages WHERE job_id='j1'"))
|
|
assert rows == [] # FK CASCADE
|
|
|
|
|
|
def test_dumps_loads_json():
|
|
assert dumps_json({"a": 1}) == '{"a": 1}'
|
|
assert loads_json('{"a": 1}') == {"a": 1}
|
|
assert loads_json(None, "d") == "d"
|
|
assert loads_json("bad", []) == []
|
|
assert dumps_json(object(), "x") == '"x"'
|