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")