zkxw/backend/tests/test_save_samples_keeps_id.py

"""协议样本保存保留原行 ID，避免下游外键被整表替换打断。"""

import unittest
from datetime import datetime
from unittest.mock import patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.crud.collection_protocol import save_samples
from app.crud.research_donation import _create_samples
from app.models.models import BizProtocolSample
from app.models.research_donation import (
    BizResearchDonationSample,
    BizResearchDonationSampleTransport,
)


def _session():
    engine = create_engine("sqlite://")
    BizProtocolSample.__table__.create(bind=engine)
    return sessionmaker(bind=engine)()


def _sample(**kwargs):
    now = datetime(2026, 10, 27, 9, 0, 0)
    data = dict(
        collection_protocol_id="proto-1",
        tenant_no="tenant-a",
        create_dept_id="",
        create_time=now,
        update_time=now,
        del_flag=0,
    )
    data.update(kwargs)
    return BizProtocolSample(**data)


class TestSaveSamplesKeepsIdentity(unittest.TestCase):
    def test_existing_sample_keeps_id_and_removed_sample_is_soft_deleted(self):
        db = _session()
        db.add(_sample(id="sample-keep", sample_code="SC001"))
        db.add(_sample(id="sample-drop", sample_code="SC002"))
        db.commit()

        with patch("app.crud.sample_freight.apply_freight_on_protocol_save"), patch(
            "app.crud.collection_protocol.save_sample_transports"
        ), patch("app.crud.collection_protocol._migrate_sample_child_refs_on_replace"):
            saved = save_samples(
                db,
                "proto-1",
                "tenant-a",
                "user-1",
                [{"id": "sample-keep", "sample_code": "SC001", "remark": "改备注"}],
            )

        self.assertEqual(saved[0].id, "sample-keep")
        self.assertEqual(saved[0].remark, "改备注")
        rows = {row.id: row for row in db.query(BizProtocolSample).all()}
        self.assertEqual(rows["sample-keep"].del_flag, 0)
        self.assertEqual(rows["sample-drop"].del_flag, 1)

    def test_same_sample_code_without_id_still_keeps_row(self):
        db = _session()
        db.add(_sample(id="sample-keep", sample_code="SC001", remark="旧"))
        db.commit()

        with patch("app.crud.sample_freight.apply_freight_on_protocol_save"), patch(
            "app.crud.collection_protocol.save_sample_transports"
        ), patch("app.crud.collection_protocol._migrate_sample_child_refs_on_replace"):
            saved = save_samples(
                db,
                "proto-1",
                "tenant-a",
                "user-1",
                [{"sample_code": "SC001", "remark": "新"}],
            )

        self.assertEqual(saved[0].id, "sample-keep")
        self.assertEqual(saved[0].remark, "新")
        self.assertEqual(db.query(BizProtocolSample).filter_by(del_flag=0).count(), 1)


class TestDonationSamplesKeepIdentity(unittest.TestCase):
    def test_existing_donation_sample_keeps_id(self):
        engine = create_engine("sqlite://")
        BizResearchDonationSample.__table__.create(bind=engine)
        BizResearchDonationSampleTransport.__table__.create(bind=engine)
        db = sessionmaker(bind=engine)()
        now = datetime(2026, 10, 27, 9, 0, 0)
        db.add(
            BizResearchDonationSample(
                id="don-keep",
                research_donation_id="don-1",
                tenant_no="tenant-a",
                sample_code="DN001",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizResearchDonationSample(
                id="don-drop",
                research_donation_id="don-1",
                tenant_no="tenant-a",
                sample_code="DN002",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.commit()

        saved = _create_samples(
            db,
            "don-1",
            "tenant-a",
            "user-1",
            [{"sample_code": "DN001", "remark": "留下"}],
            now,
        )

        self.assertEqual(saved[0].id, "don-keep")
        self.assertEqual(saved[0].remark, "留下")
        rows = {row.id: row for row in db.query(BizResearchDonationSample).all()}
        self.assertEqual(rows["don-keep"].del_flag, 0)
        self.assertEqual(rows["don-drop"].del_flag, 1)


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