88 lines
5.0 KiB
Python
88 lines
5.0 KiB
Python
"""配置持久化和鉴权使用独立 SQLite 数据库，不连接业务库。"""
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
from app.api.admin.edu import router as admin_router
|
|
from app.api.site.edu import router as site_router
|
|
from app.core.config import get_settings
|
|
from app.core.security import create_access_token
|
|
from app.db import Base, get_db
|
|
from app.models.content import SiteSetting
|
|
from app.models.rbac import AdminUser, AdminRole, AdminResource, AdminUserRole, AdminRoleResource
|
|
from app.services.association_education import validate_url
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
monkeypatch.setenv('JWT_SECRET', 'test-secret-must-be-32-characters!')
|
|
get_settings.cache_clear()
|
|
engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine, tables=[m.__table__ for m in (SiteSetting, AdminUser, AdminRole, AdminResource, AdminUserRole, AdminRoleResource)])
|
|
with Session(engine) as db:
|
|
db.add_all([AdminUser(id=i, username=f'user{i}', password_hash='unused', display_name='测试', status='disabled' if i == 3 else 'active') for i in (1,2,3,4)])
|
|
db.add_all([AdminRole(id=1, code='editor', name='教育编辑'), AdminRole(id=2, code='reader', name='只读')])
|
|
db.add_all([AdminResource(id=1, type='menu', name='edu', title='继续教育', path='/edu'), AdminResource(id=2, type='button', name='edu:edit', title='编辑', permission='edu:edit')])
|
|
db.flush()
|
|
db.add_all([AdminUserRole(user_id=i, role_id=1) for i in (1,3)])
|
|
db.add_all([AdminRoleResource(role_id=1, resource_id=i) for i in (1,2)])
|
|
db.add(AdminUserRole(user_id=4, role_id=2))
|
|
db.add(AdminRoleResource(role_id=2, resource_id=1))
|
|
db.add(SiteSetting(key='unrelated',value='preserve'))
|
|
db.commit()
|
|
def database():
|
|
with Session(engine) as db:
|
|
yield db
|
|
app=FastAPI()
|
|
app.include_router(admin_router,prefix='/api/admin')
|
|
app.include_router(site_router,prefix='/api/site')
|
|
app.dependency_overrides[get_db]=database
|
|
with TestClient(app) as api:
|
|
yield api
|
|
with Session(engine) as db:
|
|
assert db.query(SiteSetting).filter_by(key='unrelated').one().value == 'preserve'
|
|
engine.dispose()
|
|
get_settings.cache_clear()
|
|
|
|
def headers(user=1,kind='staff'):
|
|
return {'Authorization':f'Bearer {create_access_token(str(user),kind)}'}
|
|
|
|
def test_round_trip_atomic_rejection_and_clear(client):
|
|
public='/api/site/edu/association'
|
|
admin='/api/admin/edu/association'
|
|
assert client.get(public).json()=={'webUrl':'','mobileUrl':''}
|
|
data={'webUrl':'http://127.0.0.1:9800','mobileUrl':'http://127.0.0.1:9801'}
|
|
assert client.put(admin,headers=headers(),json=data).status_code==200
|
|
assert client.get(public).json()==data
|
|
assert client.get(admin,headers=headers()).json()==data
|
|
assert client.put(admin,headers=headers(),json={'webUrl':'https://example.com/new','mobileUrl':'javascript:alert(1)'}).status_code==400
|
|
assert client.get(public).json()==data
|
|
assert client.put(admin,headers=headers(),json={'webUrl':'','mobileUrl':'https://example.com/m'}).status_code==200
|
|
assert client.get(public).json()=={'webUrl':'','mobileUrl':'https://example.com/m'}
|
|
|
|
@pytest.mark.parametrize('auth',[None,(1,'member'),(2,'staff'),(3,'staff'),(99,'staff')])
|
|
def test_authorization(client,auth):
|
|
kw={'headers':headers(*auth)} if auth else {}
|
|
assert client.get('/api/admin/edu/association',**kw).status_code in (401,403)
|
|
assert client.put('/api/admin/edu/association',json={'webUrl':'https://example.com','mobileUrl':''},**kw).status_code in (401,403)
|
|
assert client.get('/api/site/edu/association').json()=={'webUrl':'','mobileUrl':''}
|
|
|
|
@pytest.mark.parametrize('url',['javascript:alert(1)','//example.com','https://','https://a:99999','https://a:bad','https://user:password@example.com','https://a\\@b','https://a\nb','data:text/html,test','https://exa mple.com'])
|
|
def test_invalid_url(url):
|
|
with pytest.raises(ValueError):validate_url(url)
|
|
|
|
@pytest.mark.parametrize('url',['','https://example.com/path?q=中文#section','http://127.0.0.1:9800','https://[::1]:9801'])
|
|
def test_valid_url(url):
|
|
assert validate_url(url)==url
|
|
|
|
|
|
def test_read_only_staff_cannot_write(client):
|
|
assert client.get('/api/admin/edu/association',headers=headers(4)).status_code == 200
|
|
assert client.put('/api/admin/edu/association',headers=headers(4),json={'webUrl':'https://example.com','mobileUrl':''}).status_code == 403
|
|
|
|
@pytest.mark.parametrize('data',[{}, {'webUrl':''}, {'webUrl':'x'*2049,'mobileUrl':''}, {'webUrl':None,'mobileUrl':''}])
|
|
def test_invalid_payload_does_not_change_settings(client,data):
|
|
assert client.put('/api/admin/edu/association',headers=headers(),json=data).status_code == 422
|
|
assert client.get('/api/site/edu/association').json() == {'webUrl':'','mobileUrl':''}
|