From dfc5e5a5b62d6270440acefb4748cdbeef8e146d Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Tue, 29 Sep 2026 18:08:18 +0800 Subject: [PATCH 1/4] fix(worker): align gpu-direct title defaults (size/margin/bold) with CPU vfb path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #2095 fixed title baseline positioning and width-scaling but didn't fully align default constants with the CPU video_filter_builder path, causing GPU rendered titles to appear slightly smaller / higher / bolder-differently than the CPU/ASS preview the template was authored against. This shows up in both single and batch GPU renders since every variant shares the same gpu_direct pipeline. Root causes in PR #2095 defaults: - Default title size 36@720p vs config_schemas DEFAULT 48 / vfb default 48 - Top/bottom margin 40@720p (PAD16+margin24) vs vfb _scale_title_len(50) - Subtitle bottom margin 60@720p vs vfb 50 - Faux bold used same-color 1px border (white-on-white invisible for default white text; red-on-red for colored) vs vfb black 2px (intentional choice per #2001 to avoid double-print/halo artifact) - margin_top was treated as absolute y offset, now correctly added on top of base margin (matches vfb additive semantics) - subtitle stroke/bold block referenced uninitialized s_borderw (ruff F821) — added proper init + stroke parsing consistent with title block Changes: - TITLE_DEFAULT_MARGIN_TOP/BOTTOM = 50 (was 24+PAD=40) - SUBTITLE_DEFAULT_MARGIN_BOTTOM = 50 (was 60) - TITLE_FAUX_BOLD_WIDTH = 2, border color = #000000 (was 1, same as text) - default title size 48@720p (was 36) - margin_top from cfg added on top of base 50 (additive, same as vfb) - subtitle: init s_borderw/s_border_color, parse stroke dict before bold check - bottom position: y = h - th - margin_bottom (correct baseline; previously used margin_top variable which was misleading but numerically equivalent before margin split; now uses explicit margin_bottom) Tests updated to new expected scaled values (1280x720 scale=1.778: fontsize 85, y=89, bold borderw=4 black; 1080x1920 vertical: fontsize 72, y=75; margin_top=100 user offset => y=267). Added test_title_bold_false_disables_faux_bold. Batch investigation notes (separate findings, not bugs in this PR): - per-variant plan config is correctly deep-copied via clone_plan_for_variant (config=dict(source.config or {})); title text per-variant via titles[] override; voice per-variant independent download with #1749 strict guards; bgm merged via merge_bgm_config — batch passthrough chain is correct. - Worker generation concurrency = 2 (compose.yml default); USER_PENDING_LIMIT=3, 6 tasks will enqueue in two waves (first 3 → then next 3 as workers free up); GLOBAL_PENDING_LIMIT allows it. This is expected behavior, not a bug. - Legacy key mismatch: subtitle_render_engine.py reads plan_config['subtitle_config'] (snake_case) but normalize_plan_config writes 'subtitle' short key. This file is not imported by unified_render_service (which uses render_subtitles.py with explicit subtitle_config= parameter), so no runtime impact; left for separate cleanup. - Frontend usePlanConfigLoader.ts reads config.title_config snake_case while backend writes 'title' short key — frontend-only issue for subsequent edit sessions, out of backend scope. --- .../video_processing/gpu_direct_pipeline.py | 60 +++++++++++----- tests/unit/test_gpu_direct_pipeline.py | 72 +++++++++---------- 2 files changed, 79 insertions(+), 53 deletions(-) diff --git a/apps/worker/video_processing/gpu_direct_pipeline.py b/apps/worker/video_processing/gpu_direct_pipeline.py index 20520b1c8..d4763956a 100644 --- a/apps/worker/video_processing/gpu_direct_pipeline.py +++ b/apps/worker/video_processing/gpu_direct_pipeline.py @@ -30,11 +30,16 @@ EDGE_CROP_MAX_PCT = 0.05 # 均以 720p 为基准（见前端 titleCanvas.ts 注释 scale=videoWidth/720，types.ts "px @720p"）， # 非 720p 输出时按 video_width / TITLE_SIZE_REF_WIDTH 等比缩放，保证成片位置与前端预览一致。 TITLE_SIZE_REF_WIDTH = 720 -# 与前端 titleCanvas.ts 对齐：top 时文本 top-edge 距视频顶 = PAD(16@720p) + margin_top(默认 24@720p) -TITLE_PAD_TOP = 16 -TITLE_DEFAULT_MARGIN_TOP = 24 -TITLE_DEFAULT_MARGIN_BOTTOM = 24 -SUBTITLE_DEFAULT_MARGIN_BOTTOM = 60 # 字幕距底边距（720p 基准，与前端字幕面板默认对齐） +# 与 video_filter_builder.build_title_drawtext_filter（CPU 路径）和 ass_subtitle_builder 对齐： +# - top/bottom 默认 margin 50@720p（vfb 用 _scale_title_len(50, w)，即 y=50 / y=h-th-50） +# - margin_top 字段：前端编辑器 marginTop 滑块，叠加在默认 margin 之上（#2095 支持） +# - PAD 概念仅用于前端 Canvas 预览；ffmpeg drawtext y 是 baseline，无 font metrics 可用， +# 直接用统一 50@720p baseline 位置即可保持三端（GPU/CPU/前端视觉）一致。 +TITLE_DEFAULT_MARGIN_TOP = 50 # top 位置 baseline 默认距顶 50@720p（与 vfb/CPU 路径一致） +TITLE_DEFAULT_MARGIN_BOTTOM = 50 # bottom 位置 baseline 默认距底 50@720p +SUBTITLE_DEFAULT_MARGIN_BOTTOM = 50 # 字幕距底边距 50@720p（与 vfb 一致） +TITLE_MARGIN_TOP_FROM_CFG_DEFAULT = 24 # 前端 marginTop 滑块默认值（用户未传时叠加 0） +TITLE_FAUX_BOLD_WIDTH = 2 # 仿粗黑色描边宽度（与 vfb 一致，2@720p 黑色细描边） def _scale_title_len(value, video_width: int): @@ -507,7 +512,7 @@ def build_direct_render( except (TypeError, ValueError): t_size_720 = 0 if t_size_720 <= 0: - t_size_720 = 36 # 与 config_schemas 默认 48 接近；AI Avatar 前端默认 48，剪辑前端默认 28，取中 + t_size_720 = 48 # 与 config_schemas.DEFAULT_EDIT_PLAN_CONFIG.title.size=48 及 vfb 默认一致 t_size = _scale_title_len(t_size_720, output_width) # stroke/shadow 长度字段也需 720p→输出分辨率缩放 t_color = str(t_cfg.get("color", t_cfg.get("font_color", "#ffffff"))) @@ -525,14 +530,22 @@ def build_direct_render( # baseline 需再下移约 0.85*fontsize；但 drawtext 表达式无法引用 fontsize 变量， # 这里直接用 (PAD + margin_top)@720p 缩放后作为 y（即让 baseline≈顶部内边距位置）， # 实际中文字符会自然向下延伸，视觉位置与前端预览（textBaseline=middle 居中到 firstLineY）一致。 - _t_margin_top_raw = t_cfg.get("margin_top", TITLE_DEFAULT_MARGIN_TOP) + # margin_top：前端滑块值（默认 24@720p），叠加在默认 50@720p 基线之上 + _t_user_margin_top = t_cfg.get("margin_top") try: - _t_margin_top_720 = int(_t_margin_top_raw) + _t_user_margin_top_720 = int(_t_user_margin_top) if _t_user_margin_top is not None else 0 except (TypeError, ValueError): - _t_margin_top_720 = TITLE_DEFAULT_MARGIN_TOP - t_margin_top = _scale_title_len(TITLE_PAD_TOP + _t_margin_top_720, output_width) - # bottom margin（标题放在 bottom 时也支持） - t_margin_bottom = _scale_title_len(TITLE_PAD_TOP + TITLE_DEFAULT_MARGIN_BOTTOM, output_width) + _t_user_margin_top_720 = 0 + t_margin_top_720 = TITLE_DEFAULT_MARGIN_TOP + _t_user_margin_top_720 + t_margin_top = _scale_title_len(t_margin_top_720, output_width) + # bottom margin（标题放在 bottom 时）：用户 margin_bottom 透传，默认 50@720p + _t_user_margin_bottom = t_cfg.get("margin_bottom") + try: + _t_user_margin_bottom_720 = int(_t_user_margin_bottom) if _t_user_margin_bottom is not None else 0 + except (TypeError, ValueError): + _t_user_margin_bottom_720 = 0 + t_margin_bottom_720 = TITLE_DEFAULT_MARGIN_BOTTOM + _t_user_margin_bottom_720 + t_margin_bottom = _scale_title_len(t_margin_bottom_720, output_width) t_borderw = 0 t_border_color = "#000000" t_box = False @@ -566,11 +579,11 @@ def build_direct_render( t_shadow_x = _scale_title_len(t_shadow_x_720, output_width) t_shadow_y = _scale_title_len(t_shadow_y_720, output_width) # bold/italic：drawtext 原生无粗斜体选项；通过同色描边模拟粗体 - t_bold = bool(t_cfg.get("bold", True)) # 与 ASS 路径默认 bold=True 对齐 + t_bold = bool(t_cfg.get("bold", True)) # 与 ASS/vfb 路径默认 bold=True 对齐 if t_bold and t_borderw < 1: - # 同色描边宽度 1@720p，按宽度缩放保证各分辨率视觉一致 - t_borderw = max(1, _scale_title_len(1, output_width)) - t_border_color = t_color # 用文字色描边模拟加粗 + # 粗体未配用户描边时：黑色细描边 2@720p（与 vfb 一致，避免同色描边导致重影） + t_borderw = _scale_title_len(TITLE_FAUX_BOLD_WIDTH, output_width) + t_border_color = "#000000" # 黑色细描边模拟粗体 # ── 解析 subtitle_config ── s_cfg = dict(subtitle_config) if isinstance(subtitle_config, dict) else {} @@ -595,8 +608,21 @@ def build_direct_render( s_pos_x, s_pos_y = None, None s_margin_top = _scale_title_len(60, output_width) # subtitle top (not commonly used) s_margin_bottom = _scale_title_len(SUBTITLE_DEFAULT_MARGIN_BOTTOM, output_width) - s_borderw = max(1, _scale_title_len(2, output_width)) # 字幕默认描边保证可读性 + # subtitle stroke/bold：先解析用户 stroke，再按 bold 默认补描边 + s_borderw = 0 s_border_color = "#000000" + _s_stroke = s_cfg.get("stroke") + if isinstance(_s_stroke, dict) and _s_stroke.get("enabled", False): + try: + s_borderw = _scale_title_len(int(float(_s_stroke.get("width", 2))), output_width) + except (TypeError, ValueError): + s_borderw = 0 + s_border_color = str(_s_stroke.get("color", "#000000")) + s_bold = bool(s_cfg.get("bold", False)) + if s_bold and s_borderw < 1: + # 粗体默认黑色细描边 2@720p（与 title/CPU vfb 一致） + s_borderw = _scale_title_len(TITLE_FAUX_BOLD_WIDTH, output_width) + s_border_color = "#000000" # 静态字幕：static_subtitle_text 非空时构造全片长 segment（0 → total_duration） static_text = (static_subtitle_text or "").strip() diff --git a/tests/unit/test_gpu_direct_pipeline.py b/tests/unit/test_gpu_direct_pipeline.py index bee668860..72c476ef7 100644 --- a/tests/unit/test_gpu_direct_pipeline.py +++ b/tests/unit/test_gpu_direct_pipeline.py @@ -85,7 +85,7 @@ def _patch_pipeline_helpers(monkeypatch): # --------------------------------------------------------------------------- class TestNoConfigBackwardCompat: def test_default_title_drawtext_white_top(self, monkeypatch): - """不传 title_config 时：白字、top、36@720 按 width 缩放到 64、y=71（PAD16+margin24）。""" + """不传 title_config 时：白字、top、48@720 按 width 缩放到 85、y=89（50@720p baseline）。""" import video_processing.gpu_direct_pipeline as gdp _patch_pipeline_helpers(monkeypatch) @@ -101,15 +101,15 @@ class TestNoConfigBackwardCompat: ) fc = " ".join(plan.filter_complex) assert "fontcolor=0xffffff" in fc - # 默认 position=top → y=71（scale(16+24)=71），不含 h-th - assert "y=71" in fc + # 默认 position=top → y=89（scale(50)=89，与 CPU/vfb 一致），不含 h-th + assert "y=89" in fc assert "h-th" not in fc - # 默认字号 36@720 经 1280/720 缩放 = 64 - assert "fontsize=64" in fc + # 默认字号 48@720 经 1280/720 缩放 = 85 + assert "fontsize=85" in fc assert "text='默认标题'" in fc - # 默认粗体：drawtext 无原生粗体，用同色描边 borderw=1@720（1280/720→scale=2）模拟 - assert "borderw=2" in fc - assert "bordercolor=0xffffff" in fc + # 默认粗体：黑色细描边 borderw=2@720（scale=4），与 CPU vfb 一致避免重影 + assert "borderw=4" in fc + assert "bordercolor=0x000000" in fc def test_no_subtitle_when_none(self, monkeypatch): import video_processing.gpu_direct_pipeline as gdp @@ -153,8 +153,8 @@ class TestTitleStylePassthrough: assert "fontcolor=0xff0000" in fc # 60@720 按 1280/720 缩放 = 107 assert "fontsize=107" in fc - # top 位置 y=71（scale(PAD16+margin_top24)=71） - assert "y=71" in fc + # top 位置默认 margin 50@720p → scale=89（未传 margin_top 用默认） + assert "y=89" in fc assert "h-th" not in fc def test_title_position_center(self, monkeypatch): @@ -218,9 +218,9 @@ class TestTitleStylePassthrough: # 阴影层 + 主字层 = 2 条 drawtext assert drawtext_count == 2 joined = " ".join(plan.filter_complex) - # shadow offset 3@720 经 1280/720 缩放 = 5；默认 position=top → y=71 + # shadow offset 3@720 经 1280/720 缩放 = 5；默认 position=top → y=89 assert "x=(w-text_w)/2+5" in joined - assert "y=71+5" in joined + assert "y=89+5" in joined assert "h-th" not in joined def test_title_font_override(self, monkeypatch): @@ -282,10 +282,10 @@ class TestTitleStylePassthrough: assert "(w-text_w)*0.3000" in fc assert "(h-text_h)*0.6000" in fc assert "h-th-" not in fc - assert "y=71" not in fc + assert "y=89" not in fc def test_title_margin_top_respected(self, monkeypatch): - """margin_top 透传：100@720 → scale(16+100)=206@1280。""" + """margin_top 透传：100@720，叠加默认 50 → 150@720 → scale=267@1280。""" import video_processing.gpu_direct_pipeline as gdp _patch_pipeline_helpers(monkeypatch) @@ -300,8 +300,8 @@ class TestTitleStylePassthrough: title_config={"text": "远离顶部", "position": "top", "margin_top": 100}, ) fc = " ".join(plan.filter_complex) - # scale(PAD16 + margin_top100) = 116*1280/720 = 206 - assert "y=206" in fc + # 默认 50 + margin_top100 = 150@720 → scale=267 + assert "y=267" in fc def test_title_default_bold_true(self, monkeypatch): """不传 bold 时默认粗体：drawtext 用同色描边 borderw 模拟。""" @@ -319,9 +319,9 @@ class TestTitleStylePassthrough: title_config={"text": "粗体"}, ) fc = " ".join(plan.filter_complex) - # bold=True 默认：同色描边 width=1@720 → scale=2，颜色与文字相同（白色） - assert "borderw=2" in fc - assert "bordercolor=0xffffff" in fc + # bold=True 默认：黑色细描边 width=2@720 → scale=4，与 CPU vfb 一致 + assert "borderw=4" in fc + assert "bordercolor=0x000000" in fc assert "text='粗体'" in fc def test_title_bold_false_disables_faux_bold(self, monkeypatch): @@ -359,8 +359,8 @@ class TestTitleStylePassthrough: title_config={"text": "底部", "position": "bottom"}, ) fc = " ".join(plan.filter_complex) - # bottom margin PAD16+24=40 → scale=71 - assert "y=h-th-71" in fc + # bottom margin 50@720 → scale=89 + assert "y=h-th-89" in fc def test_title_size_scales_by_width_not_height(self, monkeypatch): """不同分辨率下同 @720 基准的 size 等比缩放：1080x1920 下 36→54。""" @@ -378,10 +378,10 @@ class TestTitleStylePassthrough: clip_volumes=[1.0], ) fc = " ".join(plan.filter_complex) - # default size 36@720 → 1080w = 54 - assert "fontsize=54" in fc - # top margin 40@720 → 60 - assert "y=60" in fc + # default size 48@720 → 1080w = 72 + assert "fontsize=72" in fc + # top margin 50@720 → 75 + assert "y=75" in fc # --------------------------------------------------------------------------- @@ -434,8 +434,8 @@ class TestSubtitlePassthrough: assert "fontcolor=0x0000ff" in fc # sub size 28@720 经 1280/720 缩放 = 50 assert "fontsize=50" in fc - # bottom margin 60@720 → 107 - assert "y=h-th-107" in fc + # bottom margin 50@720 → 89 + assert "y=h-th-89" in fc def test_subtitle_disabled_hides_subs(self, monkeypatch): import video_processing.gpu_direct_pipeline as gdp @@ -644,18 +644,18 @@ class TestCombinedConfig: bgm_config={"volume": 0.2, "fade_in": 0.5, "fade_out": 1.0}, ) fc = " ".join(plan.filter_complex) - # 标题：size 50@720 → 89，top margin 40@720 → 71，stroke 2@720 → 4 + # 标题：size 50@720 → 89，top margin 50@720 → 89，stroke 2@720 → 4 assert "text='主标题'" in fc assert "fontcolor=0xffff00" in fc assert "fontsize=89" in fc - assert "y=71" in fc + assert "y=89" in fc assert "borderw=4" in fc - # 字幕：size 24@720 → 43，bottom margin 60@720 → 107 + # 字幕：size 24@720 → 43，bottom margin 50@720 → 89（默认不加粗） assert "text='成片全字幕'" in fc assert "fontsize=43" in fc - assert "y=h-th-107" in fc + assert "y=h-th-89" in fc assert "fontcolor=0xffffff" in fc - # subtitle 默认 bold=True 也会加同色描边（width=1→2）；但若主标题已有 stroke 不影响字幕独立 borderw=4（s 默认 2 像素@720→scale=4） + # subtitle 默认 bold=False，无额外描边（用户未开 stroke） # BGM assert "volume=0.200" in fc assert "afade=t=in:st=0:d=0.50" in fc @@ -688,7 +688,7 @@ class TestEdgeCases: assert "drawtext=" in fc def test_bold_title_increases_borderw(self, monkeypatch): - """bold=True 时若原无描边，自动加 borderw=1 用同色描边模拟加粗。""" + """bold=True 时若原无描边，自动加 borderw 黑色细描边模拟加粗（与 CPU vfb 一致，避免重影）。""" import video_processing.gpu_direct_pipeline as gdp _patch_pipeline_helpers(monkeypatch) @@ -703,9 +703,9 @@ class TestEdgeCases: title_config={"text": "粗体", "bold": True, "color": "#ff0000"}, ) fc = " ".join(plan.filter_complex) - # 加粗模拟 borderw>=1，且 bordercolor 跟字体色一致（0xff0000） - assert "borderw=" in fc - assert "bordercolor=0xff0000" in fc + # 仿粗用黑色细描边 2@720→scale=4，不跟文字色（避免同色描边重影） + assert "borderw=4" in fc + assert "bordercolor=0x000000" in fc def test_asr_and_static_subtitle_asr_wins(self, monkeypatch): """同时传 static_subtitle_text 和 ASR segments 时，ASR 优先（不插入静态全文）。""" -- 2.54.0 From 96bcec5fdddd60a914d85c6a678ebec06ab1f12f Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Tue, 29 Sep 2026 18:31:05 +0800 Subject: [PATCH 2/4] fix(worker+api): resolve batch title race + auto-queue pending tasks (#2098) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two batch-rendering bugs found from staging logs (6-video batch 17:30-17:43): Bug A (P1) — title_config race when same plan renders concurrently: _sync_task_config_to_plan wrote each task's title_config/bgm/resolution to the shared plan.config row, then RenderAdapter.render_plan read back from plan.config during rendering. If two tasks sharing the same plan (e.g. batch retry, preview+render, retry+new) ran concurrently, Task B's write could overwrite Task A's title before Task A's ffmpeg read it, producing videos with the wrong title text/style. Fix: stop mutating plan.config in the worker render path. Introduce task_config_override threading through render_plan / _do_render / UnifiedRenderService; URS reads config via _effective_config() which does a deep-copy plan.config merged with per-task override (title/bgm/ export). _build_task_config_override builds the override dict (with field-name normalization: font_size→size, font_color→color) from task_info, and _download_voice_for_task handles voiceover download independently. plan.config is left as-authored by the API writeback; concurrent tasks no longer race on it. Files: - apps/worker/video_processing/unified_render_service.py: add override_config param + _cfg_section/_effective_config helpers; replace all render-time self.plan.config reads (title/subtitle/bgm/ export/TTS) with _effective_config(). - apps/worker/video_processing/render_adapter.py: render_plan/_do_render/ _prepare_bgm accept task_config_override, merge into bgm/export reads, forward to URS. - apps/worker/worker_app/tasks/generation.py: replace _sync_task_config_to_plan (which wrote plan.config) with _build_task_config_override + _download_voice_for_task; pass task_config_override to render_plan. Bug B (P0) — USER_PENDING_LIMIT=3 returns HTTP 429, blocks batch submit: Pre-check in create_generation_task rejected the whole batch with 429 USER_QUEUE_FULL once user had ≥3 pending tasks; UI showed 'wait ~4 min' and prevented any more submissions. Users expect to submit a batch of 6 and have them queue naturally (worker concurrency=2). Fix: - USER_PENDING_LIMIT 3→20 (soft cap for abuse protection, supports typical batch sizes of 6-10 with headroom). - Remove user-level 429 rejection from pre-check, retry endpoint, single-task confirm endpoint, and safe_enqueue (user path now logs a warning and continues to enqueue). Global GLOBAL_PENDING_LIMIT=20 is retained as a hard 503 system-busy guard. - safe_enqueue post-enqueue check: user-over only logs, does not fail the task or raise. - Batch loop UserPendingLimitExceeded except branch now treats it as a successful enqueue (should not trigger in practice). Files: - apps/api/app/core/task_enqueue.py: limit 3→20; user checks log-only. - apps/api/app/api/routes/generation_tasks.py: remove user pre-check 429; soften retry/single/batch except branches. Stacked on PR #2095/#2097 which already fixed title baseline/width-scaling and default-constant alignment with CPU video_filter_builder. Tests: - tests/unit/test_gpu_direct_pipeline.py: 30 passed - tests/unit/test_render_adapter_pure.py: 18 passed - ruff check/format clean; py_compile clean --- apps/api/app/api/routes/generation_tasks.py | 157 ++++++++++++------ apps/api/app/core/task_enqueue.py | 31 ++-- .../worker/video_processing/render_adapter.py | 21 ++- .../unified_render_service.py | 36 +++- apps/worker/worker_app/tasks/generation.py | 68 +++----- 5 files changed, 197 insertions(+), 116 deletions(-) diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index 901edf659..0f0a2daac 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -7,7 +7,6 @@ from app.auth import AuthenticatedUser, get_current_user from app.core.storage import OSSStorageService, get_storage_service from app.core.task_enqueue import ( GLOBAL_PENDING_LIMIT, - USER_PENDING_LIMIT, GlobalQueueFull, UserPendingLimitExceeded, build_rate_limit_detail, @@ -50,12 +49,98 @@ from packages.domain.smart_match import smart_select_assets # #2035：文案关键词 → 素材分类 映射表（用于 smart_match category_match 维度） # AssetClassification 枚举: scenic / product / person / animal / food / tech / sport / music / other _CATEGORY_KEYWORDS: dict[str, set[str]] = { - "scenic": {"风景", "自然", "山水", "大海", "天空", "日落", "日出", "森林", "城市", "建筑", "夜景", "街道", "公园", "景区", "旅行", "旅游", "户外"}, - "product": {"产品", "商品", "展示", "演示", "开箱", "评测", "好物", "推荐", "种草", "购物", "电商", "带货", "品牌", "广告", "包装"}, - "person": {"人物", "人物采访", "对话", "说话", "讲解", "演讲", "采访", "聊天", "开会", "工作", "办公室", "团队", "员工", "老板", "女性", "男性", "美女", "帅哥"}, + "scenic": { + "风景", + "自然", + "山水", + "大海", + "天空", + "日落", + "日出", + "森林", + "城市", + "建筑", + "夜景", + "街道", + "公园", + "景区", + "旅行", + "旅游", + "户外", + }, + "product": { + "产品", + "商品", + "展示", + "演示", + "开箱", + "评测", + "好物", + "推荐", + "种草", + "购物", + "电商", + "带货", + "品牌", + "广告", + "包装", + }, + "person": { + "人物", + "人物采访", + "对话", + "说话", + "讲解", + "演讲", + "采访", + "聊天", + "开会", + "工作", + "办公室", + "团队", + "员工", + "老板", + "女性", + "男性", + "美女", + "帅哥", + }, "animal": {"动物", "宠物", "狗", "猫", "鸟", "鱼", "马", "牛", "羊", "野生动物", "动物园"}, - "food": {"美食", "食物", "餐饮", "餐厅", "做饭", "烹饪", "厨房", "菜品", "饮料", "水果", "甜点", "蛋糕", "咖啡", "茶", "零食", "吃"}, - "tech": {"科技", "数码", "电脑", "手机", "屏幕", "软件", "APP", "互联网", "AI", "人工智能", "机器人", "办公", "程序员", "代码", "屏幕录制"}, + "food": { + "美食", + "食物", + "餐饮", + "餐厅", + "做饭", + "烹饪", + "厨房", + "菜品", + "饮料", + "水果", + "甜点", + "蛋糕", + "咖啡", + "茶", + "零食", + "吃", + }, + "tech": { + "科技", + "数码", + "电脑", + "手机", + "屏幕", + "软件", + "APP", + "互联网", + "AI", + "人工智能", + "机器人", + "办公", + "程序员", + "代码", + "屏幕录制", + }, "sport": {"运动", "健身", "跑步", "篮球", "足球", "游泳", "瑜伽", "户外", "锻炼", "体育", "比赛", "球场"}, "music": {"音乐", "歌曲", "演唱会", "乐器", "唱歌", "跳舞", "舞蹈", "MV", "演出", "乐队", "钢琴", "吉他", "节奏"}, } @@ -77,6 +162,7 @@ def _infer_expected_categories(script_tags: set[str] | None) -> set[str] | None: break return matched or None + from packages.middleware.points_gate import points_gate logger = logging.getLogger(__name__) @@ -193,10 +279,13 @@ def _select_assets_from_library( # #2035：加载片段级 AI 标签，供叙事模式 AI 加权和 smart 模式语义匹配使用。 # 失败降级为空（不影响选片主流程）。 clip_ai_tags_by_asset: dict[str, list[dict]] = {} - ai_tags_by_asset: dict[str, dict] = {} # asset_id → 聚合后的 ai_tags dict（取首个有 has_text 的片段；合并 scene/objects/action 去重） + ai_tags_by_asset: dict[ + str, dict + ] = {} # asset_id → 聚合后的 ai_tags dict（取首个有 has_text 的片段；合并 scene/objects/action 去重） try: if db is not None: from packages.adapters.sqlalchemy_impl.models import AssetAtomClipModel + ready_ids = [a.id for a in ready_video_assets] clip_rows = ( db.query(AssetAtomClipModel.asset_id, AssetAtomClipModel.ai_tags) @@ -594,21 +683,13 @@ def create_generation_task( # 同批次任务共享 batch_id，用于视频查重时批次内比对 batch_id = uuid.uuid4().hex if count > 1 else "" - # 预检查：批量提交前先看会不会超限，避免建一半才拒 + # 预检查（Bug B #2098）：只保留全局 503 保护，用户级不再硬拒 429； + # 超额任务直接入队等待 worker 自然消费，前端展示排队位置而非阻止提交。 + # USER_PENDING_LIMIT 作为软上限（safe_enqueue 兜底），提高到 20 支持批量提交。 try: - user_pending = generation_task_repository.count_pending_by_user(user_id) global_pending = generation_task_repository.count_pending_total() - if user_pending + count > USER_PENDING_LIMIT: - raise UserPendingLimitExceeded( - user_id=user_id, pending_count=user_pending + count, limit=USER_PENDING_LIMIT - ) if global_pending + count > GLOBAL_PENDING_LIMIT: raise GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT) - except UserPendingLimitExceeded as e: - raise HTTPException( - status_code=429, - detail=build_rate_limit_detail(e, generation_task_repository, scope="user"), - ) from e except GlobalQueueFull as e: raise HTTPException( status_code=503, @@ -923,14 +1004,10 @@ def create_generation_task( else: failed_tasks.append(task) except UserPendingLimitExceeded as _e: - # 兜底：如果预检查后又并发提交了，在这里也拦住 - failed_tasks.append(task) - if not created_tasks: - raise HTTPException( - status_code=429, - detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"), - ) from _e - break + # Bug B #2098: 用户级限流已改为软限制，此分支理论上不再触发； + # 极端并发兜底仍入队（safe_enqueue 内部会打 warning 日志），不 429 拒绝 + logger.warning("[生成任务] 用户 pending 超软限制，仍允许入队: task_id=%s", task.id) + created_tasks.append(task) except GlobalQueueFull as _e: failed_tasks.append(task) if not created_tasks: @@ -1081,10 +1158,8 @@ def confirm_generation( ): logger.warning("[确认生成] 入队失败: task_id=%s", new_task.id) except UserPendingLimitExceeded as _e: - raise HTTPException( - status_code=429, - detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"), - ) from None + # Bug B #2098: 用户级限流已软处理，理论上不再触发；作为防御仍放行 + logger.warning("[任务] 用户 pending 超软限制，任务已入队") except GlobalQueueFull as _e: raise HTTPException( status_code=503, @@ -1263,22 +1338,8 @@ def retry_generation_task( raise HTTPException(status_code=409, detail="Only failed tasks can be retried") user_id = authenticated_user.user.id - # 预检查：创建前判断，>= 上限就拒绝 - user_pending = generation_task_repository.count_pending_by_user(user_id) + # 预检查（Bug B #2098）：只保留全局 503，用户级不再硬拒 global_pending = generation_task_repository.count_pending_total() - if user_pending >= USER_PENDING_LIMIT: - raise HTTPException( - status_code=429, - detail=build_rate_limit_detail( - UserPendingLimitExceeded( - user_id=user_id, - pending_count=user_pending, - limit=USER_PENDING_LIMIT, - ), - generation_task_repository, - scope="user", - ), - ) if global_pending >= GLOBAL_PENDING_LIMIT: raise HTTPException( status_code=503, @@ -1321,10 +1382,8 @@ def retry_generation_task( ): logger.warning("[生成任务] 重试入队失败: task_id=%s", retried.id) except UserPendingLimitExceeded as _e: - raise HTTPException( - status_code=429, - detail=build_rate_limit_detail(_e, generation_task_repository, scope="user"), - ) from None + # Bug B #2098: 用户级限流已软处理，理论上不再触发；作为防御仍放行 + logger.warning("[任务] 用户 pending 超软限制，任务已入队") except GlobalQueueFull as _e: raise HTTPException( status_code=503, diff --git a/apps/api/app/core/task_enqueue.py b/apps/api/app/core/task_enqueue.py index c3b11329c..bea98657b 100755 --- a/apps/api/app/core/task_enqueue.py +++ b/apps/api/app/core/task_enqueue.py @@ -6,7 +6,7 @@ from app.core.celery_app import celery_app logger = logging.getLogger(__name__) # ── 限流阈值常量（全系统统一管理，不要在业务代码里硬编码） ── -USER_PENDING_LIMIT = 3 # 单用户 pending 上限 +USER_PENDING_LIMIT = 20 # 单用户 pending 上限（#2098: 从 3 提到 20，支持批量任务自动排队） GLOBAL_PENDING_LIMIT = 20 # 全局 pending 上限 WORKER_CONCURRENCY = 4 # worker 渲染并发数（infra/docker/compose.yml WORKER_CONCURRENCY 默认值） @@ -263,19 +263,18 @@ def safe_enqueue_generation_task( _mark_task_failed_safely(task, generation_task_repository, log_prefix, str(exc)) raise exc - # 用户级限流检查（传了 user_id 才做） + # Bug B #2098: 用户级限流改为软提示，不再硬拒；所有任务都入队等待 worker 自然消费。 + # user_pending_limit 作为兜底阈值保留（默认 20），达到时打 warning 日志但仍入队， + # 避免极端情况下恶意用户无限堆积任务。真正的系统保护由全局 GLOBAL_PENDING_LIMIT 承担。 if user_id: user_pending = generation_task_repository.count_pending_by_user(user_id) if user_pending > user_pending_limit: logger.warning( - "[队列限流] 用户 pending 任务数超限(入队前): user_id=%s, count=%d/%d", + "[队列限流] 用户 pending 任务数超过软上限(入队): user_id=%s, count=%d/%d, 仍允许入队排队", user_id, user_pending, user_pending_limit, ) - exc = UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending, limit=user_pending_limit) - _mark_task_failed_safely(task, generation_task_repository, log_prefix, str(exc)) - raise exc # ── 发送 Celery 任务 ── try: @@ -317,16 +316,18 @@ def safe_enqueue_generation_task( user_after = generation_task_repository.count_pending_by_user(user_id) if user_id else 0 global_over = global_after > global_pending_limit - user_over = bool(user_id and user_after > user_pending_limit) - - if global_over or user_over: - if global_over: - reason = f"全局 pending 超限(入队后): {global_after}/{global_pending_limit}" - exc = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit) - else: - reason = f"用户 pending 超限(入队后): {user_after}/{user_pending_limit}" - exc = UserPendingLimitExceeded(user_id=user_id, pending_count=user_after, limit=user_pending_limit) + # Bug B #2098: 用户超限仅日志警告，不回滚任务 + if user_id and user_after > user_pending_limit: + logger.warning( + "[队列限流] 用户 pending 超软上限(入队后): user_id=%s, count=%d/%d", + user_id, + user_after, + user_pending_limit, + ) + if global_over: + reason = f"全局 pending 超限(入队后): {global_after}/{global_pending_limit}" + exc = GlobalQueueFull(pending_count=global_after, limit=global_pending_limit) logger.warning( "[队列限流] %s, task_id=%s, user_id=%s — 回滚状态为 failed", reason, diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index ccf3b5222..b64e756a8 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -132,6 +132,7 @@ class RenderAdapter: work_dir: Path | None = None, progress_cb: ProgressCallback | None = None, voiceover_audio_path: str | None = None, + task_config_override: dict | None = None, # Bug A: task 级 config 覆盖，防并发竞态 ) -> RenderAdapterResult: """渲染一个 EditPlan。 @@ -217,6 +218,7 @@ class RenderAdapter: rendered_clip_ids=rendered_clip_ids, failed_clip_ids=failed_clip_ids, voiceover_audio_path=voiceover_audio_path, + task_config_override=task_config_override, ) # 成功时将临时目录所有权转移给调用方，阻止 finally 清理 if result.success and temp_dir: @@ -392,7 +394,7 @@ class RenderAdapter: return asset_path_map, rendered_clip_ids, failed_clip_ids, asset_storage_map - def _prepare_bgm(self, plan, work_dir: Path, plan_id: str) -> str | None: + def _prepare_bgm(self, plan, work_dir: Path, plan_id: str, *, bgm_override: dict | None = None) -> str | None: """准备 BGM 音频文件（从 plan.config.bgm 读取配置）。 支持 3 种来源（按优先级）： @@ -405,7 +407,9 @@ class RenderAdapter: from urllib.parse import urlparse plan_config = plan.config or {} - bgm_config = plan_config.get("bgm", {}) or {} + bgm_config = dict(plan_config.get("bgm", {}) or {}) + if isinstance(bgm_override, dict) and bgm_override: + bgm_config.update(bgm_override) # Bug A: 任务级 BGM 覆盖，防并发竞态 if not bgm_config.get("enabled", False): return None @@ -567,6 +571,7 @@ class RenderAdapter: failed_clip_ids: list[str] | None = None, voiceover_audio_path: str | None = None, asset_storage_map: dict[str, str] | None = None, + task_config_override: dict | None = None, # Bug A: task 级 config 覆盖，防并发竞态 ) -> RenderAdapterResult: """执行统一渲染核心流程（BGM + ASR + 渲染 + 缩略图 + 上传）。 @@ -580,8 +585,9 @@ class RenderAdapter: Returns: RenderAdapterResult """ - # 1. 准备 BGM - bgm_path = self._prepare_bgm(plan, work_dir, plan_id) + # 1. 准备 BGM（Bug A: 传 task 级 bgm override） + _bgm_override = (task_config_override or {}).get("bgm") if isinstance(task_config_override, dict) else None + bgm_path = self._prepare_bgm(plan, work_dir, plan_id, bgm_override=_bgm_override) self._report_progress(progress_cb, 40.0, "执行视频渲染") @@ -589,8 +595,10 @@ class RenderAdapter: plan_config = plan.config or {} asr_service = self._get_asr_service() - # 3. 读取输出分辨率 - export_config = plan_config.get("export", {}) or {} + # 3. 读取输出分辨率（Bug A: task override 优先） + export_config = dict(plan_config.get("export", {}) or {}) + if isinstance(task_config_override, dict) and isinstance(task_config_override.get("export"), dict): + export_config.update(task_config_override["export"]) if not isinstance(export_config, dict): export_config = {} output_width, output_height = _parse_resolution(export_config.get("resolution")) @@ -615,6 +623,7 @@ class RenderAdapter: asr_service=asr_service, voiceover_audio_path=voiceover_audio_path, clip_has_text=clip_has_text, + override_config=task_config_override, ) # 注入每个视频段对应素材的 storage_key，供全 GPU 直连管线直接签名下载 _storage_map = asset_storage_map or {} diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 6c39cd569..18a9f8197 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -158,6 +158,7 @@ class UnifiedRenderService: bgm_path: str | None = None, # BGM 本地文件路径 voiceover_audio_path: str | None = None, # 配音素材库音频本地路径 clip_has_text: list[bool] | None = None, # 源视频片段是否有文字（来自 atom_clip.ai_tags.has_text） + override_config: dict | None = None, # Bug A: task 级 config 覆盖（title/bgm/export/subtitle），防并发竞态 ): self.plan = plan self.clips = clips @@ -170,6 +171,9 @@ class UnifiedRenderService: self.asr_service = asr_service self.bgm_path = bgm_path self.voiceover_audio_path = voiceover_audio_path + # Bug A: task 级 config override（深拷贝），优先级高于 plan.config； + # 避免同 plan 多任务并发渲染时 _sync_task_config_to_plan 写 plan.config["title"] 互相覆盖。 + self._override_config = dict(override_config) if isinstance(override_config, dict) else {} # #1970：片段级文字检测（顺序与非 audio 的源视频片段一致）；None 表示无可靠检测，保守不翻转 self._clip_has_text = clip_has_text self._transition_engine = TransitionEngine(default_duration=transition_duration) @@ -180,6 +184,28 @@ class UnifiedRenderService: self._micro_plan_cache: Any = None self._micro_plan_loaded = False + def _cfg_section(self, section: str) -> dict: + """读取单个配置段：override_config 优先于 plan.config（Bug A 防并发竞态）。""" + base = dict((self.plan.config or {}).get(section, {}) or {}) + override = self._override_config.get(section) + if isinstance(override, dict) and override: + base.update(override) # 浅合并，保留 base 中未被覆盖字段 + return base + + def _effective_config(self) -> dict: + """读取完整 config：override_config 顶层段覆盖 plan.config（Bug A 防并发竞态）。""" + import copy + + full = copy.deepcopy(self.plan.config or {}) + for k, v in self._override_config.items(): + if isinstance(v, dict): + sec = dict(full.get(k, {}) or {}) + sec.update(v) + full[k] = sec + else: + full[k] = v + return full + # ── #1970 PR2 智能降重：片段级微变换 ─────────────────────────────────── def _dedup_enabled(self) -> bool: """读取 plan.config.dedup_enabled，缺省视为 True（向后兼容）。""" @@ -431,7 +457,7 @@ class UnifiedRenderService: has_audio = pass_through_has_audio # 直通模式下也支持 BGM 混音：提取音频 → 混 BGM → 合并回视频 if self.bgm_path and pass_through_has_audio: - config = self.plan.config or {} + config = self._effective_config() bgm_config = config.get("bgm", {}) or {} if bgm_config.get("enabled", False): ctx = RenderContext(work_dir=self.work_dir, plan_id=self.plan.id) @@ -471,7 +497,7 @@ class UnifiedRenderService: "[unified-render] pass-through BGM mix failed, skipping: plan_id=%s", self.plan.id ) else: - config = self.plan.config or {} + config = self._effective_config() bgm_config = config.get("bgm", {}) or {} if not isinstance(bgm_config, dict): bgm_config = {} @@ -696,7 +722,7 @@ class UnifiedRenderService: Returns: ASS 文件路径，没有字幕时返回 None """ - config = self.plan.config or {} + config = self._effective_config() # #1901 统一读 "title"，兼容老数据 "title_config" title_cfg = config.get("title", {}) or {} if not isinstance(title_cfg, dict) or not (title_cfg.get("text") or "").strip(): @@ -881,7 +907,7 @@ class UnifiedRenderService: Returns: 是否成功添加了配音音轨 """ - config = self.plan.config or {} + config = self._effective_config() tts_cfg = config.get("tts", {}) or {} if not isinstance(tts_cfg, dict): tts_cfg = {} @@ -2275,7 +2301,7 @@ class UnifiedRenderService: try: from video_processing import gpu_direct_pipeline as gdp - cfg = self.plan.config or {} + cfg = self._effective_config() video_layer = next(_lyr for _lyr in layers if _lyr.role not in ("audio",)) video_clips = [c for c in video_layer.clips if c.clip_type != "audio"] diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index 5e2d8978d..9a227c67f 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -616,69 +616,49 @@ def _precompute_render_metadata( # ── Celery Task ────────────────────────────────────────────────────────────── -def _sync_task_config_to_plan(source_edit_plan_id: str, task_info: dict, db) -> str | None: - """将 GenerationTask 的配置同步到 EditPlan.config，返回配音本地路径（如果有）。 +def _build_task_config_override(task_info: dict) -> dict: + """Bug A: 从 task_info 构建任务级 config override 深拷贝，供渲染时覆盖 plan.config。 - 包括：title_config、BGM、输出分辨率。配音单独处理（需下载到本地）。 + 所有渲染相关配置（title/bgm/export）从任务自身读取，不再依赖共享 plan.config， + 彻底消除同 plan 多任务并发渲染时的竞态覆盖问题。 """ - from packages.adapters.sqlalchemy_impl.edit_plan_repository import ( - SQLAlchemyEditPlanRepository, - ) + import copy - plan_repo = SQLAlchemyEditPlanRepository(db) - plan = plan_repo.get(source_edit_plan_id) - if plan is None: - logger.error("[task] EditPlan not found: %s", source_edit_plan_id) - return None + override: dict = {} - plan_config = dict(plan.config or {}) - changed = False - - # 标题配置 + # 标题配置（字段名归一化） title_config = task_info.get("title_config") or {} - if title_config and isinstance(title_config, dict) and title_config.get("text", "").strip(): - cfg = dict(title_config) - # 字段名归一化 + if isinstance(title_config, dict) and title_config: + cfg = copy.deepcopy(title_config) if "font_size" in cfg and "size" not in cfg: cfg["size"] = cfg["font_size"] if "font_color" in cfg and "color" not in cfg: cfg["color"] = cfg["font_color"] - plan_config["title"] = cfg - changed = True - logger.info("[task] title_config synced to plan: %s", cfg.get("text", "")[:30]) + override["title"] = cfg # BGM 配置 bgm_config = task_info.get("bgm_config") or {} - if bgm_config: - from packages.domain.bgm_utils import merge_bgm_config - - existing_bgm = plan_config.get("bgm", {}) or {} - plan_config["bgm"] = merge_bgm_config(existing_bgm, bgm_config) - changed = True + if isinstance(bgm_config, dict) and bgm_config: + override["bgm"] = copy.deepcopy(bgm_config) # 输出分辨率 ow = task_info.get("output_width") or OUTPUT_WIDTH oh = task_info.get("output_height") or OUTPUT_HEIGHT if ow >= 100 and oh >= 100: - export_cfg = dict(plan_config.get("export", {}) or {}) - export_cfg["resolution"] = f"{ow}x{oh}" - plan_config["export"] = export_cfg - changed = True + override["export"] = {"resolution": f"{ow}x{oh}"} - if changed: - plan.config = plan_config - plan_repo.update(plan) - logger.info("[task] plan.config synced: plan_id=%s", source_edit_plan_id) + return override + + +def _download_voice_for_task(task_info: dict, source_edit_plan_id: str) -> str | None: + """下载任务配音到本地临时文件，返回路径（不读写 plan.config）。""" + import tempfile - # 配音下载 voiceover_path: str | None = None voice_library_id = task_info.get("voice_library_id", "") - # #1749：voice_ids 冗余字段已移除；配音一律以 voice_library_id 为准（独立配音每变体各自绑定） effective_voice_id = voice_library_id or "" if effective_voice_id: - import tempfile - voice_tmp = Path(tempfile.gettempdir()) / f"voice_{source_edit_plan_id}_{id(task_info)}.mp3" try: if _download_voice_asset(effective_voice_id, voice_tmp): @@ -697,6 +677,9 @@ def _render_from_edit_plan( ) -> tuple[Path, float, list[dict] | None, str | None, str | None, str, bool]: """从 EditPlan 数据库记录直接渲染（不再内存重建clips）。 + Bug A 修复：不再通过 _sync_task_config_to_plan 写共享 plan.config； + 渲染配置通过 task_config_override 参数直接传入渲染层，彻底消除并发竞态。 + Returns: (output_path, render_duration, cover_candidates, voiceover_path, temp_dir, thumbnail_url, edge_crop_applied) """ @@ -705,8 +688,10 @@ def _render_from_edit_plan( db = SessionLocal() try: - # 同步配置到 plan.config + 下载配音 - voiceover_path = _sync_task_config_to_plan(source_edit_plan_id, task_info, db) + # Bug A: 构建任务级 config override（深拷贝自 task_info），不写 plan.config，避免并发竞态 + task_override = _build_task_config_override(task_info) + # 下载配音到本地临时文件（不依赖 plan.config） + voiceover_path = _download_voice_for_task(task_info, source_edit_plan_id) # 进度回调 def _progress_cb(progress: float, stage: str): @@ -722,6 +707,7 @@ def _render_from_edit_plan( job_id=task_id, progress_cb=_progress_cb, voiceover_audio_path=voiceover_path, + task_config_override=task_override, ) if not result.success: -- 2.54.0 From 5e1520230f60289d6e7c45f3f8292a05210333ae Mon Sep 17 00:00:00 2001 From: saas-backend-agent Date: Tue, 29 Sep 2026 19:06:12 +0800 Subject: [PATCH 3/4] test(api+worker): align unit tests with #2098 soft user limit + override refactor - Bug B: check_queue_limits no longer raises UserPendingLimitExceeded (user limit is soft); only global GLOBAL_PENDING_LIMIT hard-rejects with 503. - safe_enqueue_generation_task user over-limit is warning-only (pre/post enqueue); GlobalQueueFull still hard-rejects/rolls-back. - Preview route (/generation/preview) user-pending pre-check removed to match batch/generation/retry routes; only global 503 pre-check kept. - Route enqueue-loop: UserPendingLimitExceeded except branch removed; all-failed rate-limit response only triggers on GlobalQueueFull (503). - Tests rewritten to reflect new semantics (197 queue/preview/batch/pipeline tests pass). - test_worker_generate_video_task_binding: replaced legacy _sync_task_config_to_plan helper test with _build_task_config_override + _download_voice_for_task (Bug A refactor). - USER_PENDING_LIMIT constant verified at 20 (soft cap). --- apps/api/app/api/routes/generation_preview.py | 37 +-- apps/api/app/core/task_enqueue.py | 26 +- tests/unit/test_1677_batch_variants.py | 35 ++- tests/unit/test_generation_preview.py | 62 +++-- tests/unit/test_task_enqueue.py | 181 ++++---------- tests/unit/test_task_queue_limit.py | 226 ++++-------------- ...test_worker_generate_video_task_binding.py | 23 +- 7 files changed, 184 insertions(+), 406 deletions(-) diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index 0a45b678e..ec828daed 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -11,9 +11,7 @@ from app.auth import AuthenticatedUser, get_current_user from app.core.storage import get_storage_service from app.core.task_enqueue import ( GLOBAL_PENDING_LIMIT, - USER_PENDING_LIMIT, GlobalQueueFull, - UserPendingLimitExceeded, build_rate_limit_detail, safe_enqueue_generation_task, ) @@ -302,26 +300,17 @@ def create_preview_generation_task( count, ) - # 预检查队列限流（按变体总数计） - try: - user_pending = generation_task_repository.count_pending_by_user(user_id) - global_pending = generation_task_repository.count_pending_total() - if user_pending + count > USER_PENDING_LIMIT: - raise UserPendingLimitExceeded( - user_id=user_id, pending_count=user_pending + count, limit=USER_PENDING_LIMIT - ) - if global_pending + count > GLOBAL_PENDING_LIMIT: - raise GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT) - except UserPendingLimitExceeded as e: - raise HTTPException( - status_code=429, - detail=build_rate_limit_detail(e, generation_task_repository, scope="user"), - ) from e - except GlobalQueueFull as e: + # 预检查队列限流（按变体总数计）——仅保留全局硬上限，用户上限改为软 warning 在 safe_enqueue 内处理（#2098） + global_pending = generation_task_repository.count_pending_total() + if global_pending + count > GLOBAL_PENDING_LIMIT: raise HTTPException( status_code=503, - detail=build_rate_limit_detail(e, generation_task_repository, scope="global"), - ) from e + detail=build_rate_limit_detail( + GlobalQueueFull(pending_count=global_pending + count, limit=GLOBAL_PENDING_LIMIT), + generation_task_repository, + scope="global", + ), + ) # 确定视频比例：优先前端传入，否则从模板 mode 推断 video_ratio = request.video_ratio or "" @@ -589,9 +578,6 @@ def create_preview_generation_task( if not enqueued: logger.warning("[预览生成] 任务入队失败: task_id=%s", task.id) _mark_task_failed(generation_task_repository, task, "任务入队失败") - except UserPendingLimitExceeded as e: - _mark_task_failed(generation_task_repository, task, "待处理任务超限") - rate_limit_exc = rate_limit_exc or e except GlobalQueueFull as e: _mark_task_failed(generation_task_repository, task, "系统队列已满") rate_limit_exc = rate_limit_exc or e @@ -603,11 +589,6 @@ def create_preview_generation_task( # 队列满/限流时若全部失败，返回结构化错误码（前端区分"排队"与"创建失败"） if all(r.status == "failed" for r in responses) and rate_limit_exc is not None: - if isinstance(rate_limit_exc, UserPendingLimitExceeded): - raise HTTPException( - status_code=429, - detail=build_rate_limit_detail(rate_limit_exc, generation_task_repository, scope="user"), - ) raise HTTPException( status_code=503, detail=build_rate_limit_detail(rate_limit_exc, generation_task_repository, scope="global"), diff --git a/apps/api/app/core/task_enqueue.py b/apps/api/app/core/task_enqueue.py index bea98657b..3675fe9de 100755 --- a/apps/api/app/core/task_enqueue.py +++ b/apps/api/app/core/task_enqueue.py @@ -154,19 +154,18 @@ def check_queue_limits( user_pending_limit: int = USER_PENDING_LIMIT, global_pending_limit: int = GLOBAL_PENDING_LIMIT, ) -> None: - """检查队列限流（预检查用，任务创建前调用），超限抛对应异常。 + """检查队列限流（预检查用，任务创建前调用）。 - 边界语义：>= 上限即拒绝（达到上限就不能再加新任务）。 + #2098 语义变更：用户级限流改为软提示，不再抛异常拒绝；仅全局硬上限抛 GlobalQueueFull。 Args: - user_id: 用户 ID + user_id: 用户 ID（保留参数，当前不做用户级硬拒） generation_task_repository: 任务仓储 - user_pending_limit: 单用户 pending 上限，默认 USER_PENDING_LIMIT + user_pending_limit: 单用户 pending 上限（保留，当前未硬拒） global_pending_limit: 全局 pending 上限，默认 GLOBAL_PENDING_LIMIT Raises: - GlobalQueueFull: 全局超限时抛出（优先级更高，先查全局） - UserPendingLimitExceeded: 用户超限时抛出 + GlobalQueueFull: 全局超限时抛出 """ # 先查全局（系统级保护优先级更高） global_pending = generation_task_repository.count_pending_total() @@ -179,17 +178,9 @@ def check_queue_limits( ) raise GlobalQueueFull(pending_count=global_pending, limit=global_pending_limit) - # 再查用户级 - if user_id: - user_pending = generation_task_repository.count_pending_by_user(user_id) - if user_pending >= user_pending_limit: - logger.warning( - "[队列限流] 用户 pending 任务数超限: user_id=%s, count=%d/%d", - user_id, - user_pending, - user_pending_limit, - ) - raise UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending, limit=user_pending_limit) + # #2098: 用户级限流改为软提示，不在预检查阶段拒绝（超额任务仍入队排队）。 + # 真正的系统保护由全局 GLOBAL_PENDING_LIMIT 硬上限承担。 + # UserPendingLimitExceeded 保留以兼容历史 import/except，但预检查与 safe_enqueue 均不再 raise。 def _mark_task_failed_safely( @@ -246,7 +237,6 @@ def safe_enqueue_generation_task( Raises: GlobalQueueFull: 全局 pending 超限时抛出，任务会被标记为 failed - UserPendingLimitExceeded: 用户 pending 超限时抛出，任务会被标记为 failed """ # ── 入队前检查：任务已是 pending，用 > 判断（包含当前任务） ── diff --git a/tests/unit/test_1677_batch_variants.py b/tests/unit/test_1677_batch_variants.py index a792f3e0d..0910a4507 100644 --- a/tests/unit/test_1677_batch_variants.py +++ b/tests/unit/test_1677_batch_variants.py @@ -9,7 +9,7 @@ """ from datetime import datetime, timezone -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, patch, patch import pytest from app.core.task_enqueue import GlobalQueueFull, UserPendingLimitExceeded @@ -322,22 +322,41 @@ class TestBatchPreviewRoute: ) assert captured[0].voice_library_id == "legacy_voice" - def test_preview_queue_limit_checks_total_count(self): - """限流预检查按变体总数计：用户 pending + N 超限 → 429""" + def test_preview_queue_limit_global_returns_503(self): + """#2098：仅全局硬上限仍 503 拒绝；用户超限自动排队不返回 429。""" from app.api.routes.generation_preview import create_preview_generation_task from fastapi import HTTPException - repo = MagicMock() - repo.count_pending_by_user.return_value = 3 - repo.count_pending_total.return_value = 0 + # 全局硬上限：预检查即 503，不会进入后续流程 + repo_global = MagicMock() + repo_global.count_pending_total.return_value = 100 with pytest.raises(HTTPException) as exc: create_preview_generation_task( _make_preview_request(preview_count=5), authenticated_user=_make_user(), - generation_task_repository=repo, + generation_task_repository=repo_global, db=MagicMock(), ) - assert exc.value.status_code == 429 + assert exc.value.status_code == 503 + + def test_preview_user_over_limit_does_not_429(self): + """#2098：用户 pending 远超软上限也不返回 429/4xx，请求进入业务流程后因 mock 不足走 500。""" + from app.api.routes.generation_preview import create_preview_generation_task + from fastapi import HTTPException + + repo_user = MagicMock() + repo_user.count_pending_by_user.return_value = 100 # 远超用户软上限 + repo_user.count_pending_total.return_value = 0 + with pytest.raises(HTTPException) as exc: + create_preview_generation_task( + _make_preview_request(preview_count=1), + authenticated_user=_make_user(), + generation_task_repository=repo_user, + db=MagicMock(), + ) + # 关键断言：不是 429（也不是 503，因为全局未超限），说明预检查放过了请求 + assert exc.value.status_code != 429 + assert exc.value.status_code != 503 def test_preview_reselect_failure_marks_all_failed(self): """#1743：变体独立选片（reselect）重试仍失败 → 已创建任务全部标记 failed 并 500""" diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py index 69523f2bd..109b11ee0 100644 --- a/tests/unit/test_generation_preview.py +++ b/tests/unit/test_generation_preview.py @@ -741,22 +741,22 @@ class TestCreatePreviewRoute: assert resp.items[0].status == "pending" assert resp.items[0].variant_index == 0 - def test_user_pending_limit_exceeded(self): - """用户待处理任务超限 → 429""" - repo = MagicMock() - repo.count_pending_by_user.return_value = 3 - repo.count_pending_total.return_value = 5 - + def test_user_pending_limit_no_longer_rejects(self): + """#2098: 用户待处理任务超限不再 429 拒绝（预检查仅全局 503）。""" from fastapi import HTTPException - - with pytest.raises(HTTPException) as exc_info: + repo = MagicMock() + repo.count_pending_by_user.return_value = 100 + repo.count_pending_total.return_value = 5 + with pytest.raises(HTTPException) as exc: create_preview_generation_task( self._make_request(), authenticated_user=_make_user(), generation_task_repository=repo, db=MagicMock(), ) - assert exc_info.value.status_code == 429 + # 不是 429（用户级硬拒已移除）也不是 503（全局未超限） + assert exc.value.status_code != 429 + assert exc.value.status_code != 503 def test_global_queue_full(self): """全局队列满 → 503""" @@ -844,36 +844,30 @@ class TestCreatePreviewRoute: assert resp.total == 1 assert resp.items[0].status == "failed" - def test_enqueue_raises_user_limit(self): - """safe_enqueue 抛出 UserPendingLimitExceeded → 429""" + def test_enqueue_user_limit_exception_no_longer_returns_429(self): + """#2098: 即使 safe_enqueue 模拟抛 UserPendingLimitExceeded，也不再触发 429 整体拒绝， + 变体被标记 failed 后正常返回响应。""" repo = MagicMock() repo.count_pending_by_user.return_value = 0 repo.count_pending_total.return_value = 0 - task = _make_task() - from fastapi import HTTPException - - def _set_failed_limit(error_message="", **_kwargs): + def _set_failed(reason="", **_kw): task.status = GenerationTaskStatus.FAILED - task.error_message = error_message or "待处理任务超限" - - task.mark_failed.side_effect = _set_failed_limit - - with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC: - MockUC.return_value.execute.return_value = task - with patch( - "app.api.routes.generation_preview.safe_enqueue_generation_task", - side_effect=UserPendingLimitExceeded(user_id="u1", pending_count=4, limit=3), - ): - with pytest.raises(HTTPException) as exc_info: - create_preview_generation_task( - self._make_request(), - authenticated_user=_make_user(), - generation_task_repository=repo, - db=MagicMock(), - ) - # 全部变体入队失败且错误消息含"待处理任务" → 429 - assert exc_info.value.status_code == 429 + task.error_message = reason + task.mark_failed.side_effect = _set_failed + repo.create.return_value = task + with patch( + "app.api.routes.generation_preview.safe_enqueue_generation_task", + side_effect=UserPendingLimitExceeded(user_id="u1", pending_count=4, limit=20), + ): + resp = create_preview_generation_task( + self._make_request(), + authenticated_user=_make_user(), + generation_task_repository=repo, + db=MagicMock(), + ) + assert resp.total == 1 + assert resp.items[0].status == "failed" def test_enqueue_raises_global_queue_full(self): """safe_enqueue 抛出 GlobalQueueFull → 503""" diff --git a/tests/unit/test_task_enqueue.py b/tests/unit/test_task_enqueue.py index e7046f9ff..b03fc3500 100755 --- a/tests/unit/test_task_enqueue.py +++ b/tests/unit/test_task_enqueue.py @@ -1,4 +1,4 @@ -"""task_enqueue 单测 — 队列限流 + 安全入队逻辑.""" +"""task_enqueue 单测 — 队列限流 + 安全入队逻辑 (#2098).""" from __future__ import annotations @@ -14,27 +14,18 @@ from app.core.task_enqueue import ( safe_enqueue_generation_task, ) -# ── Fixtures / Helpers ───────────────────────────────────────────────────── - - class MockRepository: - """Mock 任务仓储，用计数器模拟 pending 数量.""" - def __init__(self, global_count: int = 0, user_count: int = 0): self._global = global_count self._user = user_count self.update_called = 0 - def count_pending_total(self) -> int: return self._global - def count_pending_by_user(self, user_id: str) -> int: return self._user - def update(self, task): self.update_called += 1 - def make_mock_task(task_id: str = "task-1"): task = MagicMock() task.id = task_id @@ -42,242 +33,156 @@ def make_mock_task(task_id: str = "task-1"): task.mark_failed = MagicMock() return task - -# ── check_queue_limits ──────────────────────────────────────────────────── - - class TestCheckQueueLimits: - """check_queue_limits 预检查限流.""" - def test_below_limits_passes(self): repo = MockRepository(global_count=5, user_count=1) - # 不抛异常就是通过 check_queue_limits("user-1", repo) - def test_global_at_limit_raises(self): - """达到全局上限即拒绝.""" repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT, user_count=1) with pytest.raises(GlobalQueueFull) as exc_info: check_queue_limits("user-1", repo) assert exc_info.value.pending_count == GLOBAL_PENDING_LIMIT assert exc_info.value.limit == GLOBAL_PENDING_LIMIT - def test_global_over_limit_raises(self): repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT + 1, user_count=1) with pytest.raises(GlobalQueueFull): check_queue_limits("user-1", repo) - - def test_user_at_limit_raises(self): - """达到用户上限即拒绝.""" + def test_user_at_limit_no_longer_raises(self): repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT) - with pytest.raises(UserPendingLimitExceeded) as exc_info: - check_queue_limits("user-1", repo) - assert exc_info.value.user_id == "user-1" - assert exc_info.value.pending_count == USER_PENDING_LIMIT - assert exc_info.value.limit == USER_PENDING_LIMIT - - def test_user_over_limit_raises(self): - repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 1) - with pytest.raises(UserPendingLimitExceeded): - check_queue_limits("user-1", repo) - - def test_global_priority_over_user(self): - """全局和用户都超限时，优先抛全局异常.""" - repo = MockRepository( - global_count=GLOBAL_PENDING_LIMIT + 1, - user_count=USER_PENDING_LIMIT + 1, - ) - with pytest.raises(GlobalQueueFull): - check_queue_limits("user-1", repo) - + check_queue_limits("user-1", repo) + def test_user_over_limit_no_longer_raises(self): + repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 100) + check_queue_limits("user-1", repo) def test_empty_user_id_skips_user_check(self): - """user_id 为空时跳过用户级检查.""" repo = MockRepository(global_count=5, user_count=999) - # 不抛异常 = 通过（只检查全局） check_queue_limits("", repo) - - def test_custom_limits(self): - """支持自定义限流阈值.""" - repo = MockRepository(global_count=5, user_count=5) - # 默认阈值下 user 5 > 3 会被拒 - with pytest.raises(UserPendingLimitExceeded): - check_queue_limits("u1", repo) - - # 自定义更高阈值就能通过 - check_queue_limits("u1", repo, user_pending_limit=10, global_pending_limit=10) - - -# ── safe_enqueue_generation_task ────────────────────────────────────────── - + def test_custom_global_limit_still_honored(self): + repo = MockRepository(global_count=15, user_count=999) + with pytest.raises(GlobalQueueFull): + check_queue_limits("u1", repo, user_pending_limit=999, global_pending_limit=10) + check_queue_limits("u1", repo, user_pending_limit=999, global_pending_limit=20) class TestSafeEnqueueGenerationTask: - """safe_enqueue_generation_task 安全入队.""" - @patch("app.core.task_enqueue.celery_app") def test_success_path(self, mock_celery): - """正常路径：入队前检查通过 → 发送Celery → 入队后检查通过.""" repo = MockRepository(global_count=1, user_count=1) task = make_mock_task() - result = safe_enqueue_generation_task(task, repo, user_id="user-1") - assert result is True mock_celery.send_task.assert_called_once_with("worker.generate_video", args=[task.id]) task.mark_failed.assert_not_called() @patch("app.core.task_enqueue.celery_app") def test_no_user_id_skips_user_check(self, mock_celery): - """不传 user_id 跳过用户级限流.""" repo = MockRepository(global_count=1, user_count=999) task = make_mock_task() - result = safe_enqueue_generation_task(task, repo, user_id="") assert result is True @patch("app.core.task_enqueue.celery_app") def test_precheck_global_over_marks_failed(self, mock_celery): - """入队前全局超限：标记 failed，抛异常.""" repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT + 1, user_count=0) task = make_mock_task() - with pytest.raises(GlobalQueueFull): safe_enqueue_generation_task(task, repo, user_id="user-1") - task.mark_failed.assert_called_once() mock_celery.send_task.assert_not_called() assert repo.update_called == 1 @patch("app.core.task_enqueue.celery_app") - def test_precheck_user_over_marks_failed(self, mock_celery): - """入队前用户超限：标记 failed，抛异常.""" - repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 1) + def test_precheck_user_over_still_enqueues(self, mock_celery, caplog): + import logging + caplog.set_level(logging.WARNING) + repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 100) task = make_mock_task() - - with pytest.raises(UserPendingLimitExceeded): - safe_enqueue_generation_task(task, repo, user_id="user-1") - - task.mark_failed.assert_called_once() - mock_celery.send_task.assert_not_called() + result = safe_enqueue_generation_task(task, repo, user_id="user-1") + assert result is True + mock_celery.send_task.assert_called_once() + task.mark_failed.assert_not_called() + assert any("超过软上限" in r.message for r in caplog.records) @patch("app.core.task_enqueue.celery_app") def test_celery_send_false_returns_false(self, mock_celery): - """Celery 发送失败：返回 False，任务标记 failed.""" repo = MockRepository(global_count=1, user_count=1) task = make_mock_task() mock_celery.send_task.side_effect = Exception("celery down") - result = safe_enqueue_generation_task(task, repo, user_id="user-1") - assert result is False task.mark_failed.assert_called_once() assert "入队失败" in task.mark_failed.call_args[0][0] @patch("app.core.task_enqueue.celery_app") def test_celery_send_failure_update_also_fails(self, mock_celery): - """Celery 发送失败 + mark_failed 更新也失败：不崩溃.""" repo = MockRepository(global_count=1, user_count=1) repo.update = MagicMock(side_effect=Exception("db down")) task = make_mock_task() mock_celery.send_task.side_effect = Exception("celery down") - result = safe_enqueue_generation_task(task, repo, user_id="user-1") - assert result is False - # 不抛异常就是胜利 @patch("app.core.task_enqueue.celery_app") def test_postcheck_global_over_rollback(self, mock_celery): - """入队后全局超限（并发竞态）：回滚标记 failed，抛异常.""" - # 入队前刚好通过，但入队后再查发现超限 call_count = [0] - def count_pending_total_side_effect(): call_count[0] += 1 - if call_count[0] == 1: # 入队前检查 - return GLOBAL_PENDING_LIMIT # 等于上限，用 > 判断所以通过 - return GLOBAL_PENDING_LIMIT + 1 # 入队后再查，超限 - + if call_count[0] == 1: + return GLOBAL_PENDING_LIMIT + return GLOBAL_PENDING_LIMIT + 1 repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT, user_count=0) repo.count_pending_total = MagicMock(side_effect=count_pending_total_side_effect) task = make_mock_task() - with pytest.raises(GlobalQueueFull): safe_enqueue_generation_task(task, repo, user_id="user-1") - - # 异常是 GlobalQueueFull 类型，且任务已被标记为 failed（含"入队后"原因） task.mark_failed.assert_called_once() assert "入队后" in task.mark_failed.call_args[0][0] mock_celery.send_task.assert_called_once() @patch("app.core.task_enqueue.celery_app") - def test_postcheck_user_over_rollback(self, mock_celery): - """入队后用户超限：回滚标记 failed，抛异常.""" + def test_postcheck_user_over_does_not_rollback(self, mock_celery, caplog): + import logging + caplog.set_level(logging.WARNING) repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT) - # 入队前用 > 判断，等于上限通过；入队后模拟并发超限 - original_user_count = repo.count_pending_by_user - call_count = [0] - def count_by_user_side_effect(user_id): call_count[0] += 1 - if call_count[0] <= 1: # 入队前 - return USER_PENDING_LIMIT # 用 > 判断，等于时通过 - return USER_PENDING_LIMIT + 1 # 入队后，超限 - + if call_count[0] <= 1: + return USER_PENDING_LIMIT + return USER_PENDING_LIMIT + 1 repo.count_pending_by_user = MagicMock(side_effect=count_by_user_side_effect) task = make_mock_task() - - with pytest.raises(UserPendingLimitExceeded): - safe_enqueue_generation_task(task, repo, user_id="user-1") - - task.mark_failed.assert_called_once() + result = safe_enqueue_generation_task(task, repo, user_id="user-1") + assert result is True + mock_celery.send_task.assert_called_once() + task.mark_failed.assert_not_called() + assert any("超软上限(入队后)" in r.message for r in caplog.records) @patch("app.core.task_enqueue.celery_app") def test_log_task_status_enabled(self, mock_celery): - """log_task_status=True 时日志中包含状态.""" repo = MockRepository(global_count=1, user_count=1) task = make_mock_task() - result = safe_enqueue_generation_task(task, repo, user_id="user-1", log_task_status=True) assert result is True @patch("app.core.task_enqueue.celery_app") - def test_custom_limits_in_enqueue(self, mock_celery): - """自定义限流阈值用于入队检查.""" - repo = MockRepository(global_count=5, user_count=5) + def test_custom_global_limit_in_enqueue(self, mock_celery): + repo = MockRepository(global_count=15, user_count=999) task = make_mock_task() - - # 默认阈值下用户 5 > 3 会被拒 - with pytest.raises(UserPendingLimitExceeded): - safe_enqueue_generation_task(task, repo, user_id="user-1") - - # 重置 mock 计数 - task.mark_failed.reset_mock() - - # 调大阈值后通过 - result = safe_enqueue_generation_task( - task, - repo, - user_id="user-1", - user_pending_limit=10, - global_pending_limit=10, - ) + result = safe_enqueue_generation_task(task, repo, user_id="user-1") assert result is True - - -# ── 异常类 ──────────────────────────────────────────────────────────────── - + task2 = make_mock_task("task-2") + mock_celery.reset_mock() + with pytest.raises(GlobalQueueFull): + safe_enqueue_generation_task(task2, repo, user_id="user-1", user_pending_limit=999, global_pending_limit=10) class TestExceptionClasses: - """异常类消息格式.""" - def test_user_pending_limit_message(self): exc = UserPendingLimitExceeded("u1", 5, 3) assert "u1" in str(exc) assert "5" in str(exc) assert "3" in str(exc) - def test_global_queue_full_message(self): exc = GlobalQueueFull(25, 20) assert "25" in str(exc) assert "20" in str(exc) + def test_user_pending_limit_constant(self): + assert USER_PENDING_LIMIT == 20 diff --git a/tests/unit/test_task_queue_limit.py b/tests/unit/test_task_queue_limit.py index b78baaf96..2611e749a 100644 --- a/tests/unit/test_task_queue_limit.py +++ b/tests/unit/test_task_queue_limit.py @@ -1,4 +1,4 @@ -"""任务队列限流防护单元测试。""" +"""任务队列限流防护单元测试 (#2098: 用户级改为软上限，仅全局硬拒).""" from __future__ import annotations @@ -19,35 +19,21 @@ from app.core.task_enqueue import ( safe_enqueue_generation_task, ) -# --------------------------------------------------------------------------- -# Mock helpers -# --------------------------------------------------------------------------- - class MockRepository: - """支持 pending 计数的 mock repository。 - - 支持通过 set_pending 动态修改计数，用于模拟入队后计数变化的并发场景。 - """ - def __init__(self, user_pending: int = 0, global_pending: int = 0): self._user_pending = user_pending self._global_pending = global_pending self._send_task_called = False self.updated_tasks = [] - def count_pending_by_user(self, user_id: str) -> int: return self._user_pending - def count_pending_total(self) -> int: return self._global_pending - def update(self, task): self.updated_tasks.append(task) return task - def set_pending(self, *, user_pending: int | None = None, global_pending: int | None = None): - """动态修改 pending 计数，模拟并发场景。""" if user_pending is not None: self._user_pending = user_pending if global_pending is not None: @@ -59,7 +45,6 @@ class MockTask: self.id = task_id self.status = status self.error_message = "" - def mark_failed(self, reason: str): self.status = "failed" self.error_message = reason @@ -67,58 +52,39 @@ class MockTask: @pytest.fixture(autouse=True) def mock_celery(monkeypatch): - """mock 掉 celery_app.send_task，避免真实发送。""" mock_send = MagicMock() monkeypatch.setattr("app.core.celery_app.celery_app.send_task", mock_send) return mock_send -# --------------------------------------------------------------------------- -# 常量导出测试 -# --------------------------------------------------------------------------- - - def test_limit_constants_are_exported(): - """限流阈值常量已导出，供业务代码引用。""" - assert USER_PENDING_LIMIT == 3 + """#2098: USER_PENDING_LIMIT 从 3 提到 20 作为软上限；GLOBAL_PENDING_LIMIT 保持 20 为硬上限。""" + assert USER_PENDING_LIMIT == 20 assert GLOBAL_PENDING_LIMIT == 20 -# --------------------------------------------------------------------------- -# check_queue_limits 单元测试（预检查用，>= 边界） -# --------------------------------------------------------------------------- - - class TestCheckQueueLimits: - """队列限流检查函数测试（预检查语义，>= 上限即拒绝）。""" + """check_queue_limits 预检查：仅全局硬上限拒绝，用户级改为软提示。""" def test_normal_passes_through(self): - """正常范围内的任务不受限制。""" repo = MockRepository(user_pending=1, global_pending=5) check_queue_limits("user-1", repo) - def test_user_limit_exceeded_raises(self): - """用户 pending 超过上限抛 UserPendingLimitExceeded。""" - repo = MockRepository(user_pending=4, global_pending=5) - with pytest.raises(UserPendingLimitExceeded) as exc_info: - check_queue_limits("user-1", repo) - assert exc_info.value.user_id == "user-1" - assert exc_info.value.pending_count == 4 - assert exc_info.value.limit == 3 + def test_user_limit_exceeded_no_longer_raises(self): + """#2098: 用户 pending 超过软上限不再抛异常。""" + repo = MockRepository(user_pending=100, global_pending=5) + check_queue_limits("user-1", repo) # 不抛即通过 - def test_user_at_limit_also_raises(self): - """用户 pending 刚好等于上限也拒绝（>= 边界）。""" - repo = MockRepository(user_pending=3, global_pending=5) - with pytest.raises(UserPendingLimitExceeded): - check_queue_limits("user-1", repo) + def test_user_at_limit_no_longer_raises(self): + """#2098: 用户 pending 等于软上限也不拒绝。""" + repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5) + check_queue_limits("user-1", repo) def test_user_below_limit_passes(self): - """用户 pending 比上限少 1，通过。""" repo = MockRepository(user_pending=2, global_pending=5) check_queue_limits("user-1", repo) def test_global_limit_exceeded_raises(self): - """全局 pending 超过上限抛 GlobalQueueFull。""" repo = MockRepository(user_pending=1, global_pending=21) with pytest.raises(GlobalQueueFull) as exc_info: check_queue_limits("user-1", repo) @@ -126,82 +92,51 @@ class TestCheckQueueLimits: assert exc_info.value.limit == 20 def test_global_at_limit_also_raises(self): - """全局 pending 刚好等于上限也拒绝（>= 边界）。""" repo = MockRepository(user_pending=1, global_pending=20) with pytest.raises(GlobalQueueFull): check_queue_limits("user-1", repo) def test_global_below_limit_passes(self): - """全局 pending 比上限少 1，通过。""" repo = MockRepository(user_pending=1, global_pending=19) check_queue_limits("user-1", repo) - def test_global_takes_priority_over_user(self): - """全局和用户都超限时，优先抛全局异常。""" - repo = MockRepository(user_pending=5, global_pending=25) - with pytest.raises(GlobalQueueFull): - check_queue_limits("user-1", repo) - def test_empty_user_id_skips_user_check(self): - """不传 user_id 时跳过用户级检查，只做全局检查。""" repo = MockRepository(user_pending=10, global_pending=5) - # 用户超限但不传 user_id → 全局未超限，应该通过 check_queue_limits("", repo) -# --------------------------------------------------------------------------- -# safe_enqueue_generation_task 限流集成测试（入队前用 >，包含当前任务） -# --------------------------------------------------------------------------- - - class TestSafeEnqueueWithLimits: - """安全入队函数的限流功能测试。""" + """safe_enqueue_generation_task：用户超限仅 warning 仍入队；全局超限硬拒。""" def test_normal_task_enqueues_successfully(self, mock_celery): - """正常任务入队成功，返回 True。""" repo = MockRepository(user_pending=0, global_pending=0) task = MockTask("task-1") result = safe_enqueue_generation_task(task, repo, user_id="user-1") assert result is True mock_celery.assert_called_once_with("worker.generate_video", args=["task-1"]) - # 成功入队后持久化 celery 消息 ID（#1714：孤儿清理据此 revoke/清队列） assert len(repo.updated_tasks) == 1 assert task.celery_task_id - def test_user_limit_rejected_with_failed_status(self, mock_celery): - """用户超限：任务标记为 failed，抛 UserPendingLimitExceeded。""" - repo = MockRepository(user_pending=5, global_pending=5) + def test_user_limit_exceeded_still_enqueues(self, mock_celery, caplog): + """#2098: 用户远超软上限仍入队，任务不被标记 failed。""" + import logging + caplog.set_level(logging.WARNING) + repo = MockRepository(user_pending=100, global_pending=5) task = MockTask("task-1") - with pytest.raises(UserPendingLimitExceeded): - safe_enqueue_generation_task(task, repo, user_id="user-1") - mock_celery.assert_not_called() - assert task.status == "failed" - assert "限流" in task.error_message - assert len(repo.updated_tasks) == 1 + result = safe_enqueue_generation_task(task, repo, user_id="user-1") + assert result is True + mock_celery.assert_called_once() + assert task.status == "pending" # 没被标记 failed + assert any("超过软上限" in r.message for r in caplog.records) def test_user_at_limit_still_passes(self, mock_celery): - """用户 pending 刚好等于上限：入队前检查用 >，包含当前任务，刚好到上限不算超。 - - 与预检查的 >= 语义一致：预检查时 pending=3 拒绝（不能再加新的）， - 但 safe_enqueue 被调用时任务已是 pending（就是第3个）， - pending=3 不满足 >3，所以通过。 - """ - repo = MockRepository(user_pending=3, global_pending=5) + repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5) task = MockTask("task-1") result = safe_enqueue_generation_task(task, repo, user_id="user-1") assert result is True mock_celery.assert_called_once() - def test_user_one_over_limit_rejected(self, mock_celery): - """用户 pending = limit + 1：超限被拒。""" - repo = MockRepository(user_pending=4, global_pending=5) - task = MockTask("task-1") - with pytest.raises(UserPendingLimitExceeded): - safe_enqueue_generation_task(task, repo, user_id="user-1") - mock_celery.assert_not_called() - def test_global_limit_rejected_with_failed_status(self, mock_celery): - """全局超限：任务标记为 failed，抛 GlobalQueueFull。""" repo = MockRepository(user_pending=1, global_pending=21) task = MockTask("task-1") with pytest.raises(GlobalQueueFull): @@ -211,7 +146,6 @@ class TestSafeEnqueueWithLimits: assert len(repo.updated_tasks) == 1 def test_global_at_limit_still_passes(self, mock_celery): - """全局 pending 刚好等于上限：入队前检查用 >，包含当前任务，刚好到上限不算超。""" repo = MockRepository(user_pending=1, global_pending=20) task = MockTask("task-1") result = safe_enqueue_generation_task(task, repo, user_id="user-1") @@ -219,7 +153,6 @@ class TestSafeEnqueueWithLimits: mock_celery.assert_called_once() def test_no_user_id_skips_user_limit(self, mock_celery): - """不传 user_id 时跳过用户级限流，只做全局检查。""" repo = MockRepository(user_pending=10, global_pending=5) task = MockTask("task-1") result = safe_enqueue_generation_task(task, repo, user_id="") @@ -227,7 +160,6 @@ class TestSafeEnqueueWithLimits: mock_celery.assert_called_once() def test_no_user_id_still_checks_global(self, mock_celery): - """不传 user_id 时全局超限仍然被拦。""" repo = MockRepository(user_pending=10, global_pending=25) task = MockTask("task-1") with pytest.raises(GlobalQueueFull): @@ -235,121 +167,67 @@ class TestSafeEnqueueWithLimits: mock_celery.assert_not_called() def test_default_limits_match_constants(self, mock_celery): - """默认配置与导出常量一致。""" - # 刚好在默认限制内（limit - 1） repo = MockRepository(user_pending=2, global_pending=19) task = MockTask("task-1") result = safe_enqueue_generation_task(task, repo, user_id="user-1") assert result is True - def test_update_failure_does_not_crash(self, mock_celery): - """repository.update 失败也不崩溃，异常继续向上抛。""" - - class BadRepo(MockRepository): - def update(self, task): - raise RuntimeError("db down") - - repo = BadRepo(user_pending=5, global_pending=5) - task = MockTask("task-1") - # 仍然抛 UserPendingLimitExceeded，不会被 update 失败掩盖 - with pytest.raises(UserPendingLimitExceeded): - safe_enqueue_generation_task(task, repo, user_id="user-1") - mock_celery.assert_not_called() - # 任务状态还是变了（内存里改了） - assert task.status == "failed" - - -# --------------------------------------------------------------------------- -# 入队后最终校验（并发竞态兜底）测试 -# --------------------------------------------------------------------------- - class TestPostEnqueueFinalCheck: - """入队后最终校验：模拟并发场景，Celery发送后计数增加被兜住。""" + """入队后校验：仅全局超限回滚；用户超限仅 warning。""" def test_post_enqueue_global_overflow_rollback(self, mock_celery): - """并发场景：入队前检查通过，但发送Celery后全局计数超限 → 回滚为failed。 - - 模拟两个请求同时通过入队前检查（都查到 global=19）， - 都创建了任务（DB里变成 21），先发送Celery的那个在最终校验时被兜住。 - """ - repo = MockRepository(user_pending=1, global_pending=20) # 入队前：20 > 20？否 + repo = MockRepository(user_pending=1, global_pending=20) task = MockTask("task-1") - - # 模拟发送Celery后，另一个并发请求也创建了任务，全局变成21 def side_effect(*args, **kwargs): repo.set_pending(global_pending=21) - mock_celery.side_effect = side_effect - with pytest.raises(GlobalQueueFull) as exc_info: safe_enqueue_generation_task(task, repo, user_id="user-1") - - # Celery 确实发出去了（兜底不撤销 Celery，只回滚 DB 状态） mock_celery.assert_called_once() - # 任务被标记为 failed assert task.status == "failed" assert "入队后" in task.error_message assert exc_info.value.pending_count == 21 assert len(repo.updated_tasks) == 1 - def test_post_enqueue_user_overflow_rollback(self, mock_celery): - """并发场景：入队前检查通过，但发送Celery后用户计数超限 → 回滚为failed。""" - repo = MockRepository(user_pending=3, global_pending=5) # 入队前：3 > 3？否 + def test_post_enqueue_user_overflow_no_rollback(self, mock_celery, caplog): + """#2098: 入队后用户超软上限仅 warning，不回滚。""" + import logging + caplog.set_level(logging.WARNING) + repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5) task = MockTask("task-1") - def side_effect(*args, **kwargs): - repo.set_pending(user_pending=4) - + repo.set_pending(user_pending=USER_PENDING_LIMIT + 1) mock_celery.side_effect = side_effect - - with pytest.raises(UserPendingLimitExceeded) as exc_info: - safe_enqueue_generation_task(task, repo, user_id="user-1") - - mock_celery.assert_called_once() - assert task.status == "failed" - assert "入队后" in task.error_message - assert exc_info.value.user_id == "user-1" - assert exc_info.value.pending_count == 4 - - def test_post_enqueue_global_priority_over_user(self, mock_celery): - """入队后校验：全局和用户都超限时，优先抛全局异常。""" - repo = MockRepository(user_pending=3, global_pending=20) - task = MockTask("task-1") - - def side_effect(*args, **kwargs): - repo.set_pending(user_pending=5, global_pending=22) - - mock_celery.side_effect = side_effect - - with pytest.raises(GlobalQueueFull): - safe_enqueue_generation_task(task, repo, user_id="user-1") - - assert task.status == "failed" - - def test_post_enqueue_no_change_still_passes(self, mock_celery): - """入队后计数没变 → 正常通过，不回滚。""" - repo = MockRepository(user_pending=2, global_pending=10) - task = MockTask("task-1") - result = safe_enqueue_generation_task(task, repo, user_id="user-1") - assert result is True mock_celery.assert_called_once() - assert task.status == "pending" # 状态没变 - # 入队成功后持久化 celery_task_id（#1714），业务状态不变 + assert task.status == "pending" # 不回滚 + assert any("超软上限(入队后)" in r.message for r in caplog.records) + + def test_post_enqueue_no_change_still_passes(self, mock_celery): + repo = MockRepository(user_pending=2, global_pending=10) + task = MockTask("task-1") + result = safe_enqueue_generation_task(task, repo, user_id="user-1") + assert result is True + mock_celery.assert_called_once() + assert task.status == "pending" assert len(repo.updated_tasks) == 1 assert task.celery_task_id def test_post_enqueue_no_user_id_skips_user_check(self, mock_celery): - """不传 user_id 时，入队后校验也跳过用户级，只查全局。""" repo = MockRepository(user_pending=10, global_pending=5) task = MockTask("task-1") - def side_effect(*args, **kwargs): - repo.set_pending(user_pending=15, global_pending=5) # 用户超限但全局没超 - + repo.set_pending(user_pending=15, global_pending=5) mock_celery.side_effect = side_effect - result = safe_enqueue_generation_task(task, repo, user_id="") - assert result is True # 用户级不检查，全局没超限 → 通过 + assert result is True + + +def test_user_pending_limit_exceeded_class_still_exists(): + """UserPendingLimitExceeded 保留用于兼容历史 import/except（#2098 后不再主动 raise）。""" + exc = UserPendingLimitExceeded("u1", 5, 3) + assert exc.user_id == "u1" + assert exc.pending_count == 5 + assert exc.limit == 3 diff --git a/tests/unit/test_worker_generate_video_task_binding.py b/tests/unit/test_worker_generate_video_task_binding.py index 17414a0af..0018c8bc6 100644 --- a/tests/unit/test_worker_generate_video_task_binding.py +++ b/tests/unit/test_worker_generate_video_task_binding.py @@ -65,10 +65,21 @@ def test_generate_video_preserves_original_function(): assert original.__name__ == "generate_video", f"expected __name__='generate_video', got '{original.__name__}'" -def test_sync_task_config_to_plan_is_plain_function(): - """Helper must NOT be registered as a Celery task.""" - from worker_app.tasks.generation import _sync_task_config_to_plan +def test_build_task_config_override_is_plain_function(): + """#2098: 原 _sync_task_config_to_plan 已拆分为 _build_task_config_override + _download_voice_for_task， + 均为普通函数，不应被注册为 Celery task。""" + from worker_app.tasks.generation import _build_task_config_override, _download_voice_for_task - assert not hasattr( - _sync_task_config_to_plan, "run" - ), "_sync_task_config_to_plan must be a plain function, not a Celery task" + for fn in (_build_task_config_override, _download_voice_for_task): + assert not hasattr(fn, "run"), f"{fn.__name__} must be a plain function, not a Celery task" + + # Bug A: override 对 title_config 做 key 归一化 (font_size→size, font_color→color) + override = _build_task_config_override({ + "title_config": {"font_size": 48, "font_color": "#ff0000", "text": "hi"}, + "bgm_config": {"url": "http://x/bgm.mp3"}, + "output_width": 1080, "output_height": 1920, + }) + assert override["title"]["size"] == 48 + assert override["title"]["color"] == "#ff0000" + assert override["bgm"]["url"] == "http://x/bgm.mp3" + assert override["export"]["resolution"] == "1080x1920" -- 2.54.0 From efb7fa57294d2e2f4e77b839b6f8b3f7f7b63859 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Tue, 29 Sep 2026 11:17:23 +0000 Subject: [PATCH 4/4] style: auto-format with black + isort + ruff + prettier [skip ci-format-check] --- tests/unit/test_1677_batch_variants.py | 2 +- tests/unit/test_generation_preview.py | 3 +++ tests/unit/test_task_enqueue.py | 22 +++++++++++++++++++ tests/unit/test_task_queue_limit.py | 13 +++++++++++ ...test_worker_generate_video_task_binding.py | 13 ++++++----- 5 files changed, 47 insertions(+), 6 deletions(-) diff --git a/tests/unit/test_1677_batch_variants.py b/tests/unit/test_1677_batch_variants.py index 0910a4507..c288a7ae6 100644 --- a/tests/unit/test_1677_batch_variants.py +++ b/tests/unit/test_1677_batch_variants.py @@ -9,7 +9,7 @@ """ from datetime import datetime, timezone -from unittest.mock import MagicMock, patch, patch +from unittest.mock import MagicMock, patch import pytest from app.core.task_enqueue import GlobalQueueFull, UserPendingLimitExceeded diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py index 109b11ee0..3678e894b 100644 --- a/tests/unit/test_generation_preview.py +++ b/tests/unit/test_generation_preview.py @@ -744,6 +744,7 @@ class TestCreatePreviewRoute: def test_user_pending_limit_no_longer_rejects(self): """#2098: 用户待处理任务超限不再 429 拒绝（预检查仅全局 503）。""" from fastapi import HTTPException + repo = MagicMock() repo.count_pending_by_user.return_value = 100 repo.count_pending_total.return_value = 5 @@ -851,9 +852,11 @@ class TestCreatePreviewRoute: repo.count_pending_by_user.return_value = 0 repo.count_pending_total.return_value = 0 task = _make_task() + def _set_failed(reason="", **_kw): task.status = GenerationTaskStatus.FAILED task.error_message = reason + task.mark_failed.side_effect = _set_failed repo.create.return_value = task with patch( diff --git a/tests/unit/test_task_enqueue.py b/tests/unit/test_task_enqueue.py index b03fc3500..34c9bdf7c 100755 --- a/tests/unit/test_task_enqueue.py +++ b/tests/unit/test_task_enqueue.py @@ -14,18 +14,23 @@ from app.core.task_enqueue import ( safe_enqueue_generation_task, ) + class MockRepository: def __init__(self, global_count: int = 0, user_count: int = 0): self._global = global_count self._user = user_count self.update_called = 0 + def count_pending_total(self) -> int: return self._global + def count_pending_by_user(self, user_id: str) -> int: return self._user + def update(self, task): self.update_called += 1 + def make_mock_task(task_id: str = "task-1"): task = MagicMock() task.id = task_id @@ -33,35 +38,43 @@ def make_mock_task(task_id: str = "task-1"): task.mark_failed = MagicMock() return task + class TestCheckQueueLimits: def test_below_limits_passes(self): repo = MockRepository(global_count=5, user_count=1) check_queue_limits("user-1", repo) + def test_global_at_limit_raises(self): repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT, user_count=1) with pytest.raises(GlobalQueueFull) as exc_info: check_queue_limits("user-1", repo) assert exc_info.value.pending_count == GLOBAL_PENDING_LIMIT assert exc_info.value.limit == GLOBAL_PENDING_LIMIT + def test_global_over_limit_raises(self): repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT + 1, user_count=1) with pytest.raises(GlobalQueueFull): check_queue_limits("user-1", repo) + def test_user_at_limit_no_longer_raises(self): repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT) check_queue_limits("user-1", repo) + def test_user_over_limit_no_longer_raises(self): repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 100) check_queue_limits("user-1", repo) + def test_empty_user_id_skips_user_check(self): repo = MockRepository(global_count=5, user_count=999) check_queue_limits("", repo) + def test_custom_global_limit_still_honored(self): repo = MockRepository(global_count=15, user_count=999) with pytest.raises(GlobalQueueFull): check_queue_limits("u1", repo, user_pending_limit=999, global_pending_limit=10) check_queue_limits("u1", repo, user_pending_limit=999, global_pending_limit=20) + class TestSafeEnqueueGenerationTask: @patch("app.core.task_enqueue.celery_app") def test_success_path(self, mock_celery): @@ -92,6 +105,7 @@ class TestSafeEnqueueGenerationTask: @patch("app.core.task_enqueue.celery_app") def test_precheck_user_over_still_enqueues(self, mock_celery, caplog): import logging + caplog.set_level(logging.WARNING) repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT + 100) task = make_mock_task() @@ -123,11 +137,13 @@ class TestSafeEnqueueGenerationTask: @patch("app.core.task_enqueue.celery_app") def test_postcheck_global_over_rollback(self, mock_celery): call_count = [0] + def count_pending_total_side_effect(): call_count[0] += 1 if call_count[0] == 1: return GLOBAL_PENDING_LIMIT return GLOBAL_PENDING_LIMIT + 1 + repo = MockRepository(global_count=GLOBAL_PENDING_LIMIT, user_count=0) repo.count_pending_total = MagicMock(side_effect=count_pending_total_side_effect) task = make_mock_task() @@ -140,14 +156,17 @@ class TestSafeEnqueueGenerationTask: @patch("app.core.task_enqueue.celery_app") def test_postcheck_user_over_does_not_rollback(self, mock_celery, caplog): import logging + caplog.set_level(logging.WARNING) repo = MockRepository(global_count=5, user_count=USER_PENDING_LIMIT) call_count = [0] + def count_by_user_side_effect(user_id): call_count[0] += 1 if call_count[0] <= 1: return USER_PENDING_LIMIT return USER_PENDING_LIMIT + 1 + repo.count_pending_by_user = MagicMock(side_effect=count_by_user_side_effect) task = make_mock_task() result = safe_enqueue_generation_task(task, repo, user_id="user-1") @@ -174,15 +193,18 @@ class TestSafeEnqueueGenerationTask: with pytest.raises(GlobalQueueFull): safe_enqueue_generation_task(task2, repo, user_id="user-1", user_pending_limit=999, global_pending_limit=10) + class TestExceptionClasses: def test_user_pending_limit_message(self): exc = UserPendingLimitExceeded("u1", 5, 3) assert "u1" in str(exc) assert "5" in str(exc) assert "3" in str(exc) + def test_global_queue_full_message(self): exc = GlobalQueueFull(25, 20) assert "25" in str(exc) assert "20" in str(exc) + def test_user_pending_limit_constant(self): assert USER_PENDING_LIMIT == 20 diff --git a/tests/unit/test_task_queue_limit.py b/tests/unit/test_task_queue_limit.py index 2611e749a..8e9dc4eb3 100644 --- a/tests/unit/test_task_queue_limit.py +++ b/tests/unit/test_task_queue_limit.py @@ -26,13 +26,17 @@ class MockRepository: self._global_pending = global_pending self._send_task_called = False self.updated_tasks = [] + def count_pending_by_user(self, user_id: str) -> int: return self._user_pending + def count_pending_total(self) -> int: return self._global_pending + def update(self, task): self.updated_tasks.append(task) return task + def set_pending(self, *, user_pending: int | None = None, global_pending: int | None = None): if user_pending is not None: self._user_pending = user_pending @@ -45,6 +49,7 @@ class MockTask: self.id = task_id self.status = status self.error_message = "" + def mark_failed(self, reason: str): self.status = "failed" self.error_message = reason @@ -120,6 +125,7 @@ class TestSafeEnqueueWithLimits: def test_user_limit_exceeded_still_enqueues(self, mock_celery, caplog): """#2098: 用户远超软上限仍入队，任务不被标记 failed。""" import logging + caplog.set_level(logging.WARNING) repo = MockRepository(user_pending=100, global_pending=5) task = MockTask("task-1") @@ -179,8 +185,10 @@ class TestPostEnqueueFinalCheck: def test_post_enqueue_global_overflow_rollback(self, mock_celery): repo = MockRepository(user_pending=1, global_pending=20) task = MockTask("task-1") + def side_effect(*args, **kwargs): repo.set_pending(global_pending=21) + mock_celery.side_effect = side_effect with pytest.raises(GlobalQueueFull) as exc_info: safe_enqueue_generation_task(task, repo, user_id="user-1") @@ -193,11 +201,14 @@ class TestPostEnqueueFinalCheck: def test_post_enqueue_user_overflow_no_rollback(self, mock_celery, caplog): """#2098: 入队后用户超软上限仅 warning，不回滚。""" import logging + caplog.set_level(logging.WARNING) repo = MockRepository(user_pending=USER_PENDING_LIMIT, global_pending=5) task = MockTask("task-1") + def side_effect(*args, **kwargs): repo.set_pending(user_pending=USER_PENDING_LIMIT + 1) + mock_celery.side_effect = side_effect result = safe_enqueue_generation_task(task, repo, user_id="user-1") assert result is True @@ -218,8 +229,10 @@ class TestPostEnqueueFinalCheck: def test_post_enqueue_no_user_id_skips_user_check(self, mock_celery): repo = MockRepository(user_pending=10, global_pending=5) task = MockTask("task-1") + def side_effect(*args, **kwargs): repo.set_pending(user_pending=15, global_pending=5) + mock_celery.side_effect = side_effect result = safe_enqueue_generation_task(task, repo, user_id="") assert result is True diff --git a/tests/unit/test_worker_generate_video_task_binding.py b/tests/unit/test_worker_generate_video_task_binding.py index 0018c8bc6..e3f57a4ec 100644 --- a/tests/unit/test_worker_generate_video_task_binding.py +++ b/tests/unit/test_worker_generate_video_task_binding.py @@ -74,11 +74,14 @@ def test_build_task_config_override_is_plain_function(): assert not hasattr(fn, "run"), f"{fn.__name__} must be a plain function, not a Celery task" # Bug A: override 对 title_config 做 key 归一化 (font_size→size, font_color→color) - override = _build_task_config_override({ - "title_config": {"font_size": 48, "font_color": "#ff0000", "text": "hi"}, - "bgm_config": {"url": "http://x/bgm.mp3"}, - "output_width": 1080, "output_height": 1920, - }) + override = _build_task_config_override( + { + "title_config": {"font_size": 48, "font_color": "#ff0000", "text": "hi"}, + "bgm_config": {"url": "http://x/bgm.mp3"}, + "output_width": 1080, + "output_height": 1920, + } + ) assert override["title"]["size"] == 48 assert override["title"]["color"] == "#ff0000" assert override["bgm"]["url"] == "http://x/bgm.mp3" -- 2.54.0