Files
X-Financial/server/tests/test_onlyoffice_callback_summary.py
caoxiaozhu 787bc3a481 feat(platform): close AI expense value loop
Add tenant-safe value, telemetry, connector, commercial, and production-readiness foundations.
2026-07-17 14:14:08 +08:00

84 lines
2.6 KiB
Python

from io import BytesIO
from unittest.mock import MagicMock, patch
from openpyxl import Workbook
from app.services.agent_asset_onlyoffice_security import (
AgentAssetOnlyOfficeValidatedSession,
)
from app.services.agent_assets import AgentAssetService
def test_onlyoffice_callback_delegates_downloaded_workbook_to_upload():
db = MagicMock()
service = AgentAssetService(db)
service._ensure_ready = MagicMock()
asset = MagicMock()
asset.id = "test-asset"
asset.tenant_id = "tenant-a"
asset.scope = "tenant"
service._require_spreadsheet_rule = MagicMock(return_value=asset)
current_metadata = MagicMock(file_name="test.xlsx", checksum="old-checksum")
service._resolve_current_spreadsheet_meta = MagicMock(
return_value=("current", current_metadata)
)
new_wb = Workbook()
new_ws = new_wb.active
new_ws["A1"] = "new value"
new_ws["B2"] = "added"
new_content_bio = BytesIO()
new_wb.save(new_content_bio)
new_content = new_content_bio.getvalue()
document_key = service._build_onlyoffice_document_key("test-asset", current_metadata)
claimed = AgentAssetOnlyOfficeValidatedSession(
jti="callback-jti",
tenant_id="tenant-a",
resource_scope="tenant",
asset_id="test-asset",
document_key=document_key,
document_version="current",
document_fingerprint="old-checksum",
writable=True,
actor="username:test_user",
status="processing",
)
session_service = MagicMock()
session_service.claim_callback.return_value = claimed
service._onlyoffice_session_service = MagicMock(return_value=session_service)
with patch(
"app.services.agent_asset_onlyoffice.download_onlyoffice_document",
return_value=new_content,
) as secure_download:
service.upload_rule_spreadsheet = MagicMock()
payload = {
"status": 2,
"url": "http://onlyoffice/download",
"key": document_key,
}
service.handle_rule_spreadsheet_onlyoffice_callback(
"test-asset",
version="current",
payload=payload,
callback_token="signed-callback-token",
)
secure_download.assert_called_once_with(
"http://onlyoffice/download",
expected_filename="test.xlsx",
)
service.upload_rule_spreadsheet.assert_called_once_with(
"test-asset",
filename="test.xlsx",
content=new_content,
actor="username:test_user",
source="onlyoffice",
)
session_service.finish_callback.assert_called_once_with(
"callback-jti",
succeeded=True,
)