zlwuandCursor
529f36eac6
按方案 A 增加租户级 ACL（字段+人员），详情返回可编辑字段，保存改为服务端 merge；弹窗 footer 提供权限配置入口。关联 #93。 Co-authored-by: Cursor <cursoragent@cursor.com>
404 lines
17 KiB
Python
404 lines
17 KiB
Python
"""储存证质检信息：类型键、payload 与自动补空。"""
|
|
import unittest
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
|
|
from app.crud.sample_storage_cert_qc import _row_to_dict
|
|
from app.services.sample_storage_cert_qc import (
|
|
SAMPLE_TYPE_KEY_PERINATAL_HSC,
|
|
SAMPLE_TYPE_KEY_UC_MSC,
|
|
empty_detail,
|
|
merge_payload_with_acl,
|
|
normalize_payload,
|
|
resolve_sample_type_key,
|
|
sanitize_payload_for_type,
|
|
supports_storage_cert_qc,
|
|
)
|
|
from app.services.sample_storage_cert_qc_prefill import (
|
|
batch_kind_from_name,
|
|
build_suggested,
|
|
match_cd34_param,
|
|
merge_prefill,
|
|
qual_yin_yang,
|
|
sum_cryo_volume,
|
|
)
|
|
|
|
|
|
class TestResolveSampleTypeKey(unittest.TestCase):
|
|
def test_hsc(self):
|
|
self.assertEqual(resolve_sample_type_key("围产期造血干细胞"), SAMPLE_TYPE_KEY_PERINATAL_HSC)
|
|
|
|
def test_msc(self):
|
|
self.assertEqual(resolve_sample_type_key("脐带间充质干细胞"), SAMPLE_TYPE_KEY_UC_MSC)
|
|
|
|
def test_other(self):
|
|
self.assertIsNone(resolve_sample_type_key("胎盘亚全能干细胞"))
|
|
self.assertIsNone(resolve_sample_type_key(""))
|
|
|
|
def test_supports(self):
|
|
self.assertTrue(supports_storage_cert_qc("围产期造血干细胞"))
|
|
self.assertFalse(supports_storage_cert_qc("其他"))
|
|
|
|
|
|
class TestNormalizePayload(unittest.TestCase):
|
|
def test_rejects_non_object(self):
|
|
with self.assertRaises(ValueError):
|
|
normalize_payload([])
|
|
with self.assertRaises(ValueError):
|
|
normalize_payload("x")
|
|
|
|
def test_keeps_dict_stringifies_keys(self):
|
|
out = normalize_payload({"viability_pct": 90.4, 1: "bad"})
|
|
self.assertEqual(out["viability_pct"], 90.4)
|
|
self.assertNotIn(1, out)
|
|
|
|
|
|
class TestSanitizePayloadForType(unittest.TestCase):
|
|
def test_uc_msc_drops_mother_nucleic_acids(self):
|
|
out = sanitize_payload_for_type(
|
|
{
|
|
"mother_hbsag": "阴性",
|
|
"mother_hbv_dna": "阴性",
|
|
"mother_hcv_rna": "阳性",
|
|
"mother_hiv1_rna": "阴性",
|
|
"mother_cmv_igm": "阴性",
|
|
},
|
|
SAMPLE_TYPE_KEY_UC_MSC,
|
|
)
|
|
self.assertEqual(out["mother_hbsag"], "阴性")
|
|
self.assertEqual(out["mother_cmv_igm"], "阴性")
|
|
self.assertNotIn("mother_hbv_dna", out)
|
|
self.assertNotIn("mother_hcv_rna", out)
|
|
self.assertNotIn("mother_hiv1_rna", out)
|
|
|
|
def test_perinatal_keeps_mother_nucleic_acids(self):
|
|
payload = {
|
|
"mother_hbv_dna": "阴性",
|
|
"mother_hcv_rna": "阴性",
|
|
"mother_hiv1_rna": "阴性",
|
|
}
|
|
out = sanitize_payload_for_type(payload, SAMPLE_TYPE_KEY_PERINATAL_HSC)
|
|
self.assertEqual(out, payload)
|
|
|
|
|
|
class TestEmptyDetail(unittest.TestCase):
|
|
def test_shape(self):
|
|
d = empty_detail(
|
|
sample_code="B1",
|
|
lab_flow_code="LAB-001",
|
|
protocol_code="P1",
|
|
mother_name="张三",
|
|
sample_type_name="围产期造血干细胞",
|
|
sample_type_key="perinatal_hsc",
|
|
)
|
|
self.assertFalse(d["filledFlag"])
|
|
self.assertEqual(d["payload"], {})
|
|
self.assertEqual(d["sampleCode"], "B1")
|
|
self.assertEqual(d["labFlowCode"], "LAB-001")
|
|
|
|
def test_lab_flow_defaults_empty(self):
|
|
d = empty_detail(sample_code="B1")
|
|
self.assertEqual(d["labFlowCode"], "")
|
|
|
|
|
|
class TestRowToDict(unittest.TestCase):
|
|
def test_filled(self):
|
|
row = SimpleNamespace(
|
|
id="1",
|
|
sample_code="B1",
|
|
sample_type_key="perinatal_hsc",
|
|
collection_protocol_id="P",
|
|
qc_work_order_id="W",
|
|
filled_flag="1",
|
|
payload='{"viability_pct":"90"}',
|
|
create_time=None,
|
|
update_time=None,
|
|
)
|
|
d = _row_to_dict(
|
|
row,
|
|
lab_flow_code="LAB-001",
|
|
protocol_code="C1",
|
|
mother_name="李",
|
|
sample_type_name="围产期造血干细胞",
|
|
)
|
|
self.assertTrue(d["filledFlag"])
|
|
self.assertEqual(d["payload"]["viability_pct"], "90")
|
|
self.assertEqual(d["sampleCode"], "B1")
|
|
self.assertEqual(d["labFlowCode"], "LAB-001")
|
|
self.assertEqual(d["protocolCode"], "C1")
|
|
|
|
|
|
def _task(item_name, stamp, params, batch_kind=""):
|
|
values = [value for _name, value in params if str(value).strip()]
|
|
return {
|
|
"item_name": item_name,
|
|
"stamp": stamp,
|
|
"params": params,
|
|
"values": values,
|
|
"batch_kind": batch_kind,
|
|
}
|
|
|
|
|
|
class TestStorageCertPrefill(unittest.TestCase):
|
|
def test_cd34_param_names(self):
|
|
self.assertEqual(match_cd34_param("有核细胞数TNC/*108(个)"), "tnc_1e8")
|
|
self.assertEqual(match_cd34_param("CD34+细胞数/*106(个)"), "cd34_count_1e6")
|
|
self.assertEqual(match_cd34_param('CD34+ CD45dim 细胞数/*105(个)'), "cd34_cd45dim_1e5")
|
|
|
|
def test_volume_needs_both_numbers(self):
|
|
self.assertEqual(sum_cryo_volume(10.5, 2), "12.5")
|
|
self.assertEqual(sum_cryo_volume(10, None), "")
|
|
self.assertEqual(sum_cryo_volume(None, 2), "")
|
|
|
|
def test_qual_only_explicit_yin_yang(self):
|
|
self.assertEqual(qual_yin_yang(["阴性"]), "阴")
|
|
self.assertEqual(qual_yin_yang(["通过"]), "")
|
|
self.assertEqual(qual_yin_yang(["阴", "阳"]), "")
|
|
self.assertEqual(qual_yin_yang(["未检出"]), "")
|
|
|
|
def test_hsc_fills_known_values_and_skips_unconfirmed(self):
|
|
older = datetime(2026, 1, 1, 8, 0)
|
|
newer = datetime(2026, 2, 2, 9, 0)
|
|
suggested = build_suggested(
|
|
sample_type_key="perinatal_hsc",
|
|
collection_time=datetime(2026, 1, 1, 13, 45),
|
|
reception_time=datetime(2026, 1, 2, 9, 0),
|
|
enriched_blood_weight_g=20,
|
|
cryopreservation_volume_ml=9.1,
|
|
gate_tasks=[
|
|
_task(
|
|
"CD34+细胞流式",
|
|
older,
|
|
[("有核细胞数TNC/*108(个)", "1.1"), ("CD34+细胞数/*106(个)", "2.2")],
|
|
),
|
|
_task(
|
|
"CD34+细胞流式",
|
|
newer,
|
|
[
|
|
("有核细胞数TNC/*10E8", "8.27"),
|
|
("CD34+细胞数/*106(个)", "55.5"),
|
|
("CD34+ CD45dim 细胞数/*105(个)", "1.14"),
|
|
],
|
|
),
|
|
],
|
|
latest_freeze_date=datetime(2026, 3, 3, 16, 0),
|
|
)
|
|
self.assertEqual(suggested["collection_time"], "2026-01-01 13:45")
|
|
self.assertEqual(suggested["receive_time"], "2026-01-02 09:00")
|
|
self.assertEqual(suggested["cryo_volume_ml"], "29.1")
|
|
self.assertEqual(suggested["tnc_1e8"], "8.27")
|
|
self.assertEqual(suggested["cd34_count_1e6"], "55.5")
|
|
self.assertEqual(suggested["cd34_cd45dim_1e5"], "1.14")
|
|
self.assertEqual(suggested["storage_date"], "2026-03-03")
|
|
self.assertEqual(suggested["storage_years"], "20年")
|
|
self.assertEqual(suggested["bacteria_test"], "阴性")
|
|
self.assertEqual(suggested["result_comment"], "合格")
|
|
self.assertNotIn("hbsag", suggested)
|
|
self.assertNotIn("mother_hbsag", suggested)
|
|
self.assertNotIn("viability_pct", suggested)
|
|
self.assertNotIn("hla_typing", suggested)
|
|
self.assertNotIn("report_issue_date", suggested)
|
|
|
|
def test_hsc_serology_reads_batch_item_param(self):
|
|
self.assertEqual(batch_kind_from_name("B0000075-脐血"), "脐血")
|
|
self.assertEqual(batch_kind_from_name("B0000075-母血"), "母血")
|
|
stamp = datetime(2026, 7, 4, 9, 0)
|
|
suggested = build_suggested(
|
|
sample_type_key="perinatal_hsc",
|
|
gate_tasks=[
|
|
_task(
|
|
"乙肝五项定性",
|
|
stamp,
|
|
[
|
|
("乙型肝炎病毒表面抗原", "阴性"),
|
|
("抗乙型肝炎病毒表面抗体", "阳性"),
|
|
],
|
|
"脐血",
|
|
),
|
|
_task("抗丙型肝炎病毒抗体", stamp, [("抗丙型肝炎病毒抗体", "阳性")], "脐血"),
|
|
_task("抗梅毒螺旋体特异抗体", stamp, [("抗梅毒螺旋体特异抗体", "阴性")], "脐血"),
|
|
_task("抗人类免疫缺陷病毒抗体", stamp, [("抗人类免疫缺陷病毒抗体", "阴性")], "脐血"),
|
|
_task("乙肝五项定性", stamp, [("乙型肝炎病毒表面抗原", "阳性")], "母血"),
|
|
_task("乙型肝炎病毒DNA", stamp, [("乙型肝炎病毒DNA", "阴性")], "母血"),
|
|
_task(
|
|
"人类免疫缺陷病毒I型核酸（HIV-1 RNA）定量,实时荧光定量PCR",
|
|
stamp,
|
|
[],
|
|
"母血",
|
|
),
|
|
_task("丙型肝炎病毒RNA", stamp, [("丙型肝炎病毒RNA", "通过")], "母血"),
|
|
],
|
|
)
|
|
self.assertEqual(suggested["hbsag"], "阴性")
|
|
self.assertEqual(suggested["anti_hcv"], "阳性")
|
|
self.assertEqual(suggested["anti_tp"], "阴性")
|
|
self.assertEqual(suggested["anti_hiv"], "阴性")
|
|
self.assertNotIn("cmv_igm", suggested)
|
|
self.assertEqual(suggested["mother_hbsag"], "阳性")
|
|
self.assertEqual(suggested["mother_hbv_dna"], "阴性")
|
|
self.assertNotIn("mother_hiv1_rna", suggested)
|
|
self.assertNotIn("mother_hcv_rna", suggested)
|
|
self.assertEqual(suggested["bacteria_test"], "阴性")
|
|
self.assertNotIn("viability_pct", suggested)
|
|
|
|
def test_uc_msc_fixed_values_and_process_qual(self):
|
|
suggested = build_suggested(
|
|
sample_type_key="uc_msc",
|
|
receive_temperature="常温",
|
|
storage_total_tubes=17,
|
|
available_cell_count_1e7=8.5,
|
|
process_tasks=[
|
|
_task("内毒素（透析液）检测", datetime(2026, 1, 1), [("结果", "阴性")]),
|
|
_task("真菌培养及鉴定", datetime(2026, 1, 2), [("结论", "通过")]),
|
|
_task("支原体培养+药敏", datetime(2026, 1, 3), [("结果", "阳")]),
|
|
],
|
|
latest_freeze_date=datetime(2026, 4, 4),
|
|
)
|
|
self.assertEqual(suggested["transport_temp_c"], "常温")
|
|
self.assertEqual(suggested["storage_tube_count"], "17")
|
|
self.assertEqual(suggested["available_cell_count_1e7"], "8.5")
|
|
self.assertEqual(suggested["endotoxin_test"], "阴性")
|
|
self.assertNotIn("fungus_test", suggested)
|
|
self.assertEqual(suggested["mycoplasma_test"], "阳性")
|
|
self.assertEqual(suggested["bacteria_test"], "阴性")
|
|
self.assertEqual(suggested["result_judgment"], "合格")
|
|
self.assertNotIn("mother_hbsag", suggested)
|
|
self.assertEqual(suggested["available_cell_type"], "间充质干细胞")
|
|
self.assertEqual(suggested["available_cell_morphology"], "短梭/贴壁型")
|
|
self.assertEqual(suggested["available_immunophenotype"], "符合间充质干细胞表型特性")
|
|
self.assertEqual(suggested["freeze_method"], "程序降温盒降温")
|
|
self.assertEqual(suggested["storage_temp"], "-196℃")
|
|
self.assertEqual(suggested["storage_years"], "20年")
|
|
self.assertEqual(suggested["storage_date_p2"], "2026-04-04")
|
|
self.assertNotIn("available_viability_pct", suggested)
|
|
self.assertNotIn("report_issue_date", suggested)
|
|
|
|
def test_uc_msc_mother_blood_and_ignores_bacteria_culture(self):
|
|
stamp = datetime(2026, 5, 5, 8, 0)
|
|
suggested = build_suggested(
|
|
sample_type_key="uc_msc",
|
|
gate_tasks=[
|
|
_task(
|
|
"乙肝五项定性",
|
|
stamp,
|
|
[
|
|
("乙型肝炎病毒表面抗原", "阳性"),
|
|
("抗乙型肝炎病毒表面抗体", "阴性"),
|
|
],
|
|
"母血",
|
|
),
|
|
_task("乙型肝炎病毒DNA", stamp, [], "母血"),
|
|
],
|
|
process_tasks=[
|
|
_task("需氧菌培养及鉴定（相关体液）", stamp, [("需氧菌培养及鉴定（相关体液）", "阳性")]),
|
|
_task(
|
|
"支原体培养+药敏",
|
|
stamp,
|
|
[("解脲（Uu)支原体培养", "阴性"), ("人型（Mh)支原体培养", "阴性")],
|
|
),
|
|
],
|
|
)
|
|
self.assertEqual(suggested["mother_hbsag"], "阳性")
|
|
self.assertNotIn("mother_hbv_dna", suggested)
|
|
self.assertEqual(suggested["mycoplasma_test"], "阴性")
|
|
self.assertEqual(suggested["bacteria_test"], "阴性")
|
|
self.assertNotIn("hbsag", suggested)
|
|
|
|
def test_shared_mother_blood_fills_sample_without_its_own(self):
|
|
stamp = datetime(2026, 7, 4, 9, 0)
|
|
shared = [
|
|
_task(
|
|
"乙肝五项定性",
|
|
stamp,
|
|
[
|
|
("乙型肝炎病毒表面抗原", "阴性"),
|
|
("抗乙型肝炎病毒表面抗体", "阳性"),
|
|
],
|
|
"母血",
|
|
),
|
|
_task("乙型肝炎病毒DNA", stamp, [("乙型肝炎病毒DNA", "阴性")], "母血"),
|
|
_task("抗丙型肝炎病毒抗体", stamp, [("抗丙型肝炎病毒抗体", "阴性")], "母血"),
|
|
_task("丙型肝炎病毒RNA", stamp, [("丙型肝炎病毒RNA", "阴性")], "母血"),
|
|
_task("抗梅毒螺旋体特异抗体", stamp, [("抗梅毒螺旋体特异抗体", "阴性")], "母血"),
|
|
_task("抗巨细胞病毒抗体IgM定量", stamp, [("抗巨细胞病毒抗体IgM定量", "阴性")], "母血"),
|
|
_task("抗人类免疫缺陷病毒抗体", stamp, [("抗人类免疫缺陷病毒抗体", "阴性")], "母血"),
|
|
_task(
|
|
"人类免疫缺陷病毒I型核酸（HIV-1 RNA）定量,实时荧光定量PCR",
|
|
stamp,
|
|
[("人类免疫缺陷病毒核酸(1型) (HIV-1RNA)", "阴性")],
|
|
"母血",
|
|
),
|
|
_task("乙肝五项定性", stamp, [("乙型肝炎病毒表面抗原", "阳性")], "脐血"),
|
|
]
|
|
suggested = build_suggested(sample_type_key="uc_msc", shared_mother_tasks=shared)
|
|
self.assertEqual(suggested["mother_hbsag"], "阴性")
|
|
self.assertNotIn("mother_hbv_dna", suggested)
|
|
self.assertEqual(suggested["mother_anti_hcv"], "阴性")
|
|
self.assertNotIn("mother_hcv_rna", suggested)
|
|
self.assertEqual(suggested["mother_anti_tp"], "阴性")
|
|
self.assertEqual(suggested["mother_cmv_igm"], "阴性")
|
|
self.assertEqual(suggested["mother_anti_hiv"], "阴性")
|
|
self.assertNotIn("mother_hiv1_rna", suggested)
|
|
self.assertNotIn("hbsag", suggested)
|
|
|
|
def test_own_mother_blood_wins_and_shared_only_fills_gaps(self):
|
|
stamp = datetime(2026, 7, 4, 9, 0)
|
|
suggested = build_suggested(
|
|
sample_type_key="perinatal_hsc",
|
|
gate_tasks=[
|
|
_task("乙肝五项定性", stamp, [("乙型肝炎病毒表面抗原", "阳性")], "脐血"),
|
|
_task("乙肝五项定性", stamp, [("乙型肝炎病毒表面抗原", "阳性")], "母血"),
|
|
],
|
|
shared_mother_tasks=[
|
|
_task("乙肝五项定性", stamp, [("乙型肝炎病毒表面抗原", "阴性")], "母血"),
|
|
_task("乙型肝炎病毒DNA", stamp, [("乙型肝炎病毒DNA", "阴性")], "母血"),
|
|
_task("乙肝五项定性", stamp, [("乙型肝炎病毒表面抗原", "阴性")], "脐血"),
|
|
],
|
|
)
|
|
self.assertEqual(suggested["hbsag"], "阳性")
|
|
self.assertEqual(suggested["mother_hbsag"], "阳性")
|
|
self.assertEqual(suggested["mother_hbv_dna"], "阴性")
|
|
|
|
def test_merge_keeps_saved_values(self):
|
|
merged = merge_prefill(
|
|
{"storage_years": "30年", "collection_time": ""},
|
|
{"storage_years": "20年", "collection_time": "2026-01-01 13:45", "hbsag": ""},
|
|
)
|
|
self.assertEqual(merged["storage_years"], "30年")
|
|
self.assertEqual(merged["collection_time"], "2026-01-01 13:45")
|
|
self.assertNotIn("hbsag", merged)
|
|
|
|
|
|
class TestMergePayloadWithAcl(unittest.TestCase):
|
|
def test_merges_only_editable_keys(self):
|
|
merged = merge_payload_with_acl(
|
|
existing={"receive_time": "a", "bacteria_test": "阴性"},
|
|
incoming={"receive_time": "b", "bacteria_test": "阴性"},
|
|
editable_keys=["receive_time"],
|
|
sample_type_key=SAMPLE_TYPE_KEY_PERINATAL_HSC,
|
|
)
|
|
self.assertEqual(merged["receive_time"], "b")
|
|
self.assertEqual(merged["bacteria_test"], "阴性")
|
|
|
|
def test_rejects_unauthorized_change(self):
|
|
with self.assertRaises(PermissionError):
|
|
merge_payload_with_acl(
|
|
existing={"bacteria_test": "阴性"},
|
|
incoming={"bacteria_test": "阳性"},
|
|
editable_keys=["receive_time"],
|
|
sample_type_key=SAMPLE_TYPE_KEY_PERINATAL_HSC,
|
|
)
|
|
|
|
def test_rejects_empty_editable_keys(self):
|
|
with self.assertRaises(PermissionError):
|
|
merge_payload_with_acl(
|
|
existing={},
|
|
incoming={"receive_time": "x"},
|
|
editable_keys=[],
|
|
sample_type_key=SAMPLE_TYPE_KEY_PERINATAL_HSC,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|