"""系统状态来自实际资源和迁移账本；模型设置只保存草案，不能越过启用链。""" 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")