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()