zkxw/backend/tests/test_protocol_operation_log.py

"""Issue #24：协议操作过程日志 append-only 与动作映射。"""
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


def _session():
    from app.models.models import (
        BizCollectionProtocol,
        BizProtocolCollectionInfo,
        BizProtocolOperationLog,
        BizProtocolPartyCustomer,
        BizProtocolSample,
        SysUser,
    )

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


def test_record_cell_update_appends_before_after_and_maps_void_action():
    from app.crud import protocol_operation_log as op_log
    from app.models.models import BizCollectionProtocol, BizProtocolPartyCustomer, SysUser

    db, engine = _session()
    try:
        now = datetime(2026, 11, 5, 10, 0, 0)
        db.add(
            SysUser(
                id="u1",
                user_no="u1",
                nick_name="采集员甲",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizCollectionProtocol(
                id="p1",
                protocol_no="XY2401",
                protocol_status="sxz",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.add(
            BizProtocolPartyCustomer(
                id="party1",
                collection_protocol_id="p1",
                mother_name="旧姓名",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        db.commit()

        before = op_log.snapshot_cell_protocol(db, protocol_id="p1", tenant_no="zkxw")
        assert before is not None
        assert before["protocol"]["protocolStatus"] == "sxz"
        assert before["partyCustomer"]["motherName"] == "旧姓名"

        row = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == "p1").one()
        row.protocol_status = "yzf"
        party = (
            db.query(BizProtocolPartyCustomer)
            .filter(BizProtocolPartyCustomer.collection_protocol_id == "p1")
            .one()
        )
        party.mother_name = "新姓名"
        db.add(row)
        db.add(party)
        db.flush()

        op_log.record_cell_operation(
            db,
            protocol_id="p1",
            action=op_log.ACTION_UPDATE,
            tenant_no="zkxw",
            user_id="u1",
            client_channel=op_log.CHANNEL_MINI,
            before=before,
        )
        db.commit()

        items = op_log.list_protocol_operation_logs(
            db, biz_type=op_log.BIZ_CELL, biz_id="p1", tenant_no="zkxw"
        )
        assert len(items) == 1
        item = items[0]
        assert item["action"] == "void"
        assert item["clientChannel"] == "mini"
        assert item["operatorUserName"] == "采集员甲"
        assert item["before"]["partyCustomer"]["motherName"] == "旧姓名"
        assert item["after"]["partyCustomer"]["motherName"] == "新姓名"
        assert item["after"]["protocol"]["protocolStatus"] == "yzf"
    finally:
        db.close()
        engine.dispose()


def test_resolve_action_from_status_keeps_default_when_unchanged():
    from app.crud.protocol_operation_log import (
        ACTION_AUDIT,
        ACTION_UPDATE,
        ACTION_VOID,
        resolve_action_from_status,
    )

    assert (
        resolve_action_from_status(
            default_action=ACTION_UPDATE,
            before_status="sxz",
            after_status="sxz",
        )
        == ACTION_UPDATE
    )
    assert (
        resolve_action_from_status(
            default_action=ACTION_UPDATE,
            before_status="sxz",
            after_status="yzf",
        )
        == ACTION_VOID
    )
    assert (
        resolve_action_from_status(
            default_action=ACTION_AUDIT,
            before_status="dsx",
            after_status="sxz",
        )
        == ACTION_AUDIT
    )