新增全局协作台账并统一工作台管理界面
新增与项目平级的协作台账，提供空白模板、独立账号授权和在线编辑。 复用共享库的访问权限与历史版本界面，拆分信息维护入口并统一工作台卡片样式。 补充在线文档字体构建支持、数据库表结构升级和回归测试。 不包含本地台账内容、账号授权记录或历史文件迁移数据。
This commit is contained in:
Cheng Zhou
@@ -0,0 +1,463 @@
|
||||
import io
|
||||
import uuid
|
||||
import zipfile
|
||||
from datetime import timezone
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from fastapi import FastAPI, HTTPException, UploadFile
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.types import DateTime, TypeDecorator
|
||||
|
||||
from conftest import _create_test_engine
|
||||
from app.api.v1.ledgers import router
|
||||
from app.core.config import settings
|
||||
from app.core.deps import get_current_user, get_db_session
|
||||
from app.models.collaboration import CollaborationCallbackReceipt, CollaborationFile, CollaborationRevision, CollaborationSession, CollaborationShareLink
|
||||
from app.models.user import User, UserStatus
|
||||
from app.schemas.collaboration import CollaborationCallbackPayload
|
||||
from app.schemas.ledger import LedgerMemberUpsert, LedgerSettings
|
||||
from app.services import collaboration_service as collaboration
|
||||
from app.services import ledger_service as ledgers, onlyoffice_collaboration_service as office, onlyoffice_service
|
||||
|
||||
|
||||
class SQLiteUTCDateTime(TypeDecorator):
|
||||
"""Preserve the PostgreSQL UTC timestamp contract in the SQLite fixture."""
|
||||
impl = DateTime(timezone=True)
|
||||
cache_ok = True
|
||||
|
||||
def process_result_value(self, value, dialect):
|
||||
return value.replace(tzinfo=timezone.utc) if value and value.tzinfo is None else value
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def env(monkeypatch, tmp_path):
|
||||
for column in CollaborationShareLink.__table__.columns:
|
||||
if isinstance(column.type, DateTime) and column.type.timezone:
|
||||
monkeypatch.setattr(column, "type", SQLiteUTCDateTime())
|
||||
engine = await _create_test_engine()
|
||||
monkeypatch.setattr(collaboration, "COLLABORATION_ROOT", tmp_path)
|
||||
monkeypatch.setattr(settings, "ONLYOFFICE_JWT_SECRET", "ledger-test-secret-long-enough-for-tests")
|
||||
monkeypatch.setattr(onlyoffice_service, "ensure_onlyoffice_available", AsyncMock())
|
||||
async with AsyncSession(engine, expire_on_commit=False) as db:
|
||||
users = [User(id=uuid.uuid4(), email=f"ledger-{i}@example.com", password_hash="hash",
|
||||
full_name=f"Ledger user {i}", clinical_department="test", is_admin=i == 0,
|
||||
status=UserStatus.ACTIVE) for i in range(4)]
|
||||
db.add_all(users)
|
||||
await db.commit()
|
||||
initial = await ledgers.initialize_ledgers(db, users[0])
|
||||
app = FastAPI()
|
||||
app.include_router(router, prefix="/api/v1/ledgers")
|
||||
identity = SimpleNamespace(user=users[0])
|
||||
app.dependency_overrides[get_current_user] = lambda: identity.user
|
||||
app.dependency_overrides[get_db_session] = lambda: db
|
||||
async with httpx.AsyncClient(app=app, base_url="http://test") as client:
|
||||
yield SimpleNamespace(db=db, admin=users[0], editor=users[1], viewer=users[2], outsider=users[3],
|
||||
initial=initial, identity=identity, client=client, app=app)
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
async def grant(env, user, role, index=0):
|
||||
item = await ledgers.get_ledger(env.db, env.initial[index].id, env.admin, manage=True, lock=True)
|
||||
await ledgers.set_member(env.db, item, LedgerMemberUpsert(user_id=user.id, role=role), env.admin)
|
||||
return item
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initialization_preserves_both_uploaded_templates_and_is_idempotent(env):
|
||||
repeated = await ledgers.initialize_ledgers(env.db, env.admin)
|
||||
assert [r.id for r in repeated] == [r.id for r in env.initial]
|
||||
for result, (_, title, filename, _) in zip(repeated, ledgers.LEDGER_TEMPLATES):
|
||||
item = await env.db.get(CollaborationFile, result.id)
|
||||
assert item.study_id is None and item.folder_id is None and item.scope == "LEDGER"
|
||||
assert item.title == title
|
||||
revision = await env.db.get(CollaborationRevision, item.current_revision_id)
|
||||
with zipfile.ZipFile(ledgers.TEMPLATE_ROOT / filename) as original, zipfile.ZipFile(revision.file_uri) as saved:
|
||||
assert original.namelist() == saved.namelist()
|
||||
for part in original.namelist():
|
||||
if part != "xl/workbook.xml":
|
||||
assert original.read(part) == saved.read(part)
|
||||
assert b'lockStructure="1"' in saved.read("xl/workbook.xml")
|
||||
assert "ledgers" in Path(revision.file_uri).parts
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_account_grants_are_independent_and_all_file_routes_require_access(env):
|
||||
await grant(env, env.editor, "EDITOR")
|
||||
env.identity.user = env.editor
|
||||
response = await env.client.get("/api/v1/ledgers")
|
||||
assert response.status_code == 200 and response.headers["cache-control"] == "no-store"
|
||||
assert [row["id"] for row in response.json()] == [str(env.initial[0].id)]
|
||||
assert response.json()[0]["can_edit"] is True
|
||||
other = env.initial[1].id
|
||||
for suffix in ["", "/editor-config", "/revisions", "/members", "/candidates"]:
|
||||
assert (await env.client.get(f"/api/v1/ledgers/{other}{suffix}")).status_code == 404
|
||||
assert (await env.client.post("/api/v1/ledgers/initialize")).status_code == 403
|
||||
env.identity.user = env.outsider
|
||||
assert (await env.client.get("/api/v1/ledgers")).json() == []
|
||||
env.app.dependency_overrides.pop(get_current_user)
|
||||
assert (await env.client.get("/api/v1/ledgers")).status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_editor_and_viewer_share_session_but_receive_distinct_signed_permissions(env):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
await grant(env, env.viewer, "VIEWER")
|
||||
editor = await office.build_editor_config(env.db, item, env.editor)
|
||||
viewer = await office.build_editor_config(env.db, item, env.viewer)
|
||||
assert editor.config["document"]["key"] == viewer.config["document"]["key"]
|
||||
assert editor.access_mode == "edit" and viewer.access_mode == "view"
|
||||
assert editor.config["document"]["permissions"]["edit"] is True
|
||||
assert viewer.config["document"]["permissions"]["edit"] is False
|
||||
assert not editor.can_save_as and not editor.can_request_edit and not editor.can_download
|
||||
env.identity.user = env.editor
|
||||
assert (await env.client.get(f"/api/v1/ledgers/{item.id}/members")).status_code == 403
|
||||
assert (await env.client.post(f"/api/v1/ledgers/{item.id}/downloads", json={"file_type": "xlsx"})).status_code == 403
|
||||
with pytest.raises(HTTPException) as error:
|
||||
await collaboration.restore_revision(env.db, item, item.current_revision_id, env.editor, None)
|
||||
assert error.value.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_neither_editors_nor_admins_can_delete_ledger_or_revision(env):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
for user in [env.admin, env.editor]:
|
||||
env.identity.user = user
|
||||
assert (await env.client.delete(f"/api/v1/ledgers/{item.id}")).status_code == 405
|
||||
with pytest.raises(HTTPException) as error:
|
||||
await collaboration.move_to_trash(env.db, item, user)
|
||||
assert error.value.status_code == 403
|
||||
with pytest.raises(HTTPException) as error:
|
||||
await collaboration.delete_revision(env.db, item, item.current_revision_id, user)
|
||||
assert error.value.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revocation_invalidates_old_callback_before_result_download(env, monkeypatch):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
config = await office.build_editor_config(env.db, item, env.editor)
|
||||
session = await env.db.scalar(select(CollaborationSession).where(CollaborationSession.file_id == item.id))
|
||||
baseline = item.current_revision_id
|
||||
download = AsyncMock(return_value=b"should never be downloaded")
|
||||
monkeypatch.setattr(office, "_download_result", download)
|
||||
await ledgers.remove_member(env.db, item, env.editor.id, env.admin)
|
||||
result = await office.process_callback(env.db, session.id, CollaborationCallbackPayload(
|
||||
key=config.config["document"]["key"], status=6, url="http://onlyoffice/result.xlsx", users=[str(env.editor.id)],
|
||||
))
|
||||
assert result == {"error": 0} and item.current_revision_id == baseline
|
||||
download.assert_not_awaited()
|
||||
receipt = await env.db.scalar(select(CollaborationCallbackReceipt).where(CollaborationCallbackReceipt.session_id == session.id))
|
||||
assert receipt.result == "STALE"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_saved_callback_is_versioned_and_duplicate_is_idempotent(env, monkeypatch):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
config = await office.build_editor_config(env.db, item, env.editor)
|
||||
session = await env.db.scalar(select(CollaborationSession).where(CollaborationSession.file_id == item.id))
|
||||
monkeypatch.setattr(office, "_download_result", AsyncMock(return_value=collaboration.blank_file_bytes("cell")))
|
||||
payload = CollaborationCallbackPayload(key=config.config["document"]["key"], status=6,
|
||||
url="http://onlyoffice/result.xlsx", users=[str(env.editor.id)])
|
||||
await office.process_callback(env.db, session.id, payload)
|
||||
saved = item.current_revision_id
|
||||
await office.process_callback(env.db, session.id, payload)
|
||||
assert item.current_revision_id == saved
|
||||
assert len(await collaboration.list_revisions(env.db, item)) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_archive_disables_edit_and_stale_settings_are_rejected(env):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
payload = LedgerSettings(title=item.title, description="Archived", status="ARCHIVED", owner_id=item.owner_id,
|
||||
allow_export=False, allow_sheet_structure_edit=False, generation=item.generation)
|
||||
await ledgers.update_settings(env.db, item, payload, env.admin)
|
||||
assert not (await ledgers.read_ledger(env.db, item, env.editor)).can_edit
|
||||
assert (await office.build_editor_config(env.db, item, env.editor)).access_mode == "view"
|
||||
with pytest.raises(HTTPException) as error:
|
||||
await ledgers.update_settings(env.db, item, payload, env.admin)
|
||||
assert error.value.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_import_and_restore_keep_previous_versions_and_workbook_protection(env):
|
||||
item = await ledgers.get_ledger(env.db, env.initial[0].id, env.admin, manage=True)
|
||||
first = item.current_revision_id
|
||||
upload = UploadFile(filename="import.xlsx", file=io.BytesIO(collaboration.blank_file_bytes("cell")))
|
||||
await ledgers.import_content(env.db, item, upload, item.generation, env.admin)
|
||||
second = item.current_revision_id
|
||||
await ledgers.restore_revision(env.db, item, first, item.generation, env.admin)
|
||||
assert item.current_revision_id not in {first, second}
|
||||
assert len(await collaboration.list_revisions(env.db, item)) == 3
|
||||
revision = await env.db.get(CollaborationRevision, item.current_revision_id)
|
||||
with zipfile.ZipFile(revision.file_uri) as package:
|
||||
assert b'lockStructure="1"' in package.read("xl/workbook.xml")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inactive_account_cannot_be_granted_or_save_existing_session(env, monkeypatch):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
await office.build_editor_config(env.db, item, env.editor)
|
||||
session = await env.db.scalar(select(CollaborationSession).where(CollaborationSession.file_id == item.id))
|
||||
env.editor.status = UserStatus.DISABLED
|
||||
await env.db.commit()
|
||||
with pytest.raises(HTTPException) as error:
|
||||
await ledgers.set_member(env.db, item, LedgerMemberUpsert(user_id=env.editor.id, role="MANAGER"), env.admin)
|
||||
assert error.value.status_code == 422
|
||||
download = AsyncMock()
|
||||
monkeypatch.setattr(office, "_download_result", download)
|
||||
await office.process_callback(env.db, session.id, CollaborationCallbackPayload(
|
||||
key=session.document_key, status=6, url="http://onlyoffice/result.xlsx", users=[str(env.editor.id)],
|
||||
))
|
||||
download.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restore_respects_current_unlocked_workbook_policy(env):
|
||||
item = await ledgers.get_ledger(env.db, env.initial[0].id, env.admin, manage=True)
|
||||
initial = item.current_revision_id
|
||||
payload = LedgerSettings(title=item.title, status="ACTIVE", owner_id=item.owner_id,
|
||||
allow_sheet_structure_edit=True, generation=item.generation)
|
||||
await ledgers.update_settings(env.db, item, payload, env.admin)
|
||||
await ledgers.import_content(env.db, item, UploadFile(filename="blank.xlsx", file=io.BytesIO(
|
||||
collaboration.blank_file_bytes("cell"))), item.generation, env.admin)
|
||||
await ledgers.restore_revision(env.db, item, initial, item.generation, env.admin)
|
||||
revision = await env.db.get(CollaborationRevision, item.current_revision_id)
|
||||
with zipfile.ZipFile(revision.file_uri) as package:
|
||||
assert b"workbookProtection" not in package.read("xl/workbook.xml")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shared_access_uses_global_candidates_and_protects_implicit_managers(env):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
base = f"/api/v1/ledgers/{item.id}"
|
||||
members = (await env.client.get(f"{base}/members")).json()
|
||||
owner = next(row for row in members if row["user_id"] == str(env.admin.id))
|
||||
assert owner["protected_label"] == "所有者" and owner["role"] == "MANAGER"
|
||||
assert (await env.client.delete(f"{base}/members/{env.admin.id}")).status_code == 422
|
||||
candidates = (await env.client.get(f"{base}/candidates")).json()
|
||||
assert {row["user_id"] for row in candidates} == {str(user.id) for user in [env.admin, env.editor, env.viewer, env.outsider]}
|
||||
assert all(row["can_be_manager"] and row["can_be_editor"] for row in candidates)
|
||||
original_title = item.title
|
||||
result = await env.client.patch(f"{base}/access-settings", json={"generation": item.generation, "allow_edit_request": False})
|
||||
assert result.status_code == 200 and not result.json()["allow_edit_request"]
|
||||
assert result.json()["title"] == original_title and not result.json()["allow_export"]
|
||||
for actor, code in [(env.editor, 403), (env.outsider, 404)]:
|
||||
env.identity.user = actor
|
||||
for suffix in ["share-link", "edit-requests", "members", "candidates"]:
|
||||
assert (await env.client.get(f"{base}/{suffix}")).status_code == code
|
||||
assert (await env.client.patch(f"{base}/access-settings", json={"generation": item.generation, "allow_export": True})).status_code == code
|
||||
assert (await env.client.put(f"{base}/share-link", json={"enabled": True})).status_code == code
|
||||
assert (await env.client.post(f"{base}/transfer-ownership", json={"new_owner_id": str(env.viewer.id)})).status_code == code
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_viewer_edit_request_is_deduplicated_and_approval_grants_editor(env):
|
||||
item = await grant(env, env.viewer, "VIEWER")
|
||||
base = f"/api/v1/ledgers/{item.id}"
|
||||
assert (await env.client.patch(f"{base}/access-settings", json={"generation": item.generation, "allow_edit_request": True})).status_code == 200
|
||||
env.identity.user = env.viewer
|
||||
assert (await env.client.get(base)).json()["can_request_edit"]
|
||||
request = await env.client.post(f"{base}/edit-requests")
|
||||
assert request.status_code == 200
|
||||
request_id = request.json()["id"]
|
||||
assert (await env.client.post(f"{base}/edit-requests")).json()["id"] == request_id
|
||||
read = (await env.client.get(base)).json()
|
||||
assert read["edit_request_status"] == "PENDING" and not read["can_request_edit"]
|
||||
assert (await env.client.post(f"{base}/edit-requests/{request_id}/resolve", json={"status": "APPROVED"})).status_code == 403
|
||||
generation = item.generation
|
||||
env.identity.user = env.admin
|
||||
assert len((await env.client.get(f"{base}/edit-requests")).json()) == 1
|
||||
result = await env.client.post(f"{base}/edit-requests/{request_id}/resolve", json={"status": "APPROVED"})
|
||||
assert result.status_code == 200 and result.json()["status"] == "APPROVED"
|
||||
env.identity.user = env.viewer
|
||||
read = (await env.client.get(base)).json()
|
||||
assert read["role"] == "EDITOR" and read["can_edit"] and read["generation"] == generation + 1
|
||||
assert not read["can_request_edit"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoked_requester_cannot_regain_access_through_pending_approval(env):
|
||||
item = await grant(env, env.viewer, "VIEWER")
|
||||
base = f"/api/v1/ledgers/{item.id}"
|
||||
assert (await env.client.patch(f"{base}/access-settings", json={"generation": item.generation, "allow_edit_request": True})).status_code == 200
|
||||
env.identity.user = env.viewer
|
||||
request_id = (await env.client.post(f"{base}/edit-requests")).json()["id"]
|
||||
await ledgers.remove_member(env.db, item, env.viewer.id, env.admin)
|
||||
env.identity.user = env.admin
|
||||
assert (await env.client.post(f"{base}/edit-requests/{request_id}/resolve", json={"status": "APPROVED"})).status_code == 422
|
||||
assert not await collaboration.can_edit_file(env.db, item, env.viewer)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ownership_transfer_accepts_system_accounts_and_keeps_previous_owner_access(env):
|
||||
item = await grant(env, env.editor, "MANAGER")
|
||||
base = f"/api/v1/ledgers/{item.id}"
|
||||
env.identity.user = env.editor
|
||||
assert (await env.client.post(f"{base}/transfer-ownership", json={"new_owner_id": str(env.viewer.id)})).status_code == 403
|
||||
env.identity.user = env.admin
|
||||
assert (await env.client.post(f"{base}/transfer-ownership", json={"new_owner_id": str(env.viewer.id)})).status_code == 200
|
||||
env.identity.user = env.viewer
|
||||
result = await env.client.post(f"{base}/transfer-ownership", json={"new_owner_id": str(env.outsider.id)})
|
||||
assert result.status_code == 200 and result.json()["owner_id"] == str(env.outsider.id)
|
||||
assert result.json()["can_manage"] and not result.json()["can_transfer_ownership"]
|
||||
assert item.study_id is None
|
||||
members = (await env.client.get(f"{base}/members")).json()
|
||||
admin = next(row for row in members if row["user_id"] == str(env.admin.id))
|
||||
assert admin["protected_label"] == "系统管理员"
|
||||
assert (await env.client.delete(f"{base}/members/{env.admin.id}")).status_code == 422
|
||||
|
||||
|
||||
async def shared_ledger(env, *, mode="EDIT", password=None):
|
||||
from app.models.collaboration import CollaborationShareLink
|
||||
item = await ledgers.get_ledger(env.db, env.initial[0].id, env.admin)
|
||||
response = await env.client.put(f"/api/v1/ledgers/{item.id}/share-link", json={
|
||||
"enabled": True, "access_mode": mode, "expiry_policy": "SEVEN_DAYS",
|
||||
"password_mode": "SET" if password else "KEEP", **({"password": password} if password else {}),
|
||||
})
|
||||
assert response.status_code == 200
|
||||
link = await env.db.get(CollaborationShareLink, uuid.UUID(response.json()["id"]))
|
||||
return item, link, response.json()["share_token"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ledger_share_defaults_closed_and_password_change_revokes_old_credentials(env):
|
||||
from app.services import collaboration_share_service as shares
|
||||
base = f"/api/v1/ledgers/{env.initial[0].id}"
|
||||
initial = (await env.client.get(f"{base}/share-link")).json()
|
||||
assert not initial["enabled"] and initial["share_token"] is None
|
||||
item, link, token = await shared_ledger(env, password="test-password")
|
||||
grant_token = await shares.verify_share_password(env.db, token, "test-password")
|
||||
shares.validate_access_grant(link, grant_token.access_token)
|
||||
with pytest.raises(HTTPException) as wrong:
|
||||
await shares.verify_share_password(env.db, token, "incorrect")
|
||||
assert wrong.value.status_code == 401
|
||||
generation = item.generation
|
||||
await env.client.put(f"{base}/share-link", json={"enabled": True, "password_mode": "SET", "password": "new-password"})
|
||||
assert item.generation == generation + 1
|
||||
with pytest.raises(HTTPException):
|
||||
await shares.resolve_active_share(env.db, token)
|
||||
with pytest.raises(HTTPException):
|
||||
shares.validate_access_grant(link, grant_token.access_token)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_anonymous_ledger_save_keeps_share_identity_and_close_blocks_stale_callback(env, monkeypatch):
|
||||
item, link, _ = await shared_ledger(env)
|
||||
config = await office.build_shared_editor_config(env.db, item, link, client_id="test-client", display_name="访客")
|
||||
assert config.access_mode == "edit" and not config.can_save_as and not config.can_download
|
||||
session = await env.db.scalar(select(CollaborationSession).where(CollaborationSession.file_id == item.id))
|
||||
download = AsyncMock(return_value=collaboration.blank_file_bytes("cell"))
|
||||
monkeypatch.setattr(office, "_download_result", download)
|
||||
# ONLYOFFICE can send a final save with an empty user list. It must not become an owner save.
|
||||
await office.process_callback(env.db, session.id, CollaborationCallbackPayload(
|
||||
key=session.document_key, status=6, url="http://onlyoffice/first.xlsx", users=[],
|
||||
))
|
||||
revision = await env.db.get(CollaborationRevision, item.current_revision_id)
|
||||
assert revision.source == "SHARE_FORCE_SAVE" and revision.created_by is None
|
||||
saved = item.current_revision_id
|
||||
await env.client.put(f"/api/v1/ledgers/{item.id}/share-link", json={"enabled": False})
|
||||
download.reset_mock()
|
||||
await office.process_callback(env.db, session.id, CollaborationCallbackPayload(
|
||||
key=session.document_key, status=2, url="http://onlyoffice/after-close.xlsx", users=[],
|
||||
))
|
||||
download.assert_not_awaited()
|
||||
assert item.current_revision_id == saved
|
||||
with pytest.raises(HTTPException):
|
||||
await office.build_shared_editor_config(env.db, item, link, client_id="new-client", display_name="访客")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("mode,expired", [("VIEW", False), ("EDIT", True)])
|
||||
async def test_anonymous_view_or_expired_link_cannot_save_even_without_callback_users(env, monkeypatch, mode, expired):
|
||||
from datetime import datetime, timedelta, timezone
|
||||
item, link, _ = await shared_ledger(env, mode=mode)
|
||||
await office.build_shared_editor_config(env.db, item, link, client_id="test-client", display_name="访客")
|
||||
session = await env.db.scalar(select(CollaborationSession).where(CollaborationSession.file_id == item.id))
|
||||
if expired:
|
||||
link.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
|
||||
await env.db.commit()
|
||||
saved = item.current_revision_id
|
||||
download = AsyncMock()
|
||||
monkeypatch.setattr(office, "_download_result", download)
|
||||
await office.process_callback(env.db, session.id, CollaborationCallbackPayload(
|
||||
key=session.document_key, status=6, url="http://onlyoffice/denied.xlsx", users=[],
|
||||
))
|
||||
download.assert_not_awaited()
|
||||
assert item.current_revision_id == saved
|
||||
receipt = await env.db.scalar(select(CollaborationCallbackReceipt).where(CollaborationCallbackReceipt.session_id == session.id))
|
||||
assert receipt.result == "ACCESS_REVOKED"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ledger_uses_existing_public_share_http_flow_without_a_project(env):
|
||||
from app.api.v1.collaboration import public_router
|
||||
env.app.include_router(public_router, prefix="/api/v1/collaboration/shares")
|
||||
item, _, token = await shared_ledger(env, password="share-password")
|
||||
env.app.dependency_overrides.pop(get_current_user)
|
||||
headers = {"X-CTMS-Share-Token": token}
|
||||
base = "/api/v1/collaboration/shares"
|
||||
metadata = await env.client.get(f"{base}/metadata", headers=headers)
|
||||
assert metadata.status_code == 200 and metadata.json()["requires_password"]
|
||||
body = {"client_id": "test-client", "display_name": "访客"}
|
||||
assert (await env.client.post(f"{base}/editor-config", headers=headers, json=body)).status_code == 401
|
||||
access = await env.client.post(f"{base}/access", headers=headers, json={"password": "share-password"})
|
||||
assert access.status_code == 200
|
||||
config = await env.client.post(f"{base}/editor-config", headers=headers,
|
||||
json={**body, "access_token": access.json()["access_token"]})
|
||||
assert config.status_code == 200 and config.headers["cache-control"] == "no-store"
|
||||
assert config.json()["file_name"] == item.title and config.json()["access_mode"] == "edit"
|
||||
assert not config.json()["can_save_as"] and item.study_id is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revision_names_reuse_collaboration_metadata_without_replacing_content(env):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
await grant(env, env.viewer, "VIEWER")
|
||||
revision_id = item.current_revision_id
|
||||
generation = item.generation
|
||||
base = f"/api/v1/ledgers/{item.id}/revisions/{revision_id}"
|
||||
env.identity.user = env.editor
|
||||
result = await env.client.patch(base, json={"change_summary": "审核提交版"})
|
||||
assert result.status_code == 200 and result.json()["change_summary"] == "审核提交版"
|
||||
assert item.current_revision_id == revision_id and item.generation == generation
|
||||
assert len(await collaboration.list_revisions(env.db, item)) == 1
|
||||
for actor, status in [(env.viewer, 403), (env.outsider, 404)]:
|
||||
env.identity.user = actor
|
||||
assert (await env.client.patch(base, json={"change_summary": "不应保存"})).status_code == status
|
||||
env.identity.user = env.editor
|
||||
other_revision = env.initial[1].current_revision_id
|
||||
assert (await env.client.patch(f"/api/v1/ledgers/{item.id}/revisions/{other_revision}", json={"change_summary": "跨台账"})).status_code == 404
|
||||
env.identity.user = env.admin
|
||||
await env.client.patch(f"/api/v1/ledgers/{item.id}", json={
|
||||
"title": item.title, "status": "ARCHIVED", "owner_id": str(item.owner_id), "generation": item.generation,
|
||||
})
|
||||
assert (await env.client.patch(base, json={"change_summary": "归档后"})).status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_history_export_uses_live_export_permission_and_checks_revision_belongs_to_ledger(env):
|
||||
item = await grant(env, env.editor, "EDITOR")
|
||||
await grant(env, env.viewer, "VIEWER")
|
||||
base = f"/api/v1/ledgers/{item.id}"
|
||||
path = f"{base}/revisions/{item.current_revision_id}/download"
|
||||
for actor in [env.editor, env.viewer]:
|
||||
env.identity.user = actor
|
||||
assert (await env.client.get(path)).status_code == 403
|
||||
env.identity.user = env.admin
|
||||
response = await env.client.get(path)
|
||||
assert response.status_code == 200 and response.headers["cache-control"] == "no-store"
|
||||
revision = await env.db.get(CollaborationRevision, item.current_revision_id)
|
||||
assert response.content == Path(revision.file_uri).read_bytes()
|
||||
await env.client.patch(f"{base}/access-settings", json={"generation": item.generation, "allow_export": True})
|
||||
env.identity.user = env.viewer
|
||||
assert (await env.client.get(path)).status_code == 200
|
||||
assert (await env.client.get(f"{base}/revisions/{env.initial[1].current_revision_id}/download")).status_code == 404
|
||||
await ledgers.remove_member(env.db, item, env.viewer.id, env.admin)
|
||||
assert (await env.client.get(path)).status_code == 404
|
||||
env.identity.user = env.outsider
|
||||
assert (await env.client.get(path)).status_code == 404
|
||||
Reference in New Issue
Block a user