shanxia2.0/backend/tests/test_doclib.py

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