release(main): 同步台账恢复与字体合规调整

backend/tests/test_collaboration_service.py

@@ -316,6 +316,74 @@ def test_result_download_url_is_restricted_and_public_proxy_urls_are_rewritten(m
             onlyoffice_collaboration_service._validate_result_url(value)
 
 
+@pytest.mark.asyncio
+async def test_forgotten_document_command_downloads_and_validates_server_backup(monkeypatch):
+    key = "ctms-collab-forgotten-key"
+    recovered = collaboration_service.blank_file_bytes("cell")
+    request = {}
+
+    class FakeResponse:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"error": 0, "key": key, "url": "http://onlyoffice/cache/forgotten.xlsx"}
+
+    class FakeClient:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *_args):
+            return None
+
+        async def post(self, url, *, params, json):
+            request.update(url=url, params=params, body=json)
+            return FakeResponse()
+
+    monkeypatch.setattr(onlyoffice_collaboration_service.httpx, "AsyncClient", lambda **_kwargs: FakeClient())
+    download = AsyncMock(return_value=recovered)
+    monkeypatch.setattr(onlyoffice_collaboration_service, "_download_result", download)
+
+    result = await onlyoffice_collaboration_service._recover_forgotten_content(key, "cell")
+
+    assert result == recovered
+    assert request["url"] == "http://onlyoffice/command"
+    assert request["params"] == {"shardkey": key}
+    assert jwt.decode(
+        request["body"]["token"], settings.ONLYOFFICE_JWT_SECRET, algorithms=["HS256"]
+    ) == {"c": "getForgotten", "key": key}
+    download.assert_awaited_once_with("http://onlyoffice/cache/forgotten.xlsx")
+
+
+@pytest.mark.asyncio
+async def test_forgotten_document_command_rejects_a_damaged_backup(monkeypatch):
+    key = "ctms-collab-damaged-key"
+
+    class FakeResponse:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"error": 0, "key": key, "url": "http://onlyoffice/cache/damaged.xlsx"}
+
+    class FakeClient:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *_args):
+            return None
+
+        async def post(self, *_args, **_kwargs):
+            return FakeResponse()
+
+    monkeypatch.setattr(onlyoffice_collaboration_service.httpx, "AsyncClient", lambda **_kwargs: FakeClient())
+    monkeypatch.setattr(onlyoffice_collaboration_service, "_download_result", AsyncMock(return_value=b"broken"))
+
+    with pytest.raises(HTTPException) as error:
+        await onlyoffice_collaboration_service._recover_forgotten_content(key, "cell")
+    assert error.value.status_code == 502
+
+
 @pytest.mark.asyncio
 async def test_editor_config_grants_edit_only_after_collaboration_permission(monkeypatch, tmp_path):
     user_id = uuid.uuid4()
@@ -1016,6 +1084,31 @@ async def test_callback_error_status_is_recorded_and_acknowledged():
     assert db.added[0].result == "ERROR"
 
 
+@pytest.mark.asyncio
+async def test_force_save_error_keeps_the_live_session_active():
+    session = SimpleNamespace(
+        id=uuid.uuid4(),
+        file_id=uuid.uuid4(),
+        document_key="ctms-collab-force-save-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=7),
+    )
+
+    assert result == {"error": 0}
+    assert session.status == "ACTIVE"
+    assert db.added[0].result == "ERROR"
+
+
 @pytest.mark.asyncio
 async def test_force_save_updates_session_recovery_revision(monkeypatch):
     session = SimpleNamespace(