72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
"""状态变更历史：同类型追加不覆盖。"""
|
|
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()
|