881eea9195
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 36s
CI/CD Pipeline / Frontend Lint (push) Successful in 51s
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Unit Tests (push) Successful in 56s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Failing after 9s
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Successful in 1m50s
fix(api+ci): DELETE 204响应体全修复 + isort/black整理 + CI环境兼容 + 测试修复
59 lines
2.1 KiB
Python
59 lines
2.1 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
|
|
|
|
import packages.shared.config as shared_config
|
|
|
|
|
|
def _reset_settings() -> None:
|
|
app_config._settings = None
|
|
shared_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")
|