From 9c6af4dd45cff39b5ed7fbc6dbff3f3ee3fcef66 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Wed, 12 Aug 2026 22:51:20 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=88=A0=E9=99=A4=E5=AD=A4=E7=AB=8B?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E6=96=87=E4=BB=B6=20test=5Fcover=5Fservice.p?= =?UTF-8?q?y=EF=BC=88cover=5Fservice.py=20=E5=B7=B2=E5=9C=A8=20PR=20#1347?= =?UTF-8?q?=20squash=20merge=20=E4=B8=AD=E5=88=A0=E9=99=A4=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_cover_service.py | 400 ------------------------------- 1 file changed, 400 deletions(-) delete mode 100755 tests/unit/test_cover_service.py diff --git a/tests/unit/test_cover_service.py b/tests/unit/test_cover_service.py deleted file mode 100755 index 1705f1671..000000000 --- a/tests/unit/test_cover_service.py +++ /dev/null @@ -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"