import hashlib import io import uuid import zipfile from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import ANY, AsyncMock import pytest from fastapi import HTTPException from jose import jwt from app.core.config import settings from app.models.collaboration import ( CollaborationFile, CollaborationFolder, CollaborationRevision, CollaborationShareLink, ) from app.schemas.collaboration import ( CollaborationCallbackPayload, CollaborationFileUpdate, CollaborationRevisionCopyRequest, CollaborationRevisionUpdate, CollaborationShareLinkUpdate, ) from app.services import collaboration_service, collaboration_share_service, onlyoffice_collaboration_service, onlyoffice_service @pytest.fixture(autouse=True) def collaboration_settings(monkeypatch): monkeypatch.setattr(settings, "ONLYOFFICE_ENABLED", True) monkeypatch.setattr(settings, "ONLYOFFICE_JWT_SECRET", "collaboration-secret-that-is-long-enough") monkeypatch.setattr(settings, "ONLYOFFICE_INSTANCE_ID", "ctms-collaboration-test") monkeypatch.setattr(settings, "ONLYOFFICE_INTERNAL_URL", "http://onlyoffice") monkeypatch.setattr(settings, "ONLYOFFICE_STORAGE_BASE_URL", "http://backend:8000") monkeypatch.setattr(settings, "ONLYOFFICE_CONFIG_TTL_SECONDS", 300) @pytest.mark.parametrize( ("file_type", "expected_extension", "required_part"), [ ("word", "docx", "word/document.xml"), ("cell", "xlsx", "xl/worksheets/sheet1.xml"), ("slide", "pptx", "ppt/slides/slide1.xml"), ], ) def test_blank_collaboration_templates_are_ooxml_packages(file_type, expected_extension, required_part): content = collaboration_service.blank_file_bytes(file_type) assert zipfile.is_zipfile(io.BytesIO(content)) with zipfile.ZipFile(io.BytesIO(content)) as package: assert "[Content_Types].xml" in package.namelist() assert required_part in package.namelist() assert collaboration_service.FILE_TYPE_EXTENSION[file_type] == expected_extension def test_collaboration_key_is_stable_per_generation_and_is_independent_from_document_versions(): file_id = uuid.uuid4() first = onlyoffice_collaboration_service.collaboration_document_key(file_id, 1) repeated = onlyoffice_collaboration_service.collaboration_document_key(file_id, 1) next_generation = onlyoffice_collaboration_service.collaboration_document_key(file_id, 2) assert first == repeated assert first != next_generation assert first.startswith("ctms-collab-") assert len(first) <= 128 def test_collaboration_revision_preview_config_is_read_only(): revision_id = uuid.uuid4() config = onlyoffice_service.build_preview_config( resource_type="collaboration_revision", resource_id=revision_id, file_name="方案-R2.docx", file_hash="revision-hash", user_id=uuid.uuid4(), user_name="测试用户", ) assert config.resource_type == "collaboration_revision" assert config.config["editorConfig"]["mode"] == "view" assert config.config["document"]["permissions"]["edit"] is False assert f"/internal/onlyoffice/collaboration-revisions/{revision_id}/content" in config.config["document"]["url"] def test_revision_action_payloads_trim_user_input(): assert CollaborationRevisionUpdate(change_summary=" 伦理送审版 ").change_summary == "伦理送审版" folder_id = uuid.uuid4() payload = CollaborationRevisionCopyRequest(title=" 方案-R2.docx ", folder_id=folder_id) assert payload.title == "方案-R2.docx" assert payload.folder_id == folder_id @pytest.mark.asyncio async def test_copy_revision_saves_to_the_selected_workspace_folder(monkeypatch, tmp_path): source = tmp_path / "revision.docx" content = collaboration_service.blank_file_bytes("word") source.write_bytes(content) revision_id = uuid.uuid4() target_folder_id = uuid.uuid4() item = SimpleNamespace( id=uuid.uuid4(), study_id=uuid.uuid4(), folder_id=uuid.uuid4(), file_type="word", current_revision_id=uuid.uuid4(), ) revision = SimpleNamespace( id=revision_id, file_id=item.id, revision_no=2, file_uri=str(source), deleted_at=None, ) copied = SimpleNamespace(id=uuid.uuid4()) db = SimpleNamespace(get=AsyncMock(return_value=revision)) require_exporter = AsyncMock() folder_lookup = AsyncMock() create_copy = AsyncMock(return_value=copied) monkeypatch.setattr(collaboration_service, "require_file_exporter", require_exporter) monkeypatch.setattr(collaboration_service, "_folder_or_404", folder_lookup) monkeypatch.setattr(collaboration_service, "_create_file_from_bytes", create_copy) user = SimpleNamespace(id=uuid.uuid4()) result = await collaboration_service.copy_revision( db, item, revision_id, CollaborationRevisionCopyRequest(title="送审版.docx", folder_id=target_folder_id), user, ) assert result is copied require_exporter.assert_awaited_once_with(db, item, user) folder_lookup.assert_awaited_once_with(db, item.study_id, target_folder_id) create_copy.assert_awaited_once_with( db, item.study_id, title="送审版.docx", file_type="word", folder_id=target_folder_id, content=content, source="COPY", user=user, ) @pytest.mark.asyncio async def test_delete_revision_soft_deletes_a_non_current_version(monkeypatch): revision_id = uuid.uuid4() current_revision_id = uuid.uuid4() user = SimpleNamespace(id=uuid.uuid4()) item = SimpleNamespace(id=uuid.uuid4(), current_revision_id=current_revision_id) revision = SimpleNamespace( id=revision_id, file_id=item.id, revision_no=2, deleted_at=None, deleted_by=None, ) db = SimpleNamespace(get=AsyncMock(return_value=revision), commit=AsyncMock()) require_manager = AsyncMock() monkeypatch.setattr(collaboration_service, "require_file_manager", require_manager) await collaboration_service.delete_revision(db, item, revision_id, user) require_manager.assert_awaited_once_with(db, item, user) assert revision.deleted_at is not None assert revision.deleted_by == user.id db.commit.assert_awaited_once() @pytest.mark.asyncio async def test_delete_revision_rejects_the_current_version(monkeypatch): revision_id = uuid.uuid4() user = SimpleNamespace(id=uuid.uuid4()) item = SimpleNamespace(id=uuid.uuid4(), current_revision_id=revision_id) revision = SimpleNamespace( id=revision_id, file_id=item.id, revision_no=4, deleted_at=None, deleted_by=None, ) db = SimpleNamespace(get=AsyncMock(return_value=revision), commit=AsyncMock()) monkeypatch.setattr(collaboration_service, "require_file_manager", AsyncMock()) with pytest.raises(HTTPException) as exc: await collaboration_service.delete_revision(db, item, revision_id, user) assert exc.value.status_code == 409 db.commit.assert_not_awaited() @pytest.mark.asyncio async def test_revision_history_includes_creator_display_metadata(): file_id = uuid.uuid4() creator_id = uuid.uuid4() now = datetime.now(timezone.utc) revision = SimpleNamespace( id=uuid.uuid4(), file_id=file_id, revision_no=3, parent_revision_id=None, original_filename="方案.docx", file_hash="hash", file_size=2048, mime_type="application/docx", source="SESSION_CLOSE", change_summary="自动保存", created_by=creator_id, created_at=now, ) creator = SimpleNamespace(full_name="测试用户", avatar_url="/avatars/test.png") result = SimpleNamespace(all=lambda: [(revision, creator)]) db = SimpleNamespace(execute=AsyncMock(return_value=result)) history = await collaboration_service.list_revisions(db, SimpleNamespace(id=file_id)) assert len(history) == 1 assert history[0].created_by_name == "测试用户" assert history[0].created_by_avatar_url == "/avatars/test.png" statement = db.execute.await_args.args[0] assert "collaboration_revisions.source !=" in str(statement) @pytest.mark.asyncio async def test_sheet_permission_change_updates_current_file_without_creating_revision(monkeypatch, tmp_path): original = collaboration_service.blank_file_bytes("cell") revision_path = tmp_path / "current.xlsx" revision_path.write_bytes(original) revision = SimpleNamespace( id=uuid.uuid4(), file_uri=str(revision_path), file_hash="old-hash", file_size=len(original), ) item = SimpleNamespace( id=uuid.uuid4(), study_id=uuid.uuid4(), file_type="cell", extension="xlsx", title="协作表.xlsx", folder_id=None, current_revision_id=revision.id, generation=1, allow_export=True, allow_edit_request=False, allow_sheet_structure_edit=True, sheet_structure_protection_backup=None, ) user = SimpleNamespace(id=uuid.uuid4()) db = SimpleNamespace(get=AsyncMock(return_value=revision), commit=AsyncMock(), refresh=AsyncMock()) monkeypatch.setattr(collaboration_service, "require_file_manager", AsyncMock()) persist_revision = AsyncMock() monkeypatch.setattr(collaboration_service, "_persist_revision_bytes", persist_revision) result = await collaboration_service.update_file( db, item, CollaborationFileUpdate(allow_sheet_structure_edit=False), user, ) assert result is item assert item.allow_sheet_structure_edit is False assert item.generation == 2 persist_revision.assert_not_awaited() assert revision.file_hash == hashlib.sha256(revision_path.read_bytes()).hexdigest() assert revision.file_size == revision_path.stat().st_size with zipfile.ZipFile(revision_path) as package: workbook = package.read("xl/workbook.xml").decode("utf-8") assert "workbookProtection" in workbook def test_callback_token_is_bound_to_key_status_and_result_url(): result_url = "http://onlyoffice/cache/result.docx" payload = CollaborationCallbackPayload(key="collaboration-key", status=2, url=result_url) token = jwt.encode( {"payload": {"key": payload.key, "status": payload.status, "url": payload.url}}, settings.ONLYOFFICE_JWT_SECRET, algorithm="HS256", ) decoded = onlyoffice_collaboration_service.validate_callback_token(token, payload) assert decoded["payload"]["key"] == payload.key forged = payload.model_copy(update={"url": "http://onlyoffice/cache/other.docx"}) with pytest.raises(HTTPException) as mismatch: onlyoffice_collaboration_service.validate_callback_token(token, forged) assert mismatch.value.status_code == 401 config_token = jwt.encode( {"document": {"key": payload.key}, "editorConfig": {"mode": "edit"}}, settings.ONLYOFFICE_JWT_SECRET, algorithm="HS256", ) with pytest.raises(HTTPException): onlyoffice_collaboration_service.validate_callback_token(config_token, payload) def test_result_download_url_is_restricted_and_public_proxy_urls_are_rewritten(monkeypatch): monkeypatch.setattr(settings, "FRONTEND_PUBLIC_URL", "http://localhost:8888") assert onlyoffice_collaboration_service._validate_result_url( "http://onlyoffice/cache/result.docx?token=signed" ) == "http://onlyoffice/cache/result.docx?token=signed" assert onlyoffice_collaboration_service._validate_result_url( "http://localhost:8888/onlyoffice/cache/result.docx?token=signed" ) == "http://onlyoffice/cache/result.docx?token=signed" for value in ( "http://backend:8000/internal/file", "http://onlyoffice.evil.example/cache/result.docx", "http://user:password@onlyoffice/cache/result.docx", "http://onlyoffice:8080/cache/result.docx", "http://localhost:8888/api/v1/files/result.docx", "http://localhost.evil.example:8888/onlyoffice/cache/result.docx", ): with pytest.raises(HTTPException): onlyoffice_collaboration_service._validate_result_url(value) @pytest.mark.asyncio async def test_editor_config_grants_edit_only_after_collaboration_permission(monkeypatch, tmp_path): user_id = uuid.uuid4() file_id = uuid.uuid4() revision_id = uuid.uuid4() source = tmp_path / "协作表.xlsx" source.write_bytes(collaboration_service.blank_file_bytes("cell")) item = CollaborationFile( id=file_id, study_id=uuid.uuid4(), title="协作表.xlsx", file_type="cell", extension="xlsx", owner_id=user_id, current_revision_id=revision_id, generation=1, allow_export=False, ) revision = SimpleNamespace(id=revision_id, file_uri=str(source)) session = SimpleNamespace(id=uuid.uuid4(), document_key="ctms-collab-key") db = SimpleNamespace(get=AsyncMock(return_value=revision)) user = SimpleNamespace(id=user_id, full_name="测试用户") monkeypatch.setattr(onlyoffice_collaboration_service.onlyoffice_service, "ensure_onlyoffice_available", AsyncMock()) monkeypatch.setattr(onlyoffice_collaboration_service.collaboration_service, "can_edit_file", AsyncMock(return_value=True)) can_request_edit = AsyncMock(return_value=False) monkeypatch.setattr( onlyoffice_collaboration_service.collaboration_service, "can_request_edit_file", can_request_edit, ) can_export = AsyncMock(return_value=True) monkeypatch.setattr(onlyoffice_collaboration_service.collaboration_service, "can_export_file", can_export) can_create = AsyncMock(return_value=True) monkeypatch.setattr(onlyoffice_collaboration_service.collaboration_service, "can_create_file", can_create) monkeypatch.setattr(onlyoffice_collaboration_service, "_active_session", AsyncMock(return_value=session)) result = await onlyoffice_collaboration_service.build_editor_config(db, item, user) assert result.access_mode == "edit" assert result.can_save_as is True assert result.can_download is True assert result.config["editorConfig"]["mode"] == "edit" assert result.config["editorConfig"]["coEditing"] == {"mode": "fast", "change": False} assert result.config["document"]["permissions"]["edit"] is True assert result.config["document"]["permissions"]["download"] is True assert result.config["document"]["permissions"]["print"] is True assert "/internal/onlyoffice/collaboration/sessions/" in result.config["document"]["url"] assert "callbackUrl" in result.config["editorConfig"] onlyoffice_collaboration_service.collaboration_service.can_edit_file.return_value = False can_request_edit.return_value = True requestable = await onlyoffice_collaboration_service.build_editor_config(db, item, user) assert requestable.access_mode == "view" assert requestable.can_request_edit is True assert requestable.config["editorConfig"]["mode"] == "view" assert requestable.config["document"]["permissions"]["edit"] is True onlyoffice_collaboration_service.collaboration_service.can_edit_file.return_value = True can_request_edit.return_value = False can_create.return_value = False restricted = await onlyoffice_collaboration_service.build_editor_config(db, item, user) assert restricted.can_save_as is False assert restricted.can_download is True assert restricted.config["document"]["permissions"]["download"] is True can_export.return_value = False export_restricted = await onlyoffice_collaboration_service.build_editor_config(db, item, user) assert export_restricted.can_save_as is False assert export_restricted.can_download is False assert export_restricted.config["document"]["permissions"]["download"] is False @pytest.mark.asyncio async def test_shared_editor_config_respects_link_mode_and_disables_export(monkeypatch, tmp_path): owner_id = uuid.uuid4() revision_id = uuid.uuid4() source = tmp_path / "共享表.xlsx" source.write_bytes(collaboration_service.blank_file_bytes("cell")) item = CollaborationFile( id=uuid.uuid4(), study_id=uuid.uuid4(), title="共享表.xlsx", file_type="cell", extension="xlsx", owner_id=owner_id, current_revision_id=revision_id, generation=1, allow_export=False, ) link = CollaborationShareLink( id=uuid.uuid4(), file_id=item.id, enabled=True, access_mode="EDIT", expiry_policy="SEVEN_DAYS", expires_at=datetime.now(timezone.utc) + timedelta(days=7), created_by=owner_id, updated_by=owner_id, ) revision = SimpleNamespace(id=revision_id, file_uri=str(source)) session = SimpleNamespace(id=uuid.uuid4(), document_key="ctms-public-share-key") db = SimpleNamespace(get=AsyncMock(return_value=revision)) monkeypatch.setattr(onlyoffice_collaboration_service.onlyoffice_service, "ensure_onlyoffice_available", AsyncMock()) monkeypatch.setattr(onlyoffice_collaboration_service, "_active_session", AsyncMock(return_value=session)) result = await onlyoffice_collaboration_service.build_shared_editor_config( db, item, link, client_id="visitor-client-1", display_name="链接访客", ) assert result.access_mode == "edit" assert result.can_save_as is False assert result.can_download is False assert result.config["document"]["permissions"]["edit"] is True assert result.config["document"]["permissions"]["copy"] is False assert result.config["document"]["permissions"]["download"] is False assert result.config["document"]["permissions"]["print"] is False assert result.config["editorConfig"]["user"]["id"].startswith("share-") item.allow_export = True export_result = await onlyoffice_collaboration_service.build_shared_editor_config( db, item, link, client_id="visitor-client-2", display_name="链接访客", ) assert export_result.can_save_as is False assert export_result.can_download is True assert export_result.config["document"]["permissions"]["copy"] is True assert export_result.config["document"]["permissions"]["download"] is True assert export_result.config["document"]["permissions"]["print"] is True @pytest.mark.asyncio async def test_share_tokens_are_stable_revocable_and_not_accepted_after_expiry(monkeypatch): monkeypatch.setattr(settings, "JWT_SECRET_KEY", "share-token-test-secret") now = datetime.now(timezone.utc) link = CollaborationShareLink( id=uuid.uuid4(), file_id=uuid.uuid4(), enabled=True, access_mode="VIEW", expiry_policy="SEVEN_DAYS", expires_at=now + timedelta(days=7), token_version=2, created_by=uuid.uuid4(), updated_by=uuid.uuid4(), created_at=now, updated_at=now, ) item = SimpleNamespace( id=link.file_id, title="方案.docx", file_type="word", status="ACTIVE", allow_export=False, deleted_at=None, ) db = SimpleNamespace(scalar=AsyncMock(return_value=link), get=AsyncMock(return_value=item)) token = collaboration_share_service.share_token(link) repeated = collaboration_share_service.share_token(link) metadata = await collaboration_share_service.public_metadata(db, token) assert token == repeated assert metadata.file_name == "方案.docx" assert metadata.access_mode == "view" assert metadata.allow_export is False with pytest.raises(HTTPException) as tampered: replacement = "A" if token[-1] != "A" else "B" await collaboration_share_service.public_metadata(db, f"{token[:-1]}{replacement}") assert tampered.value.status_code == 404 link.token_version += 1 with pytest.raises(HTTPException) as revoked: await collaboration_share_service.public_metadata(db, token) assert revoked.value.status_code == 404 current = collaboration_share_service.share_token(link) link.expires_at = now - timedelta(seconds=1) with pytest.raises(HTTPException) as expired: await collaboration_share_service.public_metadata(db, current) assert expired.value.status_code == 410 @pytest.mark.asyncio async def test_share_link_url_stays_immutable_when_disabled_and_reenabled(monkeypatch): monkeypatch.setattr(settings, "JWT_SECRET_KEY", "immutable-share-link-test-secret") now = datetime.now(timezone.utc) link = CollaborationShareLink( id=uuid.uuid4(), file_id=uuid.uuid4(), enabled=True, access_mode="VIEW", expiry_policy="PERMANENT", expires_at=None, token_version=4, created_by=uuid.uuid4(), updated_by=uuid.uuid4(), created_at=now, updated_at=now, ) item = SimpleNamespace(id=link.file_id, allow_export=False) user = SimpleNamespace(id=uuid.uuid4()) db = SimpleNamespace( scalar=AsyncMock(return_value=link), commit=AsyncMock(), refresh=AsyncMock(), ) monkeypatch.setattr(collaboration_service, "require_file_manager", AsyncMock()) original_token = collaboration_share_service.share_token(link) await collaboration_share_service.update_share_link( db, item, CollaborationShareLinkUpdate(enabled=False, expiry_policy="PERMANENT"), user, ) assert link.enabled is False assert link.token_version == 4 assert collaboration_share_service.share_token(link) == original_token await collaboration_share_service.update_share_link( db, item, CollaborationShareLinkUpdate( enabled=True, access_mode="EDIT", expiry_policy="THIRTY_DAYS", ), user, ) assert link.enabled is True assert link.access_mode == "EDIT" assert item.allow_export is False assert link.token_version == 4 assert collaboration_share_service.share_token(link) == original_token @pytest.mark.asyncio async def test_share_password_failures_lock_the_link_and_grants_are_version_bound(monkeypatch): monkeypatch.setattr(settings, "JWT_SECRET_KEY", "share-password-test-secret") now = datetime.now(timezone.utc) link = CollaborationShareLink( id=uuid.uuid4(), file_id=uuid.uuid4(), enabled=True, access_mode="VIEW", expiry_policy="PERMANENT", password_hash="hashed-password", token_version=3, failed_attempts=0, created_by=uuid.uuid4(), updated_by=uuid.uuid4(), created_at=now, updated_at=now, ) item = SimpleNamespace(id=link.file_id, status="ACTIVE", deleted_at=None) db = SimpleNamespace( scalar=AsyncMock(return_value=link), get=AsyncMock(return_value=item), commit=AsyncMock(), ) monkeypatch.setattr(collaboration_share_service, "verify_password", lambda *_: False) token = collaboration_share_service.share_token(link) for _ in range(collaboration_share_service.PASSWORD_FAILURE_LIMIT - 1): with pytest.raises(HTTPException) as wrong: await collaboration_share_service.verify_share_password(db, token, "wrong") assert wrong.value.status_code == 401 with pytest.raises(HTTPException) as locked: await collaboration_share_service.verify_share_password(db, token, "wrong") assert locked.value.status_code == 429 assert link.locked_until is not None link.locked_until = None link.failed_attempts = 0 monkeypatch.setattr(collaboration_share_service, "verify_password", lambda *_: True) grant = await collaboration_share_service.verify_share_password(db, token, "correct") collaboration_share_service.validate_access_grant(link, grant.access_token) link.token_version += 1 with pytest.raises(HTTPException) as stale: collaboration_share_service.validate_access_grant(link, grant.access_token) assert stale.value.status_code == 401 @pytest.mark.asyncio async def test_public_share_callback_is_not_attributed_to_the_file_owner(): owner_id = uuid.uuid4() db = SimpleNamespace(get=AsyncMock()) session = SimpleNamespace(started_by=owner_id) payload = CollaborationCallbackPayload( key="shared-key", status=6, users=["share-0123456789ab-visitor-client"], ) actor = await onlyoffice_collaboration_service._callback_user(db, payload, session) assert actor is None db.get.assert_not_awaited() @pytest.mark.asyncio async def test_file_roles_define_effective_edit_manage_and_export_permissions(monkeypatch): user = SimpleNamespace(id=uuid.uuid4(), is_admin=False) item = SimpleNamespace( id=uuid.uuid4(), study_id=uuid.uuid4(), owner_id=uuid.uuid4(), allow_export=False, ) membership = SimpleNamespace(is_active=True, role_in_study="CRA") member_role = AsyncMock(return_value="EDITOR") monkeypatch.setattr(collaboration_service.member_crud, "get_member", AsyncMock(return_value=membership)) monkeypatch.setattr(collaboration_service, "_member_role", member_role) assert await collaboration_service.can_edit_file(SimpleNamespace(), item, user) is True assert await collaboration_service.can_manage_file(SimpleNamespace(), item, user) is False assert await collaboration_service.can_export_file(SimpleNamespace(), item, user) is False item.allow_export = True assert await collaboration_service.can_export_file(SimpleNamespace(), item, user) is True item.allow_export = False member_role.return_value = "MANAGER" assert await collaboration_service.can_edit_file(SimpleNamespace(), item, user) is True assert await collaboration_service.can_manage_file(SimpleNamespace(), item, user) is True assert await collaboration_service.can_export_file(SimpleNamespace(), item, user) is True member_role.return_value = None item.owner_id = user.id assert await collaboration_service.can_edit_file(SimpleNamespace(), item, user) is True assert await collaboration_service.can_manage_file(SimpleNamespace(), item, user) is True assert await collaboration_service.can_export_file(SimpleNamespace(), item, user) is True assert await collaboration_service.can_transfer_ownership(SimpleNamespace(), item, user) is True @pytest.mark.asyncio async def test_project_permissions_only_control_role_assignment_eligibility(monkeypatch): user = SimpleNamespace(id=uuid.uuid4(), is_admin=False) membership = SimpleNamespace(is_active=True, role_in_study="CTA") permission_check = AsyncMock( side_effect=lambda _db, _study_id, _role, permission: permission in { "collaboration:read", "collaboration:edit", } ) monkeypatch.setattr(collaboration_service, "role_has_api_permission", permission_check) assert await collaboration_service._role_is_assignable( SimpleNamespace(), uuid.uuid4(), user, membership, "EDITOR" ) is True assert await collaboration_service._role_is_assignable( SimpleNamespace(), uuid.uuid4(), user, membership, "MANAGER" ) is False with pytest.raises(HTTPException) as exc_info: await collaboration_service._require_role_assignable( SimpleNamespace(), uuid.uuid4(), user, membership, "MANAGER" ) assert exc_info.value.status_code == 422 assert "不具备管理者或所有者资格" in exc_info.value.detail @pytest.mark.asyncio async def test_manager_assignment_rejects_an_unqualified_project_member(monkeypatch): target_id = uuid.uuid4() item = SimpleNamespace(id=uuid.uuid4(), study_id=uuid.uuid4(), owner_id=uuid.uuid4()) actor = SimpleNamespace(id=uuid.uuid4()) target = SimpleNamespace(id=target_id, is_admin=False) membership = SimpleNamespace(is_active=True, role_in_study="CTA") db = SimpleNamespace(get=AsyncMock(return_value=target)) role_check = AsyncMock(side_effect=HTTPException(status_code=422, detail="资格不足")) monkeypatch.setattr(collaboration_service, "require_file_manager", AsyncMock()) monkeypatch.setattr(collaboration_service.member_crud, "get_member", AsyncMock(return_value=membership)) monkeypatch.setattr(collaboration_service, "_require_role_assignable", role_check) with pytest.raises(HTTPException) as exc_info: await collaboration_service.upsert_member( db, item, SimpleNamespace(user_id=target_id, role="MANAGER"), actor, ) assert exc_info.value.status_code == 422 role_check.assert_awaited_once_with(db, item.study_id, target, membership, "MANAGER") @pytest.mark.asyncio async def test_ownership_transfer_rejects_an_unqualified_new_owner(monkeypatch): target_id = uuid.uuid4() item = SimpleNamespace(id=uuid.uuid4(), study_id=uuid.uuid4(), owner_id=uuid.uuid4()) target = SimpleNamespace(id=target_id, is_admin=False) membership = SimpleNamespace(is_active=True, role_in_study="CTA") db = SimpleNamespace( scalar=AsyncMock(return_value=item), get=AsyncMock(return_value=target), ) role_check = AsyncMock(side_effect=HTTPException(status_code=422, detail="资格不足")) monkeypatch.setattr(collaboration_service, "require_ownership_transfer", AsyncMock()) monkeypatch.setattr(collaboration_service.member_crud, "get_member", AsyncMock(return_value=membership)) monkeypatch.setattr(collaboration_service, "_require_role_assignable", role_check) with pytest.raises(HTTPException) as exc_info: await collaboration_service.transfer_ownership( db, item, SimpleNamespace(new_owner_id=target_id), SimpleNamespace(id=item.owner_id), ) assert exc_info.value.status_code == 422 role_check.assert_awaited_once_with(db, item.study_id, target, membership, "MANAGER") @pytest.mark.asyncio async def test_workspace_save_as_also_requires_create_permission(monkeypatch): user = SimpleNamespace(id=uuid.uuid4(), is_admin=False) item = SimpleNamespace(study_id=uuid.uuid4()) membership = SimpleNamespace(is_active=True, role_in_study="CRA") permission_check = AsyncMock(return_value=False) monkeypatch.setattr(collaboration_service.member_crud, "get_member", AsyncMock(return_value=membership)) monkeypatch.setattr(collaboration_service, "role_has_api_permission", permission_check) assert await collaboration_service.can_create_file(SimpleNamespace(), item, user) is False permission_check.assert_awaited_once_with( ANY, item.study_id, membership.role_in_study, "collaboration:create" ) permission_check.return_value = True assert await collaboration_service.can_create_file(SimpleNamespace(), item, user) is True @pytest.mark.asyncio async def test_copy_creates_an_independent_r1_from_the_current_revision(monkeypatch, tmp_path): source = tmp_path / "source.xlsx" content = collaboration_service.blank_file_bytes("cell") source.write_bytes(content) revision = SimpleNamespace(file_uri=str(source)) item = SimpleNamespace( id=uuid.uuid4(), study_id=uuid.uuid4(), folder_id=uuid.uuid4(), title="研究数据.xlsx", file_type="cell", extension="xlsx", current_revision_id=uuid.uuid4(), ) target = SimpleNamespace(id=uuid.uuid4(), title="研究数据 - 副本.xlsx") db = SimpleNamespace(get=AsyncMock(return_value=revision)) user = SimpleNamespace(id=uuid.uuid4()) require_exporter = AsyncMock() create_copy = AsyncMock(return_value=target) monkeypatch.setattr(collaboration_service, "require_file_exporter", require_exporter) monkeypatch.setattr(collaboration_service, "_next_copy_title", AsyncMock(return_value=target.title)) monkeypatch.setattr(collaboration_service, "_create_file_from_bytes", create_copy) result = await collaboration_service.copy_file(db, item, user) assert result is target require_exporter.assert_awaited_once_with(db, item, user) create_copy.assert_awaited_once_with( db, item.study_id, title=target.title, file_type="cell", folder_id=item.folder_id, content=content, source="COPY", user=user, ) @pytest.mark.asyncio async def test_download_returns_current_revision_without_writing_audit(monkeypatch, tmp_path): source = tmp_path / "source.docx" source.write_bytes(collaboration_service.blank_file_bytes("word")) revision = SimpleNamespace( id=uuid.uuid4(), revision_no=3, file_uri=str(source), mime_type=collaboration_service.MIME_TYPES["docx"], ) item = SimpleNamespace(id=uuid.uuid4(), current_revision_id=revision.id) user = SimpleNamespace(id=uuid.uuid4()) db = SimpleNamespace(get=AsyncMock(return_value=revision), commit=AsyncMock()) require_exporter = AsyncMock() monkeypatch.setattr(collaboration_service, "require_file_exporter", require_exporter) result = await collaboration_service.prepare_download(db, item, user) assert result is revision require_exporter.assert_awaited_once_with(db, item, user) db.commit.assert_not_awaited() @pytest.mark.asyncio async def test_revision_preview_does_not_write_passive_audit(monkeypatch, tmp_path): revision_path = tmp_path / "revision.docx" revision_path.write_bytes(b"office") revision = SimpleNamespace(id=uuid.uuid4(), file_uri=str(revision_path), revision_no=2) item = SimpleNamespace(id=uuid.uuid4()) db = SimpleNamespace(commit=AsyncMock()) monkeypatch.setattr(collaboration_service, "get_revision_or_404", AsyncMock(return_value=revision)) result = await collaboration_service.prepare_revision_preview( db, item, revision.id, SimpleNamespace(id=uuid.uuid4()), ) assert result is revision db.commit.assert_not_awaited() @pytest.mark.asyncio async def test_file_read_includes_information_and_export_capability(monkeypatch): now = datetime.now(timezone.utc) owner_id = uuid.uuid4() folder_id = uuid.uuid4() revision_id = uuid.uuid4() item = CollaborationFile( id=uuid.uuid4(), study_id=uuid.uuid4(), folder_id=folder_id, title="研究方案.docx", file_type="word", extension="docx", status="ACTIVE", owner_id=owner_id, current_revision_id=revision_id, generation=2, allow_export=False, allow_edit_request=False, allow_sheet_structure_edit=True, created_at=now, updated_at=now, ) owner = SimpleNamespace(full_name="文件所有者") folder = CollaborationFolder(id=folder_id, name="方案", study_id=item.study_id, created_by=owner_id) revision = CollaborationRevision( id=revision_id, file_id=item.id, revision_no=4, file_uri="/hidden/revision.docx", original_filename=item.title, file_hash="hash", file_size=4096, mime_type=collaboration_service.MIME_TYPES["docx"], source="SESSION_CLOSE", created_at=now, ) async def get_model(model, identifier): if model is CollaborationFolder and identifier == folder_id: return folder if model is CollaborationRevision and identifier == revision_id: return revision return owner db = SimpleNamespace(get=AsyncMock(side_effect=get_model)) user = SimpleNamespace(id=owner_id, is_admin=False) monkeypatch.setattr(collaboration_service, "_member_role", AsyncMock(return_value="MANAGER")) monkeypatch.setattr(collaboration_service, "can_edit_file", AsyncMock(return_value=True)) monkeypatch.setattr(collaboration_service, "can_manage_file", AsyncMock(return_value=True)) monkeypatch.setattr(collaboration_service, "can_export_file", AsyncMock(return_value=True)) monkeypatch.setattr(collaboration_service, "edit_request_status", AsyncMock(return_value=None)) monkeypatch.setattr(collaboration_service, "can_request_edit_file", AsyncMock(return_value=False)) monkeypatch.setattr(collaboration_service, "can_transfer_ownership", AsyncMock(return_value=True)) result = await collaboration_service._file_read(db, item, user, collaborators=[]) assert result.folder_name == "方案" assert result.current_revision_no == 4 assert result.current_revision_file_size == 4096 assert result.current_revision_mime_type == collaboration_service.MIME_TYPES["docx"] assert result.current_revision_created_at == now assert result.can_export is True assert result.can_transfer_ownership is True def test_workbook_structure_policy_locks_and_restores_sheet_operations(): file_id = uuid.uuid4() original = collaboration_service.blank_file_bytes("cell") protected, backup = collaboration_service.apply_workbook_structure_policy( original, file_id=file_id, allow_sheet_structure_edit=False, protection_backup=None, ) assert backup == "" with zipfile.ZipFile(io.BytesIO(protected)) as package: workbook = package.read("xl/workbook.xml").decode("utf-8") assert "workbookProtection" in workbook assert 'lockStructure="1"' in workbook assert "workbookPassword=" in workbook restored, restored_backup = collaboration_service.apply_workbook_structure_policy( protected, file_id=file_id, allow_sheet_structure_edit=True, protection_backup=backup, ) assert restored_backup is None with zipfile.ZipFile(io.BytesIO(restored)) as package: workbook = package.read("xl/workbook.xml").decode("utf-8") assert "workbookProtection" not in workbook def test_collaboration_models_do_not_reference_attachment_or_document_tables(): foreign_keys = { foreign_key.target_fullname for table in ( CollaborationFile.__table__, ) for foreign_key in table.foreign_keys } assert all(not target.startswith("attachments.") for target in foreign_keys) assert all(not target.startswith("documents.") for target in foreign_keys) assert all(not target.startswith("document_versions.") for target in foreign_keys) @pytest.mark.asyncio async def test_collaborator_summaries_are_loaded_in_one_batch_for_the_file_list(): first_file_id = uuid.uuid4() second_file_id = uuid.uuid4() first_member = SimpleNamespace( file_id=first_file_id, user_id=uuid.uuid4(), role="MANAGER", ) second_member = SimpleNamespace( file_id=second_file_id, user_id=uuid.uuid4(), role="EDITOR", ) first_user = SimpleNamespace(full_name="文件所有者", avatar_url="/avatars/owner.png") second_user = SimpleNamespace(full_name="协作编辑者", avatar_url=None) query_result = SimpleNamespace(all=lambda: [(first_member, first_user), (second_member, second_user)]) db = SimpleNamespace(execute=AsyncMock(return_value=query_result)) result = await collaboration_service._collaborators_by_file(db, [first_file_id, second_file_id]) db.execute.assert_awaited_once() assert result[first_file_id][0].full_name == "文件所有者" assert result[first_file_id][0].avatar_url == "/avatars/owner.png" assert result[second_file_id][0].role == "EDITOR" class _CallbackDb: def __init__(self, session, item): self._scalar_results = iter((session, None)) self._item = item self.added = [] async def scalar(self, _statement): return next(self._scalar_results) async def get(self, model, _identifier): return self._item if model is CollaborationFile else None def add(self, value): self.added.append(value) async def commit(self): return None @pytest.mark.asyncio async def test_callback_error_status_is_recorded_and_acknowledged(): session = SimpleNamespace( id=uuid.uuid4(), file_id=uuid.uuid4(), document_key="ctms-collab-error-key", generation=1, status="ACTIVE", active_users=None, last_callback_at=None, ) item = SimpleNamespace(id=session.file_id, generation=1) db = _CallbackDb(session, item) result = await onlyoffice_collaboration_service.process_callback( db, session.id, CollaborationCallbackPayload(key=session.document_key, status=3), ) assert result == {"error": 0} assert session.status == "ERROR" assert db.added[0].result == "ERROR" @pytest.mark.asyncio async def test_force_save_updates_session_recovery_revision(monkeypatch): session = SimpleNamespace( id=uuid.uuid4(), file_id=uuid.uuid4(), document_key="ctms-collab-force-save-key", generation=2, status="ACTIVE", active_users=None, last_callback_at=None, base_revision_id=uuid.uuid4(), ) item = SimpleNamespace(id=session.file_id, generation=2) actor = SimpleNamespace(id=uuid.uuid4()) revision = SimpleNamespace(id=uuid.uuid4(), revision_no=4) db = _CallbackDb(session, item) monkeypatch.setattr(onlyoffice_collaboration_service, "_download_result", AsyncMock(return_value=b"saved")) monkeypatch.setattr(onlyoffice_collaboration_service, "_callback_user", AsyncMock(return_value=actor)) monkeypatch.setattr( onlyoffice_collaboration_service.collaboration_service, "append_revision", AsyncMock(return_value=(revision, True)), ) result = await onlyoffice_collaboration_service.process_callback( db, session.id, CollaborationCallbackPayload( key=session.document_key, status=6, url="http://onlyoffice/cache/force-save.docx", users=[str(actor.id)], ), ) assert result == {"error": 0} assert session.base_revision_id == revision.id assert session.status == "ACTIVE" assert db.added[0].revision_id == revision.id