test: add production task set item helper coverage

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/crud/production_task_set.py

@@ -0,0 +1,58 @@
+"""生产任务集合 CRUD"""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+
+def _strip_or_none(raw: Any) -> Optional[str]:
+    text = str(raw or "").strip()
+    return text or None
+
+
+def normalize_item_payload(raw: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+    data = raw or {}
+    task_name = str(data.get("taskName") or data.get("task_name") or "").strip()
+    if not task_name:
+        return None
+    route_raw = data.get("routeId") if "routeId" in data else data.get("route_id")
+    return {
+        "task_name": task_name,
+        "route_id": _strip_or_none(route_raw),
+        "description": _strip_or_none(data.get("description")),
+        "remark": _strip_or_none(data.get("remark")),
+    }
+
+
+def skip_empty_named_tasks(tasks: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], int]:
+    kept: List[Dict[str, Any]] = []
+    skipped = 0
+    for row in tasks or []:
+        name = str(row.get("task_name") or row.get("taskName") or "").strip()
+        if not name:
+            skipped += 1
+            continue
+        route_raw = row.get("route_id") if "route_id" in row else row.get("routeId")
+        name_raw = row.get("route_name") if "route_name" in row else row.get("routeName")
+        kept.append(
+            {
+                "task_name": name,
+                "route_id": _strip_or_none(route_raw),
+                "route_name": _strip_or_none(name_raw),
+                "description": _strip_or_none(row.get("description")),
+                "remark": _strip_or_none(row.get("remark")),
+            }
+        )
+    return kept, skipped
+
+
+def collect_missing_route_ids(items: List[Dict[str, Any]], existing_ids: Set[str]) -> List[str]:
+    missing: List[str] = []
+    seen: Set[str] = set()
+    alive = existing_ids or set()
+    for item in items or []:
+        rid = _strip_or_none(item.get("route_id"))
+        if not rid or rid in alive or rid in seen:
+            continue
+        seen.add(rid)
+        missing.append(rid)
+    return missing

backend/tests/test_production_task_set_helpers.py

@@ -0,0 +1,76 @@
+import unittest
+
+from app.crud.production_task_set import (
+    collect_missing_route_ids,
+    normalize_item_payload,
+    skip_empty_named_tasks,
+)
+
+
+class TestNormalizeItemPayload(unittest.TestCase):
+    def test_requires_task_name(self):
+        self.assertIsNone(normalize_item_payload({"taskName": "  ", "routeId": "r1"}))
+
+    def test_keeps_optional_fields(self):
+        item = normalize_item_payload(
+            {
+                "taskName": "分离",
+                "routeId": " r1 ",
+                "description": " d ",
+                "remark": " m ",
+            }
+        )
+        self.assertEqual(
+            item,
+            {
+                "task_name": "分离",
+                "route_id": "r1",
+                "description": "d",
+                "remark": "m",
+            },
+        )
+
+    def test_empty_route_becomes_none(self):
+        item = normalize_item_payload({"taskName": "培养", "routeId": ""})
+        self.assertEqual(item["route_id"], None)
+        self.assertEqual(item["description"], None)
+        self.assertEqual(item["remark"], None)
+
+
+class TestSkipEmptyNamedTasks(unittest.TestCase):
+    def test_skips_blank_names_and_counts(self):
+        kept, skipped = skip_empty_named_tasks(
+            [
+                {"task_name": "A", "route_id": "r1", "description": None, "remark": "x"},
+                {"task_name": "  ", "route_id": "r2"},
+                {"task_name": None, "route_id": None},
+            ]
+        )
+        self.assertEqual(skipped, 2)
+        self.assertEqual(len(kept), 1)
+        self.assertEqual(kept[0]["task_name"], "A")
+        self.assertEqual(kept[0]["remark"], "x")
+
+
+class TestCollectMissingRouteIds(unittest.TestCase):
+    def test_empty_route_not_missing(self):
+        self.assertEqual(
+            collect_missing_route_ids(
+                [{"route_id": None}, {"route_id": ""}],
+                {"alive"},
+            ),
+            [],
+        )
+
+    def test_lists_missing_preserving_order_unique(self):
+        self.assertEqual(
+            collect_missing_route_ids(
+                [{"route_id": "gone"}, {"route_id": "alive"}, {"route_id": "gone"}],
+                {"alive"},
+            ),
+            ["gone"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()