"""文档库权限/目录/分享的数据库集成测试；不连接开发库。""" 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 == []