fix: 删除孤立测试文件 test_cover_service.py(cover_service.py 已在 PR #1347 squash merge 中删除)
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m27s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 3m2s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 3m58s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 4m21s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 4m55s
CI/CD Pipeline / Validate - Code Quality (push) Successful in 11m20s
CI/CD Pipeline / Build Staging API Image (push) Successful in 14m9s
CI/CD Pipeline / Unit Tests (push) Successful in 14m17s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 59s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 39s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m13s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 2m52s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Successful in 7m31s
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m27s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 3m2s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 3m58s
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 4m21s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 4m55s
CI/CD Pipeline / Validate - Code Quality (push) Successful in 11m20s
CI/CD Pipeline / Build Staging API Image (push) Successful in 14m9s
CI/CD Pipeline / Unit Tests (push) Successful in 14m17s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 59s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 39s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m13s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 2m52s
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Successful in 7m31s
CI/CD Pipeline / CI Gate (push) Has been skipped
This commit is contained in:
@@ -1,400 +0,0 @@
|
||||
"""
|
||||
封面管理服务单元测试
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from apps.api.app.services.cover_service import (
|
||||
COVER_STORAGE_PREFIX,
|
||||
DEFAULT_COVER_HEIGHT,
|
||||
DEFAULT_COVER_QUALITY,
|
||||
DEFAULT_COVER_WIDTH,
|
||||
CoverService,
|
||||
)
|
||||
|
||||
|
||||
class TestGetCoverConfig:
|
||||
"""get_cover_config 静态方法测试"""
|
||||
|
||||
def test_get_cover_config_default(self):
|
||||
"""测试默认封面配置"""
|
||||
config = {}
|
||||
result = CoverService.get_cover_config(config)
|
||||
assert result["type"] == "ai_frame"
|
||||
assert result["image_url"] == ""
|
||||
assert result["frame_time"] is None
|
||||
|
||||
def test_get_cover_config_with_custom_values(self):
|
||||
"""测试自定义封面配置"""
|
||||
config = {
|
||||
"cover": {
|
||||
"type": "manual",
|
||||
"image_url": "https://example.com/cover.jpg",
|
||||
"frame_time": 5.5,
|
||||
}
|
||||
}
|
||||
result = CoverService.get_cover_config(config)
|
||||
assert result["type"] == "manual"
|
||||
assert result["image_url"] == "https://example.com/cover.jpg"
|
||||
assert result["frame_time"] == 5.5
|
||||
|
||||
def test_get_cover_config_cover_not_dict(self):
|
||||
"""测试 cover 不是 dict 时返回默认值"""
|
||||
config = {"cover": "not-a-dict"}
|
||||
result = CoverService.get_cover_config(config)
|
||||
assert result["type"] == "ai_frame"
|
||||
assert result["image_url"] == ""
|
||||
assert result["frame_time"] is None
|
||||
|
||||
def test_get_cover_config_partial_fields(self):
|
||||
"""测试部分字段存在时,其余字段用默认值"""
|
||||
config = {"cover": {"type": "custom"}}
|
||||
result = CoverService.get_cover_config(config)
|
||||
assert result["type"] == "custom"
|
||||
assert result["image_url"] == ""
|
||||
assert result["frame_time"] is None
|
||||
|
||||
def test_get_cover_config_empty_cover_dict(self):
|
||||
"""测试空的 cover dict"""
|
||||
config = {"cover": {}}
|
||||
result = CoverService.get_cover_config(config)
|
||||
assert result["type"] == "ai_frame"
|
||||
assert result["image_url"] == ""
|
||||
|
||||
|
||||
class TestExtractCoverFromClip:
|
||||
"""extract_cover_from_clip 测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_storage(self):
|
||||
storage = Mock()
|
||||
storage.download_file = Mock()
|
||||
storage.upload_file = Mock()
|
||||
storage.get_url = Mock(return_value="https://oss.example.com/covers/plan1/cover_1000.jpg")
|
||||
return storage
|
||||
|
||||
@pytest.fixture
|
||||
def mock_asset_repo(self):
|
||||
repo = Mock()
|
||||
repo.get = Mock(return_value=None)
|
||||
return repo
|
||||
|
||||
@pytest.fixture
|
||||
def video_asset(self):
|
||||
asset = Mock()
|
||||
asset.storage_key = "videos/test-video.mp4"
|
||||
asset.mime_type = "video/mp4"
|
||||
return asset
|
||||
|
||||
@pytest.fixture
|
||||
def service(self, mock_storage, mock_asset_repo):
|
||||
return CoverService(storage_service=mock_storage, asset_repository=mock_asset_repo)
|
||||
|
||||
def test_extract_cover_asset_not_found(self, service, mock_asset_repo):
|
||||
"""测试素材不存在时报错"""
|
||||
mock_asset_repo.get.return_value = None
|
||||
|
||||
with pytest.raises(ValueError, match="素材不存在"):
|
||||
service.extract_cover_from_clip(plan_id="plan-1", asset_id="nonexistent")
|
||||
|
||||
def test_extract_cover_asset_no_storage_key(self, service, mock_asset_repo):
|
||||
"""测试素材没有文件时报错"""
|
||||
asset = Mock()
|
||||
asset.storage_key = ""
|
||||
asset.mime_type = "video/mp4"
|
||||
mock_asset_repo.get.return_value = asset
|
||||
|
||||
with pytest.raises(ValueError, match="素材没有文件"):
|
||||
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-no-file")
|
||||
|
||||
def test_extract_cover_asset_not_video(self, service, mock_asset_repo):
|
||||
"""测试非视频素材报错"""
|
||||
asset = Mock()
|
||||
asset.storage_key = "images/photo.jpg"
|
||||
asset.mime_type = "image/jpeg"
|
||||
mock_asset_repo.get.return_value = asset
|
||||
|
||||
with pytest.raises(ValueError, match="素材不是视频类型"):
|
||||
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-img")
|
||||
|
||||
def test_extract_cover_download_failure(self, service, mock_asset_repo, mock_storage, video_asset):
|
||||
"""测试下载素材失败"""
|
||||
mock_asset_repo.get.return_value = video_asset
|
||||
mock_storage.download_file.side_effect = Exception("网络错误")
|
||||
|
||||
with pytest.raises(RuntimeError, match="下载素材失败"):
|
||||
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")
|
||||
|
||||
def test_extract_cover_upload_failure(self, service, mock_asset_repo, mock_storage, video_asset):
|
||||
"""测试上传封面失败"""
|
||||
mock_asset_repo.get.return_value = video_asset
|
||||
|
||||
def fake_download(storage_key, local_path):
|
||||
# 创建一个假的视频文件
|
||||
Path(local_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(local_path, "wb") as f:
|
||||
f.write(b"fake video data")
|
||||
|
||||
mock_storage.download_file.side_effect = fake_download
|
||||
mock_storage.upload_file.side_effect = Exception("上传失败")
|
||||
|
||||
# mock _extract_frame 避免真的调 ffmpeg
|
||||
with patch.object(CoverService, "_extract_frame") as mock_extract:
|
||||
|
||||
def fake_extract(video_path, output_path, **kwargs):
|
||||
# 创建假的封面文件
|
||||
with open(output_path, "wb") as f:
|
||||
f.write(b"\xff\xd8\xff\xe0fake jpeg data")
|
||||
|
||||
mock_extract.side_effect = fake_extract
|
||||
|
||||
with pytest.raises(RuntimeError, match="上传封面失败"):
|
||||
service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")
|
||||
|
||||
def test_extract_cover_get_url_falls_back_to_key(self, service, mock_asset_repo, mock_storage, video_asset):
|
||||
"""测试获取 URL 失败时降级为 storage_key"""
|
||||
mock_asset_repo.get.return_value = video_asset
|
||||
|
||||
def fake_download(storage_key, local_path):
|
||||
Path(local_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(local_path, "wb") as f:
|
||||
f.write(b"fake video data")
|
||||
|
||||
mock_storage.download_file.side_effect = fake_download
|
||||
mock_storage.get_url.side_effect = Exception("URL服务不可用")
|
||||
|
||||
with patch.object(CoverService, "_extract_frame") as mock_extract:
|
||||
|
||||
def fake_extract(video_path, output_path, **kwargs):
|
||||
with open(output_path, "wb") as f:
|
||||
f.write(b"\xff\xd8\xff\xe0fake jpeg")
|
||||
|
||||
mock_extract.side_effect = fake_extract
|
||||
|
||||
result = service.extract_cover_from_clip(plan_id="plan-abc", asset_id="asset-xyz", frame_time=2.5)
|
||||
|
||||
assert result["type"] == "manual"
|
||||
assert result["frame_time"] == 2.5
|
||||
# URL 失败时返回 storage_key
|
||||
assert COVER_STORAGE_PREFIX in result["image_url"]
|
||||
assert "plan-abc" in result["image_url"]
|
||||
|
||||
def test_extract_cover_success(self, service, mock_asset_repo, mock_storage, video_asset):
|
||||
"""测试抽帧成功完整流程"""
|
||||
mock_asset_repo.get.return_value = video_asset
|
||||
|
||||
def fake_download(storage_key, local_path):
|
||||
Path(local_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(local_path, "wb") as f:
|
||||
f.write(b"fake video data for testing")
|
||||
|
||||
mock_storage.download_file.side_effect = fake_download
|
||||
|
||||
with patch.object(CoverService, "_extract_frame") as mock_extract:
|
||||
|
||||
def fake_extract(video_path, output_path, **kwargs):
|
||||
with open(output_path, "wb") as f:
|
||||
f.write(b"\xff\xd8\xff\xe0fake jpeg image data")
|
||||
|
||||
mock_extract.side_effect = fake_extract
|
||||
|
||||
result = service.extract_cover_from_clip(
|
||||
plan_id="plan-123",
|
||||
asset_id="asset-456",
|
||||
frame_time=3.0,
|
||||
width=720,
|
||||
height=1280,
|
||||
quality=3,
|
||||
)
|
||||
|
||||
assert result["type"] == "manual"
|
||||
assert result["image_url"] == "https://oss.example.com/covers/plan1/cover_1000.jpg"
|
||||
assert result["frame_time"] == 3.0
|
||||
|
||||
# 验证上传被调用
|
||||
mock_storage.upload_file.assert_called_once()
|
||||
upload_args = mock_storage.upload_file.call_args[1]
|
||||
assert upload_args["content_type"] == "image/jpeg"
|
||||
assert "plan-123" in upload_args["storage_key"]
|
||||
assert "3000" in upload_args["storage_key"] # frame_time * 1000
|
||||
|
||||
# 验证 _extract_frame 被调用且参数正确
|
||||
mock_extract.assert_called_once()
|
||||
extract_kwargs = mock_extract.call_args[1]
|
||||
assert extract_kwargs["time_sec"] == 3.0
|
||||
assert extract_kwargs["width"] == 720
|
||||
assert extract_kwargs["height"] == 1280
|
||||
assert extract_kwargs["quality"] == 3
|
||||
|
||||
|
||||
class TestGenerateSmartCover:
|
||||
"""generate_smart_cover 测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
return CoverService(storage_service=Mock(), asset_repository=Mock())
|
||||
|
||||
def test_generate_smart_cover_calls_extract_with_default_time(self, service):
|
||||
"""测试智能封面调用 extract_cover_from_clip 并设置 type 为 ai_frame"""
|
||||
fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}
|
||||
|
||||
with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
|
||||
result = service.generate_smart_cover(plan_id="plan-1", asset_id="asset-1")
|
||||
|
||||
mock_extract.assert_called_once()
|
||||
call_kwargs = mock_extract.call_args[1]
|
||||
assert call_kwargs["plan_id"] == "plan-1"
|
||||
assert call_kwargs["asset_id"] == "asset-1"
|
||||
assert call_kwargs["frame_time"] == 3.0 # 默认第3秒
|
||||
|
||||
assert result["type"] == "ai_frame"
|
||||
assert result["image_url"] == "test.jpg"
|
||||
|
||||
def test_generate_smart_cover_passes_dimensions(self, service):
|
||||
"""测试智能封面传递尺寸和质量参数"""
|
||||
fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}
|
||||
|
||||
with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
|
||||
service.generate_smart_cover(
|
||||
plan_id="plan-1",
|
||||
asset_id="asset-1",
|
||||
width=1080,
|
||||
height=1920,
|
||||
quality=5,
|
||||
)
|
||||
|
||||
call_kwargs = mock_extract.call_args[1]
|
||||
assert call_kwargs["width"] == 1080
|
||||
assert call_kwargs["height"] == 1920
|
||||
assert call_kwargs["quality"] == 5
|
||||
|
||||
|
||||
class TestExtractFrame:
|
||||
"""_extract_frame 静态方法测试(mock subprocess)"""
|
||||
|
||||
@pytest.fixture
|
||||
def video_path(self, tmp_path):
|
||||
path = tmp_path / "test_video.mp4"
|
||||
path.write_bytes(b"fake video")
|
||||
return path
|
||||
|
||||
@pytest.fixture
|
||||
def output_path(self, tmp_path):
|
||||
return tmp_path / "cover.jpg"
|
||||
|
||||
def test_extract_frame_success(self, video_path, output_path):
|
||||
"""测试 FFmpeg 抽帧成功"""
|
||||
fake_result = Mock()
|
||||
fake_result.returncode = 0
|
||||
|
||||
with patch("subprocess.run", return_value=fake_result) as mock_run:
|
||||
CoverService._extract_frame(
|
||||
video_path=video_path,
|
||||
output_path=output_path,
|
||||
time_sec=2.5,
|
||||
width=1080,
|
||||
height=1920,
|
||||
quality=5,
|
||||
)
|
||||
|
||||
assert mock_run.call_count == 1
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert cmd[0] == "ffmpeg"
|
||||
assert "-ss" in cmd
|
||||
assert "2.500" in cmd
|
||||
assert "-vframes" in cmd
|
||||
# 验证 scale+crop 滤镜存在
|
||||
vf_index = cmd.index("-vf") + 1
|
||||
assert "scale=" in cmd[vf_index]
|
||||
assert "crop=" in cmd[vf_index]
|
||||
|
||||
def test_extract_frame_fallback_to_simple_command(self, video_path, output_path):
|
||||
"""测试主命令失败时回退到简化命令"""
|
||||
fail_result = Mock()
|
||||
fail_result.returncode = 1
|
||||
fail_result.stderr = "Filter graph error"
|
||||
|
||||
success_result = Mock()
|
||||
success_result.returncode = 0
|
||||
|
||||
call_count = 0
|
||||
|
||||
def fake_run(*args, **kwargs):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count == 1:
|
||||
return fail_result
|
||||
return success_result
|
||||
|
||||
with patch("subprocess.run", side_effect=fake_run) as mock_run:
|
||||
CoverService._extract_frame(
|
||||
video_path=video_path,
|
||||
output_path=output_path,
|
||||
time_sec=1.0,
|
||||
width=1080,
|
||||
height=1920,
|
||||
quality=5,
|
||||
)
|
||||
|
||||
assert mock_run.call_count == 2
|
||||
# 第二次是简化命令(没有 -vf 参数)
|
||||
second_cmd = mock_run.call_args_list[1][0][0]
|
||||
assert "-vf" not in second_cmd
|
||||
|
||||
def test_extract_frame_both_commands_fail(self, video_path, output_path):
|
||||
"""测试两个命令都失败时报错"""
|
||||
fail_result = Mock()
|
||||
fail_result.returncode = 1
|
||||
fail_result.stderr = "Invalid data found when processing input"
|
||||
|
||||
with patch("subprocess.run", return_value=fail_result):
|
||||
with pytest.raises(RuntimeError, match="FFmpeg 抽帧失败"):
|
||||
CoverService._extract_frame(
|
||||
video_path=video_path,
|
||||
output_path=output_path,
|
||||
time_sec=1.0,
|
||||
width=1080,
|
||||
height=1920,
|
||||
quality=5,
|
||||
)
|
||||
|
||||
def test_extract_frame_timeout(self, video_path, output_path):
|
||||
"""测试 FFmpeg 抽帧超时"""
|
||||
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="ffmpeg", timeout=60)):
|
||||
with pytest.raises(RuntimeError, match="FFmpeg 抽帧超时"):
|
||||
CoverService._extract_frame(
|
||||
video_path=video_path,
|
||||
output_path=output_path,
|
||||
time_sec=1.0,
|
||||
width=1080,
|
||||
height=1920,
|
||||
quality=5,
|
||||
)
|
||||
|
||||
def test_extract_frame_ffmpeg_not_found(self, video_path, output_path):
|
||||
"""测试 FFmpeg 不可用"""
|
||||
with patch("subprocess.run", side_effect=FileNotFoundError("ffmpeg not found")):
|
||||
with pytest.raises(RuntimeError, match="FFmpeg 不可用"):
|
||||
CoverService._extract_frame(
|
||||
video_path=video_path,
|
||||
output_path=output_path,
|
||||
time_sec=1.0,
|
||||
width=1080,
|
||||
height=1920,
|
||||
quality=5,
|
||||
)
|
||||
|
||||
|
||||
class TestDefaults:
|
||||
"""默认常量测试"""
|
||||
|
||||
def test_default_dimensions(self):
|
||||
"""测试默认尺寸常量"""
|
||||
assert DEFAULT_COVER_WIDTH == 1080
|
||||
assert DEFAULT_COVER_HEIGHT == 1920
|
||||
assert DEFAULT_COVER_QUALITY == 5
|
||||
assert COVER_STORAGE_PREFIX == "covers"
|
||||
Reference in New Issue
Block a user