Aptos 测试框架（testsuite/test_framework）源码解读：Python 集成测试与 E2E 测试工具库实战指南-创锋一号

Aptos 测试框架（testsuite/test_framework）源码解读：Python 集成测试与 E2E 测试工具库实战指南

【免费下载链接】aptos-coreAptos is a layer 1 blockchain built to support the widespread use of blockchain through better technology and user experience.项目地址: https://gitcode.com/GitHub_Trending/ap/aptos-core

导读

Aptos 仓库中大部分核心组件的单元测试都用 Rust 编写并内嵌在源码旁，但集成测试（integration）与端到端测试（e2e）还需要一套独立的、可复用的 Python 基础设施。testsuite/test_framework正是为此而生的 Python 测试工具库：它通过Shell、Filesystem、Git、Kubernetes、Time、Process、HttpClient等薄封装抽象，把"本地执行命令、读写文件、操作 Git、管理 Kubernetes 集群、发起 HTTP 请求"等真实环境依赖统一收敛为可替换的接口，并配套Fake*与Spy*测试替身（Test Double），让集成测试在无真实集群、无网络、无文件系统的环境下也能确定性运行。读完本文，你将掌握该工具库的模块划分、Fake/Spy 设计模式、每个抽象的核心 API，以及它们如何在forge.py、pangu.py等测试编排脚本中被实际使用。

一、定位：为什么 Python 测试工具独立于 Rust 单元测试

README.md 用两句话明确了该库的定位：

Library for common python test framework utilities.

As opposed to unit tests of core Rust components, which should be written in Rust along with the code, these python test utilities are for integration and e2e testing.

即：

Rust 核心组件的单元测试：随代码一起用 Rust 编写（例如各 crate 内的#[test]），负责验证单一模块的逻辑正确性；
Python 测试工具：服务于集成测试与端到端测试，负责跨组件、跨进程、跨机器（甚至跨云集群）场景下的编排与验证。

由此可以理解该目录的文件组织：testsuite/test_framework/下是一组与具体业务解耦的通用工具模块（shell.py、filesystem.py、git.py、kubernetes.py、time.py、process.py、reqwest.py、logging.py、cluster.py），而testsuite/根目录下的forge.py、pangu.py、exp、lint.py、find_latest_image.py、indexer_grpc_local.py等脚本则通过from test_framework.xxx import ...复用这些工具（详见下文"七、仓库中的实际使用"）。

二、总体设计：三层抽象与 Fake/Spy 测试替身

整个库遵循一个非常一致的抽象模式，可归纳为三层：

抽象接口层：定义操作的最小契约（基类，方法默认raise NotImplementedError()）；
真实实现层：命名上使用System*/Local*/Live*前缀，真正调用操作系统、云 CLI、Kubernetes API 等外部依赖；
测试替身层：命名上使用Fake*/Spy*前缀，Fake提供确定性返回值，Spy在Fake基础上记录调用历史（写入列表、命令列表等），供测试结束后断言。

这种设计的价值在于：集成测试的用例代码只依赖抽象层，因此可以针对真实实现跑"真正的端到端"，也可以针对Spy跑"快速、确定、无副作用"的模拟，二者共用同一套被测逻辑。

下表汇总了各模块的三层对应关系（依据各模块源码）：

模块文件	抽象层	真实实现	测试替身
shell.py	`Shell`	`LocalShell`	`FakeShell`/`SpyShell`/`FakeCommand`
filesystem.py	`Filesystem`	`LocalFilesystem`	`FakeFilesystem`/`SpyFilesystem`
kubernetes.py	`Kubernetes`（ABC）	`LiveKubernetes`	`SpyKubernetes`
time.py	`Time`	`SystemTime`	`FakeTime`
process.py	`Processes`/`Process`	`SystemProcesses`/`SystemProcess`	`FakeProcesses`/`FakeProcess`/`SpyProcesses`
reqwest.py	`HttpClient`	`SimpleHttpClient`	—（可直接替换返回值）
git.py	`Git`（组合`Shell`）	—	—（通过注入`SpyShell`模拟）
cluster.py	`ForgeCluster`/`ForgeJob`（数据类）	`list_eks_clusters`/`list_gke_clusters`/`find_forge_cluster`	—
logging.py	全局`log`与`init_logging()`	—	—

