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.pyShellLocalShellFakeShell/SpyShell/FakeCommandfilesystem.pyFilesystemLocalFilesystemFakeFilesystem/SpyFilesystemkubernetes.pyKubernetesABCLiveKubernetesSpyKubernetestime.pyTimeSystemTimeFakeTimeprocess.pyProcesses/ProcessSystemProcesses/SystemProcessFakeProcesses/FakeProcess/SpyProcessesreqwest.pyHttpClientSimpleHttpClient—可直接替换返回值git.pyGit组合Shell——通过注入SpyShell模拟cluster.pyForgeCluster/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 0exit_code为进程退出码0表示成功output以bytes形式保存 stdout 与 stderr 的合并输出stderrsubprocess.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_outputTrue时把输出实时转发到sys.stdout.buffertimeout_secs超时后process.kill()并抛出subprocess.TimeoutExpired命令以参数列表Sequence[str]传入避免 shell 注入与转义问题。3.3 FakeShell / SpyShell / FakeCommand无需真实进程的确定性模拟class FakeShell(Shell): def run(self, command, stream_outputFalse) - RunResult: return RunResult(0, boutput)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_exceptionFakeCommand甚至可以携带一个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)断言期望写入的文件确实被写入且内容逐字节一致assertMultiLineEqualassert_reads(testcase)断言期望读取的文件确实被读过且所有读过的文件都在期望集合内assert_unlinks(testcase)断言期望删除的文件确实被unlink模块还定义了特殊字节串FILE_NOT_FOUND bFILE_NOT_FOUND用于表示该文件不应存在的语义见 filesystem.py。此外SpyFilesystem.mkstemp/mkdtemp会生成可预测的递增名称temp1、temp_folder1…保证测试中临时文件路径的确定性。五、git.py 与 cluster.pyGit 与云集群的 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 --headsgit 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-count1git 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.pyForge 测试集群的抽象cluster.py 服务于 Forge 测试框架Aptos 的混沌/负载测试框架核心概念Cloud枚举AWS/GCPForgeCluster数据类字段name、cloud默认AWS、region默认us-west-2、kubeconf、is_multiregion。其__repr__输出形如AWS/us-west-2/clusterForgeJob数据类字段name、phase、cluster、num_validators、num_fullnodes并提供running()/succeeded()/failed()三个基于 Podphase的谓词对应 Kubernetes Pod 的Running/Succeeded/Failed。ForgeCluster的典型流程write→get_jobswrite(shell)把集群的 kubeconfig 写入临时文件。根据云类型选择不同命令AWS 多区域gcloud secrets versions access latest --secret karmada-kubeconfig --project forge-gcp-multiregion-test多区域走 KarmadaAWS 单区域aws eks update-kubeconfig --name cluster --kubeconfig tempGCPgcloud container clusters get-credentials cluster --zone regionget_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 --formatjson(name, location)用于校验云凭据有效。顶层函数list_eks_clusters/list_gke_clusters只返回名称以aptos-forge-开头的集群GCP 版本内置 10 次重试与 10 秒退避见 cluster.pyfind_forge_cluster(shell, cloud, name, kubeconf)则按名称查找并回填 kubeconf。六、其余工具模块time / process / reqwest / logging6.1 time.py可冻结的时间time.py 提供epoch()与now()。SystemTime.now()返回datetime.now(timezone.utc)FakeTime使用固定时间戳_now 1659078000用于让依赖当前时间的测试完全确定例如计算版本号、判断超时。6.2 process.py进程枚举与退出钩子process.py 抽象了Processname()/ppid()与Processesprocesses()生成器、get_pid()、atexit(callback)、user()SystemProcesses基于psutil.process_iter()枚举系统进程atexit注册到atexit.registerFakeProcesses固定返回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 定义全局 loggerloglogging.getLogger()与init_logging(logger, levellogging.INFO, print_metadataTrue)。默认输出格式包含时间戳、级别、文件名.函数名:行号与消息便于定位测试失败位置。七、仓库中的实际使用forge.py / pangu.py 等编排脚本该库不是孤立模块而是testsuite/下众多测试编排脚本的地基。在 forge.py 中可以看到典型用法from test_framework.shell import LocalShell, Shellforge.pyfrom test_framework.cluster import Cloud, ForgeCluster, ForgeJob, find_forge_clusterforge.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的调用链。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考