zkxw/backend/tests/test_oa_receive_account.py

import json
import unittest
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.crud import oa_receive_account as ora_crud
from app.models.base import Base
from app.models.models import OaApp, OaAppVersion, OaReceiveAccount, SysUser
from app.services.oa_engine import OaEngineError, start_instance
from app.services.oa_receive_account import (
    normalize_account_fields,
    validate_receive_account_form_values,
    validate_receive_account_ownership,
)


class TestNormalizeAccountFields(unittest.TestCase):
    def test_alipay_clears_bank(self):
        ok, err, data = normalize_account_fields(
            {
                "accountType": "alipay",
                "accountName": "张三",
                "accountNo": "13800138000",
                "bankName": "工行",
                "bankProvince": "安徽",
                "bankCity": "合肥",
                "bankBranch": "xx支行",
                "remark": "r",
            }
        )
        self.assertTrue(ok)
        self.assertIsNone(err)
        self.assertEqual(data["account_type"], "alipay")
        self.assertEqual(data["account_name"], "张三")
        self.assertEqual(data["bank_name"], "")
        self.assertEqual(data["bank_province"], "")
        self.assertEqual(data["bank_city"], "")
        self.assertEqual(data["bank_branch"], "")

    def test_personal_requires_bank(self):
        ok, err, _ = normalize_account_fields(
            {
                "accountType": "personal",
                "accountName": "李四",
                "accountNo": "6222",
            }
        )
        self.assertFalse(ok)
        self.assertTrue(err)

    def test_personal_ok(self):
        ok, err, data = normalize_account_fields(
            {
                "accountType": "personal",
                "accountName": "李四",
                "accountNo": "6222",
                "bankName": "工行",
                "bankProvince": "安徽省",
                "bankCity": "合肥市",
                "bankBranch": "长江中路支行",
            }
        )
        self.assertTrue(ok)
        self.assertEqual(data["bank_name"], "工行")


class TestValidateReceiveAccountFormValues(unittest.TestCase):
    def test_ok_array(self):
        schema = {
            "rule": [
                {
                    "type": "receiveAccount",
                    "field": "acc",
                    "title": "收款账户",
                    "$required": True,
                }
            ]
        }
        data = {
            "acc": [
                {
                    "id": "1",
                    "accountType": "alipay",
                    "accountName": "张三",
                    "accountNo": "138",
                    "bankName": "",
                    "bankProvince": "",
                    "bankCity": "",
                    "bankBranch": "",
                    "remark": "",
                }
            ]
        }
        self.assertEqual(validate_receive_account_form_values(schema, data), [])

    def test_required_empty(self):
        schema = {
            "rule": [
                {"type": "receiveAccount", "field": "acc", "title": "收款账户", "$required": True}
            ]
        }
        errs = validate_receive_account_form_values(schema, {"acc": []})
        self.assertTrue(any("收款账户" in e for e in errs))

    def test_invalid_item(self):
        schema = {"rule": [{"type": "receiveAccount", "field": "acc"}]}
        errs = validate_receive_account_form_values(schema, {"acc": [{"id": "1"}]})
        self.assertTrue(len(errs) >= 1)


def _snap(account_id: str) -> dict:
    return {
        "id": account_id,
        "accountType": "alipay",
        "accountName": "张三",
        "accountNo": "13800138000",
        "bankName": "",
        "bankProvince": "",
        "bankCity": "",
        "bankBranch": "",
        "remark": "",
    }


def _schema(*fields: str) -> dict:
    return {
        "rule": [
            {"type": "receiveAccount", "field": field, "title": field}
            for field in fields
        ]
    }


