yw1573
4ae33cbe53
- 每个文件补充模块级 docstring、类/函数 docstring、关键逻辑行内注释 - 注释风格："中文说明" 或 # 中文说明 - 代码结构和逻辑完全不变
91 lines
4.0 KiB
Python
91 lines
4.0 KiB
Python
"""测试独立批量构建（Individual Batch Build）的任务调度与工作器计数。
|
||
|
||
验证内容：
|
||
- 工作器数量根据任务数和 CPU 核心数合理计算
|
||
- 串行执行器保持成功任务顺序，同时收集失败任务信息
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
import a1_swap_mod_packer.batch as batch
|
||
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:
|
||
"""验证工作器数量 = 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):
|
||
# 任务数 20，CPU 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),
|
||
self.task("two.3mf", "two.out.3mf", copies=4),
|
||
]
|
||
|
||
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")
|
||
return BuildResult(
|
||
output_3mf=options.output_3mf,
|
||
plate_count=job.copies,
|
||
total_prediction_seconds=None,
|
||
total_weight_grams=None,
|
||
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")
|
||
self.assertIn("planned failure", result.failures[0].error)
|
||
|
||
@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)),
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|