1. 引言在软件工程领域Harness测试执行框架 / 任务编排框架是一个很常见的概念。无论是 CI/CD 流水线、自动化测试框架还是本地开发时的任务调度背后都有一个「把任务跑起来、收集结果、汇总报告」的执行器。很多同学觉得 Harness 很神秘其实它的核心逻辑并不复杂。本文将从零开始不依赖任何重量级框架用纯 Python 手写一个个人最简 Harness 项目。我们会一步步拆解 Harness 的核心能力并用可运行的代码把它们实现出来。读完本文你将拥有一个属于自己的、可扩展的最小 Harness并能理解主流测试框架背后的设计思路。2. Harness 是什么2.1 一句话定义Harness 是一个任务执行与编排框架它负责加载一组任务Task、按规则执行它们、捕获执行过程中的输出与异常并最终生成一份可读的结果报告。2.2 核心组成一个最简 Harness 通常包含以下四个部分Task任务最小执行单元可以是测试用例、构建脚本或任意可调用对象。Runner执行器真正运行 Task 的组件负责调用、计时、捕获异常。Reporter报告器把执行结果整理成文本、JSON 或 HTML 报告。Harness编排器把上面三者串起来对外提供统一入口。2.3 设计目标我们的最简 Harness 需要满足零第三方依赖只用 Python 标准库。支持同步执行多个任务。能捕获每个任务的耗时、状态通过/失败/错误和输出。能生成一份简洁的文本报告。3. 项目结构我们先规划好项目目录结构mini_harness/ ├── harness/ │ ├── __init__.py │ ├── core.py # 核心数据结构TaskResult、Task │ ├── runner.py # 执行器 │ ├── reporter.py # 报告器 │ └── harness.py # 编排器入口 ├── examples/ │ └── demo_tasks.py # 示例任务 └── main.py # 命令行入口4. 核心数据结构首先实现harness/core.py定义任务与结果的数据结构。# harness/core.pyfromdataclassesimportdataclass,fieldfromenumimportEnumfromtypingimportAny,Callable,OptionalimporttimeclassTaskStatus(Enum):任务执行状态PENDINGpendingRUNNINGrunningPASSEDpassedFAILEDfailedERRORerrordataclassclassTaskResult:单个任务的执行结果name:strstatus:TaskStatus duration:float0.0output:strerror:Optional[str]Nonepropertydefok(self)-bool:returnself.statusTaskStatus.PASSEDdataclassclassTask:任务定义一个可调用对象 元信息name:strfn:Callable[[],Any]timeout:Optional[float]Nonedef__post_init__(self):ifnotcallable(self.fn):raiseTypeError(fTask {self.name} 的 fn 必须是可调用对象)这里用dataclass让代码简洁清晰。TaskStatus枚举区分「通过 / 断言失败 / 异常错误」三种终态为后续报告提供依据。5. 执行器 Runner接下来实现harness/runner.py负责真正运行任务并捕获结果。# harness/runner.pyimporttimeimporttracebackfromtypingimportListfrom.coreimportTask,TaskResult,TaskStatusclassRunner:最简同步执行器def__init__(self):self.results:List[TaskResult][]defrun(self,task:Task)-TaskResult:运行单个任务并记录结果resultTaskResult(nametask.name,statusTaskStatus.RUNNING)starttime.perf_counter()output_lines[]try:# 捕获任务的 print 输出importioimportcontextlib bufio.StringIO()withcontextlib.redirect_stdout(buf):task.fn()output_lines.append(buf.getvalue())result.statusTaskStatus.PASSEDexceptAssertionErrorase:# 断言失败任务逻辑跑通但校验不通过result.statusTaskStatus.FAILED result.errorstr(e)exceptExceptionase:# 其他异常任务本身出错result.statusTaskStatus.ERROR result.errorf{type(e).__name__}:{e}output_lines.append(traceback.format_exc())finally:result.durationtime.perf_counter()-start result.output\n.join(output_lines).strip()self.results.append(result)returnresultdefrun_all(self,tasks:List[Task])-List[TaskResult]:顺序执行所有任务self.results[]fortaskintasks:self.run(task)returnself.results这里用contextlib.redirect_stdout捕获任务里的print输出用traceback记录异常堆栈。AssertionError单独处理是为了区分「测试没通过」和「代码写错了」。6. 报告器 Reporter实现harness/reporter.py把结果渲染成可读文本。# harness/reporter.pyfromtypingimportListfrom.coreimportTaskResult,TaskStatusclassReporter:最简文本报告器def__init__(self,results:List[TaskResult]):self.resultsresultsdefrender(self)-str:lines[]lines.append(*50)lines.append(Mini Harness Report)lines.append(*50)forrinself.results:icon{TaskStatus.PASSED:[PASS],TaskStatus.FAILED:[FAIL],TaskStatus.ERROR:[ERROR],}.get(r.status,[????])lines.append(f{icon}{r.name}({r.duration:.3f}s))ifr.output:# 缩进展示输出forlineinr.output.splitlines():lines.append(f |{line})ifr.error:lines.append(f !{r.error})lines.append(*50)passedsum(1forrinself.resultsifr.statusTaskStatus.PASSED)totallen(self.results)lines.append(fSummary:{passed}/{total}passed)lines.append(*50)return\n.join(lines)defto_json(self)-str:输出 JSON 格式报告便于机器解析importjson data{total:len(self.results),results:[{name:r.name,status:r.status.value,duration:r.duration,output:r.output,error:r.error,}forrinself.results],}returnjson.dumps(data,ensure_asciiFalse,indent2)7. 编排器 Harness最后实现harness/harness.py把 Task、Runner、Reporter 串起来。# harness/harness.pyfromtypingimportListfrom.coreimportTaskfrom.runnerimportRunnerfrom.reporterimportReporterclassHarness:最简 Harness 编排器def__init__(self):self.tasks:List[Task][]self.runnerRunner()defadd_task(self,task:Task)-Harness:注册一个任务支持链式调用self.tasks.append(task)returnselfdefadd_tasks(self,tasks:List[Task])-Harness:self.tasks.extend(tasks)returnselfdefrun(self)-Reporter:执行所有任务并返回报告器resultsself.runner.run_all(self.tasks)returnReporter(results)8. 示例任务创建examples/demo_tasks.py写几个不同类型的任务来验证 Harness。# examples/demo_tasks.pyfromharness.coreimportTaskdeftest_addition():一个会通过的任务assert112print(1 1 2)deftest_subtraction():一个断言失败的任务assert5-31,5 - 3 应该等于 2print(这行不会执行)deftest_division():一个会抛异常的任务result10/0print(result)defbuild_demo_tasks():return[Task(nametest_addition,fntest_addition),Task(nametest_subtraction,fntest_subtraction),Task(nametest_division,fntest_division),]9. 命令行入口创建main.py让用户可以从命令行运行。# main.pyfromharness.harnessimportHarnessfromexamples.demo_tasksimportbuild_demo_tasksdefmain():harnessHarness()harness.add_tasks(build_demo_tasks())reporterharness.run()print(reporter.render())# 同时输出 JSON 报告到文件withopen(report.json,w,encodingutf-8)asf:f.write(reporter.to_json())print(\nJSON 报告已写入 report.json)if__name____main__:main()10. 运行与效果在项目根目录执行python main.py预期输出 Mini Harness Report [PASS] test_addition (0.000s) | 1 1 2 [FAIL] test_subtraction (0.000s) ! 5 - 3 应该等于 2 [ERROR] test_division (0.000s) ! ZeroDivisionError: division by zero | Traceback (most recent call last): | ... Summary: 1/3 passed JSON 报告已写入 report.json可以看到三种状态通过、断言失败、异常都被正确区分并记录。11. 扩展思路我们的最简 Harness 已经能跑但它还非常「朴素」。你可以按以下方向继续扩展11.1 支持异步并发用asyncio或concurrent.futures.ThreadPoolExecutor让任务并行执行大幅提升吞吐。# 扩展思路示例线程池并发执行fromconcurrent.futuresimportThreadPoolExecutordefrun_all_parallel(self,tasks,max_workers4):withThreadPoolExecutor(max_workersmax_workers)aspool:futures[pool.submit(self.run,t)fortintasks]return[f.result()forfinfutures]11.2 支持参数化任务让 Task 支持args/kwargs同一个函数可以跑多组数据。dataclassclassTask:name:strfn:Callable[...,Any]args:tuple()kwargs:dictfield(default_factorydict)11.3 支持跳过与标签给 Task 增加skip条件和tags实现按标签筛选执行。11.4 支持超时控制用signal或子进程实现任务超时强制终止防止死循环卡死整个 Harness。11.5 支持插件化报告把 Reporter 抽象成接口支持 JSON、HTML、JUnit XML 等多种输出格式。12. 总结本文从零开始实现了一个个人最简 Harness 项目核心代码不到 150 行却完整覆盖了「任务定义 → 执行 → 结果捕获 → 报告输出」的完整链路。通过这个项目你可以清晰地理解Harness 的本质是任务编排核心是 Task / Runner / Reporter 三个角色。用contextlib.redirect_stdout捕获输出、用traceback记录异常、用dataclass组织数据都是非常实用的 Python 技巧。区分「断言失败」和「异常错误」对测试框架至关重要。希望这个最简实现能成为你理解更复杂框架pytest、unittest、Jenkins Pipeline 等的起点。动手把它跑起来然后按自己的需求扩展它吧