Cheng Zhou
d5279b124f
Storage Persistence Guard / storage-persistence-audit (push) Has been cancelled
Client Quality Gates / Shared client and Web (push) Has been cancelled
Client Quality Gates / macOS Desktop (push) Has been cancelled
Client Quality Gates / Shared client and Web (pull_request) Has been cancelled
Client Quality Gates / macOS Desktop (pull_request) Has been cancelled
Storage Persistence Guard / storage-persistence-audit (pull_request) Has been cancelled
195 lines
8.1 KiB
Python
195 lines
8.1 KiB
Python
import uuid
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from jose import jwt
|
|
|
|
from app.core.config import settings, validate_onlyoffice_configuration
|
|
from app.core.exceptions import AppException
|
|
from app.services import onlyoffice_service
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def onlyoffice_settings(monkeypatch):
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_ENABLED", True)
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_JWT_SECRET", "office-preview-secret-that-is-long-enough")
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_INSTANCE_ID", "ctms-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(
|
|
("filename", "expected"),
|
|
[
|
|
("方案.DOCX", ("docx", "word")),
|
|
("数据.xlsx", ("xlsx", "cell")),
|
|
("培训.PPTX", ("pptx", "slide")),
|
|
("国产格式.wps", ("wps", "word")),
|
|
("国产格式.et", ("et", "cell")),
|
|
("国产格式.dps", ("dps", "slide")),
|
|
("扫描件.pdf", None),
|
|
("图片.png", None),
|
|
],
|
|
)
|
|
def test_office_format_contract(filename, expected):
|
|
assert onlyoffice_service.office_format_for_filename(filename) == expected
|
|
|
|
|
|
def test_all_supported_format_groups_match_the_public_contract():
|
|
expected = {
|
|
**{extension: "word" for extension in ("doc", "docx", "docm", "dot", "dotx", "dotm", "odt", "ott", "rtf", "txt", "wps", "wpt")},
|
|
**{extension: "cell" for extension in ("xls", "xlsx", "xlsm", "xlsb", "xlt", "xltx", "xltm", "ods", "ots", "csv", "et", "ett")},
|
|
**{extension: "slide" for extension in ("ppt", "pptx", "pptm", "pps", "ppsx", "ppsm", "pot", "potx", "potm", "odp", "otp", "dps", "dpt")},
|
|
}
|
|
for extension, document_type in expected.items():
|
|
assert onlyoffice_service.office_format_for_filename(f"原始文件名.{extension.upper()}") == (
|
|
extension,
|
|
document_type,
|
|
)
|
|
|
|
|
|
def test_preview_config_is_read_only_and_uses_original_filename():
|
|
resource_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
result = onlyoffice_service.build_preview_config(
|
|
resource_type="attachment",
|
|
resource_id=resource_id,
|
|
file_name="研究方案 V1.0.docx",
|
|
user_id=user_id,
|
|
user_name="测试用户",
|
|
)
|
|
|
|
document = result.config["document"]
|
|
permissions = document["permissions"]
|
|
assert result.file_name == "研究方案 V1.0.docx"
|
|
assert document["title"] == "研究方案 V1.0.docx"
|
|
assert document["fileType"] == "docx"
|
|
assert result.config["documentType"] == "word"
|
|
assert result.config["editorConfig"]["mode"] == "view"
|
|
assert all(permissions[key] is False for key in ("copy", "comment", "download", "edit", "fillForms", "print", "review"))
|
|
assert "token=" not in document["url"]
|
|
assert "access_token=" not in document["url"]
|
|
assert document["url"].endswith(f"/internal/onlyoffice/attachments/{resource_id}/content")
|
|
|
|
payload = jwt.decode(
|
|
result.config["token"],
|
|
settings.ONLYOFFICE_JWT_SECRET,
|
|
algorithms=["HS256"],
|
|
)
|
|
assert payload["document"]["title"] == "研究方案 V1.0.docx"
|
|
assert payload["editorConfig"]["user"] == {"id": str(user_id), "name": "测试用户"}
|
|
assert payload["exp"] - payload["iat"] == settings.ONLYOFFICE_CONFIG_TTL_SECONDS
|
|
|
|
|
|
def test_document_key_is_stable_and_version_hash_changes_it():
|
|
resource_id = uuid.uuid4()
|
|
first = onlyoffice_service.onlyoffice_document_key("version", resource_id, file_hash="hash-a")
|
|
repeated = onlyoffice_service.onlyoffice_document_key("version", resource_id, file_hash="hash-a")
|
|
changed = onlyoffice_service.onlyoffice_document_key("version", resource_id, file_hash="hash-b")
|
|
|
|
assert first == repeated
|
|
assert first != changed
|
|
assert first.startswith("ctms-")
|
|
assert len(first) <= 128
|
|
|
|
|
|
def test_outbox_token_must_be_hs256_and_bound_to_exact_url():
|
|
resource_id = uuid.uuid4()
|
|
expected_url = onlyoffice_service.onlyoffice_content_url("attachment", resource_id)
|
|
token = jwt.encode({"payload": {"url": expected_url}}, settings.ONLYOFFICE_JWT_SECRET, algorithm="HS256")
|
|
|
|
assert onlyoffice_service.validate_outbox_token(token, expected_url)["payload"]["url"] == expected_url
|
|
assert onlyoffice_service.validate_outbox_token(f"Bearer {token}", expected_url)["payload"]["url"] == expected_url
|
|
|
|
with pytest.raises(AppException) as mismatch:
|
|
onlyoffice_service.validate_outbox_token(token, f"{expected_url}-other")
|
|
assert mismatch.value.code == "ONLYOFFICE_SOURCE_URL_MISMATCH"
|
|
|
|
config_token = jwt.encode(
|
|
{"document": {"url": expected_url}},
|
|
settings.ONLYOFFICE_JWT_SECRET,
|
|
algorithm="HS256",
|
|
)
|
|
with pytest.raises(AppException) as wrong_purpose:
|
|
onlyoffice_service.validate_outbox_token(config_token, expected_url)
|
|
assert wrong_purpose.value.code == "ONLYOFFICE_SOURCE_URL_MISMATCH"
|
|
|
|
legacy_top_level_url = jwt.encode(
|
|
{"url": expected_url},
|
|
settings.ONLYOFFICE_JWT_SECRET,
|
|
algorithm="HS256",
|
|
)
|
|
with pytest.raises(AppException) as wrong_shape:
|
|
onlyoffice_service.validate_outbox_token(legacy_top_level_url, expected_url)
|
|
assert wrong_shape.value.code == "ONLYOFFICE_SOURCE_URL_MISMATCH"
|
|
|
|
substituted_algorithm = jwt.encode(
|
|
{"payload": {"url": expected_url}},
|
|
settings.ONLYOFFICE_JWT_SECRET,
|
|
algorithm="HS512",
|
|
)
|
|
with pytest.raises(AppException) as invalid_algorithm:
|
|
onlyoffice_service.validate_outbox_token(substituted_algorithm, expected_url)
|
|
assert invalid_algorithm.value.code == "ONLYOFFICE_SOURCE_UNAUTHORIZED"
|
|
|
|
with pytest.raises(AppException) as missing:
|
|
onlyoffice_service.validate_outbox_token(None, expected_url)
|
|
assert missing.value.code == "ONLYOFFICE_SOURCE_UNAUTHORIZED"
|
|
|
|
with pytest.raises(AppException) as wrong_scheme:
|
|
onlyoffice_service.validate_outbox_token(f"Basic {token}", expected_url)
|
|
assert wrong_scheme.value.code == "ONLYOFFICE_SOURCE_UNAUTHORIZED"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_probe_is_cached_for_fifteen_seconds(monkeypatch):
|
|
onlyoffice_service._health_checked_at = 0.0
|
|
onlyoffice_service._health_available = False
|
|
request = type("Response", (), {"status_code": 200})()
|
|
get = AsyncMock(return_value=request)
|
|
client = MagicMock()
|
|
client.__aenter__ = AsyncMock(return_value=type("Client", (), {"get": get})())
|
|
client.__aexit__ = AsyncMock(return_value=False)
|
|
monkeypatch.setattr(onlyoffice_service.httpx, "AsyncClient", MagicMock(return_value=client))
|
|
|
|
await onlyoffice_service.ensure_onlyoffice_available()
|
|
await onlyoffice_service.ensure_onlyoffice_available()
|
|
|
|
get.assert_awaited_once()
|
|
|
|
|
|
def test_enabled_configuration_requires_a_distinct_secret_and_instance(monkeypatch):
|
|
validate_onlyoffice_configuration()
|
|
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_JWT_SECRET", settings.JWT_SECRET_KEY)
|
|
with pytest.raises(RuntimeError, match="must not reuse"):
|
|
validate_onlyoffice_configuration()
|
|
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_JWT_SECRET", "office-preview-secret-that-is-long-enough")
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_INSTANCE_ID", "")
|
|
with pytest.raises(RuntimeError, match="INSTANCE_ID"):
|
|
validate_onlyoffice_configuration()
|
|
|
|
|
|
def test_enabled_configuration_rejects_unsafe_service_urls(monkeypatch):
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_INTERNAL_URL", "http://user:password@onlyoffice")
|
|
with pytest.raises(RuntimeError, match="credentials"):
|
|
validate_onlyoffice_configuration()
|
|
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_INTERNAL_URL", "http://onlyoffice/internal-path")
|
|
with pytest.raises(RuntimeError, match="must not contain a path"):
|
|
validate_onlyoffice_configuration()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_service_returns_stable_error(monkeypatch):
|
|
monkeypatch.setattr(settings, "ONLYOFFICE_ENABLED", False)
|
|
|
|
with pytest.raises(AppException) as error:
|
|
await onlyoffice_service.ensure_onlyoffice_available()
|
|
|
|
assert error.value.code == "ONLYOFFICE_DISABLED"
|
|
assert error.value.status_code == 503
|