"""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'])