zkxw/backend/tests/test_sample_storage_cert_qc.py

"""储存证质检信息：类型键、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,
    normalize_payload,
    resolve_sample_type_key,
    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 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.assertEqual(suggested["mother_hbv_dna"], "阴性")
        self.assertEqual(suggested["mother_anti_hcv"], "阴性")
        self.assertEqual(suggested["mother_hcv_rna"], "阴性")
        self.assertEqual(suggested["mother_anti_tp"], "阴性")
        self.assertEqual(suggested["mother_cmv_igm"], "阴性")
        self.assertEqual(suggested["mother_anti_hiv"], "阴性")
        self.assertEqual(suggested["mother_hiv1_rna"], "阴性")
        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)


if __name__ == "__main__":
    unittest.main()