165 lines
7.0 KiB
Python
165 lines
7.0 KiB
Python
"""文档库权限/目录/分享的数据库集成测试；不连接开发库。"""
|
|
import io
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.models.models import SysDept, SysUser
|
|
from app.models.doclib import DoclibFolder, DoclibFile, DoclibShare
|
|
from app.services.doclib_service import DoclibService
|
|
|
|
|
|
@pytest.fixture
|
|
def db():
|
|
engine = create_engine('sqlite://', poolclass=StaticPool, connect_args={'check_same_thread': False})
|
|
for table in [SysDept.__table__, SysUser.__table__, DoclibFolder.__table__, DoclibFile.__table__, DoclibShare.__table__]:
|
|
table.create(engine)
|
|
session = sessionmaker(bind=engine, expire_on_commit=False)()
|
|
for uid, tenant in [('owner', 'a'), ('reader', 'a'), ('stranger', 'a'), ('admin', 'a'), ('outsider', 'b')]:
|
|
session.add(SysUser(id=uid, tenant_no=tenant, user_no=uid, nick_name=uid, status='1'))
|
|
session.commit()
|
|
yield session
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def svc(db, uid='owner', tenant='a', role='admin'):
|
|
return DoclibService(db, {'user_id': uid, 'tenant_no': tenant, 'role': role, 'region_id': 'dept'})
|
|
|
|
|
|
def seed_file(db, service=None, folder_id='', name='制度.pdf'):
|
|
service = service or svc(db)
|
|
return service.create_file(name=name, folder_id=folder_id, size=12, content_type='application/pdf', object_key='opaque/'+name)
|
|
|
|
|
|
def denied(fn, status=404):
|
|
with pytest.raises(HTTPException) as exc:
|
|
fn()
|
|
assert exc.value.status_code == status
|
|
|
|
|
|
def test_private_by_default_and_no_storage_url(db):
|
|
doc = seed_file(db)
|
|
assert len(svc(db).list_files()['list']) == 1
|
|
assert svc(db, 'stranger').list_files(scope='shared')['total'] == 0
|
|
denied(lambda: svc(db, 'stranger').get_file(doc['id']))
|
|
assert 'objectKey' not in doc and 'url' not in doc
|
|
assert doc['canManage'] is True and doc['shares'] == []
|
|
|
|
|
|
def test_share_receiver_read_only_and_revocation(db):
|
|
doc = seed_file(db)
|
|
svc(db).set_shares(doc['id'], ['reader'])
|
|
reader = svc(db, 'reader')
|
|
assert reader.list_files(scope='shared')['total'] == 1
|
|
assert reader.get_file(doc['id'])['canManage'] is False
|
|
denied(lambda: reader.set_shares(doc['id'], ['stranger']), 403)
|
|
denied(lambda: reader.delete_files([doc['id']]), 403)
|
|
svc(db).set_shares(doc['id'], [])
|
|
assert reader.list_files(scope='shared')['total'] == 0
|
|
denied(lambda: reader.download_record(doc['id']))
|
|
|
|
|
|
def test_admin_only_inside_current_tenant(db):
|
|
local = seed_file(db)
|
|
remote = seed_file(db, svc(db, 'outsider', 'b'))
|
|
admin = svc(db, 'admin', role='super_admin')
|
|
assert admin.list_files(scope='all')['total'] == 1
|
|
assert admin.get_file(local['id'])['canManage'] is True
|
|
denied(lambda: admin.get_file(remote['id']))
|
|
denied(lambda: svc(db).list_files(scope='all'), 403)
|
|
admin.delete_files([local['id']])
|
|
assert db.get(DoclibFile, local['id']).del_flag == 1
|
|
denied(lambda: svc(db).get_file(local['id']))
|
|
|
|
|
|
def test_cross_tenant_and_inactive_share_is_atomic(db):
|
|
doc = seed_file(db)
|
|
svc(db).set_shares(doc['id'], ['reader'])
|
|
denied(lambda: svc(db).set_shares(doc['id'], ['stranger', 'outsider']), 400)
|
|
assert [u['id'] for u in svc(db).get_file(doc['id'])['shares']] == ['reader']
|
|
db.get(SysUser, 'stranger').status = '0'
|
|
db.commit()
|
|
denied(lambda: svc(db).set_shares(doc['id'], ['stranger']), 400)
|
|
assert {u['id'] for u in svc(db).users()} == {'reader', 'admin'}
|
|
|
|
|
|
def test_nested_folder_tree_shared_only_relevant_path(db):
|
|
service = svc(db)
|
|
parent = service.create_folder('公司制度', '')
|
|
child = service.create_folder('人事', parent['folderId'])
|
|
hidden = service.create_folder('财务', parent['folderId'])
|
|
doc = seed_file(db, folder_id=child['folderId'])
|
|
seed_file(db, folder_id=hidden['folderId'], name='预算.xlsx')
|
|
service.set_shares(doc['id'], ['reader'])
|
|
reader = svc(db, 'reader')
|
|
shared = next(n for n in reader.tree()['nodes'] if n['scope'] == 'shared')
|
|
assert shared['count'] == 1
|
|
assert [n['name'] for n in shared['children']] == ['公司制度']
|
|
assert [n['name'] for n in shared['children'][0]['children']] == ['人事']
|
|
assert reader.get_file(doc['id'])['folderPath'] == '公司制度 / 人事'
|
|
assert reader.list_files(scope='shared', folder_id=hidden['folderId'])['total'] == 0
|
|
denied(lambda: reader.create_folder('越权子目录', parent['folderId']), 403)
|
|
denied(lambda: reader.create_file(name='bad.txt', folder_id=child['folderId'], size=1, content_type='text/plain', object_key='x'), 403)
|
|
|
|
|
|
def test_folder_rename_empty_delete_and_duplicates(db):
|
|
service = svc(db)
|
|
folder = service.create_folder('资料', '')
|
|
fid = folder['folderId']
|
|
denied(lambda: service.create_folder('资料', ''), 409)
|
|
assert service.rename_folder(fid, '新资料')['fullPath'] == '新资料'
|
|
doc = seed_file(db, folder_id=fid)
|
|
denied(lambda: service.delete_folder(fid), 409)
|
|
service.delete_files([doc['id']])
|
|
service.delete_folder(fid)
|
|
assert db.get(DoclibFolder, fid).del_flag == 1
|
|
assert service.create_folder('新资料', '')['folderId'] != fid
|
|
|
|
|
|
def test_parent_cannot_be_deleted_with_child(db):
|
|
service = svc(db)
|
|
f = service.create_folder('父目录', '')
|
|
service.create_folder('子目录', f['folderId'])
|
|
denied(lambda: service.delete_folder(f['folderId']), 409)
|
|
|
|
|
|
def test_batch_delete_denied_is_all_or_nothing(db):
|
|
mine = seed_file(db)
|
|
other = seed_file(db, svc(db, 'reader'), name='other.txt')
|
|
svc(db, 'reader').set_shares(other['id'], ['owner'])
|
|
denied(lambda: svc(db).delete_files([mine['id'], other['id']]), 403)
|
|
assert svc(db).get_file(mine['id'])['name'] == '制度.pdf'
|
|
|
|
|
|
def test_paging_search_sort_and_soft_deleted_shares(db):
|
|
service = svc(db)
|
|
old = seed_file(db, name='旧制度.pdf')
|
|
new = seed_file(db, name='新制度.pdf')
|
|
db.get(DoclibFile, old['id']).create_time = datetime(2025, 1, 1)
|
|
db.commit()
|
|
page = service.list_files(keyword='制度', page=1, page_size=1)
|
|
assert page['total'] == 2 and page['list'][0]['id'] == new['id']
|
|
service.set_shares(new['id'], ['reader', 'reader'])
|
|
assert len(service.get_file(new['id'])['shares']) == 1
|
|
service.set_shares(new['id'], [])
|
|
assert db.query(DoclibShare).filter_by(file_id=new['id'], del_flag=0).count() == 0
|
|
service.set_shares(new['id'], ['reader'])
|
|
assert svc(db, 'reader').list_files(scope='shared')['total'] == 1
|
|
|
|
|
|
def test_schema_rejects_client_tenant_and_invalid_names():
|
|
from pydantic import ValidationError
|
|
from app.schemas.doclib import FolderCreateBody, FolderUpdateBody, ShareBody
|
|
with pytest.raises(ValidationError):
|
|
FolderCreateBody(name='资料', tenant_no='b')
|
|
for name in ['', ' ', '../secret', 'A/B', 'A\\B']:
|
|
with pytest.raises(ValidationError):
|
|
FolderUpdateBody(name=name)
|
|
assert FolderCreateBody(name=' 资料 ').name == '资料'
|
|
assert ShareBody(userIds=[]).userIds == []
|