zkxw/backend/tests/test_oa_bank_receipt.py

"""OA 银行回执单：biz_attachment 挂载 instance。"""
from __future__ import annotations

import unittest
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.base import Base
from app.models.models import BizAttachment, OaApp, OaAppVersion, OaInstance
from app.services import oa_bank_receipt as svc


class TestOaBankReceipt(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(
            self.engine,
            tables=[
                OaApp.__table__,
                OaAppVersion.__table__,
                OaInstance.__table__,
                BizAttachment.__table__,
            ],
        )
        Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)
        self.db = Session()
        t0 = datetime(2026, 10, 16, 10, 0, 0)
        self.db.add_all(
            [
                OaApp(
                    id="app1",
                    app_name="付款申请",
                    initiator_scope="all",
                    admin_scope="all_oa_admin",
                    status="1",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t0,
                    update_time=t0,
                    del_flag=0,
                ),
                OaAppVersion(
                    id="v1",
                    app_id="app1",
                    version_no=1,
                    status="published",
                    is_latest_published="1",
                    form_schema_json="{}",
                    flow_tree_json="{}",
                    advanced_settings_json="{}",
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t0,
                    update_time=t0,
                    del_flag=0,
                ),
                OaInstance(
                    id="inst-1",
                    app_id="app1",
                    app_version_id="v1",
                    title="付款-1",
                    initiator_user_id="u1",
                    status="completed",
                    current_round=1,
                    tenant_no="zkxw",
                    create_dept_id="",
                    create_time=t0,
                    update_time=t0,
                    del_flag=0,
                ),
                OaInstance(
                    id="inst-other-tenant",
                    app_id="app1",
                    app_version_id="v1",
                    title="付款-x",
                    initiator_user_id="u1",
                    status="completed",
                    current_round=1,
                    tenant_no="other",
                    create_dept_id="",
                    create_time=t0,
                    update_time=t0,
                    del_flag=0,
                ),
            ]
        )
        self.db.commit()

    def tearDown(self):
        self.db.close()
        self.engine.dispose()

    def test_list_empty(self):
        items = svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw")
        self.assertEqual(items, [])

    def test_replace_then_list(self):
        out = svc.replace_bank_receipts(
            self.db,
            instance_id="inst-1",
            tenant_no="zkxw",
            items=[
                {
                    "fileUrl": "oa-bank-receipt/a.pdf",
                    "fileName": "回单A.pdf",
                    "fileType": "application/pdf",
                    "fileSize": 100,
                },
                {
                    "fileUrl": "oa-bank-receipt/b.jpg",
                    "fileName": "回单B.jpg",
                    "fileType": "image/jpeg",
                    "fileSize": 200,
                },
            ],
            user_id="u1",
            dept_id="d1",
        )
        self.db.commit()
        self.assertEqual(len(out), 2)
        listed = svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw")
        self.assertEqual([x["fileUrl"] for x in listed], [
            "oa-bank-receipt/a.pdf",
            "oa-bank-receipt/b.jpg",
        ])
        self.assertEqual(listed[0]["fileName"], "回单A.pdf")

    def test_replace_removes_missing(self):
        svc.replace_bank_receipts(
            self.db,
            instance_id="inst-1",
            tenant_no="zkxw",
            items=[
                {"fileUrl": "oa-bank-receipt/a.pdf", "fileName": "A.pdf"},
                {"fileUrl": "oa-bank-receipt/b.pdf", "fileName": "B.pdf"},
            ],
            user_id="u1",
            dept_id="",
        )
        self.db.commit()
        svc.replace_bank_receipts(
            self.db,
            instance_id="inst-1",
            tenant_no="zkxw",
            items=[{"fileUrl": "oa-bank-receipt/b.pdf", "fileName": "B.pdf"}],
            user_id="u1",
            dept_id="",
        )
        self.db.commit()
        listed = svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw")
        self.assertEqual([x["fileUrl"] for x in listed], ["oa-bank-receipt/b.pdf"])
        gone = (
            self.db.query(BizAttachment)
            .filter(BizAttachment.file_url == "oa-bank-receipt/a.pdf")
            .first()
        )
        self.assertEqual(gone.del_flag, 1)

    def test_replace_empty_clears_all(self):
        svc.replace_bank_receipts(
            self.db,
            instance_id="inst-1",
            tenant_no="zkxw",
            items=[{"fileUrl": "oa-bank-receipt/a.pdf", "fileName": "A.pdf"}],
            user_id="u1",
            dept_id="",
        )
        self.db.commit()
        svc.replace_bank_receipts(
            self.db,
            instance_id="inst-1",
            tenant_no="zkxw",
            items=[],
            user_id="u1",
            dept_id="",
        )
        self.db.commit()
        self.assertEqual(
            svc.list_bank_receipts(self.db, instance_id="inst-1", tenant_no="zkxw"),
            [],
        )

    def test_bind_orphan_upload_row(self):
        """通用上传已写入无 business_type 的行时，PUT 应绑定而非再插一行。"""
        t0 = datetime(2026, 10, 16, 11, 0, 0)
        orphan = BizAttachment(
            id="att-orphan",
            business_type=None,
            business_id=None,
            original_name="orphan.pdf",
            file_ext="pdf",
            file_url="oa-bank-receipt/orphan.pdf",
            file_size=10,
            file_type="application/pdf",
            tenant_no="zkxw",
            create_dept_id="",
            create_time=t0,
            update_time=t0,
            del_flag=0,
        )
        self.db.add(orphan)
        self.db.commit()
        svc.replace_bank_receipts(
            self.db,
            instance_id="inst-1",
            tenant_no="zkxw",
            items=[{"fileUrl": "oa-bank-receipt/orphan.pdf", "fileName": "orphan.pdf"}],
            user_id="u1",
            dept_id="",
        )
        self.db.commit()
        rows = (
            self.db.query(BizAttachment)
            .filter(BizAttachment.file_url == "oa-bank-receipt/orphan.pdf", BizAttachment.del_flag == 0)
            .all()
        )
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0].business_type, svc.BUSINESS_TYPE)
        self.assertEqual(rows[0].business_id, "inst-1")

    def test_missing_instance_raises(self):
        with self.assertRaises(svc.OaBankReceiptError):
            svc.list_bank_receipts(self.db, instance_id="nope", tenant_no="zkxw")

    def test_cross_tenant_raises(self):
        with self.assertRaises(svc.OaBankReceiptError):
            svc.list_bank_receipts(
                self.db, instance_id="inst-other-tenant", tenant_no="zkxw"
            )


if __name__ == "__main__":
    unittest.main()