Files
xiaoxia-saas/tests/unit/test_oss_direct_upload.py
xiaoxia 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环境兼容 + 测试修复 (#286)
fix(api+ci): DELETE 204响应体全修复 + isort/black整理 + CI环境兼容 + 测试修复
2026-07-14 07:07:51 +08:00

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