0658d178ad
- Makefile：新增 验收数据库（生成→检查→库层全量，前置校验隔离库连接串）、数据库分片、浏览器测试三个入口； 格式写入 末尾就地重新生成；数据库测试 与 验收数据库 统一排除 浏览器/网络；快检 不再静默跳过类型门； 前端旅程 预检补齐四个必需变量；pytest 目标改用仓内解释器（uv run 在嵌套检出会解析到外层环境）。 - 工具/验证锁.py：验证会话持共享锁，写入口用 --执行 在独占锁内落盘，生成/格式写入/索引生成走同一闸门。 - 工具/并行数据库测试.py：按文件分片并行，缺连接串在采集前拒绝。 - 工具/构建编排.py、环境预检.py、构建资源包.py、维护索引.py 与上述口径对齐。 - .gitignore / CI / README / AGENTS：忽略构建产物、CI 与 Makefile 单一口径、README 按实际实现陈述、AGENTS 补日常入口与验证纪律。
93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
"""开发测试显式源资源视图，不生成发布包，也不提供安装资格。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from muse.共享.错误 import 资源缺失错误
|
|
|
|
|
|
class 源码资源提供器:
|
|
"""只读取登记的工程资源；测试装配必须显式注入。"""
|
|
|
|
def __init__(self, 根: Path):
|
|
self._根 = 根
|
|
self._已加载 = False
|
|
self._资源: dict[str, bytes] = {}
|
|
self._清单: dict[str, Any] = {}
|
|
|
|
def _确保加载(self) -> None:
|
|
if self._已加载:
|
|
return
|
|
根 = self._根
|
|
from muse.接入.http.应用 import 导出合同
|
|
from 构建资源包 import 编译能力资源
|
|
|
|
登记 = json.loads((根 / "工具/资源登记.json").read_text())
|
|
for 项 in 登记["资源"]:
|
|
源 = (根 / 项["源路径"]).resolve()
|
|
if not 源.is_relative_to(根.resolve()):
|
|
raise ValueError("源码资源越过登记根")
|
|
self._资源[项["打包路径"]] = 源.read_bytes()
|
|
命令 = {
|
|
操作["operationId"]
|
|
for 路径 in 导出合同()["paths"].values()
|
|
for 操作 in 路径.values()
|
|
if isinstance(操作, dict) and "operationId" in 操作
|
|
}
|
|
目录, 能力 = 编译能力资源(根, 登记.get("能力文件", []), 命令)
|
|
self._资源.update(能力)
|
|
self._资源["能力/目录.json"] = json.dumps(
|
|
目录, ensure_ascii=False, sort_keys=True, indent=2
|
|
).encode()
|
|
for 文件 in sorted((根 / 登记["迁移目录"]).glob("V*.sql")):
|
|
self._资源[f"迁移/{文件.name}"] = 文件.read_bytes()
|
|
资源表 = {
|
|
路径: {"sha256": hashlib.sha256(数据).hexdigest()}
|
|
for 路径, 数据 in sorted(self._资源.items())
|
|
}
|
|
身份 = hashlib.sha256(json.dumps(资源表, sort_keys=True).encode()).hexdigest()
|
|
self._清单 = {
|
|
"schema_version": 1,
|
|
"资源来源": "开发源码",
|
|
"发布身份": "development-source",
|
|
"构建身份": 身份,
|
|
"代码版本": "development",
|
|
"代码哈希": "development-not-an-installation",
|
|
"合同版本": 1,
|
|
"资源": 资源表,
|
|
}
|
|
|
|
from 效果依赖 import 构建效果依赖
|
|
|
|
登记路径 = {项["打包路径"] for 项 in 登记["资源"]}
|
|
能力路径 = {*能力, "能力/目录.json"}
|
|
self._清单["效果依赖"] = 构建效果依赖(
|
|
根,
|
|
{路径: 项["sha256"] for 路径, 项 in 资源表.items()},
|
|
{
|
|
"registered": sorted(登记路径),
|
|
"capabilities": sorted(能力路径),
|
|
"migrations": sorted(set(资源表) - 登记路径 - 能力路径),
|
|
},
|
|
)
|
|
self._已加载 = True
|
|
|
|
def 加载清单(self) -> dict[str, Any]:
|
|
self._确保加载()
|
|
# 返回独立结构，测试不能污染后续用例的共享输入。
|
|
return json.loads(json.dumps(self._清单))
|
|
|
|
def 打开资源(self, 名称: str) -> bytes:
|
|
self._确保加载()
|
|
try:
|
|
return self._资源[名称]
|
|
except KeyError as exc:
|
|
raise 资源缺失错误(f"源码测试资源未登记：{名称}") from exc
|
|
|
|
def 核对资源(self) -> dict[str, Any]:
|
|
return self.加载清单()
|