From 4f808d4f8a8ada1bc89252a6a9419c1dca7ee75d Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 14 Jul 2026 20:38:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(ci):=20url=5Fsecurity=E6=A0=BC=E5=BC=8F?= =?UTF-8?q?=E5=8C=96=20+=20=E4=BF=AE=E5=A4=8D8=E4=B8=AA=E5=8D=95=E6=B5=8Bm?= =?UTF-8?q?ock=E8=B7=AF=E5=BE=84=E9=80=82=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - packages/shared/url_security.py: black格式化 - test_generation_p3_optimizations.py: mock从urllib.request.urlopen改为OpenerDirector.open （适配手动重定向实现，新旧实现均兼容） - test_oneclick_gen_p0_fixes.py: 同上，3个_verify_url_accessible测试 --- packages/shared/url_security.py | 26 ++++++------ .../unit/test_generation_p3_optimizations.py | 40 +++++++++---------- tests/unit/test_oneclick_gen_p0_fixes.py | 6 +-- 3 files changed, 35 insertions(+), 37 deletions(-) diff --git a/packages/shared/url_security.py b/packages/shared/url_security.py index a8a02eb83..d2220376f 100755 --- a/packages/shared/url_security.py +++ b/packages/shared/url_security.py @@ -98,15 +98,15 @@ _MAX_REDIRECTS = 5 _MAGIC_NUMBERS: dict[str, list[list[tuple[int, bytes]]]] = { # ── 音频 ── "audio/mpeg": [ - [(0, b"ID3")], # ID3v2 标签 - [(0, b"\xff\xfb")], # MPEG1 Layer3 - [(0, b"\xff\xf3")], # MPEG2 Layer3 - [(0, b"\xff\xf2")], # MPEG2.5 Layer3 - [(0, b"\xff\xfa")], # MPEG1 Layer2 - [(0, b"\xff\xf9")], # 其他 MPEG ADTS + [(0, b"ID3")], # ID3v2 标签 + [(0, b"\xff\xfb")], # MPEG1 Layer3 + [(0, b"\xff\xf3")], # MPEG2 Layer3 + [(0, b"\xff\xf2")], # MPEG2.5 Layer3 + [(0, b"\xff\xfa")], # MPEG1 Layer2 + [(0, b"\xff\xf9")], # 其他 MPEG ADTS ], "audio/wav": [ - [(0, b"RIFF"), (8, b"WAVE")], # RIFF + WAVE + [(0, b"RIFF"), (8, b"WAVE")], # RIFF + WAVE ], "audio/x-wav": [ [(0, b"RIFF"), (8, b"WAVE")], @@ -121,29 +121,28 @@ _MAGIC_NUMBERS: dict[str, list[list[tuple[int, bytes]]]] = { [(0, b"fLaC")], ], "audio/aac": [ - [(0, b"\xff\xf1")], # ADTS MPEG-4 - [(0, b"\xff\xf9")], # ADTS MPEG-2 + [(0, b"\xff\xf1")], # ADTS MPEG-4 + [(0, b"\xff\xf9")], # ADTS MPEG-2 ], "audio/aacp": [ [(0, b"\xff\xf1")], [(0, b"\xff\xf9")], ], "audio/mp4": [ - [(4, b"ftyp")], # ISO Base Media (M4A) + [(4, b"ftyp")], # ISO Base Media (M4A) ], "audio/x-m4a": [ [(4, b"ftyp")], ], - # ── 视频 ── "video/mp4": [ - [(4, b"ftyp")], # ISO Base Media (MP4) + [(4, b"ftyp")], # ISO Base Media (MP4) ], "video/quicktime": [ [(4, b"ftyp")], ], "video/x-matroska": [ - [(0, b"\x1a\x45\xdf\xa3")], # EBML header + [(0, b"\x1a\x45\xdf\xa3")], # EBML header ], "video/webm": [ [(0, b"\x1a\x45\xdf\xa3")], @@ -151,7 +150,6 @@ _MAGIC_NUMBERS: dict[str, list[list[tuple[int, bytes]]]] = { "video/x-msvideo": [ [(0, b"RIFF"), (8, b"AVI ")], ], - # ── 图片 ── "image/jpeg": [ [(0, b"\xff\xd8\xff")], diff --git a/tests/unit/test_generation_p3_optimizations.py b/tests/unit/test_generation_p3_optimizations.py index d4c379b2a..ff4229083 100644 --- a/tests/unit/test_generation_p3_optimizations.py +++ b/tests/unit/test_generation_p3_optimizations.py @@ -25,8 +25,8 @@ class TestVerifyUrlAccessibleRetry: """_verify_url_accessible 重试逻辑.""" @patch("time.sleep") - @patch("urllib.request.urlopen") - def test_first_attempt_success(self, mock_urlopen, mock_sleep): + @patch("urllib.request.OpenerDirector.open") + def test_first_attempt_success(self, mock_open, mock_sleep): """首次成功，不重试.""" from worker_app.tasks.generation import _verify_url_accessible @@ -34,15 +34,15 @@ class TestVerifyUrlAccessibleRetry: mock_resp.status = 200 mock_resp.__enter__ = MagicMock(return_value=mock_resp) mock_resp.__exit__ = MagicMock(return_value=False) - mock_urlopen.return_value = mock_resp + mock_open.return_value = mock_resp assert _verify_url_accessible("https://example.com/file.mp4") is True - assert mock_urlopen.call_count == 1 + assert mock_open.call_count == 1 mock_sleep.assert_not_called() @patch("time.sleep") - @patch("urllib.request.urlopen") - def test_retry_then_success(self, mock_urlopen, mock_sleep): + @patch("urllib.request.OpenerDirector.open") + def test_retry_then_success(self, mock_open, mock_sleep): """首次失败，重试后成功.""" from worker_app.tasks.generation import _verify_url_accessible @@ -52,31 +52,31 @@ class TestVerifyUrlAccessibleRetry: mock_resp_ok.__enter__ = MagicMock(return_value=mock_resp_ok) mock_resp_ok.__exit__ = MagicMock(return_value=False) - mock_urlopen.side_effect = [ + mock_open.side_effect = [ OSError("connection reset"), mock_resp_ok, ] assert _verify_url_accessible("https://example.com/file.mp4") is True - assert mock_urlopen.call_count == 2 + assert mock_open.call_count == 2 mock_sleep.assert_called_once_with(1) @patch("time.sleep") - @patch("urllib.request.urlopen") - def test_all_retries_exhausted(self, mock_urlopen, mock_sleep): + @patch("urllib.request.OpenerDirector.open") + def test_all_retries_exhausted(self, mock_open, mock_sleep): """全部重试耗尽，返回 False.""" from worker_app.tasks.generation import _verify_url_accessible - mock_urlopen.side_effect = OSError("connection refused") + mock_open.side_effect = OSError("connection refused") assert _verify_url_accessible("https://example.com/file.mp4") is False # 1 首次 + 2 重试 = 3 次 - assert mock_urlopen.call_count == 3 + assert mock_open.call_count == 3 assert mock_sleep.call_count == 2 @patch("time.sleep") - @patch("urllib.request.urlopen") - def test_http_500_then_success(self, mock_urlopen, mock_sleep): + @patch("urllib.request.OpenerDirector.open") + def test_http_500_then_success(self, mock_open, mock_sleep): """HTTP 500 后重试成功.""" from worker_app.tasks.generation import _verify_url_accessible @@ -90,21 +90,21 @@ class TestVerifyUrlAccessibleRetry: mock_resp_200.__enter__ = MagicMock(return_value=mock_resp_200) mock_resp_200.__exit__ = MagicMock(return_value=False) - mock_urlopen.side_effect = [mock_resp_500, mock_resp_200] + mock_open.side_effect = [mock_resp_500, mock_resp_200] assert _verify_url_accessible("https://example.com/file.mp4") is True - assert mock_urlopen.call_count == 2 + assert mock_open.call_count == 2 @patch("time.sleep") - @patch("urllib.request.urlopen") - def test_custom_retries_zero(self, mock_urlopen, mock_sleep): + @patch("urllib.request.OpenerDirector.open") + def test_custom_retries_zero(self, mock_open, mock_sleep): """retries=0 时不重试.""" from worker_app.tasks.generation import _verify_url_accessible - mock_urlopen.side_effect = OSError("timeout") + mock_open.side_effect = OSError("timeout") assert _verify_url_accessible("https://example.com/file.mp4", retries=0) is False - assert mock_urlopen.call_count == 1 + assert mock_open.call_count == 1 mock_sleep.assert_not_called() diff --git a/tests/unit/test_oneclick_gen_p0_fixes.py b/tests/unit/test_oneclick_gen_p0_fixes.py index efde21b16..9f259f484 100644 --- a/tests/unit/test_oneclick_gen_p0_fixes.py +++ b/tests/unit/test_oneclick_gen_p0_fixes.py @@ -162,14 +162,14 @@ class TestOSSUploadAndVerify: mock_response.__enter__ = MagicMock(return_value=mock_response) mock_response.__exit__ = MagicMock(return_value=False) - with patch("urllib.request.urlopen", return_value=mock_response): + with patch("urllib.request.OpenerDirector.open", return_value=mock_response): assert _verify_url_accessible("https://example.com/test.mp4") is True def test_verify_url_accessible_failure(self): """URL 不可访问时返回 False.""" from worker_app.tasks.generation import _verify_url_accessible - with patch("urllib.request.urlopen", side_effect=Exception("connection refused")): + with patch("urllib.request.OpenerDirector.open", side_effect=Exception("connection refused")): assert _verify_url_accessible("https://example.com/test.mp4") is False def test_verify_url_404(self): @@ -181,7 +181,7 @@ class TestOSSUploadAndVerify: mock_response.__enter__ = MagicMock(return_value=mock_response) mock_response.__exit__ = MagicMock(return_value=False) - with patch("urllib.request.urlopen", return_value=mock_response): + with patch("urllib.request.OpenerDirector.open", return_value=mock_response): assert _verify_url_accessible("https://example.com/test.mp4") is False