docs: 为全部13个源码模块和5个测试文件添加简体中文注释

- 每个文件补充模块级 docstring、类/函数 docstring、关键逻辑行内注释
- 注释风格:"中文说明" 或 # 中文说明
- 代码结构和逻辑完全不变
This commit is contained in:
2026-07-28 09:39:55 +08:00
parent b7a6f42189
commit 4ae33cbe53
18 changed files with 1528 additions and 76 deletions
+30 -3
View File
@@ -1,3 +1,9 @@
"""测试独立批量构建(Individual Batch Build)的任务调度与工作器计数。
验证内容:
- 工作器数量根据任务数和 CPU 核心数合理计算
- 串行执行器保持成功任务顺序,同时收集失败任务信息
"""
from __future__ import annotations
import unittest
@@ -9,16 +15,23 @@ from a1_swap_mod_packer.models import BuildOptions, BuildResult, PlateJob
class IndividualBatchBuildTest(unittest.TestCase):
"""测试独立批量构建的工作器计数和运行器行为。"""
def test_worker_count_uses_task_count_and_caps_cpu_count(self) -> None:
self.assertEqual(batch.individual_batch_worker_count(0), 0)
self.assertEqual(batch.individual_batch_worker_count(3, max_workers=99), 3)
self.assertEqual(batch.individual_batch_worker_count(3, max_workers=0), 1)
"""验证工作器数量 = min(任务数, CPU 核心数),且受上下限约束。"""
self.assertEqual(batch.individual_batch_worker_count(0), 0) # 无任务 → 0 个工作器
self.assertEqual(batch.individual_batch_worker_count(3, max_workers=99), 3) # 任务数 < 上限 → 按任务数
self.assertEqual(batch.individual_batch_worker_count(3, max_workers=0), 1) # 上限为 0 → 退化为 1
with patch.object(batch.os, "cpu_count", return_value=32):
# 任务数 20CPU 32 核,上限 = INDIVIDUAL_BATCH_MAX_WORKERS
self.assertEqual(batch.individual_batch_worker_count(20), batch.INDIVIDUAL_BATCH_MAX_WORKERS)
with patch.object(batch.os, "cpu_count", return_value=None):
# cpu_count 返回 None → 回退为 1
self.assertEqual(batch.individual_batch_worker_count(20), 1)
def test_serial_runner_keeps_success_order_and_collects_failures(self) -> None:
"""验证串行执行器按提交顺序返回成功结果,并收集失败任务。"""
# 准备 3 个任务,其中第 2 个计划失败
tasks = [
self.task("one.3mf", "one.out.3mf", copies=2),
self.task("fail.3mf", "fail.out.3mf", copies=3),
@@ -26,6 +39,7 @@ class IndividualBatchBuildTest(unittest.TestCase):
]
def fake_build(jobs: list[PlateJob], options: BuildOptions) -> BuildResult:
"""模拟构建:fail.3mf 触发 ValueError,其余正常返回。"""
job = jobs[0]
if job.source_3mf.name == "fail.3mf":
raise ValueError("planned failure")
@@ -37,15 +51,18 @@ class IndividualBatchBuildTest(unittest.TestCase):
gcode_md5=job.source_3mf.stem,
)
# 使用单工作器串行运行,验证顺序与失败收集
with patch.object(batch, "build_packed_3mf", side_effect=fake_build):
result = batch.run_individual_batch_builds(tasks, max_workers=1)
self.assertEqual(result.worker_count, 1)
# 成功结果按原始顺序排列,跳过失败项
self.assertEqual(
[item.output_3mf.name for item in result.results],
["one.out.3mf", "two.out.3mf"],
)
self.assertEqual([item.plate_count for item in result.results], [2, 4])
# 失败详情:索引、任务源文件名、错误信息
self.assertEqual(len(result.failures), 1)
self.assertEqual(result.failures[0].index, 1)
self.assertEqual(result.failures[0].job.source_3mf.name, "fail.3mf")
@@ -53,6 +70,16 @@ class IndividualBatchBuildTest(unittest.TestCase):
@staticmethod
def task(source_name: str, output_name: str, copies: int) -> batch.IndividualBuildTask:
"""构造独立构建任务辅助方法。
Args:
source_name: 源 3MF 文件名
output_name: 输出 3MF 文件名
copies: 打印板份数
Returns:
包装好的 IndividualBuildTask
"""
return batch.IndividualBuildTask(
PlateJob(Path(source_name), copies),
BuildOptions(swap_gcode=Path("swap.gcode"), output_3mf=Path(output_name)),