9712589bfe
- 删除 数据库/旧库迁移 工具树、旧库迁移用例与旧库样本夹具；摘除 Makefile/pyproject/ruff/pyright 与 PYTHONPATH 引用，同步 README/AGENTS/设计文档与索引、分片工具 - 清空业务数据改按 种子表 登记过滤：未知 schema 与新增运行时表默认清空，迁移账本/索引代次单例/原文命名空间/结构种子保留；迁移新增种子表漏登记由新单测机械拦截 - 共享库复用前兜底清空，防上例强杀残留；例间只清数据不删表，个案探针表须自带 DROP TABLE IF EXISTS（规则与迁移用例同步） - 底座生命周期单测由 15 例精简为 5 例（保留清理、未封存模板、失败清理相关用例），其余 11 例随本批移除
146 lines
5.3 KiB
Python
146 lines
5.3 KiB
Python
"""不连接 PG，验证模板失败、复用及清理所有权的生命周期。"""
|
||
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from 支持.数据库底座 import 库身份, 数据库底座, 种子表
|
||
|
||
|
||
class 环境替身:
|
||
def __init__(self):
|
||
self.库 = {}
|
||
self.迁移次数 = 0
|
||
self.角色次数 = 0
|
||
self.序号 = 0
|
||
self.失败迁移 = False
|
||
self.拒绝清理 = set()
|
||
|
||
def 初始化角色(self, 定义):
|
||
self.角色次数 += 1
|
||
return True
|
||
|
||
def 环境身份(self):
|
||
return {"major": 16, "extensions": [["plpgsql", "1.0"]]}
|
||
|
||
def 创建(self, 名称, 注记, 模板=None):
|
||
if 模板:
|
||
assert self.库[模板.名称][0] == 模板
|
||
if not self.库[模板.名称][1]:
|
||
raise RuntimeError("模板未封存")
|
||
self.序号 += 1
|
||
身份 = 库身份(名称, self.序号, 42, 注记)
|
||
self.库[名称] = (身份, False)
|
||
return 身份
|
||
|
||
def 迁移(self, 身份, 目录, 引用目录):
|
||
self.迁移次数 += 1
|
||
if self.失败迁移:
|
||
raise RuntimeError("迁移失败")
|
||
return len(list(目录.glob("*.sql")))
|
||
|
||
def 封存(self, 身份):
|
||
self.库[身份.名称] = (身份, True)
|
||
|
||
def 删除(self, 身份):
|
||
if 身份.名称 in self.拒绝清理:
|
||
raise RuntimeError("连接仍占用")
|
||
if self.库[身份.名称][0] != 身份:
|
||
raise RuntimeError("身份漂移")
|
||
del self.库[身份.名称]
|
||
|
||
def 工厂(self, 名称, 目录):
|
||
return {"database": 名称}
|
||
|
||
def 清空业务数据(self, 身份):
|
||
self.清空次数 = getattr(self, "清空次数", 0) + 1
|
||
|
||
|
||
@pytest.fixture
|
||
def 环境(tmp_path):
|
||
根 = tmp_path / "repo"
|
||
(根 / "数据库/初始化").mkdir(parents=True)
|
||
(根 / "数据库/初始化/用途角色.sql").write_text("SELECT 1;")
|
||
(根 / "数据库/迁移").mkdir()
|
||
for 版本 in (1, 2, 4, 16):
|
||
(根 / f"数据库/迁移/V{版本:04d}__test.sql").write_text(f"SELECT {版本};")
|
||
后端 = 环境替身()
|
||
底座 = 数据库底座(后端, 根, tmp_path / "baseline")
|
||
yield 底座, 后端, 根
|
||
if 底座.存活:
|
||
后端.拒绝清理.clear()
|
||
底座.关闭()
|
||
|
||
|
||
@pytest.mark.case_id("NC-database-baseline-unit-15")
|
||
def test_共享库同指纹复用同一库且例间清空业务数据(环境, tmp_path):
|
||
底座, 后端, _ = 环境
|
||
名称 = []
|
||
for 行 in range(3):
|
||
with 底座.共享库(tmp_path / str(行), (1, 4, 16)) as 工厂:
|
||
名称.append(工厂["database"])
|
||
assert len(后端.库) == 2 # 1个模板 + 1个共享库
|
||
assert len(后端.库) == 2 # 例后不销毁共享库
|
||
assert len(set(名称)) == 1 # 全部复用同一个库名
|
||
动作 = [事件["action"] for 事件 in 底座.事件 if 事件["action"].startswith("shared")]
|
||
assert 动作.count("shared_clean") == 3 # 每次例后清空
|
||
assert 动作.count("shared_clean_before") == 2 # 复用前兑底清空；首用是新建库不清
|
||
底座.关闭()
|
||
assert 后端.库 == {} # 关闭时清理全部库
|
||
|
||
|
||
@pytest.mark.case_id("NC-database-baseline-unit-16")
|
||
def test_迁移种子表全部登记无漏项():
|
||
"""迁移 SQL 中 INSERT 的表必须出现在 种子表 登记；新增种子表漏登记在此被机械拦截。"""
|
||
迁移目录 = Path(__file__).resolve().parents[2] / "数据库/迁移"
|
||
模式 = re.compile(
|
||
r"INSERT\s+INTO\s+(?:%1\$I\s*\.\s*|[A-Za-z_]\w*\s*\.\s*)?([A-Za-z_]\w*)",
|
||
re.I,
|
||
)
|
||
插入表 = {
|
||
表名 for 文件 in sorted(迁移目录.glob("*.sql")) for 表名 in 模式.findall(文件.read_text())
|
||
}
|
||
已登记 = {表 for _, 表 in 种子表}
|
||
未登记 = 插入表 - 已登记
|
||
assert not 未登记, (
|
||
f"迁移向 {未登记} 插入种子数据但未登记；"
|
||
"请在 tests/支持/数据库底座.py 的 种子表 登记，否则共享库例间清空会破坏种子"
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id("NC-database-baseline-unit-05")
|
||
def test_测试异常仍清理而且不会删其他worker(环境, tmp_path):
|
||
底座, 后端, 根 = 环境
|
||
另一个 = 数据库底座(后端, 根, tmp_path / "worker2", worker="gw1")
|
||
另一个.基线()
|
||
with pytest.raises(ValueError), 底座.借库(tmp_path):
|
||
raise ValueError("调用失败")
|
||
底座.关闭()
|
||
assert len(后端.库) == 1
|
||
assert next(iter(后端.库)) == next(iter(另一个.存活))
|
||
另一个.关闭()
|
||
|
||
|
||
@pytest.mark.case_id("NC-database-baseline-unit-08")
|
||
def test_失败清理保留精确对象且继续清理其他库(环境):
|
||
底座, 后端, _ = 环境
|
||
一 = 底座.基线((1,))
|
||
二 = 底座.基线((1, 2))
|
||
后端.拒绝清理.add(二.名称)
|
||
with pytest.raises(ExceptionGroup):
|
||
底座.关闭()
|
||
assert 一.名称 not in 后端.库
|
||
assert list(底座.存活) == [二.名称]
|
||
assert any(e["action"] == "cleanup_failed" and e["database"] == 二.名称 for e in 底座.事件)
|
||
|
||
|
||
@pytest.mark.case_id("NC-database-baseline-unit-07")
|
||
def test_未封存模板不能克隆(环境, tmp_path):
|
||
底座, 后端, _ = 环境
|
||
模板 = 底座.基线()
|
||
后端.库[模板.名称] = (模板, False)
|
||
with pytest.raises(RuntimeError, match="未封存"), 底座.借库(tmp_path):
|
||
pass
|
||
assert len(后端.库) == 1
|