136 lines
6.3 KiB
Python
136 lines
6.3 KiB
Python
"""独立现场只读入口：真实菜单多角色并集、数据范围及 HTTP 越权回归。"""
|
|
from datetime import date
|
|
from importlib.util import find_spec
|
|
|
|
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.core.auth import get_current_user
|
|
from app.core.database import get_db
|
|
from app.models.models import SysMenu, SysRole, SysRoleMenu, SysUserRole, SysRoleDataScopeDetail
|
|
from app.models.pl_dispatch_calendar import PlDispatchCalendar
|
|
|
|
PATH = '/api/transport/site-dispatch-calendar'
|
|
ORIGINAL = '/api/transport/dispatch-calendar'
|
|
PERMISSION = 'mini:pl:site-dispatch-calendar:list'
|
|
|
|
|
|
@pytest.fixture
|
|
def setup():
|
|
assert find_spec('app.api.pl.site_dispatch_calendar') is not None, '独立现场只读 API 尚未实现'
|
|
from app.api.pl.site_dispatch_calendar import router
|
|
from app.api.pl.dispatch_calendar import router as original_router
|
|
engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
|
|
for model in (PlDispatchCalendar, SysMenu, SysRole, SysRoleMenu, SysUserRole, SysRoleDataScopeDetail):
|
|
model.__table__.create(engine)
|
|
factory = sessionmaker(bind=engine)
|
|
with factory() as db:
|
|
db.add_all([
|
|
SysRole(id='r1', role_name='基础角色', tenant_no='t1', data_scope='4'),
|
|
SysRole(id='r2', role_name='现场人员', tenant_no='t1', data_scope='4'),
|
|
SysUserRole(id='ur1', user_id='u1', role_id='r1', tenant_no='t1'),
|
|
SysUserRole(id='ur2', user_id='u1', role_id='r2', tenant_no='t1'),
|
|
SysMenu(id='site-menu', menu_name='现场派车日历', parent_id='0', menu_type='C',
|
|
perms=PERMISSION, tenant_no='t1'),
|
|
])
|
|
for rid, kind, tenant, owner, deleted, vehicle in [
|
|
('site-own', 'site', 't1', 'u1', 0, 'v1'),
|
|
('site-other', 'site', 't1', 'u2', 0, 'v2'),
|
|
('site-t2', 'site', 't2', 'u1', 0, 'v1'),
|
|
('site-deleted', 'site', 't1', 'u1', 1, 'v1'),
|
|
('acid', 'acid', 't1', 'u1', 0, 'v1'),
|
|
('alkali', 'alkali', 't1', 'u1', 0, 'v1'),
|
|
('other', 'other', 't1', 'u1', 0, 'v1'),
|
|
]:
|
|
db.add(PlDispatchCalendar(id=rid, calendar_kind=kind, tenant_no=tenant, create_by=owner,
|
|
del_flag=deleted, dispatch_date=date(2026, 10, 30), vehicle_id=vehicle,
|
|
license_plate='鄂E12345', product_id='p1', product_name='液碱', origin='山峡', destination='重庆'))
|
|
db.commit()
|
|
def database():
|
|
with factory() as db:
|
|
yield db
|
|
app = FastAPI()
|
|
app.include_router(router, prefix='/api')
|
|
app.include_router(original_router, prefix='/api')
|
|
app.dependency_overrides[get_db] = database
|
|
with TestClient(app) as client:
|
|
yield app, client, factory
|
|
engine.dispose()
|
|
|
|
|
|
def login(app, **kwargs):
|
|
app.dependency_overrides[get_current_user] = lambda: {
|
|
'role': 'admin', 'user_id': 'u1', 'tenant_no': 't1', 'region_id': 'd1', **kwargs,
|
|
}
|
|
|
|
|
|
def grant(factory):
|
|
with factory() as db:
|
|
db.add(SysRoleMenu(id='grant', role_id='r2', menu_id='site-menu', tenant_no='t1'))
|
|
db.commit()
|
|
|
|
|
|
def test_auth_permission_second_role_and_tenant(setup):
|
|
app, client, factory = setup
|
|
assert client.get(PATH, params={'month': '2026-10'}).status_code == 401
|
|
login(app)
|
|
assert client.get(PATH, params={'month': '2026-10'}).status_code == 403
|
|
grant(factory)
|
|
response = client.get(PATH, params={'month': '2026-10'})
|
|
assert response.status_code == 200, response.text
|
|
assert [r['id'] for r in response.json()['list']] == ['site-own']
|
|
login(app, tenant_no='t2')
|
|
assert client.get(PATH, params={'month': '2026-10'}).status_code == 403
|
|
login(app, role='super_admin')
|
|
assert client.get(PATH, params={'month': '2026-10'}).json()['total'] == 2
|
|
|
|
|
|
def test_fixed_kind_tenant_scope_and_details(setup):
|
|
app, client, factory = setup
|
|
login(app)
|
|
grant(factory)
|
|
response = client.get(PATH, params={'month': '2026-10', 'calendarKind': 'acid', 'tenantNo': 't2',
|
|
'tenant_no': 't2', 'vehicleId': 'v1'})
|
|
assert response.status_code == 200
|
|
assert [r['id'] for r in response.json()['list']] == ['site-own']
|
|
assert client.get(PATH + '/site-own').json()['calendarKind'] == 'site'
|
|
for rid in ('acid', 'alkali', 'other', 'site-t2', 'site-other', 'site-deleted', 'missing', 'vehicles', 'products'):
|
|
assert client.get(PATH + '/' + rid).status_code == 404
|
|
assert client.get(PATH, params={'month': '2026-10', 'vehicleId': 'v2'}).json()['total'] == 0
|
|
with factory() as db:
|
|
db.get(SysRole, 'r1').data_scope = '1'
|
|
db.commit()
|
|
# 菜单授权来自第二角色，全部数据范围来自第一角色，均按并集生效。
|
|
assert client.get(PATH, params={'month': '2026-10'}).json()['total'] == 2
|
|
assert client.get(PATH + '/site-other').status_code == 200
|
|
assert client.get(PATH + '/site-t2').status_code == 404
|
|
page1 = client.get(PATH, params={'month': '2026-10', 'pageSize': 1, 'page': 1}).json()
|
|
page2 = client.get(PATH, params={'month': '2026-10', 'pageSize': 1, 'page': 2}).json()
|
|
assert page1['total'] == page2['total'] == 2
|
|
assert page1['list'][0]['id'] != page2['list'][0]['id']
|
|
|
|
|
|
def test_new_permission_does_not_allow_write_original_or_master_options(setup):
|
|
app, client, factory = setup
|
|
login(app)
|
|
grant(factory)
|
|
for method, path in [('post', PATH), ('put', PATH + '/site-own'), ('delete', PATH + '/site-own')]:
|
|
assert getattr(client, method)(path).status_code == 405
|
|
for method, path in [('get', ORIGINAL), ('get', ORIGINAL + '/acid'), ('get', ORIGINAL + '/vehicles'),
|
|
('get', ORIGINAL + '/products'), ('post', ORIGINAL),
|
|
('put', ORIGINAL + '/site-own'), ('delete', ORIGINAL + '/site-own')]:
|
|
assert getattr(client, method)(path).status_code == 403
|
|
|
|
|
|
@pytest.mark.parametrize('params', [{}, {'month': '2026-13'}, {'month': '2026-10', 'page': 0},
|
|
{'month': '2026-10', 'pageSize': 501}])
|
|
def test_query_validation(setup, params):
|
|
app, client, factory = setup
|
|
login(app)
|
|
grant(factory)
|
|
assert client.get(PATH, params=params).status_code == 422
|