diff --git a/tests/unit/test_pagination.py b/tests/unit/test_pagination.py index ae2b2fd14..37ee01da8 100755 --- a/tests/unit/test_pagination.py +++ b/tests/unit/test_pagination.py @@ -12,13 +12,12 @@ import pytest from pydantic import ValidationError from packages.application.common.pagination import ( - PaginationParams, - PaginationMeta, PaginatedResponse, + PaginationMeta, + PaginationParams, paginate, ) - # ============================================================ # PaginationParams # ============================================================ diff --git a/tests/unit/test_storage_service.py b/tests/unit/test_storage_service.py index 9ce15dac9..605ba3a8c 100755 --- a/tests/unit/test_storage_service.py +++ b/tests/unit/test_storage_service.py @@ -26,7 +26,6 @@ from packages.shared.storage import ( get_storage_service, ) - # ============================================================ # Fixtures # ============================================================ @@ -49,9 +48,10 @@ def _make_service( mock_bucket = MagicMock() if with_bucket else None - with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \ - patch.dict(os.environ, {"GENERATED_FILES_URL_PREFIX": local_url_prefix}, clear=False): - + with ( + patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), + patch.dict(os.environ, {"GENERATED_FILES_URL_PREFIX": local_url_prefix}, clear=False), + ): if with_bucket: with patch("packages.shared.storage.oss2") as mock_oss2: mock_oss2.Auth.return_value = MagicMock() @@ -85,51 +85,37 @@ class TestNormalizeStorageKey: def test_https_url_extracts_path(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/clip.mp4" - ) + result = service._normalize_storage_key("https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/clip.mp4") assert result == "videos/clip.mp4" def test_http_url_extracts_path(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "http://test-bucket.oss-cn-hangzhou.aliyuncs.com/audio/voice.mp3" - ) + result = service._normalize_storage_key("http://test-bucket.oss-cn-hangzhou.aliyuncs.com/audio/voice.mp3") assert result == "audio/voice.mp3" def test_url_with_query_strips_query(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://bucket.oss-cn.com/file.mp4?signature=abc&expires=123" - ) + result = service._normalize_storage_key("https://bucket.oss-cn.com/file.mp4?signature=abc&expires=123") assert result == "file.mp4" def test_url_with_leading_slash_in_path(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://bucket.oss.com//double/slash.jpg" - ) + result = service._normalize_storage_key("https://bucket.oss.com//double/slash.jpg") assert result == "double/slash.jpg" def test_url_decodes_percent_encoded_spaces(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://bucket.oss.com/my%20video.mp4" - ) + result = service._normalize_storage_key("https://bucket.oss.com/my%20video.mp4") assert result == "my video.mp4" def test_url_decodes_percent_encoded_chinese(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://bucket.oss.com/%E4%B8%AD%E6%96%87.mp4" - ) + result = service._normalize_storage_key("https://bucket.oss.com/%E4%B8%AD%E6%96%87.mp4") assert result == "中文.mp4" def test_url_with_special_chars_decoded(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://bucket.oss.com/file%281%29.jpg" - ) + result = service._normalize_storage_key("https://bucket.oss.com/file%281%29.jpg") assert result == "file(1).jpg" def test_plain_key_with_percent_not_decoded(self): @@ -151,16 +137,12 @@ class TestNormalizeStorageKey: def test_nested_path_url(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://bucket.oss.com/a/b/c/d/file.txt" - ) + result = service._normalize_storage_key("https://bucket.oss.com/a/b/c/d/file.txt") assert result == "a/b/c/d/file.txt" def test_url_with_port(self): service, _, _ = _make_service() - result = service._normalize_storage_key( - "https://bucket.oss.com:443/file.txt" - ) + result = service._normalize_storage_key("https://bucket.oss.com:443/file.txt") assert result == "file.txt" @@ -183,9 +165,7 @@ class TestIsLocalGeneratedUrl: def test_full_url_with_local_path_returns_true(self): service, _, _ = _make_service() - assert service._is_local_generated_url( - "https://example.com/generated-files/abc.mp4" - ) is True + assert service._is_local_generated_url("https://example.com/generated-files/abc.mp4") is True def test_other_path_returns_false(self): service, _, _ = _make_service() @@ -252,31 +232,23 @@ class TestCreateDirectUploadPost: def test_method_is_post(self): service, _, _ = _make_service() - result = service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 1024, 3600 - ) + result = service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 1024, 3600) assert result["method"] == "POST" def test_url_is_public_url(self): service, _, _ = _make_service() - result = service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 1024, 3600 - ) + result = service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 1024, 3600) assert result["url"] == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com" def test_storage_key_normalized(self): service, _, _ = _make_service() - result = service.create_direct_upload_post( - "/uploads/test.jpg", "image/jpeg", 1024, 3600 - ) + result = service.create_direct_upload_post("/uploads/test.jpg", "image/jpeg", 1024, 3600) assert result["storage_key"] == "uploads/test.jpg" assert result["fields"]["key"] == "uploads/test.jpg" def test_fields_contain_required_keys(self): service, _, _ = _make_service() - result = service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 1024, 3600 - ) + result = service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 1024, 3600) fields = result["fields"] assert fields["key"] == "uploads/a.jpg" assert fields["OSSAccessKeyId"] == "test-key-id" @@ -289,9 +261,7 @@ class TestCreateDirectUploadPost: """验证 HMAC-SHA1 签名是否正确""" secret = "my-secret-key-123" service, _, _ = _make_service(access_key_secret=secret) - result = service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 1024, 3600 - ) + result = service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 1024, 3600) policy = result["fields"]["policy"] signature = result["fields"]["Signature"] @@ -303,9 +273,7 @@ class TestCreateDirectUploadPost: def test_policy_contains_bucket_and_key(self): service, _, _ = _make_service(bucket_name="my-bucket") - result = service.create_direct_upload_post( - "uploads/photo.png", "image/png", 2048, 1800 - ) + result = service.create_direct_upload_post("uploads/photo.png", "image/png", 2048, 1800) policy = json.loads(base64.b64decode(result["fields"]["policy"])) conditions = policy["conditions"] @@ -314,9 +282,7 @@ class TestCreateDirectUploadPost: def test_policy_contains_content_length_range(self): service, _, _ = _make_service() - result = service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 5242880, 3600 - ) + result = service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 5242880, 3600) policy = json.loads(base64.b64decode(result["fields"]["policy"])) conditions = policy["conditions"] @@ -327,9 +293,7 @@ class TestCreateDirectUploadPost: def test_policy_content_type_starts_with(self): service, _, _ = _make_service() - result = service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 1024, 3600 - ) + result = service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 1024, 3600) policy = json.loads(base64.b64decode(result["fields"]["policy"])) conditions = policy["conditions"] @@ -340,9 +304,7 @@ class TestCreateDirectUploadPost: def test_policy_has_expiration(self): service, _, _ = _make_service() - result = service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 1024, 3600 - ) + result = service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 1024, 3600) policy = json.loads(base64.b64decode(result["fields"]["policy"])) assert "expiration" in policy # ISO 8601 格式 @@ -351,16 +313,12 @@ class TestCreateDirectUploadPost: def test_non_uploads_key_raises_value_error(self): service, _, _ = _make_service() with pytest.raises(ValueError, match="uploads/"): - service.create_direct_upload_post( - "videos/a.mp4", "video/mp4", 1024, 3600 - ) + service.create_direct_upload_post("videos/a.mp4", "video/mp4", 1024, 3600) def test_no_credentials_raises_runtime_error(self): service, _, _ = _make_service(access_key_id="", access_key_secret="", with_bucket=False) with pytest.raises(RuntimeError, match="not configured"): - service.create_direct_upload_post( - "uploads/a.jpg", "image/jpeg", 1024, 3600 - ) + service.create_direct_upload_post("uploads/a.jpg", "image/jpeg", 1024, 3600) def test_url_normalized_key_in_uploads(self): service, _, _ = _make_service() @@ -394,9 +352,7 @@ class TestGetDownloadUrlFallback: def test_no_bucket_url_input_normalized(self): service, _, _ = _make_service(with_bucket=False) - result = service.get_download_url( - "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4" - ) + result = service.get_download_url("https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4") assert result == "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/videos/test.mp4" def test_with_bucket_calls_sign_url(self): @@ -534,8 +490,10 @@ class TestInitEndpointHandling: mock_settings.oss_access_key_id = "key-id" mock_settings.oss_access_key_secret = "key-secret" - with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \ - patch("packages.shared.storage.oss2") as mock_oss2: + with ( + patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), + patch("packages.shared.storage.oss2") as mock_oss2, + ): mock_oss2.Bucket.return_value = MagicMock() service = SharedStorageService() @@ -551,8 +509,10 @@ class TestInitEndpointHandling: mock_settings.oss_access_key_id = "key-id" mock_settings.oss_access_key_secret = "key-secret" - with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \ - patch("packages.shared.storage.oss2") as mock_oss2: + with ( + patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), + patch("packages.shared.storage.oss2") as mock_oss2, + ): mock_oss2.Bucket.return_value = MagicMock() service = SharedStorageService() @@ -567,8 +527,10 @@ class TestInitEndpointHandling: mock_settings.oss_access_key_id = "key-id" mock_settings.oss_access_key_secret = "key-secret" - with patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), \ - patch("packages.shared.storage.oss2") as mock_oss2: + with ( + patch("packages.shared.storage.get_shared_settings", return_value=mock_settings), + patch("packages.shared.storage.oss2") as mock_oss2, + ): mock_oss2.Bucket.return_value = MagicMock() service = SharedStorageService() diff --git a/tests/unit/test_url_security.py b/tests/unit/test_url_security.py index e0a01d6ad..20bdcb026 100755 --- a/tests/unit/test_url_security.py +++ b/tests/unit/test_url_security.py @@ -11,28 +11,27 @@ url_security URL安全校验单元测试 import os import tempfile +from unittest.mock import MagicMock, patch import pytest -from unittest.mock import patch, MagicMock from packages.shared.url_security import ( - ALLOWED_SCHEMES, - ALLOWED_PORTS, - MAX_URL_LENGTH, - UrlSecurityError, - NoRedirectHandler, - validate_url_safety, - is_url_safe, - _validate_magic_number, - _check_internal_hostnames, - _is_trusted_domain, - safe_download_file, - safe_download_bytes, ALLOWED_AUDIO_MIME_TYPES, ALLOWED_IMAGE_MIME_TYPES, + ALLOWED_PORTS, + ALLOWED_SCHEMES, + MAX_URL_LENGTH, + NoRedirectHandler, + UrlSecurityError, + _check_internal_hostnames, + _is_trusted_domain, + _validate_magic_number, + is_url_safe, + safe_download_bytes, + safe_download_file, + validate_url_safety, ) - # ── validate_url_safety 基础校验 ───────────────────────────────────────────── @@ -203,9 +202,10 @@ class TestTrustedDomains: def test_is_trusted_domain_exact_match(self): with patch("packages.shared.url_security.TRUSTED_DOMAINS", {"example.com", "cdn.example.org"}): # 重新加载模块以应用环境变量不太现实，直接测函数 - from packages.shared.url_security import _is_trusted_domain # 直接改全局状态再还原 import packages.shared.url_security as mod + from packages.shared.url_security import _is_trusted_domain + original = mod.TRUSTED_DOMAINS mod.TRUSTED_DOMAINS = {"example.com", "cdn.example.org"} try: @@ -216,6 +216,7 @@ class TestTrustedDomains: def test_is_trusted_domain_subdomain(self): import packages.shared.url_security as mod + original = mod.TRUSTED_DOMAINS mod.TRUSTED_DOMAINS = {"example.com"} try: @@ -226,6 +227,7 @@ class TestTrustedDomains: def test_is_trusted_domain_no_match(self): import packages.shared.url_security as mod + original = mod.TRUSTED_DOMAINS mod.TRUSTED_DOMAINS = {"example.com"} try: @@ -237,6 +239,7 @@ class TestTrustedDomains: def test_validate_with_trusted_domains_restricted(self): """白名单非空时，不在白名单中的域名被拒""" import packages.shared.url_security as mod + original = mod.TRUSTED_DOMAINS mod.TRUSTED_DOMAINS = {"trusted.com"} try: @@ -297,8 +300,12 @@ class TestNoRedirectHandler: def test_redirect_request_returns_none(self): handler = NoRedirectHandler() result = handler.redirect_request( - MagicMock(), MagicMock(), 302, "Found", - {"Location": "http://other.com"}, "http://other.com", + MagicMock(), + MagicMock(), + 302, + "Found", + {"Location": "http://other.com"}, + "http://other.com", ) assert result is None @@ -312,80 +319,80 @@ class TestMagicNumberValidation: def test_valid_png(self, tmp_path): f = tmp_path / "test.png" # PNG 文件头: 89 50 4E 47 0D 0A 1A 0A - f.write_bytes(b'\x89PNG\r\n\x1a\n' + b'\x00' * 100) + f.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100) # 不抛异常 = 通过 _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) def test_valid_jpeg(self, tmp_path): f = tmp_path / "test.jpg" # JPEG 文件头: FF D8 FF - f.write_bytes(b'\xff\xd8\xff\xe0' + b'\x00' * 100) + f.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) def test_valid_gif87a(self, tmp_path): f = tmp_path / "test.gif" - f.write_bytes(b'GIF87a' + b'\x00' * 100) + f.write_bytes(b"GIF87a" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) def test_valid_gif89a(self, tmp_path): f = tmp_path / "test.gif" - f.write_bytes(b'GIF89a' + b'\x00' * 100) + f.write_bytes(b"GIF89a" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) def test_valid_webp(self, tmp_path): f = tmp_path / "test.webp" # RIFF....WEBP - data = bytearray(b'RIFF') - data += b'\x00\x00\x00\x00' # size placeholder - data += b'WEBP' - data += b'\x00' * 100 + data = bytearray(b"RIFF") + data += b"\x00\x00\x00\x00" # size placeholder + data += b"WEBP" + data += b"\x00" * 100 f.write_bytes(bytes(data)) _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) def test_valid_bmp(self, tmp_path): f = tmp_path / "test.bmp" - f.write_bytes(b'BM' + b'\x00' * 100) + f.write_bytes(b"BM" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) def test_valid_wav(self, tmp_path): f = tmp_path / "test.wav" # RIFF....WAVE - data = bytearray(b'RIFF') - data += b'\x00\x00\x00\x00' - data += b'WAVE' - data += b'\x00' * 100 + data = bytearray(b"RIFF") + data += b"\x00\x00\x00\x00" + data += b"WAVE" + data += b"\x00" * 100 f.write_bytes(bytes(data)) _validate_magic_number(str(f), ALLOWED_AUDIO_MIME_TYPES) def test_valid_mp3_id3(self, tmp_path): f = tmp_path / "test.mp3" - f.write_bytes(b'ID3\x03\x00\x00\x00\x00\x00\x00' + b'\x00' * 100) + f.write_bytes(b"ID3\x03\x00\x00\x00\x00\x00\x00" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_AUDIO_MIME_TYPES) def test_valid_mp3_adts(self, tmp_path): f = tmp_path / "test.mp3" - f.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 100) + f.write_bytes(b"\xff\xfb\x90\x00" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_AUDIO_MIME_TYPES) def test_valid_ogg(self, tmp_path): f = tmp_path / "test.ogg" - f.write_bytes(b'OggS\x00\x02\x00\x00' + b'\x00' * 100) + f.write_bytes(b"OggS\x00\x02\x00\x00" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_AUDIO_MIME_TYPES) def test_valid_flac(self, tmp_path): f = tmp_path / "test.flac" - f.write_bytes(b'fLaC' + b'\x00' * 100) + f.write_bytes(b"fLaC" + b"\x00" * 100) _validate_magic_number(str(f), ALLOWED_AUDIO_MIME_TYPES) def test_invalid_file_content_raises(self, tmp_path): f = tmp_path / "test.bin" - f.write_bytes(b'this is not an image file at all') + f.write_bytes(b"this is not an image file at all") with pytest.raises(UrlSecurityError, match="魔数"): _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) def test_empty_file_raises(self, tmp_path): f = tmp_path / "empty.bin" - f.write_bytes(b'') + f.write_bytes(b"") with pytest.raises(UrlSecurityError, match="为空"): _validate_magic_number(str(f), ALLOWED_IMAGE_MIME_TYPES) @@ -396,14 +403,14 @@ class TestMagicNumberValidation: def test_no_allowed_mime_types_skips(self, tmp_path): """allowed_mime_types 为空时跳过校验""" f = tmp_path / "test.bin" - f.write_bytes(b'random data here') + f.write_bytes(b"random data here") # 不抛异常 _validate_magic_number(str(f), set()) def test_unknown_mime_types_skips(self, tmp_path): """没有已知魔数的 MIME 类型跳过校验""" f = tmp_path / "test.bin" - f.write_bytes(b'random data') + f.write_bytes(b"random data") _validate_magic_number(str(f), {"application/x-unknown-type"}) @@ -441,7 +448,7 @@ class TestSafeDownloadFile: def test_download_with_mime_check_passes(self, tmp_path): # PNG 文件 - test_content = b'\x89PNG\r\n\x1a\n' + b'\x00' * 200 + test_content = b"\x89PNG\r\n\x1a\n" + b"\x00" * 200 dest = str(tmp_path / "test.png") mock_resp = MagicMock() @@ -463,7 +470,7 @@ class TestSafeDownloadFile: assert size == len(test_content) def test_download_mime_type_rejected(self, tmp_path): - test_content = b'GIF89a' + b'\x00' * 50 + test_content = b"GIF89a" + b"\x00" * 50 dest = str(tmp_path / "test.gif") mock_resp = MagicMock()