0658d178ad
- Makefile：新增 验收数据库（生成→检查→库层全量，前置校验隔离库连接串）、数据库分片、浏览器测试三个入口； 格式写入 末尾就地重新生成；数据库测试 与 验收数据库 统一排除 浏览器/网络；快检 不再静默跳过类型门； 前端旅程 预检补齐四个必需变量；pytest 目标改用仓内解释器（uv run 在嵌套检出会解析到外层环境）。 - 工具/验证锁.py：验证会话持共享锁，写入口用 --执行 在独占锁内落盘，生成/格式写入/索引生成走同一闸门。 - 工具/并行数据库测试.py：按文件分片并行，缺连接串在采集前拒绝。 - 工具/构建编排.py、环境预检.py、构建资源包.py、维护索引.py 与上述口径对齐。 - .gitignore / CI / README / AGENTS：忽略构建产物、CI 与 Makefile 单一口径、README 按实际实现陈述、AGENTS 补日常入口与验证纪律。
165 lines
6.7 KiB
Python
165 lines
6.7 KiB
Python
"""验收指定实际 wheel：隔离解压，仓外运行，缺失及损坏资源必须拒绝。
|
||
|
||
指定 wheel 时不重新构建；省略参数则增量构建并消费本批 wheel。
|
||
不安装依赖、不启动服务；PG/浏览器/恢复验收仍由各自环境承担。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import sysconfig
|
||
import tempfile
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
_探针 = r"""
|
||
import hashlib, json, os, pathlib, socket, sys
|
||
def deny_connection(self, address):
|
||
raise AssertionError("包入口解析期间不得连接外部服务")
|
||
socket.socket.connect = deny_connection
|
||
prefix = pathlib.Path(sys.argv[1]).resolve()
|
||
source = pathlib.Path(sys.argv[3]).resolve()
|
||
def reject_source_reads(event, args):
|
||
if event in {"open", "os.listdir", "os.scandir"} and args and isinstance(args[0], (str, bytes)):
|
||
target = pathlib.Path(os.fsdecode(args[0])).resolve()
|
||
if target.is_relative_to(source) or ".git" in target.parts:
|
||
raise AssertionError("installed package attempted to read source or Git")
|
||
sys.addaudithook(reject_source_reads)
|
||
sys.path.insert(0, str(prefix))
|
||
sys.path.append(sys.argv[2])
|
||
import muse, muse.启动, muse.配置
|
||
from muse.启动 import 应用装配, 构建
|
||
assert callable(应用装配) and callable(构建)
|
||
assert pathlib.Path(muse.__file__).resolve().is_relative_to(prefix), muse.__file__
|
||
from muse.资源加载 import 核对资源
|
||
from muse.共享.错误 import 资源缺失错误
|
||
try:
|
||
manifest = 核对资源()
|
||
except 资源缺失错误 as error:
|
||
print(json.dumps({"拒绝": type(error).__name__, "原因": str(error)}, ensure_ascii=False))
|
||
sys.exit(23)
|
||
files = {"src/" + f.relative_to(prefix).as_posix(): hashlib.sha256(f.read_bytes()).hexdigest()
|
||
for f in (prefix / "muse").rglob("*")
|
||
if f.suffix in {".py", ".ts"} and "资源" not in f.relative_to(prefix / "muse").parts}
|
||
code_hash = hashlib.sha256(json.dumps(files, sort_keys=True).encode()).hexdigest()
|
||
assert code_hash == manifest["代码哈希"]
|
||
from muse.编排.生成规划 import 规划模板
|
||
from muse.审校修订.完整审校 import 文学模板
|
||
from muse.资源加载 import 读取能力
|
||
for kind, resource in [("outline", "规划/生成作品规划.md"),
|
||
("fine_outline", "规划/细化章节计划.md")]:
|
||
text, digest = 规划模板(kind)
|
||
assert text and digest == manifest["资源"][resource]["sha256"]
|
||
text, digest = 文学模板()
|
||
assert text and digest == manifest["资源"]["审校/创作质量审阅.md"]["sha256"]
|
||
for capability in ("write-next-chapter", "check-consistency"):
|
||
assert 读取能力("operation", capability)["正文"]
|
||
print(json.dumps({"构建身份": manifest["构建身份"], "资源数": len(manifest["资源"]),
|
||
"加载位置": str(muse.__file__)}, ensure_ascii=False))
|
||
"""
|
||
|
||
|
||
def 验收(轮子: Path) -> dict:
|
||
if not 轮子.is_file() or 轮子.suffix != ".whl":
|
||
raise ValueError("必须提供本批实际 wheel 文件")
|
||
包摘要 = hashlib.sha256(轮子.read_bytes()).hexdigest()
|
||
环境 = {k: v for k, v in os.environ.items() if k != "PYTHONPATH" and not k.startswith("MUSE_")}
|
||
环境["PYTHONDONTWRITEBYTECODE"] = "1"
|
||
with tempfile.TemporaryDirectory(prefix="muse-包验收-") as 临时:
|
||
根 = Path(临时)
|
||
前缀 = 根 / "installed"
|
||
with zipfile.ZipFile(轮子) as zf:
|
||
名称们 = zf.namelist()
|
||
if len(名称们) != len(set(名称们)):
|
||
raise ValueError("wheel 存在重复路径")
|
||
for 名 in 名称们:
|
||
if Path(名).is_absolute() or {
|
||
"..",
|
||
".git",
|
||
"tests",
|
||
".agents.local",
|
||
".env",
|
||
}.intersection(Path(名).parts):
|
||
raise ValueError(f"wheel 包含不允许的路径：{名}")
|
||
zf.extractall(前缀)
|
||
结果 = []
|
||
|
||
def 运行(场景: str, 期望: int) -> None:
|
||
进程 = subprocess.run(
|
||
[
|
||
sys.executable,
|
||
"-I",
|
||
"-S",
|
||
"-B",
|
||
"-c",
|
||
_探针,
|
||
str(前缀),
|
||
sysconfig.get_path("purelib"),
|
||
str(Path(__file__).resolve().parents[1]),
|
||
],
|
||
cwd=根,
|
||
env=环境,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=120,
|
||
)
|
||
if 进程.returncode != 期望:
|
||
raise RuntimeError(
|
||
f"{场景} exit={进程.returncode}，期望={期望}\n{进程.stdout}\n{进程.stderr}"
|
||
)
|
||
结果.append({"场景": 场景, "exit": 进程.returncode, "输出": json.loads(进程.stdout)})
|
||
|
||
运行("完整实际wheel", 0)
|
||
清单文件 = 前缀 / "muse/资源/清单.json"
|
||
原清单 = 清单文件.read_bytes()
|
||
清单文件.unlink()
|
||
运行("缺清单必须拒绝", 23)
|
||
清单文件.write_bytes(原清单)
|
||
资源 = 前缀 / "muse/资源/运行资源.zip"
|
||
原包 = 资源.read_bytes()
|
||
资源.unlink()
|
||
运行("缺资源必须拒绝", 23)
|
||
资源.write_bytes(原包)
|
||
with zipfile.ZipFile(资源) as zf:
|
||
条目 = [(info, zf.read(info.filename)) for info in zf.infolist()]
|
||
with zipfile.ZipFile(资源, "w") as zf:
|
||
for 序号, (信息, 数据) in enumerate(条目):
|
||
zf.writestr(信息, 数据 + b"tamper" if 序号 == 0 else 数据)
|
||
运行("资源字节损坏必须拒绝", 23)
|
||
if hashlib.sha256(轮子.read_bytes()).hexdigest() != 包摘要:
|
||
raise RuntimeError("验收期间原 wheel 发生变化")
|
||
return {"wheel": str(轮子), "sha256": 包摘要, "场景": 结果}
|
||
|
||
|
||
def 主() -> int:
|
||
解析器 = argparse.ArgumentParser(description=__doc__)
|
||
解析器.add_argument("wheel", type=Path, nargs="?")
|
||
解析器.add_argument("--报告", type=Path)
|
||
参数 = 解析器.parse_args()
|
||
构建报告 = None
|
||
if 参数.wheel is None:
|
||
from 构建编排 import 执行构建
|
||
|
||
构建报告 = 执行构建(轮子=True)
|
||
轮子 = Path(构建报告["wheel"])
|
||
else:
|
||
轮子 = 参数.wheel.resolve()
|
||
报告 = 验收(轮子)
|
||
if 构建报告 is not None:
|
||
报告["同批构建"] = 构建报告
|
||
文本 = json.dumps(报告, ensure_ascii=False, indent=2) + "\n"
|
||
if 参数.报告:
|
||
参数.报告.parent.mkdir(parents=True, exist_ok=True)
|
||
参数.报告.write_text(文本)
|
||
print(文本)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(主())
|