zlwuandCursor
e2e9059c17
When a cell-storage or research-donation protocol is voided, mark all related sample receptions as discarded; when every sample is discarded, auto-void the protocol as 实验室废弃 (sysfq). Covers issue #31. Co-authored-by: Cursor <cursoragent@cursor.com>
329 lines
9.6 KiB
Python
329 lines
9.6 KiB
Python
"""协议作废 ↔ 样本接收废弃双向联动。"""
|
|
|
|
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()
|