test: 锁定收款账户按登记人过滤和发起归属校验
这些用例先失败，用来约束后续的选项接口和管理页展示。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed
1 parent
11a46bf1eb
commit
96468f3c48
1 file changed
+242
@@ -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()
|
||||
Reference in new issue
Block a user