feat(oa): add permission-controlled shared finance remarks to done center
Share one finance remark per approval instance across Web and mobile. Require the explicit permission and the current tenant user's own done task for every read and write; keep general instance responses free of financial remarks. Add idempotent MySQL migration without automatic role grants, cross-end editors, isolated backend and frontend tests, and rollout requirements. Refs #7; platform validation and target database migration remain pending.
This commit is contained in:
zlwu committed
1 parent
86df360f35
commit
0e64b413ce
15 files changed
+1188
-78
No files matched your search
@@ -0,0 +1,195 @@
|
||||
"""OA 财务备注：真实隔离数据库和 HTTP 路由的读写/权限契约。"""
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.api.oa_approval import router
|
||||
from app.core.auth import get_current_user
|
||||
from app.core.database import Base, get_db
|
||||
from app.models.models import (
|
||||
OaApp, OaAppVersion, OaInstance, OaTask,
|
||||
SysUser, SysRole, SysUserRole, SysMenu, SysRoleMenu,
|
||||
SysPost, SysPostRole, SysUserPost,
|
||||
)
|
||||
|
||||
PERM = 'oa:done:finance-remark'
|
||||
PREFIX = '/web/admin/oa-approval'
|
||||
URL = PREFIX + '/instances/i1/finance-remark'
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def case():
|
||||
engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
|
||||
tables = (OaApp, OaAppVersion, OaInstance, OaTask, SysUser, SysRole,
|
||||
SysUserRole, SysMenu, SysRoleMenu, SysPost, SysPostRole, SysUserPost)
|
||||
Base.metadata.create_all(engine, tables=[m.__table__ for m in tables])
|
||||
db = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)()
|
||||
for uid, tenant in [('u1', 'a'), ('u2', 'a'), ('outsider', 'a'), ('other', 'b')]:
|
||||
db.add(SysUser(id=uid, user_no=uid, nick_name=uid, password='unused', tenant_no=tenant, status='1'))
|
||||
db.add_all([
|
||||
OaApp(id='app', app_name='报销', tenant_no='a'),
|
||||
OaAppVersion(id='v1', app_id='app', version_no=1, tenant_no='a'),
|
||||
SysRole(id='finance', role_name='财务', tenant_no='a', status='1'),
|
||||
SysMenu(id='menu', menu_name='财务备注', parent_id='done', menu_type='F', perms=PERM, tenant_no='a', status='1'),
|
||||
SysRoleMenu(id='rm', role_id='finance', menu_id='menu', tenant_no='a'),
|
||||
])
|
||||
for uid in ['u1', 'u2', 'outsider']:
|
||||
db.add(SysUserRole(id='ur-'+uid, user_id=uid, role_id='finance', tenant_no='a'))
|
||||
for iid, tenant in [('i1', 'a'), ('i2', 'a'), ('foreign', 'b')]:
|
||||
inst = OaInstance(id=iid, app_id='app', app_version_id='v1', title=iid,
|
||||
initiator_user_id='u1', tenant_no=tenant, status='completed', form_data_json='{}')
|
||||
inst.finance_remark = '原备注'
|
||||
db.add(inst)
|
||||
for tid, uid, iid, tenant, status in [
|
||||
('t1', 'u1', 'i1', 'a', 'approved'), ('t2', 'u2', 'i1', 'a', 'rejected'),
|
||||
('pending', 'u1', 'i2', 'a', 'pending'), ('foreign-task', 'u1', 'foreign', 'b', 'approved'),
|
||||
]:
|
||||
db.add(OaTask(id=tid, instance_id=iid, assignee_user_id=uid, tenant_no=tenant,
|
||||
task_type='approval', status=status))
|
||||
db.commit()
|
||||
user = {'user_id': 'u1', 'tenant_no': 'a', 'role': 'admin'}
|
||||
app = FastAPI()
|
||||
app.include_router(router)
|
||||
app.dependency_overrides[get_current_user] = lambda: user
|
||||
app.dependency_overrides[get_db] = lambda: db
|
||||
with TestClient(app) as client:
|
||||
yield db, client, user
|
||||
db.close()
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def test_shared_save_reload_and_clear(case):
|
||||
db, client, user = case
|
||||
result = client.get(URL)
|
||||
assert result.status_code == 200
|
||||
assert result.json()['data'] == {'instanceId': 'i1', 'financeRemark': '原备注'}
|
||||
for text in ['财务核对完成\n<标签按原文显示>', '汉' * 512, '']:
|
||||
user['user_id'] = 'u1'
|
||||
response = client.put(URL, json={'financeRemark': text})
|
||||
assert response.status_code == 200
|
||||
assert response.json()['code'] == 200
|
||||
db.expire_all()
|
||||
assert db.get(OaInstance, 'i1').finance_remark == text
|
||||
assert db.get(OaInstance, 'i1').update_by == 'u1'
|
||||
user['user_id'] = 'u2'
|
||||
assert client.get(URL).json()['data']['financeRemark'] == text
|
||||
rows = client.get(PREFIX + '/tasks/done').json()['data']
|
||||
assert rows['canManageFinanceRemark'] is True
|
||||
assert rows['list'][0]['financeRemark'] == text
|
||||
|
||||
|
||||
@pytest.mark.parametrize('role', ['admin', 'super_admin'])
|
||||
def test_no_permission_hides_list_and_denies_both_routes(case, role):
|
||||
db, client, user = case
|
||||
db.get(SysRoleMenu, 'rm').del_flag = 1
|
||||
db.commit()
|
||||
user['role'] = role
|
||||
data = client.get(PREFIX + '/tasks/done').json()['data']
|
||||
assert data['canManageFinanceRemark'] is False
|
||||
assert len(data['list']) == 1
|
||||
assert 'financeRemark' not in data['list'][0]
|
||||
assert client.get(URL).status_code == 403
|
||||
assert client.put(URL, json={'financeRemark': '泄漏'}).status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.parametrize('target', ['i2', 'foreign', 'missing'])
|
||||
def test_only_own_done_instances_are_accessible(case, target):
|
||||
_, client, _ = case
|
||||
url = PREFIX + '/instances/' + target + '/finance-remark'
|
||||
assert client.get(url).status_code == 404
|
||||
assert client.put(url, json={'financeRemark': '越权'}).status_code == 404
|
||||
|
||||
|
||||
def test_authorized_outsider_cannot_read_or_write(case):
|
||||
_, client, user = case
|
||||
user['user_id'] = 'outsider'
|
||||
assert client.get(URL).status_code == 404
|
||||
assert client.put(URL, json={'financeRemark': '越权'}).status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.parametrize('model,key', [(OaInstance, 'i1'), (OaTask, 't1')])
|
||||
def test_deleted_instance_or_done_task_denies_access(case, model, key):
|
||||
db, client, _ = case
|
||||
db.get(model, key).del_flag = 1
|
||||
db.commit()
|
||||
assert client.get(URL).status_code == 404
|
||||
assert client.put(URL, json={'financeRemark': '越权'}).status_code == 404
|
||||
data = client.get(PREFIX + '/tasks/done').json()['data']
|
||||
assert all('financeRemark' not in row for row in data['list'])
|
||||
|
||||
|
||||
@pytest.mark.parametrize('status', ['approved', 'rejected', 'returned', 'transferred'])
|
||||
def test_all_existing_done_statuses_can_manage(case, status):
|
||||
db, client, _ = case
|
||||
db.get(OaTask, 't1').status = status
|
||||
db.commit()
|
||||
assert client.get(URL).status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.parametrize('status', ['pending', 'done', 'cancelled'])
|
||||
def test_non_done_statuses_cannot_manage(case, status):
|
||||
db, client, _ = case
|
||||
db.get(OaTask, 't1').status = status
|
||||
db.commit()
|
||||
assert client.get(URL).status_code == 404
|
||||
|
||||
|
||||
def test_forged_tenant_uses_real_user_tenant(case):
|
||||
_, client, user = case
|
||||
user['tenant_no'] = 'b'
|
||||
assert client.get(URL).json()['data']['financeRemark'] == '原备注'
|
||||
data = client.get(PREFIX + '/tasks/done').json()['data']
|
||||
assert [r['instanceId'] for r in data['list']] == ['i1']
|
||||
assert client.get(PREFIX + '/instances/foreign/finance-remark').status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.parametrize('model,key,field,value', [
|
||||
(SysUser, 'u1', 'status', '0'), (SysUser, 'u1', 'del_flag', 1),
|
||||
(SysRole, 'finance', 'status', '0'), (SysRole, 'finance', 'tenant_no', 'b'),
|
||||
(SysMenu, 'menu', 'status', '0'), (SysMenu, 'menu', 'tenant_no', 'b'),
|
||||
(SysUserRole, 'ur-u1', 'tenant_no', 'b'),
|
||||
])
|
||||
def test_inactive_or_cross_tenant_grants_deny_access(case, model, key, field, value):
|
||||
db, client, _ = case
|
||||
setattr(db.get(model, key), field, value)
|
||||
db.commit()
|
||||
assert client.get(URL).status_code == 403
|
||||
assert client.put(URL, json={'financeRemark': '越权'}).status_code == 403
|
||||
|
||||
|
||||
def test_post_role_permission_union(case):
|
||||
db, client, _ = case
|
||||
db.get(SysUserRole, 'ur-u1').del_flag = 1
|
||||
db.add_all([
|
||||
SysPost(id='post', post_name='财务岗', post_code='finance', tenant_no='a', status='1'),
|
||||
SysUserPost(id='up', user_id='u1', post_id='post', tenant_no='a'),
|
||||
SysPostRole(id='pr', post_id='post', role_id='finance', tenant_no='a'),
|
||||
])
|
||||
db.commit()
|
||||
assert client.get(URL).status_code == 200
|
||||
db.get(SysPost, 'post').status = '0'
|
||||
db.commit()
|
||||
assert client.get(URL).status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.parametrize('body', [
|
||||
{}, {'financeRemark': None}, {'financeRemark': 123}, {'financeRemark': 'a' * 513},
|
||||
{'financeRemark': 'note', 'tenantNo': 'b'}, {'financeRemark': 'note', 'status': 'running'},
|
||||
])
|
||||
def test_invalid_body_rejected_without_writes(case, body):
|
||||
db, client, _ = case
|
||||
assert client.put(URL, json=body).status_code == 422
|
||||
db.expire_all()
|
||||
assert db.get(OaInstance, 'i1').finance_remark == '原备注'
|
||||
|
||||
|
||||
def test_generic_instance_serialization_does_not_expose_finance_remark(case):
|
||||
db, client, _ = case
|
||||
from app.services.oa_engine import _instance_to_dict
|
||||
assert 'financeRemark' not in _instance_to_dict(db.get(OaInstance, 'i1'))
|
||||
for kind in ['mine', 'pending']:
|
||||
data = client.get(PREFIX + '/tasks/' + kind).json()['data']
|
||||
assert all('financeRemark' not in row for row in data['list'])
|
||||
Reference in new issue
Block a user