67 lines
1.9 KiB
Plaintext
67 lines
1.9 KiB
Plaintext
# -*- coding: utf-8 -*-
|
||
"""旧库废弃表迁移样例（复制到 tests/ 并按技能改写）。
|
||
|
||
验收：给定旧库含废弃物理表 + 元数据 + 历史行 → init_db() 后表与元数据均清除。
|
||
见 references/SCHEMA.md「表增删改与退役」。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import unittest
|
||
|
||
from _support import IsolatedDataRoot
|
||
|
||
|
||
class TestTableDeprecationMigration(unittest.TestCase):
|
||
def test_init_db_drops_legacy_task_logs(self) -> None:
|
||
with IsolatedDataRoot():
|
||
from db.connection import get_conn, init_db
|
||
from db.display_metadata import iter_user_tables
|
||
|
||
# 1) 模拟升级前的旧库
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"""
|
||
CREATE TABLE task_logs (
|
||
id INTEGER PRIMARY KEY,
|
||
task_type TEXT,
|
||
status TEXT
|
||
)
|
||
"""
|
||
)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO _jiangchang_tables
|
||
(table_name, display_name, visible, readonly)
|
||
VALUES ('task_logs', '任务日志', 1, 0)
|
||
"""
|
||
)
|
||
cur.execute("INSERT INTO task_logs (task_type, status) VALUES ('run', 'ok')")
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
# 2) 新代码 init_db() 应幂等清除（在 connection.init_db 中调用 drop_deprecated_user_table）
|
||
init_db()
|
||
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='task_logs'"
|
||
)
|
||
self.assertIsNone(cur.fetchone())
|
||
cur.execute(
|
||
"SELECT COUNT(*) FROM _jiangchang_tables WHERE table_name='task_logs'"
|
||
)
|
||
self.assertEqual(int(cur.fetchone()[0]), 0)
|
||
self.assertNotIn("task_logs", list(iter_user_tables(cur)))
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|