Cheng Zhou
570 lines
31 KiB
Python
570 lines
31 KiB
Python
import io
|
|
import uuid
|
|
import zipfile
|
|
from datetime import datetime, 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())
|
|
monkeypatch.setattr(office, "_document_server_users", AsyncMock(return_value=["live-user"]))
|
|
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_failed_ledger_session_recovers_server_backup_under_a_new_document_key(env, monkeypatch):
|
|
item = await env.db.get(CollaborationFile, env.initial[0].id)
|
|
first = await office.build_editor_config(env.db, item, env.admin)
|
|
failed = await env.db.scalar(select(CollaborationSession).where(
|
|
CollaborationSession.file_id == item.id,
|
|
CollaborationSession.generation == item.generation,
|
|
))
|
|
failed.status = "ERROR"
|
|
await env.db.commit()
|
|
|
|
current = await env.db.get(CollaborationRevision, item.current_revision_id)
|
|
recovered_buffer = io.BytesIO(Path(current.file_uri).read_bytes())
|
|
with zipfile.ZipFile(recovered_buffer, "a") as package:
|
|
package.comment = b"document-server-recovery"
|
|
recovery = AsyncMock(return_value=recovered_buffer.getvalue())
|
|
monkeypatch.setattr(office, "_recover_forgotten_content", recovery)
|
|
|
|
reopened = await office.build_editor_config(env.db, item, env.admin)
|
|
await env.db.refresh(item)
|
|
await env.db.refresh(failed)
|
|
sessions = (await env.db.scalars(select(CollaborationSession).where(
|
|
CollaborationSession.file_id == item.id,
|
|
).order_by(CollaborationSession.generation))).all()
|
|
recovered_revision = await env.db.get(CollaborationRevision, item.current_revision_id)
|
|
|
|
assert first.config["document"]["key"] != reopened.config["document"]["key"]
|
|
assert item.generation == 2
|
|
assert failed.status == "RECOVERED"
|
|
assert [session.generation for session in sessions] == [1, 2]
|
|
assert recovered_revision.source == "SERVER_RECOVERY"
|
|
assert recovered_revision.change_summary == "自动恢复在线文档服务器备份"
|
|
recovery.assert_awaited_once_with(failed.document_key, "cell")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_closed_ledger_session_starts_a_new_key_instead_of_reopening_server_cache(env, monkeypatch):
|
|
item = await env.db.get(CollaborationFile, env.initial[0].id)
|
|
first = await office.build_editor_config(env.db, item, env.admin)
|
|
closed = await env.db.scalar(select(CollaborationSession).where(
|
|
CollaborationSession.file_id == item.id,
|
|
CollaborationSession.generation == item.generation,
|
|
))
|
|
await office.process_callback(env.db, closed.id, CollaborationCallbackPayload(
|
|
key=closed.document_key,
|
|
status=4,
|
|
))
|
|
recovery = AsyncMock(return_value=None)
|
|
monkeypatch.setattr(office, "_recover_forgotten_content", recovery)
|
|
|
|
reopened = await office.build_editor_config(env.db, item, env.admin)
|
|
await env.db.refresh(item)
|
|
await env.db.refresh(closed)
|
|
sessions = (await env.db.scalars(select(CollaborationSession).where(
|
|
CollaborationSession.file_id == item.id,
|
|
).order_by(CollaborationSession.generation))).all()
|
|
|
|
assert first.config["document"]["key"] != reopened.config["document"]["key"]
|
|
assert item.generation == 2
|
|
assert closed.status == "CLOSED"
|
|
assert [session.generation for session in sessions] == [1, 2]
|
|
recovery.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stale_active_ledger_session_starts_a_new_key_when_document_server_has_no_users(
|
|
env, monkeypatch
|
|
):
|
|
item = await env.db.get(CollaborationFile, env.initial[0].id)
|
|
first = await office.build_editor_config(env.db, item, env.admin)
|
|
stale = await env.db.scalar(select(CollaborationSession).where(
|
|
CollaborationSession.file_id == item.id,
|
|
CollaborationSession.generation == item.generation,
|
|
))
|
|
stale.last_callback_at = datetime.now(timezone.utc)
|
|
await env.db.commit()
|
|
live_users = AsyncMock(return_value=[])
|
|
recovery = AsyncMock(return_value=None)
|
|
monkeypatch.setattr(office, "_document_server_users", live_users)
|
|
monkeypatch.setattr(office, "_recover_forgotten_content", recovery)
|
|
|
|
reopened = await office.build_editor_config(env.db, item, env.admin)
|
|
await env.db.refresh(item)
|
|
await env.db.refresh(stale)
|
|
|
|
assert first.config["document"]["key"] != reopened.config["document"]["key"]
|
|
assert item.generation == 2
|
|
assert stale.status == "CLOSED"
|
|
live_users.assert_awaited_once_with(stale.document_key)
|
|
recovery.assert_awaited_once_with(stale.document_key, "cell")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_active_session_is_reused_during_browser_connection_grace(env, monkeypatch):
|
|
item = await env.db.get(CollaborationFile, env.initial[0].id)
|
|
first = await office.build_editor_config(env.db, item, env.admin)
|
|
live_users = AsyncMock(return_value=[])
|
|
monkeypatch.setattr(office, "_document_server_users", live_users)
|
|
|
|
repeated = await office.build_editor_config(env.db, item, env.admin)
|
|
|
|
assert first.config["document"]["key"] == repeated.config["document"]["key"]
|
|
live_users.assert_not_awaited()
|
|
|
|
|
|
@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
|