fix(ci): url_security格式化 + 修复8个单测mock路径适配
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 2m12s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 2m24s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 3m20s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m28s

- 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测试
This commit is contained in:
CI Bot
2026-07-14 20:38:29 +08:00
parent fb33b0647f
commit 4f808d4f8a
3 changed files with 35 additions and 37 deletions
+12 -14
View File
@@ -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")],
+20 -20
View File
@@ -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()
+3 -3
View File
@@ -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