style: auto-format with black + isort + ruff + prettier [skip ci-format-check]
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2s
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 3s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 1m3s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker 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 / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m11s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m6s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m5s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 5m6s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 5m47s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 6m21s
AI Code Review / AI Code Review (pull_request) Successful in 6m44s
CI/CD Pipeline / Validate - Style (pull_request) Failing after 8m41s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 10m42s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 15m45s
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 / CI Gate (pull_request) Failing after 1s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped

This commit is contained in:
CI Bot
2026-10-04 09:43:40 +00:00
parent 661883c427
commit 0da7faaa6f
2 changed files with 33 additions and 19 deletions
+1 -1
View File
@@ -453,7 +453,7 @@ class TestViralVideoPipeline:
def test_review_pass_v16(self, mock_review, mock_job):
"""v1.6 _step_review 接收 copy_result dict。"""
from apps.worker.worker_app.tasks.viral_video import _step_review
from packages.application.viral_video.reviewer import ReviewResult, ReviewIssue
from packages.application.viral_video.reviewer import ReviewIssue, ReviewResult
mock_review.return_value = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[])
cr = {"voiceover_script": "大家好", "shots": []}
+32 -18
View File
@@ -87,6 +87,7 @@ STORYBOARD_XML = """
@pytest.fixture(autouse=True)
def invalidate_loader_cache():
from packages.application.viral_video import prompt_loader as pl
pl.invalidate()
yield
pl.invalidate()
@@ -120,8 +121,7 @@ class TestIntentParsingWiring:
def test_uses_loader_and_parses_xml(self, job):
from apps.worker.worker_app.tasks import viral_video as vv
img_result = {"products": [{"name": "lipstick", "brand": "品牌X",
"key_features": ["显白", "持久"]}]}
img_result = {"products": [{"name": "lipstick", "brand": "品牌X", "key_features": ["显白", "持久"]}]}
with patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML) as mock_llm:
result = vv._step_intent_parsing(job, img_result)
@@ -154,8 +154,7 @@ class TestScriptGenerationWiring:
result = vv._step_script_generation(job, intent, {})
# fusion_level 对应的指令文本被注入到 system prompt 中
assert FUSION_INSTRUCTIONS[level] in captured_system["final"], \
f"fusion_level {level} 指令未注入 system_prompt"
assert FUSION_INSTRUCTIONS[level] in captured_system["final"], f"fusion_level {level} 指令未注入 system_prompt"
# 输出保持 Seedance 兼容结构
assert "overview" in result
assert "shots" in result
@@ -201,7 +200,7 @@ class TestReviewWiring:
"""审核不通过时触发自动重写,并更新 job.copy_result"""
from apps.worker.worker_app.tasks import viral_video as vv
from packages.application.viral_video.reviewer import Reviewer, ReviewResult
from packages.application.viral_video.schemas import FusionResult, ScriptSegment, ReviewIssue
from packages.application.viral_video.schemas import FusionResult, ReviewIssue, ScriptSegment
copy_result = {
"title": "原标题",
@@ -212,19 +211,25 @@ class TestReviewWiring:
job.intent_result = {"key_messages": ["显白"], "intent": "推广"}
fail_result = ReviewResult(
passed=False, score=50,
passed=False,
score=50,
issues=[ReviewIssue(dimension="违规词", severity="high", location="开头", text="绝了")],
rewrite_suggestions=["去掉夸大词"],
)
rewritten = FusionResult(
title="新标题", hook="修改后钩子",
script_segments=[ScriptSegment(text="修改后口播正文")], cta="行动号召",
word_count=10, estimated_duration=10,
title="新标题",
hook="修改后钩子",
script_segments=[ScriptSegment(text="修改后口播正文")],
cta="行动号召",
word_count=10,
estimated_duration=10,
)
pass_after = ReviewResult(passed=True, score=88, issues=[], rewrite_suggestions=[])
with patch.object(Reviewer, "review", side_effect=[fail_result, pass_after]), \
patch.object(Reviewer, "rewrite", return_value=rewritten):
with (
patch.object(Reviewer, "review", side_effect=[fail_result, pass_after]),
patch.object(Reviewer, "rewrite", return_value=rewritten),
):
out = vv._step_review(job, copy_result)
assert out["passed"] is True
@@ -247,9 +252,11 @@ class TestEndToEndLoaderUsed:
called_types.append(prompt_type)
return real_get(prompt_type, **kwargs)
with patch.object(pl, "get_template", side_effect=spy_get), \
patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML), \
patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML):
with (
patch.object(pl, "get_template", side_effect=spy_get),
patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML),
patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML),
):
# 1) image
img_res = vv._analyze_single_image(0, "https://img/1.jpg", "vlm", 15)
# 2) intent
@@ -261,26 +268,33 @@ class TestEndToEndLoaderUsed:
# script 和 review 单独验证(需要不同的 LLM 返回)
called_types_2 = []
def spy_get_2(prompt_type, **kwargs):
called_types_2.append(prompt_type)
return real_get(prompt_type, **kwargs)
with patch.object(pl, "get_template", side_effect=spy_get_2), \
patch("packages.shared.ai_service.call_llm", return_value=STORYBOARD_XML):
with (
patch.object(pl, "get_template", side_effect=spy_get_2),
patch("packages.shared.ai_service.call_llm", return_value=STORYBOARD_XML),
):
copy_res = vv._step_script_generation(job, intent_res, {"products": [img_res]})
assert "storyboard" in called_types_2
called_types_3 = []
def spy_get_3(prompt_type, **kwargs):
called_types_3.append(prompt_type)
return real_get(prompt_type, **kwargs)
from packages.application.viral_video.reviewer import Reviewer, ReviewResult
pass_result = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[])
job.intent_result = intent_res
job.copy_result = copy_res
with patch.object(pl, "get_template", side_effect=spy_get_3), \
patch.object(Reviewer, "review", return_value=pass_result) as mock_review:
with (
patch.object(pl, "get_template", side_effect=spy_get_3),
patch.object(Reviewer, "review", return_value=pass_result) as mock_review,
):
review_res = vv._step_review(job, copy_res)
# review 步骤内部直接调用 Reviewer.review,该方法被 mock,因此 get_template 不会被调用;
# 此处验证 Reviewer.review 被调用即可说明 review 步骤走通了。