From a789872c793f07b16e7288d3a5b5692c89c6f25a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E8=83=9C=E5=AE=87?= <17412268+zzzzz11122222@user.noreply.gitee.com> Date: Thu, 10 Sep 2026 17:11:52 +0800 Subject: [PATCH] test: record pre-migration workspace evidence --- .../foundation-migration-preflight.json | 54 ++++++++ .../test_foundation_migration_preflight.py | 95 ++++++++++++++ tools/foundation_migration_preflight.py | 117 ++++++++++++++++++ 3 files changed, 266 insertions(+) create mode 100644 docs/evidence/foundation-migration-preflight.json create mode 100644 tests/unit/tools/test_foundation_migration_preflight.py create mode 100644 tools/foundation_migration_preflight.py diff --git a/docs/evidence/foundation-migration-preflight.json b/docs/evidence/foundation-migration-preflight.json new file mode 100644 index 0000000..4f3ecab --- /dev/null +++ b/docs/evidence/foundation-migration-preflight.json @@ -0,0 +1,54 @@ +[ + { + "path": "D:\\桌面\\财富项目\\group_fqcd_jr", + "branch": "develop", + "head": "c6be99078e2be5d00397449c826fa049b60910ba", + "status": [ + "?? \"docs/15-\\347\\237\\245\\350\\257\\206\\346\\243\\200\\347\\264\\242\\346\\216\\245\\345\\205\\245\\346\\226\\271\\346\\241\\210.md\"", + "?? docs/superpowers/plans/2026-09-10-customer-service-rag-plan-a.md" + ] + }, + { + "path": "D:\\桌面\\财富项目\\group_fqcd_jr\\.worktrees\\customer-service-rag", + "branch": "feature/customer-service-rag", + "head": "ad7172367a0d69903e8bb92916f5d927084d8970", + "status": [ + " M app/core/config.py", + " M app/core/contracts.py", + " M app/main.py", + " M app/service/agent/bootstrap.py", + " M app/service/agent/factory.py", + " M app/service/agent/governance.py", + " M app/service/agent_run_application_service.py", + " M app/service/model_gateway.py", + " M app/worker/runtime.py", + " M tests/conftest.py", + " M tests/integration/test_agent_run_acceptance.py", + " M tests/integration/test_worker_runtime_mysql.py", + " M tests/unit/service/test_model_gateway.py", + "?? app/core/knowledge_contracts.py", + "?? app/infrastructure/milvus_knowledge_adapter.py", + "?? app/model/knowledge.py", + "?? app/service/agent/customer_service_agent.py", + "?? app/service/agent/customer_service_routing.py", + "?? app/service/knowledge_authority.py", + "?? app/service/knowledge_config.py", + "?? app/service/knowledge_retrieval_service.py", + "?? app/service/knowledge_tool_service.py", + "?? app/static/", + "?? tests/unit/api/test_customer_service_test_page.py", + "?? tests/unit/core/test_knowledge_contracts.py", + "?? tests/unit/infrastructure/test_milvus_knowledge_adapter.py", + "?? tests/unit/service/test_customer_service_agent.py", + "?? tests/unit/service/test_knowledge_authority.py", + "?? tests/unit/service/test_knowledge_config.py", + "?? tests/unit/service/test_knowledge_retrieval.py" + ] + }, + { + "path": "D:\\桌面\\财富项目\\111\\qyqy_develop", + "branch": "qyqy_develop", + "head": "6516ccb385024f7c9fe9e2207de69cd924e26262", + "status": [] + } +] diff --git a/tests/unit/tools/test_foundation_migration_preflight.py b/tests/unit/tools/test_foundation_migration_preflight.py new file mode 100644 index 0000000..0c76e2f --- /dev/null +++ b/tests/unit/tools/test_foundation_migration_preflight.py @@ -0,0 +1,95 @@ +"""验证迁移前状态证据采集器只读取 Git 元数据。""" + +# 导入测试所需的标准库 JSON、路径和子进程结果类型。 +import json +from pathlib import Path +from subprocess import CompletedProcess + +# 导入 pytest 补丁类型以及待验证的预检工具接口。 +from pytest import MonkeyPatch + +from tools.foundation_migration_preflight import ( + collect_workspace_state, + run_git, + write_preflight_report, +) + + +# 验证未跟踪客服文件会被记录，且采集结果不含环境变量名称或值。 +def test_collect_workspace_state_records_untracked_paths_without_environment_values( + tmp_path: Path, +) -> None: + # 构造确定性的 Git 命令替身，不调用真实 Git 或环境变量。 + def fake_git_runner(_worktree: Path, *args: str) -> str: + # 为分支查询返回客服功能分支名称。 + if args == ("branch", "--show-current"): + return "feature/customer-service-rag\n" + # 为提交查询返回固定的非敏感提交标识。 + if args == ("rev-parse", "HEAD"): + return "abc123\n" + # 为状态查询返回一个未跟踪客服文件。 + if args == ("status", "--short"): + return "?? app/service/agent/customer_service_agent.py\n" + # 防止测试静默接受未定义的 Git 查询。 + raise AssertionError(f"unexpected git arguments: {args}") + + # 使用替身采集临时工作区的只读状态。 + state = collect_workspace_state(tmp_path, runner=fake_git_runner) + + # 断言采集到预期的分支名称。 + assert state["branch"] == "feature/customer-service-rag" + # 断言未跟踪客服文件完整保留在状态清单中。 + assert state["status"] == ["?? app/service/agent/customer_service_agent.py"] + # 断言序列化结果不包含任何环境变量敏感字段。 + assert "MYSQL_PASSWORD" not in json.dumps(state) + + +# 验证报告写入器只输出传入的非敏感 Git 状态结构。 +def test_write_preflight_report_persists_utf8_json(tmp_path: Path) -> None: + # 指定临时报告文件，避免写入任何真实工作目录。 + report_path = tmp_path / "preflight.json" + # 构造只含安全 Git 元数据的状态条目。 + states = [{"path": "D:/workspace", "branch": "develop", "head": "abc123", "status": []}] + + # 写入迁移前报告。 + write_preflight_report(report_path, states) + + # 以 UTF-8 读取并解析报告正文。 + report = json.loads(report_path.read_text(encoding="utf-8")) + # 断言报告保留原始状态条目。 + assert report == states + + +# 验证每次 Git 查询只信任当前显式工作区，而不写入全局 Git 配置。 +def test_run_git_scopes_safe_directory_to_the_requested_worktree( + tmp_path: Path, + monkeypatch: MonkeyPatch, +) -> None: + # 保存被测函数交给子进程层的命令参数。 + captured_commands: list[list[str]] = [] + + # 构造返回固定分支名的无副作用子进程替身。 + def fake_run(command: list[str], **_kwargs: object) -> CompletedProcess[str]: + # 记录命令以便后续断言安全目录范围。 + captured_commands.append(command) + # 返回模拟的 Git 成功结果。 + return CompletedProcess(command, 0, "develop\n", "") + + # 让测试不执行真实 Git。 + monkeypatch.setattr("tools.foundation_migration_preflight.subprocess.run", fake_run) + + # 执行一个受限 Git 分支查询。 + output = run_git(tmp_path, "branch", "--show-current") + + # 断言调用仍返回 Git 输出。 + assert output == "develop\n" + # 断言命令仅为该临时工作区附加安全目录。 + assert captured_commands == [[ + "git", + "-c", + f"safe.directory={tmp_path.resolve().as_posix()}", + "-C", + str(tmp_path.resolve()), + "branch", + "--show-current", + ]] diff --git a/tools/foundation_migration_preflight.py b/tools/foundation_migration_preflight.py new file mode 100644 index 0000000..10aa99a --- /dev/null +++ b/tools/foundation_migration_preflight.py @@ -0,0 +1,117 @@ +"""为无损底座迁移保存并校验工作区的只读 Git 状态证据。""" + +# 导入命令行参数解析器以支持生成和校验报告。 +import argparse +# 导入 JSON 序列化工具以写入 UTF-8 状态证据。 +import json +# 导入子进程工具以直接调用 Git 而不经过 shell。 +import subprocess +# 导入可调用协议和类型别名支持。 +from collections.abc import Callable +# 导入路径类型以约束工作区和报告位置。 +from pathlib import Path +# 导入任意 JSON 对象的静态类型。 +from typing import Any + +# 定义只允许执行的 Git 只读子命令首参数集合。 +READ_ONLY_GIT_COMMANDS = frozenset({"branch", "rev-parse", "status"}) +# 定义便于测试注入的 Git 调用函数类型。 +GitRunner = Callable[..., str] + + +# 执行受限的 Git 只读命令并返回标准输出文本。 +def run_git(worktree: Path, *args: str) -> str: + # 解析目标工作区以确保 Git 和安全目录使用同一个绝对路径。 + resolved = worktree.resolve() + # 拒绝空命令，避免形成未约束的 Git 调用。 + if not args: + raise ValueError("git command is required") + # 拒绝任何不在白名单中的 Git 子命令。 + if args[0] not in READ_ONLY_GIT_COMMANDS: + raise ValueError("git command is not read-only") + # 将安全目录限定为当前查询工作区，避免写入全局 Git 配置。 + safe_directory = resolved.as_posix() + # 以参数数组执行 Git，禁止 shell 解释路径或输入内容。 + result = subprocess.run( + ["git", "-c", f"safe.directory={safe_directory}", "-C", str(resolved), *args], + check=True, + capture_output=True, + encoding="utf-8", + errors="replace", + ) + # 返回 Git 的标准输出，供调用方以确定性方式解析。 + return result.stdout + + +# 收集一个工作区的分支、提交与简短状态，不读取环境变量或业务数据。 +def collect_workspace_state(worktree: Path, runner: GitRunner = run_git) -> dict[str, object]: + # 解析绝对路径，防止报告中出现随当前目录变化的相对路径。 + resolved = worktree.resolve() + # 依次执行已白名单化的 Git 查询并构造安全状态字典。 + return { + "path": str(resolved), + "branch": runner(resolved, "branch", "--show-current").strip(), + "head": runner(resolved, "rev-parse", "HEAD").strip(), + "status": runner(resolved, "status", "--short").splitlines(), + } + + +# 将已收集的状态以 UTF-8 JSON 写入调用者指定的报告文件。 +def write_preflight_report(target: Path, states: list[dict[str, object]]) -> None: + # 确保报告父目录存在，但不创建或改动任何工作区内容。 + target.parent.mkdir(parents=True, exist_ok=True) + # 使用稳定缩进和 UTF-8 编码写入仅由调用方提供的状态数据。 + target.write_text(json.dumps(states, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + +# 从磁盘读取先前报告并验证指定工作区状态完全一致。 +def verify_preflight_report(target: Path, worktrees: list[Path]) -> list[str]: + # 以 UTF-8 读取 JSON 报告，避免系统默认编码影响比较。 + expected_value: Any = json.loads(target.read_text(encoding="utf-8")) + # 拒绝非列表报告，防止错误文件被误当作迁移证据。 + if not isinstance(expected_value, list): + raise ValueError("preflight report must be a list") + # 为每个当前工作区重新采集只读 Git 状态。 + current = [collect_workspace_state(worktree) for worktree in worktrees] + # 返回 JSON 表示不同的工作区路径，空列表代表完全一致。 + return [ + str(item["path"]) + for item, expected in zip(current, expected_value, strict=True) + if item != expected + ] + + +# 解析 CLI 工作区参数并执行报告生成或校验。 +def main() -> int: + # 创建命令行解析器并限定所有输入为显式路径。 + parser = argparse.ArgumentParser(description=__doc__) + # 要求调用者指定报告文件路径。 + parser.add_argument("--report", type=Path, required=True) + # 允许多次提供要采集或校验的工作区路径。 + parser.add_argument("--worktree", type=Path, action="append", required=True) + # 启用校验模式时不覆盖报告。 + parser.add_argument("--verify", action="store_true") + # 解析用户传入的参数。 + arguments = parser.parse_args() + # 在校验模式下输出差异并返回非零状态。 + if arguments.verify: + # 比较当前状态和报告状态。 + differences = verify_preflight_report(arguments.report, arguments.worktree) + # 输出机器和人工都可识别的校验结果。 + print("UNCHANGED" if not differences else f"CHANGED: {', '.join(differences)}") + # 有任何差异时返回失败状态。 + return 0 if not differences else 1 + # 收集调用方明确列出的工作区状态。 + states = [collect_workspace_state(worktree) for worktree in arguments.worktree] + # 写入新的迁移前证据报告。 + write_preflight_report(arguments.report, states) + # 输出报告保存位置，避免输出任何敏感运行配置。 + print(f"WROTE: {arguments.report}") + # 报告创建成功时返回零状态。 + return 0 + + +# 仅在脚本直接执行时运行命令行入口。 +if __name__ == "__main__": + # 用 main 的返回值作为进程退出码。 + raise SystemExit(main())