三、shell.py：命令执行的统一封装与 SpyShell 命令断言

shell.py 是所有工具中使用最广泛的基础模块，负责把"执行一条命令并收集输出"抽象成可替换的接口。

3.1 RunResult：命令结果的数据类

@dataclass class RunResult: exit_code: int output: bytes def output_str(self) -> str: return self.output.decode("utf-8") def unwrap(self) -> bytes: if not self.succeeded(): raise Exception(self.output_str()) return self.output def succeeded(self) -> bool: return self.exit_code == 0

exit_code为进程退出码，0表示成功；
output以bytes形式保存 stdout 与 stderr 的合并输出（stderr=subprocess.STDOUT），由调用方按需.decode()；
unwrap()仿照 RustResult::unwrap语义：失败时抛出包含输出内容的异常，成功时返回原始字节。

3.2 Shell 抽象与 LocalShell 真实实现

class Shell: def run(self, command: Sequence[str], stream_output: bool = False, timeout_secs: Optional[float] = None) -> RunResult: raise NotImplementedError() async def gen_run(self, command: Sequence[str], stream_output: bool = False) -> RunResult: raise NotImplementedError()

Shell提供同步run与异步gen_run两个入口（gen_run内部基于asyncio.create_subprocess_exec实现，见 shell.py）。LocalShell是真实实现，其同步版本逐行读取子进程输出，支持：

stream_output=True时把输出实时转发到sys.stdout.buffer；
timeout_secs超时后process.kill()并抛出subprocess.TimeoutExpired；
命令以参数列表（Sequence[str]）传入，避免 shell 注入与转义问题。

3.3 FakeShell / SpyShell / FakeCommand：无需真实进程的确定性模拟

class FakeShell(Shell): def run(self, command, stream_output=False) -> RunResult: return RunResult(0, b"output")

FakeShell无脑返回成功与固定输出；SpyShell则更进一步，它要求调用方预先声明"期望被执行的命令序列"（expected_command_list），然后在每次run时：

记录实际执行的命令（self.commands）；
按命令出现次数从期望列表中匹配对应的FakeCommand（支持同一命令重复出现并按顺序返回不同结果，见 shell_test.py 中的用例）；
命中失败或命令未声明时抛出明确异常；
assert_commands(testcase)在测试结束时断言"实际执行序列 == 期望执行序列"。

class FakeCommand: def __init__(self, command: str, result_or_exception: Union[RunResult, Exception]): self.command = command self.result_or_exception = result_or_exception

FakeCommand甚至可以携带一个Exception，用于模拟命令执行失败的分支。这一设计使得上层逻辑（如 Git 操作、集群管理）的测试可以精确验证"命令以正确的顺序、正确的参数被执行"。

四、filesystem.py：文件系统操作抽象与读写断言

filesystem.py 将文件/目录操作抽象为Filesystem接口，方法包括write、read、mkstemp、mkdtemp、mkdir、rmtree、copyfile、rlimit、unlink、exists。

LocalFilesystem直接映射到open/os.mkdir/shutil.rmtree/resource.setrlimit等真实调用；
FakeFilesystem所有操作均为空操作或固定返回值；
SpyFilesystem是关键测试替身，它维护writes（文件名→内容）与reads（读取历史）字典，并提供断言方法：
- assert_writes(testcase)：断言期望写入的文件确实被写入且内容逐字节一致（assertMultiLineEqual）；
- assert_reads(testcase)：断言期望读取的文件确实被读过，且所有读过的文件都在期望集合内；
- assert_unlinks(testcase)：断言期望删除的文件确实被unlink；
- 模块还定义了特殊字节串FILE_NOT_FOUND = b"FILE_NOT_FOUND"，用于表示"该文件不应存在"的语义（见 filesystem.py）。

此外SpyFilesystem.mkstemp/mkdtemp会生成可预测的递增名称（temp1、temp_folder1…），保证测试中临时文件路径的确定性。

