muse-agent-example/tests/单元/test_备份发布身份.py
zizi b9af240b68 R2 收尾:W30 唯一写入观察与最终验收(C08 章后接通 / 构建身份重建 / 最终备份与恢复演练)
- 章后处理:事实提案命令身份加入类型、批内去重改为先校验后去重(REPEAT_OBJECT_TYPE 记跳过、不再静默丢重复);查询章后状态回传事实跳过;可按需发布/核对可登记任务与 config_not_declared 语义收紧
- 模型治理:推理字符数计量(ChatCompletions/Messages/执行合同)、探针在声明关闭推理却仍有推理时拒绝、计价支持缓存协议与未知模型兜底
- 运行态:资源包重建(发布身份 21d850f7…),三处配置同步,post-extract v5 代次 5 启用,服务 PID 96675(health/ready 通过)
- 门禁:make 检查通过;离线 1008 passed;定点数据库 101 passed;最终完整备份 bkp-1f5591d8… 与空库恢复演练 verified(152 表/158 关系/迁移 45,演练库已 DROP)
- 文档与证据:运行手册、功能覆盖、工作包清单 W30 证据、测试用例清单(17 条 execution_evidence 回填 + 历史死指针说明)、目标文件清单
- 旧实现与旧库直连材料退出(含明文凭据文件移除);新配置 配置/本机正式.toml、配置/本机维护.toml、配置/计费/、配置/运行配置/ 入库,均只含受控引用
- R2 执行证据在 .agents.local/改造/R2-20260909/(不入库)
2026-09-16 08:23:39 +08:00

79 lines
3.1 KiB
Python

"""备份恢复不因迁移集相同就接受另一程序资源构建。"""
import configparser
import subprocess
from dataclasses import replace
from pathlib import Path
import pytest
from muse.共享.调用身份 import 用途
from muse.基础设施 import 备份恢复
from muse.基础设施.备份恢复 import 创建备份, 备份恢复错误, 恢复备份
from muse.资源加载 import 加载清单
from muse.配置 import 应用配置, 数据库引用
def test_相同发布标签不能掩盖不同构建且拒绝发生在目标连接前__29a011(monkeypatch, tmp_path):
installed = 加载清单()
config = 应用配置(
数据库引用("受控存储", str(tmp_path / "missing")), installed["构建身份"], 运行用途=用途.维护
)
monkeypatch.setattr(
备份恢复,
"核对备份",
lambda _: {
"backup_id": "chosen",
"resource_release": installed["发布身份"],
"resource_build_id": "0" * 64,
},
)
with pytest.raises(备份恢复错误, match="构建身份"):
恢复备份(tmp_path / "backup", config, "chosen")
with pytest.raises(备份恢复错误, match="程序、资源或配置"):
创建备份(replace(config, 资源发布身份="wrong-release"), tmp_path / "new")
assert not (tmp_path / "new").exists()
@pytest.mark.parametrize("失败", [False, True])
def test_备份客户端保活参数传递且临时配置总被清理__c0f001(monkeypatch, tmp_path, 失败):
凭据 = tmp_path / "连接.txt"
凭据.write_text(
"host=127.0.0.1 dbname=muse user=muse_maint password=test-secret "
"keepalives=1 keepalives_idle=15 keepalives_interval=5 keepalives_count=3"
)
凭据.chmod(0o600)
引用 = 数据库引用("受控存储", str(凭据))
临时文件 = []
monkeypatch.setenv("PGSERVICE", "untrusted")
def 执行(命令, **参数):
环境 = 参数["env"]
文件 = Path(环境["PGSERVICEFILE"])
临时文件.append(文件)
assert 文件.stat().st_mode & 0o777 == 0o600
配置 = configparser.ConfigParser()
配置.read(文件)
assert dict(配置[环境["PGSERVICE"]]) == {
"keepalives": "1",
"keepalives_idle": "15",
"keepalives_interval": "5",
"keepalives_count": "3",
}
assert 环境["PGUSER"] == "muse_maint"
assert 环境["PGPASSWORD"] == "test-secret"
assert "test-secret" not in " ".join(命令)
assert "test-secret" not in 文件.read_text()
if 失败:
raise subprocess.TimeoutExpired(命令, 300)
return subprocess.CompletedProcess(命令, 0, stderr=b"")
monkeypatch.setattr(备份恢复.subprocess, "run", 执行)
if 失败:
with pytest.raises(备份恢复错误, match="客户端无法完成"):
备份恢复._运行客户端("pg_dump", (), 引用, "数据库备份")
else:
备份恢复._运行客户端("pg_dump", (), 引用, "数据库备份")
assert len(临时文件) == 1
assert not 临时文件[0].exists()