73 lines
4.4 KiB
Python
73 lines
4.4 KiB
Python
from sqlalchemy import select
|
|
|
|
from app import main
|
|
from app.models import AdminUser, RefundPaymentAccount, RefundRecord
|
|
from test_refund_amounts import setup
|
|
|
|
|
|
def test_operator_cannot_grant_or_use_another_account(setup):
|
|
db, client, owner, _, task = setup
|
|
other = RefundPaymentAccount(name="其他账户", alipay_account="other@example.com", single_limit_fen=50000)
|
|
db.add(other)
|
|
db.commit()
|
|
url = f"/api/admin/users/{owner.id}/payment-account"
|
|
assert client.get("/api/admin/users").status_code == 403
|
|
assert client.patch(url, json={"account_id": other.id}).status_code == 403
|
|
available = client.get("/api/admin/refund-payment-accounts/available").json()["data"]
|
|
assert [row["id"] for row in available] == [owner.refund_payment_account_id]
|
|
assert client.post(f"/api/admin/task-board/{task.id}/confirm-refund", data={"payer_account_id": other.id}).status_code == 403
|
|
assert task.status == "已通过"
|
|
assert db.scalar(select(RefundRecord.id)) is None
|
|
|
|
|
|
def test_manager_department_scope_and_revocation(setup):
|
|
db, client, owner, _, _ = setup
|
|
owner.dingtalk_dept_id = 10
|
|
manager = AdminUser(username="manager", display_name="主管", password_hash="x", role_code="dept_manager", dingtalk_dept_id=10)
|
|
child = AdminUser(username="child", display_name="下级部门", password_hash="x", dingtalk_dept_id=11, dingtalk_dept_path=",10,11,")
|
|
outsider = AdminUser(username="outsider", display_name="其他部门", password_hash="x", dingtalk_dept_id=110, dingtalk_dept_path=",110,")
|
|
db.add_all([manager, child, outsider])
|
|
db.commit()
|
|
account_id = owner.refund_payment_account_id
|
|
main.app.dependency_overrides[main.current_user] = lambda: manager
|
|
rows = client.get("/api/admin/users").json()["data"]
|
|
assert {row["id"] for row in rows} == {owner.id, child.id}
|
|
for target in (owner, child):
|
|
assert client.patch(f"/api/admin/users/{target.id}/payment-account", json={"account_id": account_id}).status_code == 200
|
|
for target in (outsider, manager):
|
|
assert client.patch(f"/api/admin/users/{target.id}/payment-account", json={"account_id": account_id}).status_code == 403
|
|
assert client.patch(f"/api/admin/users/{owner.id}", json={"role_code": "super_admin"}).status_code == 403
|
|
assert client.patch(f"/api/admin/users/{owner.id}/payment-account", json={"account_id": None}).status_code == 200
|
|
main.app.dependency_overrides[main.current_user] = lambda: owner
|
|
assert client.get("/api/admin/refund-payment-accounts/available").json()["data"] == []
|
|
manager.dingtalk_dept_id = None
|
|
db.commit()
|
|
main.app.dependency_overrides[main.current_user] = lambda: manager
|
|
assert client.get("/api/admin/users").json()["data"] == []
|
|
assert client.patch(f"/api/admin/users/{owner.id}/payment-account", json={"account_id": account_id}).status_code == 403
|
|
|
|
|
|
def test_admin_grant_validation_and_account_deletion(setup):
|
|
db, client, owner, _, task = setup
|
|
admin = AdminUser(username="admin-test", display_name="管理员", password_hash="x", role_code="super_admin")
|
|
disabled = RefundPaymentAccount(name="停用", alipay_account="off@example.com", is_active=False, single_limit_fen=50000)
|
|
db.add_all([admin, disabled])
|
|
db.commit()
|
|
main.app.dependency_overrides[main.current_user] = lambda: admin
|
|
url = f"/api/admin/users/{owner.id}/payment-account"
|
|
for account_id in (disabled.id, 99999):
|
|
assert client.patch(url, json={"account_id": account_id}).status_code == 400
|
|
for payload in ({}, {"account_id": "1"}, {"account_id": True}, {"account_id": 0}):
|
|
assert client.patch(url, json=payload).status_code == 422
|
|
account_id = owner.refund_payment_account_id
|
|
assert client.patch(url, json={"account_id": None}).status_code == 200
|
|
main.app.dependency_overrides[main.current_user] = lambda: owner
|
|
assert client.post(f"/api/admin/task-board/{task.id}/confirm-refund", data={"payer_account_id": account_id}).status_code == 403
|
|
main.app.dependency_overrides[main.current_user] = lambda: admin
|
|
assert client.patch(url, json={"account_id": account_id}).status_code == 200
|
|
rows = client.get("/api/admin/users").json()["data"]
|
|
assert next(row for row in rows if row["id"] == owner.id)["refund_payment_account_name"] == "公司账户"
|
|
assert client.delete(f"/api/admin/refund-payment-accounts/{account_id}").status_code == 200
|
|
db.refresh(owner)
|
|
assert owner.refund_payment_account_id is None
|