xiaoxia-saas/tests/unit/test_oss_direct_upload.py

import base64
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

from app.core.storage import OSSStorageService

from packages.config import reload_settings_cache


def _reset_settings() -> None:
    """清空所有配置缓存，让测试设置的环境变量生效。"""
    reload_settings_cache()


def test_create_direct_upload_post_limits_key_and_size(monkeypatch):
    monkeypatch.setenv("OSS_ACCESS_KEY_ID", "test-access")
    monkeypatch.setenv("OSS_ACCESS_KEY_SECRET", "test-secret")
    monkeypatch.setenv("OSS_BUCKET_NAME", "test-bucket")
    monkeypatch.setenv("OSS_ENDPOINT", "oss-cn-hangzhou.aliyuncs.com")
    _reset_settings()

    service = OSSStorageService()
    payload = service.create_direct_upload_post(
        storage_key="uploads/abc/sample.MOV",
        content_type="video/quicktime",
        max_size_bytes=800 * 1024 * 1024,
        expires_seconds=900,
    )

    assert payload["url"] == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com"
    assert payload["method"] == "POST"
    assert payload["storage_key"] == "uploads/abc/sample.MOV"
    fields = payload["fields"]
    assert fields["key"] == "uploads/abc/sample.MOV"
    assert fields["OSSAccessKeyId"] == "test-access"
    assert fields["success_action_status"] == "201"
    policy = json.loads(base64.b64decode(fields["policy"]).decode("utf-8"))
    assert {"bucket": "test-bucket"} in policy["conditions"]
    assert {"key": "uploads/abc/sample.MOV"} in policy["conditions"]
    assert ["content-length-range", 1, 800 * 1024 * 1024] in policy["conditions"]


def test_create_direct_upload_post_rejects_non_upload_key(monkeypatch):
    monkeypatch.setenv("OSS_ACCESS_KEY_ID", "test-access")
    monkeypatch.setenv("OSS_ACCESS_KEY_SECRET", "test-secret")
    _reset_settings()

    service = OSSStorageService()
    try:
        service.create_direct_upload_post("generated/bad.mov", "video/quicktime", 100, 900)
    except ValueError as error:
        assert "uploads" in str(error)
    else:
        raise AssertionError("expected ValueError")