企业官网建设流程全解析

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):"""任务执行状态"""PENDING="pending"RUNNING="running"PASSED="passed"FAILED="failed"ERROR="error"@dataclassclassTaskResult:"""单个任务的执行结果"""name:strstatus:TaskStatus duration:float=0.0output:str=""error:Optional[str]=None@propertydefok(self)->bool:returnself.status==TaskStatus.PASSED@dataclassclassTask:"""任务定义：一个可调用对象 + 元信息"""name:strfn:Callable[[],Any]timeout:Optional[float]=Nonedef__post_init__(self):ifnotcallable(self.fn):raiseTypeError(f"Task '{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:"""运行单个任务并记录结果"""result=TaskResult(name=task.name,status=TaskStatus.RUNNING)start=time.perf_counter()output_lines=[]try:# 捕获任务的 print 输出importioimportcontextlib buf=io.StringIO()withcontextlib.redirect_stdout(buf):task.fn()output_lines.append(buf.getvalue())result.status=TaskStatus.PASSEDexceptAssertionErrorase:# 断言失败：任务逻辑跑通但校验不通过result.status=TaskStatus.FAILED result.error=str(e)exceptExceptionase:# 其他异常：任务本身出错result.status=TaskStatus.ERROR result.error=f"{type(e).__name__}:{e}"output_lines.append(traceback.format_exc())finally:result.duration=time.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.results=resultsdefrender(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)passed=sum(1forrinself.resultsifr.status==TaskStatus.PASSED)total=len(self.results)lines.append(f"Summary:{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_ascii=False,indent=2)

7. 编排器 Harness

最后实现harness/harness.py，把 Task、Runner、Reporter 串起来。

# harness/harness.pyfromtypingimportListfrom.coreimportTaskfrom.runnerimportRunnerfrom.reporterimportReporterclassHarness:"""最简 Harness 编排器"""def__init__(self):self.tasks:List[Task]=[]self.runner=Runner()defadd_task(self,task:Task)->"Harness":"""注册一个任务，支持链式调用"""self.tasks.append(task)returnselfdefadd_tasks(self,tasks:List[Task])->"Harness":self.tasks.extend(tasks)returnselfdefrun(self)->Reporter:"""执行所有任务并返回报告器"""results=self.runner.run_all(self.tasks)returnReporter(results)

8. 示例任务

创建examples/demo_tasks.py，写几个不同类型的任务来验证 Harness。

# examples/demo_tasks.pyfromharness.coreimportTaskdeftest_addition():"""一个会通过的任务"""assert1+1==2print("1 + 1 = 2")deftest_subtraction():"""一个断言失败的任务"""assert5-3==1,"5 - 3 应该等于 2"print("这行不会执行")deftest_division():"""一个会抛异常的任务"""result=10/0print(result)defbuild_demo_tasks():return[Task(name="test_addition",fn=test_addition),Task(name="test_subtraction",fn=test_subtraction),Task(name="test_division",fn=test_division),]

9. 命令行入口

创建main.py，让用户可以从命令行运行。

# main.pyfromharness.harnessimportHarnessfromexamples.demo_tasksimportbuild_demo_tasksdefmain():harness=Harness()harness.add_tasks(build_demo_tasks())reporter=harness.run()print(reporter.render())# 同时输出 JSON 报告到文件withopen("report.json","w",encoding="utf-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_workers=4):withThreadPoolExecutor(max_workers=max_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:dict=field(default_factory=dict)

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 等）的起点。动手把它跑起来，然后按自己的需求扩展它吧！