feat(backend): resolve temporary QC target cell product vs external code

Extract pure mutual-exclusion helper to avoid circular imports in unit tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-08-20 10:38:56 +08:00
1 parent 83d417b6e6
commit 562f35a839
4 files changed
+68

No files matched your search

+1
View File
@@ -23,6 +23,7 @@ from app.crud.qc_inbound_lab_storage_sort import (
from app.crud.collection_protocol import urls_to_attachment_list
from app.utils.sample_special import apply_special_to_item_from_samples
from app.crud import sample_reception as sample_reception_svc
from app.crud.temp_qc_target import resolve_temp_qc_target
from app.crud.sample_reception import _donation_sample_id_col
from app.crud import sample_reception_check_result as check_result_svc
from app.crud.cell_product import _enrich_product_dicts as _cp_enrich, _row_to_dict as _cp_row_to_dict, get_cell_product
+22
View File
@@ -0,0 +1,22 @@
"""临时质检对象:系统产品 / 外部编号互斥归一化。"""
from typing import Optional, Tuple
def resolve_temp_qc_target(
*,
cell_product_id: Optional[str] = None,
external_target_code: Optional[str] = None,
) -> Tuple[Optional[str], Optional[str]]:
"""临时质检对象互斥归一化。返回 (cell_product_id|None, external_target_code|None)。"""
cp = (cell_product_id or "").strip() or None
ext = (external_target_code or "").strip() or None
if cp and ext:
raise ValueError("质检对象不能同时选择系统产品并填写外部编号")
if not cp and not ext:
raise ValueError("请选择质检对象或填写外部编号")
if ext and len(ext) > 64:
raise ValueError("外部编号最长 64 个字符")
if ext:
return None, ext
return cp, None
+1
View File
@@ -1685,6 +1685,7 @@ class BizQcWorkOrder(Base):
infusion_schedule_id = Column(String(64), nullable=True, comment="中试输注计划ID(出库质检关联)")
biz_type = Column(String(20), nullable=True, comment="业务类型: pilot_service=中试服务, clinical_application=临床应用")
cell_product_id = Column(String(64), nullable=True, comment="质检对象-细胞产品ID")
external_target_code = Column(String(64), nullable=True, comment="外部质检对象编号(手写)")
production_work_order_id = Column(String(64), nullable=True, comment="关联生产工单ID(生产过程触发时)")
production_task_id = Column(String(64), nullable=True, comment="关联生产任务ID(可选)")
production_step_output_id = Column(String(64), nullable=True, comment="关联生产任务步骤产出物ID(生产过程触发时)")
@@ -0,0 +1,44 @@
"""临时质检对象:系统产品 / 外部编号互斥归一化。"""
import unittest
from app.crud.temp_qc_target import resolve_temp_qc_target
class TestResolveTempQcTarget(unittest.TestCase):
def test_cell_product_only(self):
cp, ext = resolve_temp_qc_target(
cell_product_id=" pid-1 ", external_target_code=" "
)
self.assertEqual(cp, "pid-1")
self.assertIsNone(ext)
def test_external_only(self):
cp, ext = resolve_temp_qc_target(
cell_product_id="", external_target_code=" EXT-1 "
)
self.assertIsNone(cp)
self.assertEqual(ext, "EXT-1")
def test_both_raises(self):
with self.assertRaises(ValueError) as ctx:
resolve_temp_qc_target(
cell_product_id="pid", external_target_code="EXT"
)
self.assertIn("不能同时", str(ctx.exception))
def test_neither_raises(self):
with self.assertRaises(ValueError) as ctx:
resolve_temp_qc_target(cell_product_id="", external_target_code="")
self.assertIn("质检对象", str(ctx.exception))
def test_external_too_long(self):
with self.assertRaises(ValueError) as ctx:
resolve_temp_qc_target(
cell_product_id="", external_target_code="x" * 65
)
self.assertIn("64", str(ctx.exception))
if __name__ == "__main__":
unittest.main()