import unittest from app.services.lab_sample_processing_fields import ( DATASET_HSC, DATASET_MSC, DATASET_UBMC, DEPT_MARKET, DEPT_PRODUCTION, DEPT_QC, editable_depts_for_user, field_map, resolve_dataset_by_type_name, schema_payload, ) class LabSampleProcessingFieldsTest(unittest.TestCase): def test_resolve_dataset(self): self.assertEqual(resolve_dataset_by_type_name("围产期造血干细胞", "B0000001"), DATASET_HSC) self.assertEqual(resolve_dataset_by_type_name("脐带间充质干细胞", "BC0000003"), DATASET_MSC) self.assertEqual(resolve_dataset_by_type_name("", "UBMC2025001"), DATASET_UBMC) self.assertEqual(resolve_dataset_by_type_name("", "UC2026001"), DATASET_MSC) def test_schema_not_empty(self): self.assertGreater(len(schema_payload(DATASET_HSC)), 10) self.assertGreater(len(schema_payload(DATASET_UBMC)), 10) self.assertGreater(len(schema_payload(DATASET_MSC)), 10) def test_dept_acl(self): self.assertEqual( editable_depts_for_user( is_manager=True, dept_flags={DEPT_MARKET: False, DEPT_PRODUCTION: False, DEPT_QC: False}, ), {DEPT_MARKET, DEPT_PRODUCTION, DEPT_QC}, ) self.assertEqual( editable_depts_for_user( is_manager=False, dept_flags={DEPT_MARKET: False, DEPT_PRODUCTION: True, DEPT_QC: False}, ), {DEPT_PRODUCTION}, ) def test_hsc_market_and_qc_fields(self): fmap = field_map(DATASET_HSC) self.assertEqual(fmap["motherName"].dept, DEPT_MARKET) self.assertEqual(fmap["cd34Count"].dept, DEPT_QC) self.assertEqual(fmap["originalBloodWeightG"].dept, DEPT_PRODUCTION) def test_source_fields_not_overridden_by_extra(self): from datetime import datetime from app.services.lab_sample_processing import _build_row from types import SimpleNamespace sample = SimpleNamespace(id="S1", sample_code="ZKZX34000865", collection_protocol_id="P1") party = SimpleNamespace(mother_name="真实妈妈") coll = SimpleNamespace( sample_collect_time=None, maternal_blood_collect_time=datetime(2026, 11, 4, 5, 20), ) hospital = SimpleNamespace(hospital_name="真实医院") protocol = SimpleNamespace( consent_form_urls="protocol-consent/a", health_survey_form_urls="", ) reception = SimpleNamespace( id="R1", reception_time=None, receive_temperature=None, handler_name="接收人", remark=None, lab_storage_code="B0000103", ) production = SimpleNamespace( id="PR1", original_blood_weight_g=72.08, hes_volume_ml=None, blood_type="B", infectious_four_test_result="zc", enriched_blood_weight_g=None, cryopreservation_volume_ml=None, handler_name="操作人", remark=None, original_blood_volume_ml=None, nucleated_cell_count_1e8=None, inoculation_amount=None, cryopreservation_time=None, storage_total_tubes=None, available_cell_count_1e7=None, ) row = _build_row( dataset=DATASET_HSC, sample=sample, type_name="围产期造血干细胞", party=party, coll=coll, hospital=hospital, reception=reception, production=production, extra={ "motherName": "Excel妈妈", "originalBloodWeightG": 1, "cd34Count": "11.4", }, protocol=protocol, check_flags={ "健康调查表": True, "样本信息正确，标识清晰": True, "针路密闭": True, "容器无破损": True, }, ) self.assertEqual(row["sampleCode"], "B0000103") self.assertEqual(row["protocolSampleCode"], "ZKZX34000865") self.assertEqual(row["motherName"], "真实妈妈") self.assertEqual(row["originalBloodWeightG"], 72.08) self.assertEqual(row["bloodType"], "B") self.assertEqual(row["cd34Count"], "11.4") self.assertEqual(row["hasMaternalBlood"], "是") self.assertEqual(row["hasConsent"], "是") self.assertEqual(row["hasHealthSurvey"], "是") self.assertEqual(row["labelClear"], "是") self.assertEqual(row["infectiousFourCard"], "正常") self.assertIn("针路密闭", row["receiveCheck"] or "") self.assertFalse(row["fromExtraOnly"]) def test_infectious_code_roundtrip(self): from app.services.lab_sample_processing import _infectious_code, _infectious_label self.assertEqual(_infectious_label("zc"), "正常") self.assertEqual(_infectious_label("bzc"), "不正常") self.assertEqual(_infectious_code("正常"), "zc") self.assertEqual(_infectious_code("不正常"), "bzc") def test_extra_field_keys_include_fungus(self): from app.services.lab_sample_processing import _extra_field_keys keys = _extra_field_keys(DATASET_MSC) self.assertIn("fungusTest", keys) self.assertIn("endotoxinTest", keys) self.assertIn("bacteriaTest", keys) self.assertIn("mycoplasmaTest", keys) def test_msc_donation_collect_and_transport_fallback(self): from datetime import datetime from app.services.lab_sample_processing import _build_row from types import SimpleNamespace sample = SimpleNamespace(id="D1", sample_code="KYJZ1", collection_protocol_id=None) party = SimpleNamespace(mother_name="孙玲") coll = SimpleNamespace( sample_collect_time=datetime(2026, 10, 10, 10, 5), maternal_blood_collect_time=None, ) reception = SimpleNamespace( id="R1", reception_time=datetime(2026, 10, 10, 15, 40, 57), receive_temperature="23.1", handler_name="侯以婷", remark=None, lab_storage_code="UC2026006", ) row = _build_row( dataset=DATASET_MSC, sample=sample, type_name="脐带间充质干细胞", party=party, coll=coll, hospital=None, reception=reception, production=None, extra={}, ) self.assertEqual(row["sampleCode"], "UC2026006") self.assertEqual(row["collectTime"], "2026-10-10 10:05:00") self.assertEqual(row["transportTempC"], "23.1") self.assertNotIn("导出自存", row["motherName"] or "") def test_legacy_supplement_fills_empty_source_fields(self): from app.services.lab_sample_processing import _build_row, _merge_extra_with_legacy from types import SimpleNamespace extra = _merge_extra_with_legacy( {}, {"infectiousFourTestLabel": "正常", "labMaterialCode": "202611006"}, ) sample = SimpleNamespace(id="S1", sample_code="UBMC202611006", collection_protocol_id=None) reception = SimpleNamespace( id="R1", reception_time=None, receive_temperature=None, handler_name=None, remark=None, lab_storage_code="UBMC202611006", ) production = SimpleNamespace( id="PR1", original_blood_weight_g=None, hes_volume_ml=None, blood_type=None, infectious_four_test_result=None, enriched_blood_weight_g=None, cryopreservation_volume_ml=None, handler_name=None, remark=None, original_blood_volume_ml=None, nucleated_cell_count_1e8=None, inoculation_amount=None, cryopreservation_time=None, storage_total_tubes=None, available_cell_count_1e7=None, ) row = _build_row( dataset=DATASET_UBMC, sample=sample, type_name="脐血单个核细胞", party=None, coll=None, hospital=None, reception=reception, production=production, extra=extra, ) self.assertEqual(row["infectiousFourCard"], "正常") self.assertEqual(row["labMaterialCode"], "202611006") if __name__ == "__main__": unittest.main()