Files
xiaoxia-saas/tests/unit/test_oss_direct_upload.py
T
2026-06-23 15:47:03 +08:00

56 lines
2.0 KiB
Python

import base64
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
import app.config as app_config
from app.core.storage import OSSStorageService
def _reset_settings() -> None:
app_config._settings = None
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")