重置项目上下文与规范文档
This commit is contained in:
Binary file not shown.
@@ -1,127 +0,0 @@
|
||||
"""批量回收设备脚本测试。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SCRIPT_PATH = Path(__file__).with_name("batch_device_recall.py")
|
||||
SPEC = importlib.util.spec_from_file_location("batch_device_recall", SCRIPT_PATH)
|
||||
assert SPEC is not None and SPEC.loader is not None
|
||||
batch_device_recall = importlib.util.module_from_spec(SPEC)
|
||||
sys.modules[SPEC.name] = batch_device_recall
|
||||
SPEC.loader.exec_module(batch_device_recall)
|
||||
|
||||
|
||||
class LoadDevicesTest(unittest.TestCase):
|
||||
"""验证单列 CSV 输入。"""
|
||||
|
||||
def write_csv(self, content: str) -> Path:
|
||||
"""创建临时 CSV。"""
|
||||
directory = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(directory.cleanup)
|
||||
path = Path(directory.name) / "devices.csv"
|
||||
path.write_text(content, encoding="utf-8")
|
||||
return path
|
||||
|
||||
def test_loads_header_and_rows(self) -> None:
|
||||
"""支持表头并保留行号。"""
|
||||
path = self.write_csv("device_identifier\nIMEI001\nVIRTUAL001\n")
|
||||
|
||||
devices = batch_device_recall.load_devices(path)
|
||||
|
||||
self.assertEqual(
|
||||
devices,
|
||||
[
|
||||
batch_device_recall.DeviceInput(2, "IMEI001"),
|
||||
batch_device_recall.DeviceInput(3, "VIRTUAL001"),
|
||||
],
|
||||
)
|
||||
|
||||
def test_rejects_duplicate_identifier(self) -> None:
|
||||
"""同一标识不能重复回收。"""
|
||||
path = self.write_csv("IMEI001\nIMEI001\n")
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "重复"):
|
||||
batch_device_recall.load_devices(path)
|
||||
|
||||
|
||||
class ExactDeviceMatchesTest(unittest.TestCase):
|
||||
"""验证模糊查询结果必须再次精确匹配。"""
|
||||
|
||||
def test_keeps_only_exact_virtual_no_or_imei(self) -> None:
|
||||
"""排除仅包含关键字的候选设备。"""
|
||||
body = {
|
||||
"data": {
|
||||
"items": [
|
||||
{"id": 1, "virtual_no": "VIRTUAL001", "imei": "IMEI001"},
|
||||
{"id": 2, "virtual_no": "VIRTUAL001-X", "imei": "IMEI002"},
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
matches = batch_device_recall.exact_device_matches(body, "VIRTUAL001")
|
||||
|
||||
self.assertEqual([item["id"] for item in matches], [1])
|
||||
|
||||
|
||||
class RecallBatchRowsTest(unittest.TestCase):
|
||||
"""验证批量接口结果映射。"""
|
||||
|
||||
def test_maps_failed_items_to_source_rows(self) -> None:
|
||||
"""接口失败明细应对应回原 CSV 标识。"""
|
||||
devices = [
|
||||
batch_device_recall.ResolvedDevice(
|
||||
batch_device_recall.DeviceInput(2, "IMEI001"), 1, "V001", "IMEI001"
|
||||
),
|
||||
batch_device_recall.ResolvedDevice(
|
||||
batch_device_recall.DeviceInput(3, "IMEI002"), 2, "V002", "IMEI002"
|
||||
),
|
||||
]
|
||||
result = batch_device_recall.HTTPResult(
|
||||
200,
|
||||
{
|
||||
"code": 0,
|
||||
"msg": "success",
|
||||
"data": {
|
||||
"success_count": 1,
|
||||
"fail_count": 1,
|
||||
"failed_items": [{"device_id": 2, "reason": "设备已在平台库存中"}],
|
||||
},
|
||||
},
|
||||
"",
|
||||
)
|
||||
|
||||
rows, success = batch_device_recall.recall_batch_rows(devices, result)
|
||||
|
||||
self.assertFalse(success)
|
||||
self.assertEqual([row["status"] for row in rows], ["成功", "失败"])
|
||||
self.assertEqual(rows[1]["msg"], "设备已在平台库存中")
|
||||
|
||||
def test_rejects_incomplete_failed_items(self) -> None:
|
||||
"""失败数量没有对应明细时不能把设备误记为成功。"""
|
||||
device = batch_device_recall.ResolvedDevice(
|
||||
batch_device_recall.DeviceInput(2, "IMEI001"), 1, "V001", "IMEI001"
|
||||
)
|
||||
result = batch_device_recall.HTTPResult(
|
||||
200,
|
||||
{
|
||||
"code": 0,
|
||||
"msg": "success",
|
||||
"data": {"success_count": 0, "fail_count": 1, "failed_items": []},
|
||||
},
|
||||
"",
|
||||
)
|
||||
|
||||
rows, success = batch_device_recall.recall_batch_rows([device], result)
|
||||
|
||||
self.assertFalse(success)
|
||||
self.assertEqual(rows[0]["status"], "失败")
|
||||
self.assertIn("统计", rows[0]["msg"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user