From 96468f3c48a147cdb3105578c27099c3141cca6c Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Wed, 21 Oct 2026 18:37:52 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E9=94=81=E5=AE=9A=E6=94=B6=E6=AC=BE?= =?UTF-8?q?=E8=B4=A6=E6=88=B7=E6=8C=89=E7=99=BB=E8=AE=B0=E4=BA=BA=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=E5=92=8C=E5=8F=91=E8=B5=B7=E5=BD=92=E5=B1=9E=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 这些用例先失败，用来约束后续的选项接口和管理页展示。 Co-authored-by: Cursor --- backend/tests/test_oa_receive_account.py | 242 +++++++++++++++++++++++ 1 file changed, 242 insertions(+) diff --git a/backend/tests/test_oa_receive_account.py b/backend/tests/test_oa_receive_account.py index 2d4b4985..1fd0d207 100644 --- a/backend/tests/test_oa_receive_account.py +++ b/backend/tests/test_oa_receive_account.py @@ -1,8 +1,18 @@ +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, ) @@ -100,5 +110,237 @@ class TestValidateReceiveAccountFormValues(unittest.TestCase): 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()