713cc45c63
- 51 个增量迁移文件压扁为 V0001__基线.sql 完整快照（结构+种子+授权）， 等价门禁：旧链全量执行库与基线库 pg_dump 逐字节一致 - 剔除 pg_dump 固化的 public schema 超级用户归属断言（muse_maint 无权执行） - 测试删减至保留集：备份往返 2 + 预算 2 + 租约 2 + 基线建库 1 - 租约/预算夹具改共享库，消除按例克隆建库 - 修复共享库三类既有污染：账本注入残留（系统管理）、失败触发器残留（评测）、 建表残留（正式变更事务），发布包迁移文件名硬编码改动态核对 - 全量数据库验收 722 passed / 0 failed / 0 errors（main 基线为 24F+291E） - 运行手册登记基线模式改表流程与账本校验和同步
233 lines
10 KiB
Python
233 lines
10 KiB
Python
"""系统状态来自实际资源和迁移账本；模型设置只保存草案，不能越过启用链。"""
|
||
|
||
from dataclasses import asdict, replace
|
||
from uuid import uuid4
|
||
|
||
import psycopg
|
||
import pytest
|
||
import test_定稿与导出 as 定稿测试
|
||
import test_生产评测权限隔离 as 接入测试
|
||
from fastapi.testclient import TestClient
|
||
|
||
from muse.任务运行.接口 import 凭据引用, 提供方配置, 运行配置内容, 配置版本管理
|
||
from muse.作品规划.接口 import 档案保存
|
||
from muse.共享.调用身份 import 用途
|
||
from muse.接入.cli.入口 import main
|
||
from muse.接入.http.应用 import 创建应用
|
||
from muse.资源加载 import 加载清单
|
||
from muse.配置 import 读取配置
|
||
|
||
pytestmark = pytest.mark.数据库
|
||
候选环境 = 定稿测试.候选环境
|
||
交付环境 = 定稿测试.交付环境
|
||
|
||
|
||
@pytest.fixture
|
||
def 脏账本还原(应用测试库):
|
||
"""账本污染用例的例后还原；muse_migration 属种子表不被例间清空，必须自愈。"""
|
||
from muse.基础设施.数据库.迁移 import 列出迁移
|
||
|
||
yield 应用测试库
|
||
基线 = 列出迁移()[0]
|
||
with 应用测试库[用途.维护].连接() as conn, conn.transaction():
|
||
conn.execute("DELETE FROM muse_migration WHERE version > %s", (基线.版本,))
|
||
conn.execute(
|
||
"UPDATE muse_migration SET name=%s, checksum=%s WHERE version=%s",
|
||
(基线.文件名, 基线.校验和, 基线.版本),
|
||
)
|
||
|
||
|
||
def _配置(pool, path):
|
||
cfg = 接入测试._配置文件(pool, path)
|
||
cfg.write_text(
|
||
cfg.read_text().replace(
|
||
'"发布身份"="test"',
|
||
'"发布身份"="' + 加载清单()["构建身份"] + '"',
|
||
)
|
||
)
|
||
return 读取配置(cfg)
|
||
|
||
|
||
def _草案(tmp_path):
|
||
return 运行配置内容(
|
||
"direct",
|
||
"1",
|
||
"1",
|
||
"muse-foundation-r2",
|
||
"draft-budget",
|
||
{"writer": {"provider": "draft-provider", "model": "draft-only-model", "thinking": "high"}},
|
||
(凭据引用("draft-key", "受控存储", str(tmp_path / "private-provider-key")),),
|
||
(
|
||
提供方配置(
|
||
"draft-provider", "responses", "https://example.invalid/provider", "draft-key"
|
||
),
|
||
),
|
||
"draft-pricing",
|
||
)
|
||
|
||
|
||
def _登录(client):
|
||
assert (
|
||
client.post(
|
||
"/api/v1/session",
|
||
json={"password": "synthetic-evaluation-only"},
|
||
).status_code
|
||
== 200
|
||
)
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w28-28b001",
|
||
environment="隔离PG",
|
||
given="明确隔离配置、发布包和受控测试资料",
|
||
when="经实际入口执行对应操作并读回",
|
||
then=["系统页核实际代码资源版本且配置草案不启用"],
|
||
contract="docs/系统架构/新版设计/文件设计/工程配置与部署.md",
|
||
)
|
||
def test_系统页核实际代码资源版本且配置草案不启用__28b001(应用测试库, tmp_path):
|
||
production = 应用测试库[用途.生产]
|
||
config = _配置(production, tmp_path)
|
||
app = 创建应用(config)
|
||
with TestClient(创建应用(replace(config, 资源发布身份=加载清单()["发布身份"]))) as legacy:
|
||
assert legacy.get("/ready").status_code == 503
|
||
with TestClient(app, headers={"origin": "http://testserver"}) as client:
|
||
assert client.get("/api/v1/system").status_code == 401
|
||
_登录(client)
|
||
status = client.get("/api/v1/system")
|
||
assert status.status_code == 200, status.text
|
||
report = status.json()
|
||
assert report["ready"] is True
|
||
assert report["release"]["code_matches"] is True
|
||
assert report["database"]["compatible"] is True
|
||
assert report["database"]["installed_version"] == report["database"]["database_version"]
|
||
assert str(tmp_path) not in status.text
|
||
policy = client.get("/api/v1/system/model-policy").json()
|
||
assert set(policy["roles"]) == {"writer", "planner", "detector", "extractor", "judge"}
|
||
assert policy["models"][policy["roles"]["writer"]["model_policy"]]
|
||
assert policy["roles"]["writer"]["readonly_tools"] is True
|
||
request = {"config_id": "draft", "version": "1", "content": asdict(_草案(tmp_path))}
|
||
result = client.post("/api/v1/system/runtime-configs", json=request)
|
||
assert result.status_code == 200, result.text
|
||
assert result.json()["state"] == "draft"
|
||
configs = client.get("/api/v1/system/runtime-configs")
|
||
rows = configs.json()
|
||
assert len(rows) == 1 and not rows[0]["active"] and rows[0]["generation"] == 0
|
||
assert rows[0]["roles"]["writer"]["model"] == "draft-only-model"
|
||
for private in ["private-provider-key", "example.invalid/provider", "draft-key", "凭据"]:
|
||
assert private not in configs.text
|
||
invalid = client.post(
|
||
"/api/v1/system/runtime-configs",
|
||
json={
|
||
"config_id": "bad",
|
||
"version": "1",
|
||
"content": {},
|
||
},
|
||
)
|
||
assert invalid.status_code == 422
|
||
assert 配置版本管理(应用测试库[用途.评测]).列出摘要() == []
|
||
with production.连接() as conn, pytest.raises(psycopg.errors.InsufficientPrivilege):
|
||
conn.execute("DELETE FROM muse_migration")
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w28-28b002",
|
||
environment="隔离PG",
|
||
given="明确隔离配置、发布包和受控测试资料",
|
||
when="经实际入口执行对应操作并读回",
|
||
then=["未知数据库版本或校验和漂移拒绝HTTP写入保留只读核对"],
|
||
contract="docs/系统架构/新版设计/文件设计/工程配置与部署.md",
|
||
)
|
||
@pytest.mark.parametrize("mode", ["future", "checksum"])
|
||
def test_未知数据库版本或校验和漂移拒绝HTTP写入保留只读核对__28b002(
|
||
应用测试库,
|
||
脏账本还原,
|
||
tmp_path,
|
||
mode,
|
||
):
|
||
app = 创建应用(_配置(应用测试库[用途.生产], tmp_path))
|
||
with 应用测试库[用途.维护].连接() as conn, conn.transaction():
|
||
if mode == "future":
|
||
conn.execute(
|
||
"INSERT INTO muse_migration(version,name,checksum) VALUES(999,'future',%s)",
|
||
("0" * 64,),
|
||
)
|
||
else:
|
||
conn.execute("UPDATE muse_migration SET checksum=%s WHERE version=1", ("0" * 64,))
|
||
with TestClient(app, headers={"origin": "http://testserver"}) as client:
|
||
_登录(client)
|
||
assert client.get("/ready").status_code == 503
|
||
report = client.get("/api/v1/system").json()
|
||
assert report["ready"] is False
|
||
field = "unknown_versions" if mode == "future" else "changed_versions"
|
||
assert report["database"][field] == ([999] if mode == "future" else [1])
|
||
result = client.post(
|
||
"/api/v1/system/runtime-configs",
|
||
json={
|
||
"config_id": "blocked",
|
||
"version": "1",
|
||
"content": asdict(_草案(tmp_path)),
|
||
},
|
||
)
|
||
assert result.status_code == 409, result.text
|
||
assert result.json()["code"] == "MUSE_SCHEMA_MISMATCH"
|
||
assert client.get("/api/v1/system/runtime-configs").json() == []
|
||
assert client.delete("/api/v1/session").status_code == 200
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w28-28b003",
|
||
environment="隔离PG",
|
||
given="明确隔离配置、发布包和受控测试资料",
|
||
when="经实际入口执行对应操作并读回",
|
||
then=["不兼容时CLI拒绝配置写入而系统诊断可读"],
|
||
contract="docs/系统架构/新版设计/文件设计/工程配置与部署.md",
|
||
)
|
||
def test_不兼容时CLI拒绝配置写入而系统诊断可读__28b003(应用测试库, 脏账本还原, tmp_path, capsys):
|
||
_配置(应用测试库[用途.生产], tmp_path)
|
||
cfg = next(tmp_path.glob("*.toml"))
|
||
with 应用测试库[用途.维护].连接() as conn, conn.transaction():
|
||
conn.execute(
|
||
"INSERT INTO muse_migration(version,name,checksum) VALUES(999,'future',%s)",
|
||
("0" * 64,),
|
||
)
|
||
assert main(["配置", str(cfg), "保存", "blocked", "1", "--内容", "missing.toml"]) == 1
|
||
assert "MUSE_SCHEMA_MISMATCH" in capsys.readouterr().err
|
||
assert main(["管理", str(cfg), "系统状态"]) == 0
|
||
assert '"unknown_versions": [999]' in capsys.readouterr().out
|
||
assert 配置版本管理(应用测试库[用途.生产]).列出摘要() == []
|
||
|
||
|
||
@pytest.mark.case_id(
|
||
"NC-w28-28b004",
|
||
environment="隔离PG",
|
||
given="明确隔离配置、发布包和受控测试资料",
|
||
when="经实际入口执行对应操作并读回",
|
||
then=["交付跨作品引用数据库拒绝且任务盘点按用途隔离"],
|
||
contract="docs/系统架构/新版设计/文件设计/工程配置与部署.md",
|
||
)
|
||
def test_交付跨作品引用数据库拒绝且任务盘点按用途隔离__28b004(交付环境):
|
||
app, author, pools, request = 交付环境
|
||
app.要求交付().冻结定稿(author, "cross-reference", request)
|
||
work = app.要求作品()
|
||
schema = work.新档案结构(author, "other", "work_core", 1)
|
||
work.保存档案(author, "other-work", 档案保存("other", 0, {"名称": "另一合成作品"}, schema))
|
||
production = pools[用途.生产]
|
||
with production.连接() as conn, conn.transaction():
|
||
with pytest.raises(psycopg.errors.ForeignKeyViolation), conn.transaction():
|
||
conn.execute(
|
||
"INSERT INTO muse_release(release_id,author_id,work_id,delivery_id,payload,"
|
||
"content_hash) VALUES(%s,%s,'other',%s,'{}',%s)",
|
||
(uuid4(), author.作者, request.delivery_id, "0" * 64),
|
||
)
|
||
from muse.编排.定稿交付 import 发起定稿导出
|
||
|
||
发起定稿导出(app, author, "count", request.delivery_id, ["txt"])
|
||
assert app.任务运行.盘点任务(author.作者) == {"queued": 1}
|
||
assert app.任务运行.盘点任务("other-author") == {}
|
||
with pools[用途.评测].连接(只读=True) as conn:
|
||
assert (
|
||
conn.execute("SELECT count(*) FROM evaluation.muse_task_inventory").fetchone()[0] == 0
|
||
)
|
||
with pytest.raises(psycopg.errors.InsufficientPrivilege):
|
||
conn.execute("SELECT * FROM public.muse_task_inventory")
|