347 lines
12 KiB
Python
347 lines
12 KiB
Python
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()
|