"""协议作废 ↔ 样本接收废弃双向联动。"""

from __future__ import annotations

import unittest
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.database import Base
from app.models.models import (
    BizCollectionProtocol,
    BizProtocolSample,
    BizProtocolStatusChange,
    BizSampleReception,
    SysDictData,
)
from app.models.research_donation import (
    BizDonationStatusChange,
    BizResearchDonation,
    BizResearchDonationSample,
)
from app.services.protocol_reception_discard_sync import (
    LAB_VOID_TYPE_CODE,
    after_protocol_voided,
    after_sample_reception_saved,
    ensure_lab_void_type,
)


def _session():
    engine = create_engine("sqlite://")
    Base.metadata.create_all(
        bind=engine,
        tables=[
            SysDictData.__table__,
            BizCollectionProtocol.__table__,
            BizProtocolSample.__table__,
            BizProtocolStatusChange.__table__,
            BizSampleReception.__table__,
            BizResearchDonation.__table__,
            BizResearchDonationSample.__table__,
            BizDonationStatusChange.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _now():
    return datetime(2026, 11, 5, 12, 0, 0)


class TestProtocolReceptionDiscardSync(unittest.TestCase):
    def test_ensure_lab_void_type_creates_dict_row(self):
        db = _session()
        code = ensure_lab_void_type(db, tenant_no="tenant-a")
        self.assertEqual(code, LAB_VOID_TYPE_CODE)
        row = (
            db.query(SysDictData)
            .filter(
                SysDictData.tenant_no == "tenant-a",
                SysDictData.dict_type == "xyzflx",
                SysDictData.dict_value == LAB_VOID_TYPE_CODE,
            )
            .one()
        )
        self.assertEqual(row.dict_label, "实验室废弃")
        # idempotent
        self.assertEqual(ensure_lab_void_type(db, tenant_no="tenant-a"), LAB_VOID_TYPE_CODE)

    def test_protocol_void_marks_all_sample_receptions_discarded(self):
        db = _session()
        now = _now()
        db.add(
            BizCollectionProtocol(
                id="p1",
                protocol_no="CP-1",
                protocol_status="sxz",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizProtocolSample(
                id="s1",
                collection_protocol_id="p1",
                sample_code="ZK001",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizProtocolSample(
                id="s2",
                collection_protocol_id="p1",
                sample_code="ZK002",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizSampleReception(
                id="r1",
                protocol_sample_id="s1",
                collection_protocol_id="p1",
                reception_time=now,
                accept_conclusion="js",
                reception_source="cell_storage",
                sample_code="ZK001",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
                check_packaging=0,
                check_sealing=0,
                check_no_damage=0,
                check_no_leakage=0,
                check_color=0,
                check_tying_rope=0,
                check_no_stratification=0,
            )
        )
        db.commit()

        after_protocol_voided(
            db,
            source="cell_storage",
            protocol_id="p1",
            tenant_no="tenant-a",
            user_id="u1",
        )
        db.commit()

        rows = (
            db.query(BizSampleReception)
            .filter(BizSampleReception.tenant_no == "tenant-a", BizSampleReception.del_flag == 0)
            .all()
        )
        self.assertEqual(len(rows), 2)
        by_code = {r.sample_code: r.accept_conclusion for r in rows}
        self.assertEqual(by_code["ZK001"], "bj")
        self.assertEqual(by_code["ZK002"], "bj")

    def test_all_samples_discarded_auto_voids_protocol_with_lab_type(self):
        db = _session()
        now = _now()
        db.add(
            BizCollectionProtocol(
                id="p2",
                protocol_no="CP-2",
                protocol_status="sxz",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizProtocolSample(
                id="s3",
                collection_protocol_id="p2",
                sample_code="ZK003",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        rec = BizSampleReception(
            id="r2",
            protocol_sample_id="s3",
            collection_protocol_id="p2",
            reception_time=now,
            accept_conclusion="bj",
            reception_source="cell_storage",
            sample_code="ZK003",
            tenant_no="tenant-a",
            create_dept_id="",
            create_time=now,
            update_time=now,
            del_flag=0,
            check_packaging=0,
            check_sealing=0,
            check_no_damage=0,
            check_no_leakage=0,
            check_color=0,
            check_tying_rope=0,
            check_no_stratification=0,
        )
        db.add(rec)
        db.commit()

        after_sample_reception_saved(
            db, reception=rec, tenant_no="tenant-a", user_id="u1"
        )
        db.commit()

        proto = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == "p2").one()
        self.assertEqual(proto.protocol_status, "yzf")
        self.assertEqual(proto.void_type, LAB_VOID_TYPE_CODE)
        self.assertIn("全部样本接收", proto.void_desc or "")

    def test_partial_discard_does_not_void_protocol(self):
        db = _session()
        now = _now()
        db.add(
            BizCollectionProtocol(
                id="p3",
                protocol_no="CP-3",
                protocol_status="sxz",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        for sid, code in (("s4", "ZK004"), ("s5", "ZK005")):
            db.add(
                BizProtocolSample(
                    id=sid,
                    collection_protocol_id="p3",
                    sample_code=code,
                    tenant_no="tenant-a",
                    create_dept_id="",
                    create_time=now,
                    update_time=now,
                    del_flag=0,
                )
            )
        rec = BizSampleReception(
            id="r3",
            protocol_sample_id="s4",
            collection_protocol_id="p3",
            reception_time=now,
            accept_conclusion="bj",
            reception_source="cell_storage",
            sample_code="ZK004",
            tenant_no="tenant-a",
            create_dept_id="",
            create_time=now,
            update_time=now,
            del_flag=0,
            check_packaging=0,
            check_sealing=0,
            check_no_damage=0,
            check_no_leakage=0,
            check_color=0,
            check_tying_rope=0,
            check_no_stratification=0,
        )
        db.add(rec)
        db.commit()

        after_sample_reception_saved(
            db, reception=rec, tenant_no="tenant-a", user_id="u1"
        )
        db.commit()

        proto = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == "p3").one()
        self.assertEqual(proto.protocol_status, "sxz")
        self.assertFalse(proto.void_type)

    def test_donation_void_creates_discard_receptions(self):
        db = _session()
        now = _now()
        db.add(
            BizResearchDonation(
                id="d1",
                donation_no="RD-1",
                donation_status="sxz",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizResearchDonationSample(
                id="ds1",
                research_donation_id="d1",
                sample_code="RD001",
                tenant_no="tenant-a",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.commit()

        after_protocol_voided(
            db,
            source="research_donation",
            protocol_id="d1",
            tenant_no="tenant-a",
            user_id="u1",
        )
        db.commit()

        rec = (
            db.query(BizSampleReception)
            .filter(
                BizSampleReception.sample_code == "RD001",
                BizSampleReception.reception_source == "research_donation",
            )
            .one()
        )
        self.assertEqual(rec.accept_conclusion, "bj")
        self.assertEqual(rec.research_donation_id, "d1")


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