五、git.py 与 cluster.py：Git 与云集群的 Forge 抽象

5.1 Git：以 Shell 为底层组合出来的高层操作

git.py 不直接执行git，而是组合注入的Shell实例：

@dataclass class Git: shell: Shell def run(self, command) -> RunResult: return self.shell.run(["git", *command])

其提供的高层方法包括：

方法	底层命令	用途
`last(limit)`	`git rev-parse HEAD~i`	取最近 N 个 commit 哈希
`branch()`	`git rev-parse --abbrev-ref HEAD`	当前分支名
`branch_exists(branch)`	`git rev-parse --verify [origin/]branch`	本地/远程分支是否存在
`status()`	`git status --porcelain`	工作区是否干净
`branch_matches_remote(remote, ref)`	`git ls-remote --heads`+`git rev-parse`	本地分支是否与远端一致
`get_remote_branches_matching_pattern(remote, pattern, regex)`	`git ls-remote --heads`	按正则筛选远端分支（如`aptos-release-v*`）
`get_commit_hashes(branch, max_commits)`	`git log -n --format=%H`	获取分支提交哈希列表
`get_branch_creation_time(branch)`	`git rev-list --first-parent --max-count=1`+`git show -s --format=%ci`	计算分支创建时间
`get_repo_from_remote(remote_name)`	`git remote get-url`	从 remote URL 解析`org/repo`

由于Git依赖注入的Shell，测试时传入SpyShell即可在不触碰真实 Git 仓库的情况下验证命令序列。

5.2 cluster.py：Forge 测试集群的抽象

cluster.py 服务于 Forge 测试框架（Aptos 的混沌/负载测试框架），核心概念：

Cloud枚举：AWS/GCP；
ForgeCluster数据类：字段name、cloud（默认AWS）、region（默认"us-west-2"）、kubeconf、is_multiregion。其__repr__输出形如AWS/us-west-2/<cluster>；
ForgeJob数据类：字段name、phase、cluster、num_validators、num_fullnodes，并提供running()/succeeded()/failed()三个基于 Podphase的谓词（对应 Kubernetes Pod 的Running/Succeeded/Failed）。

ForgeCluster的典型流程（write→get_jobs）：

write(shell)把集群的 kubeconfig 写入临时文件。根据云类型选择不同命令：
- AWS 多区域：gcloud secrets versions access latest --secret karmada-kubeconfig --project forge-gcp-multiregion-test（多区域走 Karmada）；
- AWS 单区域：aws eks update-kubeconfig --name <cluster> --kubeconfig <temp>；
- GCP：gcloud container clusters get-credentials <cluster> --zone <region>；
get_jobs(shell)通过kubectl get pods -n default -o json --kubeconfig <conf>枚举以forge-开头且带有forge-namespace标签的测试运行 Pod，再进入对应命名空间统计validator/fullnodePod 数量，组装出ForgeJob列表——这正是 Forge 汇总"每个测试任务跑了几台验证器/全节点"的实现基础；
assert_auth(shell)在 AWS 上调用aws eks list-clusters、在 GCP 上调用gcloud container clusters list --format=json(name, location)，用于校验云凭据有效。

顶层函数list_eks_clusters/list_gke_clusters只返回名称以aptos-forge-开头的集群（GCP 版本内置 10 次重试与 10 秒退避，见 cluster.py）；find_forge_cluster(shell, cloud, name, kubeconf)则按名称查找并回填 kubeconf。

六、其余工具模块：time / process / reqwest / logging

6.1 time.py：可冻结的时间

time.py 提供epoch()与now()。SystemTime.now()返回datetime.now(timezone.utc)；FakeTime使用固定时间戳_now = 1659078000，用于让依赖当前时间的测试完全确定（例如计算版本号、判断超时）。

6.2 process.py：进程枚举与退出钩子

process.py 抽象了Process（name()/ppid()）与Processes（processes()生成器、get_pid()、atexit(callback)、user()）：

SystemProcesses基于psutil.process_iter()枚举系统进程，atexit注册到atexit.register；
FakeProcesses固定返回FakeProcess("concensus", 1)等假进程；
SpyProcesses.run_atexit()可手动触发所有注册的退出回调，便于测试"进程退出时的清理逻辑"。

