Cheng Zhou
1057 lines
42 KiB
Python
1057 lines
42 KiB
Python
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
|