shanxia2.0/backend/tests/test_bx_advance_batch_issue.py

"""备用金批量发放：同一备注和附件写入每张待发放单，实际金额按单提交。"""
from datetime import datetime

from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.bx_auth import BxUserContext
from app.models.bx_models import BxAdvanceApply, BxAttachment
from app.services.bx.advance import AdvanceService

TENANT = "shanxia2.0"
NOW = datetime(2026, 11, 4, 15, 0, 0)


def _session():
    engine = create_engine("sqlite://")
    BxAdvanceApply.__table__.create(engine)
    BxAttachment.__table__.create(engine)
    return sessionmaker(bind=engine)()


def _user() -> BxUserContext:
    return BxUserContext(
        id="finance-1",
        name="财务",
        dept_id="dept",
        role="super_admin",
        tenant_no=TENANT,
    )


def _apply(*, apply_id: str, amount: float, status: str = "approved", issue_status: str = None) -> BxAdvanceApply:
    return BxAdvanceApply(
        id=apply_id,
        apply_code=apply_id,
        applicant_id="applicant",
        applicant_name="汪璇",
        reason="备用金",
        amount=amount,
        status=status,
        issue_status=issue_status,
        tenant_no=TENANT,
        create_dept_id="dept",
        create_time=NOW,
        update_time=NOW,
        del_flag=0,
    )


def _attachment() -> BxAttachment:
    return BxAttachment(
        id="att-1",
        biz_type="advance_issue",
        biz_id="0",
        file_name="发放截图.png",
        file_path="bx/advance_issue/shot.png",
        file_type="image/png",
        file_size=12,
        tenant_no=TENANT,
        create_dept_id="dept",
        create_time=NOW,
        del_flag=0,
    )


def test_batch_issue_writes_shared_remark_and_attachment_with_per_bill_amount():
    db = _session()
    db.add_all([_apply(apply_id="BYJ-A", amount=100), _apply(apply_id="BYJ-B", amount=80.5)])
    db.add(_attachment())
    db.commit()

    rows = AdvanceService.batch_issue(
        db,
        [{"id": "BYJ-A", "actual_amount": 100}, {"id": "BYJ-B", "actual_amount": 70}],
        _user(),
        issue_remark="同一批次发放",
        attachment_ids=["att-1"],
    )
    db.commit()

    assert [row.id for row in rows] == ["BYJ-A", "BYJ-B"]
    saved = {row.id: row for row in db.query(BxAdvanceApply).all()}
    assert saved["BYJ-A"].status == "issued"
    assert saved["BYJ-A"].issue_status == "issued"
    assert float(saved["BYJ-A"].actual_amount) == 100
    assert saved["BYJ-A"].issue_remark == "同一批次发放"
    assert float(saved["BYJ-B"].actual_amount) == 70
    assert saved["BYJ-B"].issue_remark == "同一批次发放"

    copies = (
        db.query(BxAttachment)
        .filter(BxAttachment.del_flag == 0, BxAttachment.biz_type == "advance_issue")
        .order_by(BxAttachment.biz_id)
        .all()
    )
    assert [(row.biz_id, row.file_path) for row in copies] == [
        ("BYJ-A", "bx/advance_issue/shot.png"),
        ("BYJ-B", "bx/advance_issue/shot.png"),
    ]
    assert db.query(BxAttachment).filter(BxAttachment.id == "att-1").one().del_flag == 1


def test_batch_issue_requires_attachment_and_rejects_issued_bill():
    db = _session()
    db.add_all(
        [
            _apply(apply_id="BYJ-A", amount=100),
            _apply(apply_id="BYJ-B", amount=80, status="issued", issue_status="issued"),
        ]
    )
    db.commit()

    try:
        AdvanceService.batch_issue(
            db,
            [{"id": "BYJ-A", "actual_amount": 100}, {"id": "BYJ-B", "actual_amount": 80}],
            _user(),
            issue_remark="不应写入",
            attachment_ids=["att-1"],
        )
    except HTTPException as exc:
        assert exc.status_code == 400
        db.rollback()
    else:
        raise AssertionError("expected batch issue to reject an issued bill")

    saved = {row.id: row for row in db.query(BxAdvanceApply).all()}
    assert saved["BYJ-A"].status == "approved"
    assert saved["BYJ-A"].issue_remark is None

    try:
        AdvanceService.batch_issue(
            db,
            [{"id": "BYJ-A", "actual_amount": 100}],
            _user(),
            issue_remark="没有附件",
            attachment_ids=[],
        )
    except HTTPException as exc:
        assert exc.status_code == 400
        assert "附件" in str(exc.detail)
        db.rollback()
    else:
        raise AssertionError("expected batch issue to require an attachment")

    assert db.query(BxAdvanceApply).filter(BxAdvanceApply.id == "BYJ-A").one().status == "approved"