class TestReceiveAccountOwner(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(
            self.engine,
            tables=[
                OaReceiveAccount.__table__,
                SysUser.__table__,
                OaApp.__table__,
                OaAppVersion.__table__,
            ],
        )
        self.db = sessionmaker(bind=self.engine, autoflush=False, autocommit=False)()
        now = datetime(2026, 10, 21, 10, 0, 0)
        self.db.add_all(
            [
                SysUser(id="me", user_no="me", nick_name="我自己", password="", tenant_no="zkxw", create_dept_id="", create_time=now, update_time=now, del_flag=0),
                SysUser(id="other", user_no="other", nick_name="别人", password="", tenant_no="zkxw", create_dept_id="", create_time=now, update_time=now, del_flag=0),
            ]
        )
        self.db.add_all(
            [
                self._account("mine", "me", 0),
                self._account("padded", "  me  ", 0),
                self._account("theirs", "other", 0),
                self._account("blank", "", 0),
                self._account("none", None, 0),
                self._account("gone", "me", 1),
                self._account("other-tenant", "me", 0, tenant_no="other"),
            ]
        )
        self.db.commit()

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

    def _account(self, account_id, create_by, del_flag, tenant_no="zkxw"):
        now = datetime(2026, 10, 21, 10, 0, 0)
        return OaReceiveAccount(
            id=account_id,
            account_type="alipay",
            account_name=account_id,
            account_no=account_id,
            tenant_no=tenant_no,
            create_by=create_by,
            create_dept_id="",
            create_time=now,
            update_time=now,
            del_flag=del_flag,
        )

    def test_options_only_current_owner(self):
        ids = {row["id"] for row in ora_crud.list_options(self.db, tenant_no="zkxw", user_id="me")}
        self.assertEqual(ids, {"padded", "mine"})

    def test_list_keeps_every_tenant_account_and_owner_name(self):
        rows, total = ora_crud.list_accounts(self.db, tenant_no="zkxw", page=1, page_size=20)
        by_id = {row["id"]: row for row in rows}
        self.assertEqual(total, 5)
        self.assertEqual(by_id["mine"]["ownerUserId"], "me")
        self.assertEqual(by_id["mine"]["ownerName"], "我自己")
        self.assertEqual(by_id["padded"]["ownerUserId"], "me")
        self.assertEqual(by_id["padded"]["ownerName"], "我自己")
        self.assertEqual(by_id["blank"]["ownerUserId"], "")
        self.assertEqual(by_id["blank"]["ownerName"], "—")
        self.assertEqual(by_id["none"]["ownerName"], "—")
        self.assertNotIn("gone", by_id)
        self.assertNotIn("other-tenant", by_id)

    def test_missing_user_still_selectable_with_dash_name(self):
        self.db.add(self._account("orphan", "ghost", 0))
        self.db.commit()
        ids = [row["id"] for row in ora_crud.list_options(self.db, tenant_no="zkxw", user_id="ghost")]
        self.assertEqual(ids, ["orphan"])
        row = ora_crud.get_account(self.db, "orphan", "zkxw")
        self.assertEqual(row["ownerUserId"], "ghost")
        self.assertEqual(row["ownerName"], "—")

    def test_create_sets_current_user_as_owner(self):
        created = ora_crud.create_account(
            self.db,
            tenant_no="zkxw",
            user_id="me",
            dept_id="",
            fields={
                "account_type": "alipay",
                "account_name": "新户",
                "account_no": "6222",
                "bank_name": "",
                "bank_province": "",
                "bank_city": "",
                "bank_branch": "",
                "remark": "",
            },
        )
        self.assertEqual(created["ownerUserId"], "me")
        self.assertEqual(created["ownerName"], "我自己")
        ids = {row["id"] for row in ora_crud.list_options(self.db, tenant_no="zkxw", user_id="me")}
        self.assertIn(created["id"], ids)

    def test_update_does_not_change_owner(self):
        ora_crud.update_account(
            self.db,
            "mine",
            "zkxw",
            "other",
            {
                "account_type": "alipay",
                "account_name": "已改户名",
                "account_no": "mine",
                "bank_name": "",
                "bank_province": "",
                "bank_city": "",
                "bank_branch": "",
                "remark": "",
            },
        )
        row = self.db.query(OaReceiveAccount).filter(OaReceiveAccount.id == "mine").one()
        self.assertEqual(row.create_by, "me")
        self.assertEqual(row.account_name, "已改户名")

    def test_ownership_accepts_own_account(self):
        errs = validate_receive_account_ownership(
            self.db,
            _schema("acc"),
            {"acc": [_snap("mine")]},
            tenant_no="zkxw",
            initiator_user_id="me",
        )
        self.assertEqual(errs, [])

    def test_ownership_rejects_in_schema_order(self):
        errs = validate_receive_account_ownership(
            self.db,
            _schema("first", "second"),
            {"first": [_snap("mine")], "second": [_snap("theirs"), _snap("blank")]},
            tenant_no="zkxw",
            initiator_user_id="me",
        )
        self.assertEqual(errs, ["「second」不属于当前用户"])

    def test_ownership_error_messages(self):
        cases = [
            ("gone", "「acc」不存在或已删除"),
            ("other-tenant", "「acc」不存在或已删除"),
            ("missing", "「acc」不存在或已删除"),
            ("theirs", "「acc」不属于当前用户"),
            ("blank", "「acc」不属于当前用户"),
            ("none", "「acc」不属于当前用户"),
        ]
        for account_id, message in cases:
            errs = validate_receive_account_ownership(
                self.db,
                _schema("acc"),
                {"acc": [_snap(account_id)]},
                tenant_no="zkxw",
                initiator_user_id="me",
            )
            self.assertEqual(errs, [message], account_id)

    def _published_app(self):
        now = datetime(2026, 10, 21, 10, 0, 0)
        schema = {"rule": [{"type": "receiveAccount", "field": "acc", "title": "收款账户"}]}
        self.db.add(
            OaApp(
                id="app1",
                app_name="付款",
                initiator_scope="all",
                admin_scope="all_oa_admin",
                status="1",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.add(
            OaAppVersion(
                id="v1",
                app_id="app1",
                version_no=1,
                status="published",
                is_latest_published="1",
                form_schema_json=json.dumps(schema, ensure_ascii=False),
                flow_tree_json="{}",
                advanced_settings_json="{}",
                tenant_no="zkxw",
                create_dept_id="",
                create_time=now,
                update_time=now,
                del_flag=0,
            )
        )
        self.db.commit()

    def test_start_instance_rejects_other_owner_before_flow(self):
        self._published_app()
        with self.assertRaises(OaEngineError) as ctx:
            start_instance(
                self.db,
                tenant_no="zkxw",
                app_id="app1",
                initiator_user_id="me",
                form_data={"acc": [_snap("theirs")]},
            )
        self.assertEqual(str(ctx.exception), "「收款账户」不属于当前用户")


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