c08a064e55
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
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 / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
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 / 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 / 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
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m59s
AI Code Review / AI Code Review (pull_request) Successful in 4m8s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 4m12s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 5m25s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 5m46s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 5m54s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 6m27s
CI/CD Pipeline / Validate - Security (pull_request) Successful in 7m30s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Failing after 8m24s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 8m48s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 18m30s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
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
CI/CD Pipeline / CI Gate (pull_request) Failing after 2s
244 lines
9.4 KiB
Python
244 lines
9.4 KiB
Python
"""_get_template_segments 回退路径测试.
|
||
|
||
验证三级回退链:
|
||
1. 新模板系统(tpl_svc.list_clip_configs)正常 → 直接返回
|
||
2. 新模板系统主表不存在(ValueError)→ 直接查 template_clip_configs 表兜底
|
||
3. 直接查表也失败 → 回退旧模板系统(template_segments)
|
||
4. 全部失败 → 返回空列表
|
||
|
||
覆盖 P0 修复:自建模板在 edit_templates 主表不存在但在 template_clip_configs 有记录时,
|
||
from-assets 流程不再 400。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, PropertyMock
|
||
|
||
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
|
||
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
|
||
|
||
from app.api.routes.templates_editor.clips import _get_template_segments
|
||
|
||
TEST_TEMPLATE_ID = "tmpl-orphan-001"
|
||
DEFAULT_DUR = 5.0 # _DEFAULT_EDITOR_CLIP_DURATION
|
||
|
||
|
||
def _make_clip_config(order: int, min_dur: float = 3.0, max_dur: float = 8.0):
|
||
"""构造 mock TemplateClipConfig 领域实体."""
|
||
cc = MagicMock()
|
||
cc.order = order
|
||
cc.min_duration = min_dur
|
||
cc.max_duration = max_dur
|
||
return cc
|
||
|
||
|
||
def _make_old_segment(segment_order: int, dur_min: float = 4.0, dur_max: float = 7.0):
|
||
"""构造 mock 旧 TemplateSegment."""
|
||
s = MagicMock()
|
||
s.segment_order = segment_order
|
||
s.duration_min = dur_min
|
||
s.duration_max = dur_max
|
||
return s
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGetTemplateSegmentsFallback:
|
||
"""_get_template_segments 三级回退链."""
|
||
|
||
def test_new_system_works(self):
|
||
"""路径1:新模板系统正常返回 → 直接使用."""
|
||
configs = [_make_clip_config(0, 2.0, 6.0), _make_clip_config(1, 3.0, 9.0)]
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.return_value = configs
|
||
db = MagicMock()
|
||
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
|
||
assert len(result) == 2
|
||
assert result[0] == (0, 2.0, 6.0)
|
||
assert result[1] == (1, 3.0, 9.0)
|
||
tpl_svc.list_clip_configs.assert_called_once_with(TEST_TEMPLATE_ID)
|
||
|
||
def test_main_table_missing_direct_query_succeeds(self):
|
||
"""路径2(P0修复):主表不存在 ValueError → 直接查表成功.
|
||
|
||
模拟自建模板在 edit_templates 主表已删除/不存在,
|
||
但 template_clip_configs 表有记录。
|
||
"""
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.side_effect = ValueError(f"模板不存在: {TEST_TEMPLATE_ID}")
|
||
db = MagicMock()
|
||
|
||
# Mock SQLAlchemyTemplateClipConfigRepository
|
||
direct_configs = [
|
||
_make_clip_config(0, 2.0, 5.0),
|
||
_make_clip_config(1, 3.0, 7.0),
|
||
_make_clip_config(2, 4.0, 8.0),
|
||
]
|
||
with (
|
||
__import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
|
||
) as mock_repo_cls,
|
||
):
|
||
mock_repo = MagicMock()
|
||
mock_repo.list_by_template.return_value = direct_configs
|
||
mock_repo_cls.return_value = mock_repo
|
||
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
|
||
assert len(result) == 3
|
||
assert result[0] == (0, 2.0, 5.0)
|
||
assert result[1] == (1, 3.0, 7.0)
|
||
assert result[2] == (2, 4.0, 8.0)
|
||
mock_repo.list_by_template.assert_called_once_with(TEST_TEMPLATE_ID)
|
||
|
||
def test_main_table_missing_direct_query_empty_falls_to_old(self):
|
||
"""路径2→3:主表不存在 + 直接查表为空 → 回退旧系统."""
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
|
||
db = MagicMock()
|
||
|
||
old_segments = [_make_old_segment(0, 3.0, 6.0)]
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
|
||
) as mock_repo_cls:
|
||
mock_repo = MagicMock()
|
||
mock_repo.list_by_template.return_value = [] # 新表也没记录
|
||
mock_repo_cls.return_value = mock_repo
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository"
|
||
) as mock_old_cls:
|
||
mock_old = MagicMock()
|
||
mock_old.list_segments.return_value = old_segments
|
||
mock_old_cls.return_value = mock_old
|
||
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
|
||
assert len(result) == 1
|
||
assert result[0] == (0, 3.0, 6.0)
|
||
|
||
def test_all_fail_returns_empty(self):
|
||
"""路径4:三级全部失败 → 返回空列表."""
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
|
||
db = MagicMock()
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
|
||
) as mock_repo_cls:
|
||
mock_repo = MagicMock()
|
||
mock_repo.list_by_template.side_effect = Exception("DB error")
|
||
mock_repo_cls.return_value = mock_repo
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateRepository"
|
||
) as mock_old_cls:
|
||
mock_old = MagicMock()
|
||
mock_old.list_segments.return_value = [] # 旧表也空
|
||
mock_old_cls.return_value = mock_old
|
||
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
|
||
assert result == []
|
||
|
||
def test_direct_query_sorts_by_order(self):
|
||
"""直接查表返回的结果按 order 排序."""
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
|
||
db = MagicMock()
|
||
|
||
# 故意乱序
|
||
configs = [
|
||
_make_clip_config(2, 5.0, 10.0),
|
||
_make_clip_config(0, 2.0, 4.0),
|
||
_make_clip_config(1, 3.0, 6.0),
|
||
]
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
|
||
) as mock_repo_cls:
|
||
mock_repo = MagicMock()
|
||
mock_repo.list_by_template.return_value = configs
|
||
mock_repo_cls.return_value = mock_repo
|
||
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
|
||
assert [r[0] for r in result] == [0, 1, 2]
|
||
assert result[0] == (0, 2.0, 4.0)
|
||
assert result[1] == (1, 3.0, 6.0)
|
||
assert result[2] == (2, 5.0, 10.0)
|
||
|
||
def test_direct_query_handles_none_durations(self):
|
||
"""直接查表时 min/max_duration 为 None → 使用默认值."""
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.side_effect = ValueError("模板不存在")
|
||
db = MagicMock()
|
||
|
||
cc = MagicMock()
|
||
cc.order = 0
|
||
cc.min_duration = None
|
||
cc.max_duration = None
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
|
||
) as mock_repo_cls:
|
||
mock_repo = MagicMock()
|
||
mock_repo.list_by_template.return_value = [cc]
|
||
mock_repo_cls.return_value = mock_repo
|
||
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
|
||
assert len(result) == 1
|
||
# None → default (5.0), max(None or None) → default (5.0)
|
||
assert result[0] == (0, DEFAULT_DUR, DEFAULT_DUR)
|
||
|
||
def test_new_system_returns_empty_tries_direct(self):
|
||
"""新模板系统返回空列表(非异常)→ 继续尝试直接查表."""
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.return_value = [] # 空列表,非异常
|
||
db = MagicMock()
|
||
|
||
direct_configs = [_make_clip_config(0, 3.0, 6.0)]
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
|
||
) as mock_repo_cls:
|
||
mock_repo = MagicMock()
|
||
mock_repo.list_by_template.return_value = direct_configs
|
||
mock_repo_cls.return_value = mock_repo
|
||
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
|
||
# 新系统返回空 → 不走 except → 但也没 return → 继续往下走
|
||
# 直接查表有数据 → 返回
|
||
assert len(result) == 1
|
||
assert result[0] == (0, 3.0, 6.0)
|
||
|
||
def test_existing_template_unaffected(self):
|
||
"""正常模板(主表存在)行为不变."""
|
||
configs = [_make_clip_config(0, 2.0, 5.0)]
|
||
tpl_svc = MagicMock()
|
||
tpl_svc.list_clip_configs.return_value = configs
|
||
db = MagicMock()
|
||
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch(
|
||
"app.api.routes.templates_editor.clips.SQLAlchemyTemplateClipConfigRepository"
|
||
) as mock_repo_cls:
|
||
result = _get_template_segments(TEST_TEMPLATE_ID, tpl_svc, db)
|
||
# 直接查表不应被调用(新系统已返回)
|
||
mock_repo_cls.assert_not_called()
|
||
|
||
assert len(result) == 1
|
||
assert result[0] == (0, 2.0, 5.0)
|