6.3 reqwest.py：极简 HTTP 客户端

reqwest.py 仿照 Rust 的reqwest命名，提供HttpClient.get(url, headers)；SimpleHttpClient直接委托给requests.get。测试中替换为返回固定Response的桩即可。

6.4 logging.py：统一日志

logging.py 定义全局 loggerlog（logging.getLogger("")）与init_logging(logger, level=logging.INFO, print_metadata=True)。默认输出格式包含时间戳、级别、文件名.函数名:行号与消息，便于定位测试失败位置。

七、仓库中的实际使用：forge.py / pangu.py 等编排脚本

该库不是孤立模块，而是testsuite/下众多测试编排脚本的地基。在 forge.py 中可以看到典型用法：

from test_framework.shell import LocalShell, Shell（forge.py）；
from test_framework.cluster import Cloud, ForgeCluster, ForgeJob, find_forge_cluster（forge.py）；
实例化shell = LocalShell()（如 forge.py），构造ForgeCluster(name=..., ...)并通过find_forge_cluster(...)解析真实集群，再调用config.get_jobs(context.shell)汇总所有测试任务状态（forge.py）。

同样，testsuite/pangu_lib/node_commands/*.py、testsuite/pangu_lib/testnet_commands/*.py（启动/停止/重启节点、创建/删除测试网等命令）以及testsuite/exp、testsuite/lint.py、testsuite/find_latest_image.py、testsuite/indexer_grpc_local.py等脚本均以from test_framework.xxx import ...的方式复用Shell、Kubernetes、Git、Time等抽象。配套的单元测试 shell_test.py、git_test.py、kubernetes_test.py 则用SpyShell、SpyKubernetes等验证工具库自身的行为。

八、编写基于 test_framework 的测试：推荐模式

综合上述模块，一个典型的集成测试编写模式是：

注入抽象：被测代码通过构造函数/参数接收Shell、Filesystem、Git、Time等，而不是直接调用subprocess/os/gitCLI /datetime；
真实路径：需要真正执行命令或写文件时，注入LocalShell、LocalFilesystem；
模拟路径：需要确定性测试时，注入SpyShell（预先用FakeCommand声明期望命令与返回结果，结束后assert_commands）与SpyFilesystem（结束后assert_writes/assert_reads校验文件读写）；需要测试 Kubernetes 编排时注入SpyKubernetes，其内部用"命名空间 → 资源类型 → 资源名"的多层字典模拟资源生命周期（kubernetes.py）；
时间与进程：涉及时间敏感或进程清理逻辑时，注入FakeTime与SpyProcesses。

该模式在testsuite/下的测试脚本中反复出现：例如forge_test.py、exp_test.py、lint_test.py、indexer_grpc_local_test.py以及pangu_lib/tests/下各命令的*_test.py，它们大多先构造SpyShell/SpyKubernetes/SpyFilesystem，再驱动被测函数并做断言。

九、小结

testsuite/test_framework用约 9 个 Python 模块，为 Aptos 的 Python 集成/E2E 测试提供了四类核心资产：

统一的环境抽象：命令、文件系统、Git、Kubernetes、时间、进程、HTTP、日志，全部收敛为可替换接口；
真实的系统实现：LocalShell、LocalFilesystem、LiveKubernetes、SystemTime等，保证真实环境下的端到端能力；
确定性的测试替身：Fake*提供固定行为，Spy*记录调用历史并支持事后断言，让测试无需真实集群/网络/文件系统即可运行；
Forge 专属抽象：ForgeCluster/ForgeJob与list_eks_clusters/list_gke_clusters/find_forge_cluster，支撑跨 AWS/GCP 的负载与混沌测试编排。

理解这套"抽象接口 + 系统实现 + Fake/Spy 替身"的模式，不仅有助于读懂forge.py、pangu.py等测试编排脚本，也可以直接复用到你自己的集成测试工程中。如需深入，可继续阅读 shell.py 与 shell_test.py 的配对实现，以及 forge.py 对ForgeCluster.get_jobs的调用链。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析