zkxw/backend/tests/test_protocol_status_change_history.py

"""状态变更历史：同类型追加不覆盖。"""
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


def _session():
    from app.models.models import BizCollectionProtocol, BizProtocolStatusChange, SysUser

    engine = create_engine("sqlite:///:memory:")
    BizCollectionProtocol.__table__.create(engine)
    BizProtocolStatusChange.__table__.create(engine)
    SysUser.__table__.create(engine)
    return sessionmaker(bind=engine, autoflush=False, autocommit=False)(), engine


def test_same_type_status_change_appends_instead_of_overwrite():
    from app.crud.collection_protocol import (
        list_protocol_status_changes,
        update_collection_protocol,
    )
    from app.models.models import BizCollectionProtocol

    db, engine = _session()
    try:
        now = datetime(2026, 3, 1, 10, 0, 0)
        db.add(
            BizCollectionProtocol(
                id="p1",
                protocol_no="XY0001",
                protocol_status="sxz",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.commit()

        update_collection_protocol(
            db,
            "p1",
            tenant_no="zkxw",
            user_id="u1",
            protocol_status="ygq",
            expire_time=datetime(2026, 6, 1, 12, 0, 0),
            expire_desc="第一次过期",
        )
        update_collection_protocol(
            db,
            "p1",
            tenant_no="zkxw",
            user_id="u2",
            protocol_status="ygq",
            expire_time=datetime(2026, 9, 15, 18, 0, 0),
            expire_desc="第二次过期",
        )

        row = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == "p1").one()
        assert row.expire_desc == "第二次过期"
        items = list_protocol_status_changes(db, collection_protocol_id="p1", tenant_no="zkxw")
        assert [item["changeDesc"] for item in items] == ["第一次过期", "第二次过期"]
        assert [item["changeType"] for item in items] == ["expire", "expire"]
        assert items[0]["fromStatus"] == "sxz"
        assert items[1]["fromStatus"] == "ygq"
        assert items[1]["toStatus"] == "ygq"
    finally:
        db.close()
        engine.dispose()