zkxw/backend/tests/test_cell_product_chain.py

"""细胞产品来源链路 — 科研捐赠样本解析与多进一回归"""
import unittest
from unittest.mock import MagicMock, patch

from app.crud.cell_product import (
    _ancestor_sample_chain_link,
    _donation_sample_chain_node,
    _edge_meta,
    _find_donation_sample,
    _nest_ancestor_parent,
)


class TestDonationSampleChainResolution(unittest.TestCase):
    def _mock_wo(self, **kwargs):
        wo = MagicMock()
        wo.reception_source = kwargs.get("reception_source", "research_donation")
        wo.protocol_sample_id = kwargs.get("protocol_sample_id", "ds-001")
        wo.sample_code = kwargs.get("sample_code", "JZ202605270001")
        wo.protocol_no = kwargs.get("protocol_no", "JZCC20260527000001")
        return wo

    def _mock_donation_sample(self):
        s = MagicMock()
        s.id = "ds-001"
        s.sample_code = "JZ202605270001"
        s.research_donation_id = "don-001"
        s.sample_type_id = "type-1"
        return s

    @patch("app.crud.cell_product._find_protocol_sample", return_value=None)
    @patch("app.crud.cell_product._find_donation_sample")
    @patch("app.crud.cell_product._donation_sample_customer_dict")
    @patch("app.crud.cell_product._sample_type_name_from_id", return_value="脐带血")
    def test_ancestor_link_from_donation_sample_by_code(
        self, _type_name, customer_dict, find_donation, _find_proto
    ):
        find_donation.return_value = self._mock_donation_sample()
        customer_dict.return_value = {"motherName": "测试母亲"}

        db = MagicMock()
        sample_input = MagicMock(
            input_id="",
            input_code="JZ202605270001",
            input_name="",
        )
        wo = self._mock_wo()
        edge = {"taskNo": "SCRW001", "workOrderNo": "SCGD001"}

        link = _ancestor_sample_chain_link(db, "zkxw", sample_input, wo, edge)

        self.assertIsNotNone(link)
        self.assertEqual(link["node"]["nodeType"], "sample")
        self.assertEqual(link["node"]["receptionSource"], "research_donation")
        self.assertEqual(link["node"]["sampleCode"], "JZ202605270001")
        self.assertEqual(link["node"]["customer"]["motherName"], "测试母亲")
        find_donation.assert_called_once()

    @patch("app.crud.cell_product._find_protocol_sample", return_value=None)
    @patch("app.crud.cell_product._donation_sample_customer_dict", return_value={})
    @patch("app.crud.cell_product._sample_type_name_from_id", return_value="")
    @patch("app.crud.cell_product._find_donation_sample")
    def test_ancestor_link_fallback_to_work_order_sample_id(
        self, find_donation, _type_name, _customer, _find_proto
    ):
        find_donation.return_value = self._mock_donation_sample()

        db = MagicMock()
        sample_input = MagicMock(input_id="", input_code="", input_name="")
        wo = self._mock_wo()

        link = _ancestor_sample_chain_link(db, "zkxw", sample_input, wo, {})

        self.assertIsNotNone(link)
        self.assertEqual(link["node"]["id"], "ds-001")
        self.assertEqual(find_donation.call_args[0][4], wo)

    def test_donation_sample_chain_node_fields(self):
        db = MagicMock()
        with patch(
            "app.crud.cell_product._donation_sample_customer_dict",
            return_value={"motherName": "A"},
        ):
            link = _donation_sample_chain_node(
                db,
                "zkxw",
                self._mock_donation_sample(),
                wo=self._mock_wo(),
                edge={"taskId": "t1"},
                sample_type_name="脐带",
            )
        node = link["node"]
        self.assertEqual(node["collectionProtocolId"], "don-001")
        self.assertEqual(node["receptionSourceLabel"], "科研捐赠")
        self.assertEqual(node["sampleTypeName"], "脐带")


class TestChainEdgeMeta(unittest.TestCase):
    def test_edge_meta_work_order_type_labels(self):
        task = MagicMock(task_no="T001", task_name="制备", id="task-1")
        wo = MagicMock(work_order_no="WO001", work_order_type="temporary")
        edge = _edge_meta(task, wo)
        self.assertEqual(edge["workOrderType"], "temporary")
        self.assertEqual(edge["workOrderTypeLabel"], "临时生产")

        wo.work_order_type = "outbound"
        edge = _edge_meta(task, wo)
        self.assertEqual(edge["workOrderType"], "outbound")
        self.assertEqual(edge["workOrderTypeLabel"], "出库生产")

        wo.work_order_type = "inbound"
        edge = _edge_meta(task, wo)
        self.assertEqual(edge["workOrderType"], "inbound")
        self.assertEqual(edge["workOrderTypeLabel"], "入库生产")


class TestNestAncestorParent(unittest.TestCase):
    def test_nest_empty(self):
        self.assertIsNone(_nest_ancestor_parent([]))

    def test_nest_single(self):
        item = {"node": {"id": "a"}}
        self.assertIs(_nest_ancestor_parent([item]), item)

    def test_nest_multi(self):
        items = [{"node": {"id": "a"}}, {"node": {"id": "b"}}]
        self.assertEqual(_nest_ancestor_parent(items), items)


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