diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index b123db26a..f750eab65 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -14,70 +14,18 @@ from packages.adapters.sqlalchemy_impl import ( SQLAlchemyIngestJobRepository, ) from packages.domain import Asset, AssetStatus, IngestJobStatus +from packages.domain.media_validation import ( + MIN_AUDIO_FILE_SIZE, + MIN_IMAGE_FILE_SIZE, + MIN_VIDEO_FILE_SIZE, + SUPPORTED_VIDEO_CODECS, + is_valid_media as _is_valid_media, + safe_parse_fps as _safe_parse_fps, +) logger = get_task_logger(__name__) -# 最小有效文件大小（字节）：小于此值的直接判为无效，避免文本/空文件伪装成媒体 -MIN_VIDEO_FILE_SIZE = 1024 # 1KB -MIN_AUDIO_FILE_SIZE = 100 # 100B -MIN_IMAGE_FILE_SIZE = 100 # 100B - -# 支持的视频编码格式（白名单，尽可能放宽） -# 渲染引擎会在 concat 前统一转码为 h264，因此只要 ffprobe 能识别的视频编码都允许 ingested -SUPPORTED_VIDEO_CODECS = { - "h264", - "avc1", - "avc", # H.264 / AVC - "hevc", - "h265", - "hev1", - "hvc1", # H.265 / HEVC - "vp9", - "vp09", # VP9 - "av1", - "av01", # AV1 - "vp8", - "vp08", # VP8 - "mpeg4", - "mp4v", # MPEG-4 - "mpeg2video", - "mpg2", # MPEG-2 - "wmv2", - "wmv1", - "vc1", # WMV / VC-1 - "flv1", - "flv", - "vp6f", # Flash / FLV - "theora", - "ogg", # Theora - "prores", - "prores_ks", - "apcn", - "apch", - "apco", - "apcs", - "ap4h", - "ap4x", # Apple ProRes - "dnxhd", - "dnxhr", # DNxHD / DNxHR -} - - -def _safe_parse_fps(fps_str: str) -> float: - """Safely parse fps from a fraction string like \"30/1\" or \"30000/1001\".""" - try: - if "/" in fps_str: - num, den = fps_str.split("/", 1) - den_val = float(den) - if den_val == 0: - return 0.0 - return float(num) / den_val - return float(fps_str) - except (ValueError, ZeroDivisionError): - return 0.0 - - def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]: """ 提取媒体文件的元数据。 @@ -207,38 +155,6 @@ def extract_media_metadata(file_url: str, media_type: str) -> tuple[dict, bool]: return metadata, success -def _is_valid_media(metadata: dict, media_type: str) -> bool: - """根据元数据判断文件是否为有效媒体文件。 - - Args: - metadata: extract_media_metadata 返回的元数据 - media_type: 媒体类型 - - Returns: - True 表示文件有效 - """ - size = int(metadata.get("size_bytes", 0)) - - if media_type == "video": - duration = float(metadata.get("duration", 0)) - if size < MIN_VIDEO_FILE_SIZE or duration <= 0: - return False - # 编码格式校验：只排除明确非视频的编码格式，只要 ffprobe 能识别的视频编码都允许 - # 渲染引擎会在 concat 前统一转码为 h264 yuv420p，ingest 层不再做严格的编码拦截 - codec = str(metadata.get("codec", "")).lower() - if codec and codec not in SUPPORTED_VIDEO_CODECS: - logger.info("检测到非白名单视频编码 %s，仍允许 ingested，渲染层会统一转码", codec) - return True - if media_type == "audio": - duration = float(metadata.get("duration", 0)) - return size >= MIN_AUDIO_FILE_SIZE and duration > 0 - if media_type == "image": - width = int(metadata.get("width", 0)) - height = int(metadata.get("height", 0)) - return size >= MIN_IMAGE_FILE_SIZE and width > 0 and height > 0 - return False - - @celery_app.task(name="worker.ingest_asset") def ingest_asset(job_id: str) -> dict: """ diff --git a/packages/domain/media_validation.py b/packages/domain/media_validation.py new file mode 100755 index 000000000..32eb36e8e --- /dev/null +++ b/packages/domain/media_validation.py @@ -0,0 +1,110 @@ +"""媒体文件有效性校验与元数据解析工具。 + +从 worker ingest 任务中抽取的纯逻辑模块，包含： +- FPS 解析：从分数格式字符串（如 30000/1001）安全解析帧率 +- 媒体有效性校验：根据元数据判断视频/音频/图片文件是否有效 +- 常量定义：最小文件大小、支持的视频编码白名单 +""" + +from __future__ import annotations + +# 最小有效文件大小（字节）：小于此值的直接判为无效，避免文本/空文件伪装成媒体 +MIN_VIDEO_FILE_SIZE = 1024 # 1KB +MIN_AUDIO_FILE_SIZE = 100 # 100B +MIN_IMAGE_FILE_SIZE = 100 # 100B + +# 支持的视频编码格式（白名单，尽可能放宽） +# 渲染引擎会在 concat 前统一转码为 h264，因此只要 ffprobe 能识别的视频编码都允许 ingested +SUPPORTED_VIDEO_CODECS: frozenset[str] = frozenset( + { + "h264", + "avc1", + "avc", # H.264 / AVC + "hevc", + "h265", + "hev1", + "hvc1", # H.265 / HEVC + "vp9", + "vp09", # VP9 + "av1", + "av01", # AV1 + "vp8", + "vp08", # VP8 + "mpeg4", + "mp4v", # MPEG-4 + "mpeg2video", + "mpg2", # MPEG-2 + "wmv2", + "wmv1", + "vc1", # WMV / VC-1 + "flv1", + "flv", + "vp6f", # Flash / FLV + "theora", + "ogg", # Theora + "prores", + "prores_ks", + "apcn", + "apch", + "apco", + "apcs", + "ap4h", + "ap4x", # Apple ProRes + "dnxhd", + "dnxhr", # DNxHD / DNxHR + } +) + + +def safe_parse_fps(fps_str: str) -> float: + """Safely parse fps from a fraction string like "30/1" or "30000/1001". + + Args: + fps_str: FPS 字符串，支持小数格式（"30.0"）或分数格式（"30000/1001"） + + Returns: + 解析得到的帧率浮点数；解析失败或分母为0时返回 0.0 + """ + try: + if "/" in fps_str: + num, den = fps_str.split("/", 1) + den_val = float(den) + if den_val == 0: + return 0.0 + return float(num) / den_val + return float(fps_str) + except (ValueError, ZeroDivisionError): + return 0.0 + + +def is_valid_media(metadata: dict, media_type: str) -> bool: + """根据元数据判断文件是否为有效媒体文件。 + + Args: + metadata: 媒体元数据字典，可能包含 size_bytes / duration / codec / width / height 等 + media_type: 媒体类型（video / audio / image） + + Returns: + True 表示文件有效 + """ + size = int(metadata.get("size_bytes", 0)) + + if media_type == "video": + duration = float(metadata.get("duration", 0)) + if size < MIN_VIDEO_FILE_SIZE or duration <= 0: + return False + # 编码格式校验：只排除明确非视频的编码格式，只要 ffprobe 能识别的视频编码都允许 + # 渲染引擎会在 concat 前统一转码为 h264 yuv420p，ingest 层不再做严格的编码拦截 + codec = str(metadata.get("codec", "")).lower() + if codec and codec not in SUPPORTED_VIDEO_CODECS: + # 非白名单编码仍允许通过，仅记录日志（调用方负责日志） + pass + return True + if media_type == "audio": + duration = float(metadata.get("duration", 0)) + return size >= MIN_AUDIO_FILE_SIZE and duration > 0 + if media_type == "image": + width = int(metadata.get("width", 0)) + height = int(metadata.get("height", 0)) + return size >= MIN_IMAGE_FILE_SIZE and width > 0 and height > 0 + return False diff --git a/tests/unit/test_application_modules.py b/tests/unit/test_application_modules.py new file mode 100755 index 000000000..09711b97a --- /dev/null +++ b/tests/unit/test_application_modules.py @@ -0,0 +1,499 @@ +"""Application 层零测试模块合集 — 第100波里程碑。 + +覆盖: +- packages/application/generated_videos.py (8个UseCase) +- packages/application/assets.py (ListAssets + CreateAsset) +- packages/application/asset_libraries.py (ListLibraries + CreateLibrary) + +策略: Mock repository，测参数校验 + 委托行为 +""" + +from unittest.mock import MagicMock + +import pytest + +from packages.application.asset_libraries import ( + CreateAssetLibraryCommand, + CreateAssetLibraryUseCase, + ListAssetLibrariesUseCase, +) +from packages.application.assets import ( + CreateAssetCommand, + CreateAssetUseCase, + ListAssetsUseCase, +) +from packages.application.generated_videos import ( + GetGeneratedVideoDownloadUrlUseCase, + GetGeneratedVideoUseCase, + GetVideosByIdsUseCase, + ListGeneratedVideosByTaskUseCase, + ListGeneratedVideosPaginatedUseCase, + ListGeneratedVideosUseCase, + UpdateVideoReviewStatusUseCase, +) +from packages.domain import AssetLibraryKind, AssetStatus, ClassificationStatus, GeneratedVideo + +# ── generated_videos.py ──────────────────────────────────────────────────────── + + +class TestListGeneratedVideosUseCase: + def test_success(self): + mock_repo = MagicMock() + mock_repo.list_by_project.return_value = [MagicMock(spec=GeneratedVideo)] + use_case = ListGeneratedVideosUseCase(mock_repo) + + result = use_case.execute("proj1") + + assert len(result) == 1 + mock_repo.list_by_project.assert_called_once_with("proj1") + + def test_strips_project_id(self): + mock_repo = MagicMock() + use_case = ListGeneratedVideosUseCase(mock_repo) + + use_case.execute(" proj1 ") + + mock_repo.list_by_project.assert_called_once_with("proj1") + + def test_empty_project_id_raises(self): + mock_repo = MagicMock() + use_case = ListGeneratedVideosUseCase(mock_repo) + + with pytest.raises(ValueError, match="project_id 不能为空"): + use_case.execute("") + + def test_whitespace_project_id_raises(self): + mock_repo = MagicMock() + use_case = ListGeneratedVideosUseCase(mock_repo) + + with pytest.raises(ValueError, match="project_id 不能为空"): + use_case.execute(" \t ") + + +class TestListGeneratedVideosPaginatedUseCase: + def test_default_params(self): + mock_repo = MagicMock() + mock_repo.list_paginated.return_value = ([], 0) + use_case = ListGeneratedVideosPaginatedUseCase(mock_repo) + + result, total = use_case.execute() + + assert total == 0 + assert result == [] + mock_repo.list_paginated.assert_called_once_with( + user_id=None, + project_id=None, + status=None, + review_status=None, + page=1, + page_size=20, + ) + + def test_page_below_1_clamps_to_1(self): + mock_repo = MagicMock() + mock_repo.list_paginated.return_value = ([], 0) + use_case = ListGeneratedVideosPaginatedUseCase(mock_repo) + + use_case.execute(page=0) + + mock_repo.list_paginated.assert_called_once() + call_kwargs = mock_repo.list_paginated.call_args.kwargs + assert call_kwargs["page"] == 1 + + def test_negative_page_clamps(self): + mock_repo = MagicMock() + mock_repo.list_paginated.return_value = ([], 0) + use_case = ListGeneratedVideosPaginatedUseCase(mock_repo) + + use_case.execute(page=-5) + + assert mock_repo.list_paginated.call_args.kwargs["page"] == 1 + + def test_page_size_zero_clamps(self): + mock_repo = MagicMock() + mock_repo.list_paginated.return_value = ([], 0) + use_case = ListGeneratedVideosPaginatedUseCase(mock_repo) + + use_case.execute(page_size=0) + + assert mock_repo.list_paginated.call_args.kwargs["page_size"] == 20 + + def test_page_size_over_100_clamps(self): + mock_repo = MagicMock() + mock_repo.list_paginated.return_value = ([], 0) + use_case = ListGeneratedVideosPaginatedUseCase(mock_repo) + + use_case.execute(page_size=200) + + assert mock_repo.list_paginated.call_args.kwargs["page_size"] == 20 + + def test_page_size_50_ok(self): + mock_repo = MagicMock() + mock_repo.list_paginated.return_value = ([], 0) + use_case = ListGeneratedVideosPaginatedUseCase(mock_repo) + + use_case.execute(page_size=50) + + assert mock_repo.list_paginated.call_args.kwargs["page_size"] == 50 + + def test_with_all_filters(self): + mock_repo = MagicMock() + mock_repo.list_paginated.return_value = ([], 0) + use_case = ListGeneratedVideosPaginatedUseCase(mock_repo) + + use_case.execute( + user_id="u1", + project_id="p1", + status="completed", + review_status="approved", + page=2, + page_size=10, + ) + + mock_repo.list_paginated.assert_called_once_with( + user_id="u1", + project_id="p1", + status="completed", + review_status="approved", + page=2, + page_size=10, + ) + + +class TestGetGeneratedVideoUseCase: + def test_found(self): + mock_repo = MagicMock() + expected = MagicMock(spec=GeneratedVideo) + mock_repo.get.return_value = expected + use_case = GetGeneratedVideoUseCase(mock_repo) + + result = use_case.execute("vid1") + + assert result == expected + mock_repo.get.assert_called_once_with("vid1") + + def test_not_found(self): + mock_repo = MagicMock() + mock_repo.get.return_value = None + use_case = GetGeneratedVideoUseCase(mock_repo) + + result = use_case.execute("vid1") + + assert result is None + + +class TestListGeneratedVideosByTaskUseCase: + def test_success(self): + mock_repo = MagicMock() + mock_repo.list_by_generation_task.return_value = [MagicMock()] + use_case = ListGeneratedVideosByTaskUseCase(mock_repo) + + result = use_case.execute("task1") + + assert len(result) == 1 + mock_repo.list_by_generation_task.assert_called_once_with("task1") + + def test_strips_task_id(self): + mock_repo = MagicMock() + use_case = ListGeneratedVideosByTaskUseCase(mock_repo) + + use_case.execute(" task1 ") + + mock_repo.list_by_generation_task.assert_called_once_with("task1") + + def test_empty_task_id_raises(self): + mock_repo = MagicMock() + use_case = ListGeneratedVideosByTaskUseCase(mock_repo) + + with pytest.raises(ValueError, match="generation_task_id 不能为空"): + use_case.execute("") + + +class TestGetGeneratedVideoDownloadUrlUseCase: + def test_found(self): + mock_repo = MagicMock() + mock_item = MagicMock() + mock_item.file_url = "https://cdn/v.mp4" + mock_repo.get.return_value = mock_item + use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo) + + result = use_case.execute("vid1") + + assert result == "https://cdn/v.mp4" + + def test_not_found_returns_none(self): + mock_repo = MagicMock() + mock_repo.get.return_value = None + use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo) + + result = use_case.execute("vid1") + + assert result is None + + +class TestUpdateVideoReviewStatusUseCase: + def test_pending_review(self): + mock_repo = MagicMock() + mock_repo.update_review_status.return_value = MagicMock() + use_case = UpdateVideoReviewStatusUseCase(mock_repo) + + use_case.execute("vid1", "pending_review") + + mock_repo.update_review_status.assert_called_once_with("vid1", "pending_review") + + def test_approved(self): + mock_repo = MagicMock() + use_case = UpdateVideoReviewStatusUseCase(mock_repo) + + use_case.execute("vid1", "approved") + + mock_repo.update_review_status.assert_called_once_with("vid1", "approved") + + def test_rejected(self): + mock_repo = MagicMock() + use_case = UpdateVideoReviewStatusUseCase(mock_repo) + + use_case.execute("vid1", "rejected") + + mock_repo.update_review_status.assert_called_once_with("vid1", "rejected") + + def test_strips_video_id(self): + mock_repo = MagicMock() + use_case = UpdateVideoReviewStatusUseCase(mock_repo) + + use_case.execute(" vid1 ", "approved") + + mock_repo.update_review_status.assert_called_once_with("vid1", "approved") + + def test_empty_video_id_raises(self): + mock_repo = MagicMock() + use_case = UpdateVideoReviewStatusUseCase(mock_repo) + + with pytest.raises(ValueError, match="video_id 不能为空"): + use_case.execute("", "approved") + + def test_invalid_status_raises(self): + mock_repo = MagicMock() + use_case = UpdateVideoReviewStatusUseCase(mock_repo) + + with pytest.raises(ValueError, match="无效的 review_status"): + use_case.execute("vid1", "invalid_status") + + def test_not_found_returns_none(self): + mock_repo = MagicMock() + mock_repo.update_review_status.return_value = None + use_case = UpdateVideoReviewStatusUseCase(mock_repo) + + result = use_case.execute("vid1", "approved") + + assert result is None + + +class TestGetVideosByIdsUseCase: + def test_success(self): + mock_repo = MagicMock() + mock_repo.get_by_ids.return_value = [MagicMock(), MagicMock()] + use_case = GetVideosByIdsUseCase(mock_repo) + + result = use_case.execute(["id1", "id2", "id3"]) + + assert len(result) == 2 + mock_repo.get_by_ids.assert_called_once_with(["id1", "id2", "id3"]) + + def test_empty_list(self): + mock_repo = MagicMock() + mock_repo.get_by_ids.return_value = [] + use_case = GetVideosByIdsUseCase(mock_repo) + + result = use_case.execute([]) + + assert result == [] + mock_repo.get_by_ids.assert_called_once_with([]) + + +# ── assets.py ────────────────────────────────────────────────────────────────── + + +class TestCreateAssetCommand: + def test_minimal(self): + cmd = CreateAssetCommand( + project_id="p1", + library_id="l1", + name="test.mp4", + storage_key="k", + mime_type="video/mp4", + ) + assert cmd.project_id == "p1" + assert cmd.library_id == "l1" + assert cmd.name == "test.mp4" + assert cmd.storage_key == "k" + assert cmd.mime_type == "video/mp4" + assert cmd.file_size == 0 + assert cmd.status == AssetStatus.UPLOADING + assert cmd.classification_status == ClassificationStatus.PENDING + + def test_full(self): + cmd = CreateAssetCommand( + project_id="p1", + library_id="l1", + name="test.mp4", + storage_key="k", + mime_type="video/mp4", + metadata={"k": "v"}, + file_size=1024, + duration=10.0, + width=1920, + height=1080, + fps=30.0, + codec="h264", + status=AssetStatus.READY, + quality_score=0.9, + uploaded_by_user_id="u1", + ) + assert cmd.file_size == 1024 + assert cmd.duration == 10.0 + assert cmd.status == AssetStatus.READY + assert cmd.quality_score == 0.9 + + +class TestListAssetsUseCase: + def test_success(self): + mock_repo = MagicMock() + mock_repo.find_by_library.return_value = [] + use_case = ListAssetsUseCase(mock_repo) + + result = use_case.execute("lib1") + + assert result == [] + mock_repo.find_by_library.assert_called_once_with("lib1") + + def test_strips_library_id(self): + mock_repo = MagicMock() + use_case = ListAssetsUseCase(mock_repo) + + use_case.execute(" lib1 ") + + mock_repo.find_by_library.assert_called_once_with("lib1") + + def test_empty_library_id_raises(self): + mock_repo = MagicMock() + use_case = ListAssetsUseCase(mock_repo) + + with pytest.raises(ValueError, match="library_id 不能为空"): + use_case.execute("") + + +class TestCreateAssetUseCase: + def test_creates_asset_via_repo(self): + mock_repo = MagicMock() + mock_repo.create.return_value = MagicMock() + use_case = CreateAssetUseCase(mock_repo) + + cmd = CreateAssetCommand( + project_id="p1", + library_id="l1", + name="test.mp4", + storage_key="videos/t.mp4", + mime_type="video/mp4", + file_size=1024, + ) + result = use_case.execute(cmd) + + assert result is not None + mock_repo.create.assert_called_once() + created_asset = mock_repo.create.call_args[0][0] + assert created_asset.project_id == "p1" + assert created_asset.name == "test.mp4" + assert created_asset.file_size == 1024 + assert created_asset.status == AssetStatus.UPLOADING + + def test_asset_create_validation_propagates(self): + mock_repo = MagicMock() + use_case = CreateAssetUseCase(mock_repo) + + cmd = CreateAssetCommand( + project_id="p1", + library_id="l1", + name="", + storage_key="k", + mime_type="video/mp4", + ) + + with pytest.raises(ValueError, match="素材名称不能为空"): + use_case.execute(cmd) + + +# ── asset_libraries.py ──────────────────────────────────────────────────────── + + +class TestCreateAssetLibraryCommand: + def test_creation(self): + cmd = CreateAssetLibraryCommand( + project_id="p1", + name="我的库", + kind=AssetLibraryKind.VIDEO, + ) + assert cmd.project_id == "p1" + assert cmd.name == "我的库" + assert cmd.kind == AssetLibraryKind.VIDEO + + +class TestListAssetLibrariesUseCase: + def test_success(self): + mock_repo = MagicMock() + mock_repo.find_by_project.return_value = [] + use_case = ListAssetLibrariesUseCase(mock_repo) + + result = use_case.execute("p1") + + assert result == [] + mock_repo.find_by_project.assert_called_once_with("p1") + + def test_strips_project_id(self): + mock_repo = MagicMock() + use_case = ListAssetLibrariesUseCase(mock_repo) + + use_case.execute(" p1 ") + + mock_repo.find_by_project.assert_called_once_with("p1") + + def test_empty_project_id_raises(self): + mock_repo = MagicMock() + use_case = ListAssetLibrariesUseCase(mock_repo) + + with pytest.raises(ValueError, match="project_id 不能为空"): + use_case.execute("") + + +class TestCreateAssetLibraryUseCase: + def test_creates_library_via_repo(self): + mock_repo = MagicMock() + mock_repo.create.return_value = MagicMock() + use_case = CreateAssetLibraryUseCase(mock_repo) + + cmd = CreateAssetLibraryCommand( + project_id="p1", + name="视频库", + kind=AssetLibraryKind.VIDEO, + ) + result = use_case.execute(cmd) + + assert result is not None + mock_repo.create.assert_called_once() + created = mock_repo.create.call_args[0][0] + assert created.project_id == "p1" + assert created.name == "视频库" + assert created.kind == AssetLibraryKind.VIDEO + + def test_validation_propagates(self): + mock_repo = MagicMock() + use_case = CreateAssetLibraryUseCase(mock_repo) + + cmd = CreateAssetLibraryCommand( + project_id="p1", + name="", + kind=AssetLibraryKind.VIDEO, + ) + + with pytest.raises(ValueError, match="素材库名称不能为空"): + use_case.execute(cmd) diff --git a/tests/unit/test_domain_entities.py b/tests/unit/test_domain_entities.py new file mode 100755 index 000000000..95184c0fe --- /dev/null +++ b/tests/unit/test_domain_entities.py @@ -0,0 +1,488 @@ +"""Domain entities 单元测试。""" + +from datetime import datetime, timezone + +import pytest + +from packages.domain.classification import ( + AssetLibraryKind, + ClassificationStatus, + IngestJobStatus, +) +from packages.domain.entities import ( + Asset, + AssetLibrary, + AssetStatus, + IngestJob, + Project, + User, +) + + +class TestProjectCreate: + def test_create_success(self): + project = Project.create(owner_user_id="user1", name="我的项目") + assert project.id is not None + assert len(project.id) == 32 + assert project.owner_user_id == "user1" + assert project.name == "我的项目" + assert project.description == "" + assert project.shared_users == [] + assert isinstance(project.created_at, datetime) + + def test_create_with_description(self): + project = Project.create("u1", "Test Project", "A test description") + assert project.description == "A test description" + + def test_create_strips_name(self): + project = Project.create("u1", " 带空格的项目 ") + assert project.name == "带空格的项目" + + def test_create_strips_description(self): + project = Project.create("u1", "P1", " desc ") + assert project.description == "desc" + + def test_create_empty_name(self): + with pytest.raises(ValueError, match="项目名称不能为空"): + Project.create("u1", "") + + def test_create_whitespace_name(self): + with pytest.raises(ValueError, match="项目名称不能为空"): + Project.create("u1", " \t ") + + def test_create_unique_ids(self): + p1 = Project.create("u1", "P1") + p2 = Project.create("u1", "P2") + assert p1.id != p2.id + + +class TestProjectAccess: + def test_is_owner_true(self): + project = Project.create("owner1", "P1") + assert project.is_owner("owner1") is True + + def test_is_owner_false(self): + project = Project.create("owner1", "P1") + assert project.is_owner("other") is False + + def test_is_shared_with_true(self): + project = Project.create("owner1", "P1") + project.shared_users = ["user_a", "user_b"] + assert project.is_shared_with("user_a") is True + assert project.is_shared_with("user_b") is True + + def test_is_shared_with_false(self): + project = Project.create("owner1", "P1") + project.shared_users = ["user_a"] + assert project.is_shared_with("user_c") is False + + def test_can_access_owner(self): + project = Project.create("owner1", "P1") + assert project.can_access("owner1") is True + + def test_can_access_shared_user(self): + project = Project.create("owner1", "P1") + project.shared_users = ["shared_user"] + assert project.can_access("shared_user") is True + + def test_cannot_access_other(self): + project = Project.create("owner1", "P1") + assert project.can_access("stranger") is False + + def test_empty_shared_users(self): + project = Project.create("owner1", "P1") + assert project.shared_users == [] + assert project.is_shared_with("anyone") is False + + +class TestAssetLibraryCreate: + def test_create_video_library(self): + lib = AssetLibrary.create("proj1", "视频素材库", AssetLibraryKind.VIDEO) + assert lib.id is not None + assert len(lib.id) == 32 + assert lib.project_id == "proj1" + assert lib.name == "视频素材库" + assert lib.kind == AssetLibraryKind.VIDEO + assert lib.asset_count == 0 + assert lib.total_size == 0 + + def test_create_voice_library(self): + lib = AssetLibrary.create("proj1", "音乐库", AssetLibraryKind.VOICE) + assert lib.kind == AssetLibraryKind.VOICE + + def test_create_image_library(self): + lib = AssetLibrary.create("proj1", "图片库", AssetLibraryKind.IMAGE) + assert lib.kind == AssetLibraryKind.IMAGE + + def test_create_strips_name(self): + lib = AssetLibrary.create("p1", " 我的库 ", AssetLibraryKind.VIDEO) + assert lib.name == "我的库" + + def test_create_empty_name(self): + with pytest.raises(ValueError, match="素材库名称不能为空"): + AssetLibrary.create("p1", "", AssetLibraryKind.VIDEO) + + def test_create_whitespace_name(self): + with pytest.raises(ValueError, match="素材库名称不能为空"): + AssetLibrary.create("p1", " \t ", AssetLibraryKind.VIDEO) + + +class TestAssetStatusEnum: + def test_basic_values(self): + assert AssetStatus.UPLOADING.value == "uploading" + assert AssetStatus.READY.value == "ready" + assert AssetStatus.PROCESSING.value == "processing" + assert AssetStatus.ERROR.value == "error" + assert AssetStatus.DELETED.value == "deleted" + + def test_missing_uploaded_maps_to_ready(self): + assert AssetStatus("uploaded") == AssetStatus.READY + + def test_missing_success_maps_to_ready(self): + assert AssetStatus("success") == AssetStatus.READY + + def test_missing_ok_maps_to_ready(self): + assert AssetStatus("ok") == AssetStatus.READY + + def test_missing_done_maps_to_ready(self): + assert AssetStatus("done") == AssetStatus.READY + + def test_missing_complete_maps_to_ready(self): + assert AssetStatus("complete") == AssetStatus.READY + + def test_missing_upload_maps_to_uploading(self): + assert AssetStatus("upload") == AssetStatus.UPLOADING + + def test_missing_uploading_start_maps_to_uploading(self): + assert AssetStatus("uploading_start") == AssetStatus.UPLOADING + + def test_missing_upload_start_maps_to_uploading(self): + assert AssetStatus("upload_start") == AssetStatus.UPLOADING + + def test_missing_failed_maps_to_error(self): + assert AssetStatus("failed") == AssetStatus.ERROR + + def test_missing_fail_maps_to_error(self): + assert AssetStatus("fail") == AssetStatus.ERROR + + def test_missing_err_maps_to_error(self): + assert AssetStatus("err") == AssetStatus.ERROR + + def test_missing_process_maps_to_processing(self): + assert AssetStatus("process") == AssetStatus.PROCESSING + + def test_missing_running_maps_to_processing(self): + assert AssetStatus("running") == AssetStatus.PROCESSING + + def test_missing_run_maps_to_processing(self): + assert AssetStatus("run") == AssetStatus.PROCESSING + + def test_missing_unknown_value_falls_back_to_ready(self): + assert AssetStatus("completely_unknown_status") == AssetStatus.READY + + def test_missing_empty_string_falls_back_to_ready(self): + assert AssetStatus("") == AssetStatus.READY + + def test_missing_case_insensitive(self): + assert AssetStatus("UPLOADED") == AssetStatus.READY + assert AssetStatus("Success") == AssetStatus.READY + assert AssetStatus("FAILED") == AssetStatus.ERROR + + def test_missing_with_whitespace(self): + assert AssetStatus(" uploaded ") == AssetStatus.READY + assert AssetStatus("\tfailed\n") == AssetStatus.ERROR + + def test_missing_non_string_value(self): + assert AssetStatus(None) == AssetStatus.READY + assert AssetStatus(123) == AssetStatus.READY + + def test_known_values_still_work(self): + assert AssetStatus("uploading") == AssetStatus.UPLOADING + assert AssetStatus("ready") == AssetStatus.READY + assert AssetStatus("processing") == AssetStatus.PROCESSING + assert AssetStatus("error") == AssetStatus.ERROR + assert AssetStatus("deleted") == AssetStatus.DELETED + + +class TestAssetCreate: + def test_create_minimal(self): + asset = Asset.create( + project_id="proj1", + library_id="lib1", + name="test.mp4", + storage_key="videos/test.mp4", + mime_type="video/mp4", + ) + assert asset.id is not None + assert len(asset.id) == 32 + assert asset.project_id == "proj1" + assert asset.library_id == "lib1" + assert asset.name == "test.mp4" + assert asset.storage_key == "videos/test.mp4" + assert asset.mime_type == "video/mp4" + assert asset.file_size == 0 + assert asset.thumbnail_url is None + assert asset.duration is None + assert asset.width is None + assert asset.height is None + assert asset.status == AssetStatus.UPLOADING + assert asset.classification_status == ClassificationStatus.PENDING + assert asset.quality_score is None + assert asset.tag_ids == [] + assert isinstance(asset.created_at, datetime) + assert isinstance(asset.updated_at, datetime) + + def test_create_with_all_fields(self): + asset = Asset.create( + project_id="proj1", + library_id="lib1", + name="movie.mp4", + storage_key="v/m.mp4", + mime_type="video/mp4", + metadata={"key": "val"}, + file_size=1024000, + thumbnail_url="http://cdn/thumb.jpg", + duration=120.5, + width=1920, + height=1080, + fps=30.0, + codec="h264", + status=AssetStatus.READY, + classification_status=ClassificationStatus.COMPLETED, + quality_score=0.85, + uploaded_by_user_id="user1", + file_hash="abc123", + ) + assert asset.file_size == 1024000 + assert asset.thumbnail_url == "http://cdn/thumb.jpg" + assert asset.duration == 120.5 + assert asset.width == 1920 + assert asset.height == 1080 + assert asset.fps == 30.0 + assert asset.codec == "h264" + assert asset.status == AssetStatus.READY + assert asset.classification_status == ClassificationStatus.COMPLETED + assert asset.quality_score == 0.85 + assert asset.uploaded_by_user_id == "user1" + assert asset.file_hash == "abc123" + assert asset.metadata == {"key": "val"} + + def test_create_strips_name(self): + asset = Asset.create("p1", "l1", " test.mp4 ", "k", "video/mp4") + assert asset.name == "test.mp4" + + def test_create_strips_storage_key(self): + asset = Asset.create("p1", "l1", "n", " key.mp4 ", "video/mp4") + assert asset.storage_key == "key.mp4" + + def test_create_strips_mime_type(self): + asset = Asset.create("p1", "l1", "n", "k", " video/mp4 ") + assert asset.mime_type == "video/mp4" + + def test_create_empty_name(self): + with pytest.raises(ValueError, match="素材名称不能为空"): + Asset.create("p1", "l1", "", "k", "video/mp4") + + def test_create_empty_storage_key(self): + with pytest.raises(ValueError, match="storage_key 不能为空"): + Asset.create("p1", "l1", "n", "", "video/mp4") + + def test_create_empty_mime_type(self): + with pytest.raises(ValueError, match="mime_type 不能为空"): + Asset.create("p1", "l1", "n", "k", "") + + def test_create_whitespace_storage_key(self): + with pytest.raises(ValueError, match="storage_key 不能为空"): + Asset.create("p1", "l1", "n", " \t ", "video/mp4") + + def test_create_none_metadata_defaults_to_empty_dict(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4", metadata=None) + assert asset.metadata == {} + + def test_create_unique_ids(self): + a1 = Asset.create("p1", "l1", "n1", "k1", "video/mp4") + a2 = Asset.create("p1", "l1", "n2", "k2", "video/mp4") + assert a1.id != a2.id + + +class TestAssetFileType: + def test_video_mime(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + assert asset.file_type == "video" + + def test_audio_mime(self): + asset = Asset.create("p1", "l1", "n", "k", "audio/mpeg") + assert asset.file_type == "audio" + + def test_image_mime(self): + asset = Asset.create("p1", "l1", "n", "k", "image/jpeg") + assert asset.file_type == "image" + + def test_simple_mime_no_slash(self): + asset = Asset.create("p1", "l1", "n", "k", "application") + assert asset.file_type == "application" + + +class TestAssetTags: + def test_add_tag(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("tag1") + assert "tag1" in asset.tag_ids + assert len(asset.tag_ids) == 1 + + def test_add_tag_strips(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag(" tag_trim ") + assert "tag_trim" in asset.tag_ids + + def test_add_tag_duplicate_prevented(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("tag1") + asset.add_tag("tag1") + assert asset.tag_ids.count("tag1") == 1 + assert len(asset.tag_ids) == 1 + + def test_add_tag_empty(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + with pytest.raises(ValueError, match="标签 ID 不能为空"): + asset.add_tag("") + + def test_add_tag_whitespace(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + with pytest.raises(ValueError, match="标签 ID 不能为空"): + asset.add_tag(" \t ") + + def test_add_multiple_tags(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + asset.add_tag("t2") + asset.add_tag("t3") + assert asset.tag_ids == ["t1", "t2", "t3"] + + def test_remove_tag(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + asset.add_tag("t2") + asset.remove_tag("t1") + assert asset.tag_ids == ["t2"] + + def test_remove_nonexistent_tag_idempotent(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + # 删除不存在的标签不报错 + asset.remove_tag("nonexistent") + assert asset.tag_ids == ["t1"] + + def test_remove_tag_strips(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + asset.remove_tag(" t1 ") + assert asset.tag_ids == [] + + def test_add_tag_updates_updated_at(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + old_time = asset.updated_at + asset.add_tag("t1") + assert asset.updated_at >= old_time + + def test_remove_tag_updates_updated_at(self): + asset = Asset.create("p1", "l1", "n", "k", "video/mp4") + asset.add_tag("t1") + old_time = asset.updated_at + asset.remove_tag("t1") + assert asset.updated_at >= old_time + + +class TestIngestJobCreate: + def test_create_success(self): + job = IngestJob.create( + project_id="proj1", + library_id="lib1", + storage_key="videos/test.mp4", + ) + assert job.id is not None + assert len(job.id) == 32 + assert job.project_id == "proj1" + assert job.library_id == "lib1" + assert job.storage_key == "videos/test.mp4" + assert job.status == IngestJobStatus.PENDING + assert job.error_message == "" + assert job.result_asset_id == "" + assert job.file_hash == "" + + def test_create_with_hash(self): + job = IngestJob.create("p1", "l1", "k", file_hash="abcdef123456") + assert job.file_hash == "abcdef123456" + + def test_create_strips_project_id(self): + job = IngestJob.create(" p1 ", "l1", "k") + assert job.project_id == "p1" + + def test_create_strips_library_id(self): + job = IngestJob.create("p1", " l1 ", "k") + assert job.library_id == "l1" + + def test_create_strips_storage_key(self): + job = IngestJob.create("p1", "l1", " k ") + assert job.storage_key == "k" + + def test_create_strips_file_hash(self): + job = IngestJob.create("p1", "l1", "k", file_hash=" hash ") + assert job.file_hash == "hash" + + def test_create_empty_project_id(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + IngestJob.create("", "l1", "k") + + def test_create_empty_library_id(self): + with pytest.raises(ValueError, match="library_id 不能为空"): + IngestJob.create("p1", "", "k") + + def test_create_empty_storage_key(self): + with pytest.raises(ValueError, match="storage_key 不能为空"): + IngestJob.create("p1", "l1", "") + + def test_create_whitespace_project_id(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + IngestJob.create(" \t ", "l1", "k") + + def test_create_unique_ids(self): + j1 = IngestJob.create("p1", "l1", "k1") + j2 = IngestJob.create("p1", "l1", "k2") + assert j1.id != j2.id + + +class TestUserDataclass: + def test_default_values(self): + user = User(id="u1", email="test@example.com", display_name="Test User") + assert user.id == "u1" + assert user.email == "test@example.com" + assert user.display_name == "Test User" + assert user.username == "" + assert user.password_hash == "" + assert user.email_verified is False + assert user.subscription_plan == "free" + assert user.subscription_status == "active" + assert user.max_projects == 3 + assert user.max_storage_gb == 10 + assert user.used_storage_gb == 0.0 + assert user.is_admin is False + assert user.wechat_openid is None + assert user.phone is None + assert user.phone_verified is False + assert isinstance(user.created_at, datetime) + + def test_admin_user(self): + user = User(id="admin", email="admin@test.com", display_name="Admin", is_admin=True) + assert user.is_admin is True + + def test_pro_subscription(self): + user = User( + id="u1", + email="u@t.com", + display_name="U", + subscription_plan="pro", + max_storage_gb=100, + ) + assert user.subscription_plan == "pro" + assert user.max_storage_gb == 100 diff --git a/tests/unit/test_domain_small_modules.py b/tests/unit/test_domain_small_modules.py new file mode 100755 index 000000000..ab018f167 --- /dev/null +++ b/tests/unit/test_domain_small_modules.py @@ -0,0 +1,391 @@ +"""Domain 小模块合集单元测试。 + +覆盖零测试的小 domain 模块： +- EditingMode 枚举 +- Template / TemplateSegment +- TemplateClipConfig + ClipType + TransitionEffect +- EditTemplateVersion +- VoiceLibraryItem +- TitleLibraryItem +- Recipe / RecipeItem +""" + +from datetime import datetime, timezone + +import pytest + +from packages.domain.editing_mode import EditingMode +from packages.domain.recipe import RecipeItem +from packages.domain.template import TemplateSegment +from packages.domain.template_clip_config import ( + ClipType, + TemplateClipConfig, + TransitionEffect, +) +from packages.domain.template_version import EditTemplateVersion +from packages.domain.title_library import TitleLibraryItem +from packages.domain.voice_library import VoiceLibraryItem + + +class TestEditingMode: + def test_all_modes_exist(self): + assert EditingMode.ONE_TAKE.value == "one_take" + assert EditingMode.PIP.value == "pip" + assert EditingMode.VOICE_OVER.value == "voice_over" + assert EditingMode.VOICE_PIP.value == "voice_pip" + + def test_from_string(self): + assert EditingMode("one_take") == EditingMode.ONE_TAKE + assert EditingMode("voice_over") == EditingMode.VOICE_OVER + + def test_invalid_mode_raises(self): + with pytest.raises(ValueError): + EditingMode("invalid_mode") + + def test_is_str_enum(self): + # StrEnum 的值是字符串，可以直接比较 + assert EditingMode.ONE_TAKE == "one_take" + + +class TestTemplateSegment: + def test_create_minimal(self): + seg = TemplateSegment( + id="seg1", + template_id="tpl1", + segment_order=1, + duration_min=5.0, + duration_max=10.0, + ) + assert seg.id == "seg1" + assert seg.template_id == "tpl1" + assert seg.segment_order == 1 + assert seg.duration_min == 5.0 + assert seg.duration_max == 10.0 + assert seg.material_type is None + assert isinstance(seg.created_at, datetime) + + def test_create_with_material_type(self): + seg = TemplateSegment( + id="seg2", + template_id="tpl1", + segment_order=2, + duration_min=3.0, + duration_max=8.0, + material_type="人物", + ) + assert seg.material_type == "人物" + + +class TestClipType: + def test_basic_types_exist(self): + assert hasattr(ClipType, "MAIN") + assert hasattr(ClipType, "INTRO") + assert hasattr(ClipType, "OUTRO") + assert hasattr(ClipType, "TRANSITION") + + def test_values_are_strings(self): + for ct in ClipType: + assert isinstance(ct.value, str) + + +class TestTransitionEffect: + def test_effects_exist(self): + assert TransitionEffect.CUT.value == "cut" + assert TransitionEffect.FADE.value == "fade" + assert TransitionEffect.DISSOLVE.value == "dissolve" + # 至少有 5 种以上转场效果 + assert len(list(TransitionEffect)) >= 5 + + +class TestTemplateClipConfig: + def test_create_minimal(self): + config = TemplateClipConfig.create( + template_id="tpl1", + clip_type=ClipType.MAIN, + order=1, + min_duration=3.0, + max_duration=8.0, + ) + assert config.id is not None + assert config.template_id == "tpl1" + assert config.clip_type == ClipType.MAIN + assert config.order == 1 + assert config.min_duration == 3.0 + assert config.max_duration == 8.0 + + def test_create_with_string_type(self): + config = TemplateClipConfig.create( + template_id="tpl1", + clip_type="intro", + order=0, + min_duration=2.0, + max_duration=5.0, + ) + assert config.clip_type == ClipType.INTRO + + def test_has_duration_range_true(self): + config = TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + min_duration=3.0, + max_duration=8.0, + ) + assert config.has_duration_range is True + + def test_has_duration_range_false_when_both_zero(self): + config = TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + ) + assert config.has_duration_range is False + + def test_default_duration_midpoint(self): + config = TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + min_duration=4.0, + max_duration=6.0, + ) + assert config.default_duration == pytest.approx(5.0) + + def test_default_duration_when_only_max(self): + config = TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + max_duration=5.0, + ) + assert config.default_duration == 5.0 + + def test_create_negative_min_duration_raises(self): + with pytest.raises(ValueError, match="min_duration"): + TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + min_duration=-1.0, + ) + + def test_create_min_greater_than_max_raises(self): + with pytest.raises(ValueError, match="min_duration.*max_duration"): + TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + min_duration=10.0, + max_duration=5.0, + ) + + def test_create_empty_template_id_raises(self): + with pytest.raises(ValueError, match="template_id"): + TemplateClipConfig.create( + template_id="", + clip_type=ClipType.MAIN, + order=1, + ) + + def test_default_transition_is_cut(self): + config = TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + ) + assert config.transition_effect == TransitionEffect.CUT + + def test_custom_transition_effect(self): + config = TemplateClipConfig.create( + template_id="t1", + clip_type=ClipType.MAIN, + order=1, + transition_effect="fade", + ) + assert config.transition_effect == TransitionEffect.FADE + + +class TestEditTemplateVersion: + def test_create_minimal(self): + version = EditTemplateVersion.create( + template_id="tpl1", + version=1, + ) + assert version.id is not None + assert len(version.id) == 32 + assert version.template_id == "tpl1" + assert version.version == 1 + assert version.config == {} + assert version.clip_configs == [] + assert version.published_by == "" + assert version.change_note == "" + assert version.name == "" + assert version.editing_mode == "one_take" + assert isinstance(version.created_at, datetime) + + def test_create_with_config_and_clip_configs(self): + version = EditTemplateVersion.create( + template_id="tpl1", + version=2, + config={"layout": "one_take"}, + clip_configs=[{"clip_id": "c1", "type": "main"}], + published_by="user1", + change_note="添加了片头效果", + ) + assert version.config == {"layout": "one_take"} + assert len(version.clip_configs) == 1 + assert version.published_by == "user1" + assert version.change_note == "添加了片头效果" + + def test_create_with_name_and_mode(self): + version = EditTemplateVersion.create( + template_id="t1", + version=1, + name="v1.0 正式版", + editing_mode="voice_over", + ) + assert version.name == "v1.0 正式版" + assert version.editing_mode == "voice_over" + + def test_create_unique_ids(self): + v1 = EditTemplateVersion.create("t1", 1) + v2 = EditTemplateVersion.create("t1", 2) + assert v1.id != v2.id + + def test_none_config_defaults_to_empty_dict(self): + version = EditTemplateVersion.create("t1", 1, config=None) + assert version.config == {} + + def test_none_clip_configs_defaults_to_empty_list(self): + version = EditTemplateVersion.create("t1", 1, clip_configs=None) + assert version.clip_configs == [] + + +class TestVoiceLibraryItem: + def test_create_minimal(self): + item = VoiceLibraryItem( + id="v1", + user_id="u1", + name="我的配音", + ) + assert item.id == "v1" + assert item.user_id == "u1" + assert item.name == "我的配音" + assert item.text == "" + assert item.voice_provider == "" + assert item.duration == 0 + assert item.status == "completed" + assert item.tags == [] + assert item.project_id is None + assert isinstance(item.created_at, datetime) + + def test_create_with_all_fields(self): + item = VoiceLibraryItem( + id="v2", + user_id="u1", + name="产品介绍", + text="欢迎来到我们的产品", + voice_provider="cosyvoice", + voice_id="voice_001", + voice_name="温柔女声", + audio_url="https://cdn/v2.mp3", + duration=30.5, + file_size=102400, + status="processing", + project_id="proj1", + tags=["产品", "介绍"], + ) + assert item.text == "欢迎来到我们的产品" + assert item.voice_provider == "cosyvoice" + assert item.voice_id == "voice_001" + assert item.audio_url == "https://cdn/v2.mp3" + assert item.duration == 30.5 + assert item.file_size == 102400 + assert item.status == "processing" + assert item.project_id == "proj1" + assert item.tags == ["产品", "介绍"] + + +class TestTitleLibraryItem: + def test_create_minimal(self): + item = TitleLibraryItem( + id="t1", + user_id="u1", + name="爆款标题1", + text="这是一个爆款标题", + ) + assert item.id == "t1" + assert item.user_id == "u1" + assert item.name == "爆款标题1" + assert item.text == "这是一个爆款标题" + assert item.category == "default" + assert item.description == "" + assert item.tags == [] + assert item.usage_count == 0 + assert item.is_active is True + + def test_create_with_category(self): + item = TitleLibraryItem( + id="t2", + user_id="u1", + name="美食标题", + text="太好吃了！", + category="美食", + ) + assert item.category == "美食" + + def test_inactive_item(self): + item = TitleLibraryItem( + id="t3", + user_id="u1", + name="旧标题", + text="旧文案", + is_active=False, + ) + assert item.is_active is False + + def test_usage_count_increment(self): + item = TitleLibraryItem( + id="t4", + user_id="u1", + name="T", + text="T", + ) + item.usage_count += 1 + assert item.usage_count == 1 + + +class TestRecipeItem: + def test_create_minimal(self): + item = RecipeItem( + id="ri1", + recipe_id="r1", + item_type="asset", + item_id="asset_001", + ) + assert item.id == "ri1" + assert item.recipe_id == "r1" + assert item.item_type == "asset" + assert item.item_id == "asset_001" + assert item.position == 0 + assert item.metadata_ == {} + + def test_create_with_position_and_metadata(self): + item = RecipeItem( + id="ri2", + recipe_id="r1", + item_type="title", + item_id="title_001", + position=2, + metadata_={"style": "bold"}, + ) + assert item.position == 2 + assert item.metadata_ == {"style": "bold"} + + def test_item_types_variety(self): + asset_item = RecipeItem(id="a", recipe_id="r", item_type="asset", item_id="i1") + title_item = RecipeItem(id="t", recipe_id="r", item_type="title", item_id="i2") + voice_item = RecipeItem(id="v", recipe_id="r", item_type="voice", item_id="i3") + assert asset_item.item_type == "asset" + assert title_item.item_type == "title" + assert voice_item.item_type == "voice" diff --git a/tests/unit/test_job_domain.py b/tests/unit/test_job_domain.py index ba0cc9aa1..948ccc105 100755 --- a/tests/unit/test_job_domain.py +++ b/tests/unit/test_job_domain.py @@ -1,4 +1,6 @@ -"""Job 领域层单元测试 - job.py""" +"""Job 领域模型单元测试。""" + +from datetime import datetime, timezone import pytest @@ -10,45 +12,45 @@ from packages.domain.job import ( ) -class TestJobType: - """JobType 枚举测试""" +class TestJobTypeEnum: + def test_all_types_exist(self): + assert JobType.VIDEO_COMPOSE.value == "video_compose" + assert JobType.RENDER_EDIT_PLAN.value == "render_edit_plan" + assert JobType.ASSET_INGEST.value == "asset_ingest" + assert JobType.CLASSIFICATION.value == "classification" + assert JobType.VOICE_EXTRACTION.value == "voice_extraction" + assert JobType.GENERATION.value == "generation" - def test_all_types_have_values(self): - """所有枚举成员都有字符串值""" - for jt in JobType: - assert isinstance(jt.value, str) - assert jt.value + def test_from_string(self): + assert JobType("video_compose") == JobType.VIDEO_COMPOSE + assert JobType("generation") == JobType.GENERATION - def test_str_enum_behavior(self): - """是 str 枚举""" - assert JobType.VIDEO_COMPOSE == "video_compose" - assert isinstance(JobType.VIDEO_COMPOSE, str) - - def test_known_types_exist(self): - """核心任务类型都存在""" - assert JobType.VIDEO_COMPOSE - assert JobType.RENDER_EDIT_PLAN - assert JobType.ASSET_INGEST - assert JobType.CLASSIFICATION - assert JobType.GENERATION + def test_invalid_type_raises(self): + with pytest.raises(ValueError): + JobType("invalid_type") -class TestJobStatus: - """JobStatus 枚举测试""" +class TestJobStatusEnum: + def test_all_statuses_exist(self): + assert JobStatus.PENDING.value == "pending" + assert JobStatus.RUNNING.value == "running" + assert JobStatus.SUCCESS.value == "success" + assert JobStatus.FAILED.value == "failed" + assert JobStatus.CANCELLED.value == "cancelled" - def test_all_statuses_have_values(self): - for js in JobStatus: - assert isinstance(js.value, str) - assert js.value + def test_from_string(self): + assert JobStatus("pending") == JobStatus.PENDING + assert JobStatus("success") == JobStatus.SUCCESS - def test_str_enum_behavior(self): - assert JobStatus.PENDING == "pending" - assert isinstance(JobStatus.PENDING, str) - def test_terminal_statuses(self): - """终态集合包含成功/失败/取消""" +class TestTerminalStatuses: + def test_success_is_terminal(self): assert JobStatus.SUCCESS in TERMINAL_STATUSES + + def test_failed_is_terminal(self): assert JobStatus.FAILED in TERMINAL_STATUSES + + def test_cancelled_is_terminal(self): assert JobStatus.CANCELLED in TERMINAL_STATUSES def test_pending_not_terminal(self): @@ -59,372 +61,376 @@ class TestJobStatus: class TestJobCreate: - """Job.create 工厂方法测试""" - - def test_create_basic(self): - """基本创建""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - ) - assert job.id - assert len(job.id) == 32 # uuid4 hex - assert job.project_id == "proj-1" + def test_create_minimal(self): + job = Job.create(project_id="proj1", job_type=JobType.VIDEO_COMPOSE) + assert job.id is not None + assert len(job.id) == 32 + assert job.project_id == "proj1" assert job.job_type == JobType.VIDEO_COMPOSE assert job.status == JobStatus.PENDING assert job.progress == 0.0 + assert job.current_stage == "" assert job.payload == {} assert job.result == {} + assert job.error_message == "" assert job.retry_count == 0 assert job.max_retries == 3 - assert job.created_at - assert job.updated_at + assert job.celery_task_id == "" + assert job.source_id == "" + assert job.created_by_user_id == "" + assert job.started_at is None + assert job.completed_at is None + assert isinstance(job.created_at, datetime) + assert isinstance(job.updated_at, datetime) - def test_create_with_string_job_type(self): - """用字符串创建任务类型""" - job = Job.create( - project_id="proj-1", - job_type="video_compose", - ) + def test_create_with_enum_type(self): + job = Job.create("p1", JobType.GENERATION) + assert job.job_type == JobType.GENERATION + + def test_create_with_string_type(self): + job = Job.create("p1", "video_compose") assert job.job_type == JobType.VIDEO_COMPOSE - def test_create_invalid_string_job_type_raises(self): - """无效的任务类型字符串抛 ValueError""" - with pytest.raises(ValueError, match="不支持的任务类型"): - Job.create(project_id="proj-1", job_type="invalid_type") - - def test_create_empty_project_id_raises(self): - """空 project_id 抛 ValueError""" - with pytest.raises(ValueError, match="project_id 不能为空"): - Job.create(project_id=" ", job_type=JobType.VIDEO_COMPOSE) - def test_create_with_payload(self): - """带 payload 创建""" - payload = {"video_id": "v1", "quality": "1080p"} - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - payload=payload, - ) + payload = {"edit_plan_id": "plan123", "resolution": "1080p"} + job = Job.create("p1", JobType.VIDEO_COMPOSE, payload=payload) assert job.payload == payload - def test_create_with_source_id(self): - """带 source_id 创建""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - source_id="plan-123", - ) - assert job.source_id == "plan-123" - - def test_create_with_created_by(self): - """带创建人""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - created_by_user_id="user-1", - ) - assert job.created_by_user_id == "user-1" - - def test_create_with_custom_max_retries(self): - """自定义最大重试次数""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - max_retries=5, - ) - assert job.max_retries == 5 - - def test_create_project_id_stripped(self): - """project_id 会被 strip""" - job = Job.create( - project_id=" proj-1 ", - job_type=JobType.VIDEO_COMPOSE, - ) - assert job.project_id == "proj-1" - - def test_create_source_id_stripped(self): - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - source_id=" src-1 ", - ) - assert job.source_id == "src-1" - - def test_create_created_by_stripped(self): - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - created_by_user_id=" user-1 ", - ) - assert job.created_by_user_id == "user-1" - - def test_create_none_payload_defaults_to_empty_dict(self): - """payload=None 时默认为空 dict""" - job = Job.create( - project_id="proj-1", - job_type=JobType.VIDEO_COMPOSE, - payload=None, - ) + def test_create_with_none_payload(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, payload=None) assert job.payload == {} + def test_create_with_source_id(self): + job = Job.create("p1", JobType.GENERATION, source_id="gen123") + assert job.source_id == "gen123" -class TestJobIsTerminal: - """is_terminal 属性测试""" + def test_create_with_user_id(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, created_by_user_id="user1") + assert job.created_by_user_id == "user1" + def test_create_with_custom_max_retries(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=5) + assert job.max_retries == 5 + + def test_create_strips_project_id(self): + job = Job.create(" proj1 ", JobType.VIDEO_COMPOSE) + assert job.project_id == "proj1" + + def test_create_strips_source_id(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, source_id=" src1 ") + assert job.source_id == "src1" + + def test_create_strips_user_id(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, created_by_user_id=" u1 ") + assert job.created_by_user_id == "u1" + + def test_create_empty_project_id(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + Job.create("", JobType.VIDEO_COMPOSE) + + def test_create_whitespace_project_id(self): + with pytest.raises(ValueError, match="project_id 不能为空"): + Job.create(" \t ", JobType.VIDEO_COMPOSE) + + def test_create_invalid_job_type_string(self): + with pytest.raises(ValueError, match="不支持的任务类型"): + Job.create("p1", "invalid_type") + + def test_create_unique_ids(self): + j1 = Job.create("p1", JobType.VIDEO_COMPOSE) + j2 = Job.create("p1", JobType.VIDEO_COMPOSE) + assert j1.id != j2.id + + +class TestIsTerminal: def test_pending_not_terminal(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) assert job.is_terminal is False def test_running_not_terminal(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.RUNNING) assert job.is_terminal is False def test_success_is_terminal(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.RUNNING) job.transition_to(JobStatus.SUCCESS) assert job.is_terminal is True def test_failed_is_terminal(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.RUNNING) job.transition_to(JobStatus.FAILED) assert job.is_terminal is True def test_cancelled_is_terminal(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.CANCELLED) assert job.is_terminal is True -class TestJobTransitions: - """状态转换测试""" +class TestIsRetryable: + def test_pending_not_retryable(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + assert job.is_retryable is False + def test_running_not_retryable(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + assert job.is_retryable is False + + def test_success_not_retryable(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_success() + assert job.is_retryable is False + + def test_failed_within_limit_is_retryable(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3) + job.mark_running() + job.mark_failed("error") + assert job.is_retryable is True + + def test_failed_at_limit_not_retryable(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3) + job.mark_running() + job.mark_failed("error") + job.retry_count = 3 # 已达到上限 + assert job.is_retryable is False + + def test_failed_over_limit_not_retryable(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3) + job.retry_count = 5 + job.status = JobStatus.FAILED + assert job.is_retryable is False + + def test_zero_max_retries_not_retryable(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=0) + job.status = JobStatus.FAILED + assert job.is_retryable is False + + +class TestTransitionTo: def test_pending_to_running(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.RUNNING) assert job.status == JobStatus.RUNNING assert job.started_at is not None def test_pending_to_success(self): - """pending 可以直接到 success（快速成功）""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.SUCCESS) assert job.status == JobStatus.SUCCESS assert job.completed_at is not None def test_pending_to_cancelled(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.CANCELLED) assert job.status == JobStatus.CANCELLED - def test_running_to_success(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.SUCCESS) - assert job.status == JobStatus.SUCCESS - assert job.completed_at is not None - - def test_running_to_failed(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.FAILED) - assert job.status == JobStatus.FAILED - assert job.completed_at is not None - - def test_running_to_cancelled(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.CANCELLED) - assert job.status == JobStatus.CANCELLED - - def test_failed_to_pending_retry(self): - """失败后可以回到 pending（重试）""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.transition_to(JobStatus.RUNNING) - job.transition_to(JobStatus.FAILED) - job.transition_to(JobStatus.PENDING) - assert job.status == JobStatus.PENDING - - def test_invalid_transition_raises(self): - """非法状态转换抛 ValueError""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - # pending 不能直接到 failed + def test_pending_to_failed_invalid(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) with pytest.raises(ValueError, match="非法状态转换"): job.transition_to(JobStatus.FAILED) - def test_success_to_pending_raises(self): - """成功后不能回到 pending""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_running_to_success(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + job.transition_to(JobStatus.SUCCESS) + assert job.status == JobStatus.SUCCESS + + def test_running_to_failed(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + job.transition_to(JobStatus.FAILED) + assert job.status == JobStatus.FAILED + + def test_running_to_cancelled(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + job.transition_to(JobStatus.CANCELLED) + assert job.status == JobStatus.CANCELLED + + def test_running_to_pending_invalid(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + with pytest.raises(ValueError, match="非法状态转换"): + job.transition_to(JobStatus.PENDING) + + def test_failed_to_pending(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.transition_to(JobStatus.RUNNING) + job.transition_to(JobStatus.FAILED) + # 注意：_VALID_TRANSITIONS 中 FAILED → PENDING 是允许的 + job.transition_to(JobStatus.PENDING) + assert job.status == JobStatus.PENDING + + def test_success_to_anything_invalid(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.RUNNING) job.transition_to(JobStatus.SUCCESS) with pytest.raises(ValueError): - job.transition_to(JobStatus.PENDING) + job.transition_to(JobStatus.FAILED) + with pytest.raises(ValueError): + job.transition_to(JobStatus.RUNNING) def test_transition_with_string_status(self): - """用字符串做状态转换""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to("running") assert job.status == JobStatus.RUNNING - def test_transition_invalid_string_raises(self): - """无效状态字符串抛 ValueError""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_transition_with_invalid_string(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) with pytest.raises(ValueError, match="无效状态"): job.transition_to("invalid_status") def test_transition_updates_updated_at(self): - """状态转换更新 updated_at""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - old_updated = job.updated_at - import time - - time.sleep(0.001) + job = Job.create("p1", JobType.VIDEO_COMPOSE) + old_time = job.updated_at job.transition_to(JobStatus.RUNNING) - assert job.updated_at >= old_updated + assert job.updated_at >= old_time def test_started_at_only_set_once(self): - """started_at 只在第一次 RUNNING 时设置""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.transition_to(JobStatus.RUNNING) - first_started = job.started_at - job.transition_to(JobStatus.SUCCESS) - # 回到 pending 再 running（模拟重试场景，但started_at是None时才设置） - # 注意：正常重试是通过 prepare_retry 重置的 - assert first_started is not None + first_start = job.started_at + # 再次 RUNNING 不合法，但我们测试 started_at 在多次 running→success→retry→running 时的行为 + # 先失败重试 + job.transition_to(JobStatus.FAILED) + job.transition_to(JobStatus.PENDING) + job.started_at = None # 模拟 prepare_retry 的重置 + job.transition_to(JobStatus.RUNNING) + assert job.started_at is not None + assert job.started_at != first_start -class TestJobMarkMethods: - """便捷标记方法测试""" - - def test_mark_running(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running("合成中") - assert job.status == JobStatus.RUNNING - assert job.current_stage == "合成中" - - def test_mark_running_no_stage(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) +class TestMarkRunning: + def test_mark_running_basic(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.mark_running() assert job.status == JobStatus.RUNNING - assert job.current_stage == "" + assert job.started_at is not None - def test_mark_success(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running() - job.mark_success({"output_url": "http://..."}) - assert job.status == JobStatus.SUCCESS - assert job.progress == 100.0 - assert job.current_stage == "完成" - assert job.result == {"output_url": "http://..."} + def test_mark_running_with_stage(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.mark_running(stage="下载素材") + assert job.status == JobStatus.RUNNING + assert job.current_stage == "下载素材" - def test_mark_success_no_result(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_mark_running_empty_stage_unchanged(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.current_stage = "已有阶段" + job.mark_running() # 不传 stage + assert job.current_stage == "已有阶段" + + +class TestMarkSuccess: + def test_mark_success_basic(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.mark_running() job.mark_success() assert job.status == JobStatus.SUCCESS - assert job.result == {} + assert job.progress == 100.0 + assert job.current_stage == "完成" + assert job.completed_at is not None - def test_mark_failed(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_mark_success_with_result(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.mark_running() + result = {"video_url": "https://...", "duration": 30} + job.mark_success(result=result) + assert job.result == result + + def test_mark_success_without_result(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.mark_running() + original_result = job.result.copy() + job.mark_success() + assert job.result == original_result # 不变 + + +class TestMarkFailed: + def test_mark_failed_basic(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.mark_running() job.mark_failed("网络超时") assert job.status == JobStatus.FAILED assert job.error_message == "网络超时" assert job.current_stage == "失败" + assert job.completed_at is not None - def test_mark_cancelled(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_mark_failed_empty_message(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_failed("") + assert job.error_message == "" + + +class TestMarkCancelled: + def test_mark_cancelled_from_pending(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.mark_cancelled() assert job.status == JobStatus.CANCELLED assert job.current_stage == "已取消" + def test_mark_cancelled_from_running(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.mark_running() + job.mark_cancelled() + assert job.status == JobStatus.CANCELLED -class TestJobProgress: - """进度更新测试""" - def test_update_progress(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.update_progress(50.0, "渲染中") +class TestUpdateProgress: + def test_update_progress_valid(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.update_progress(50.0) assert job.progress == 50.0 - assert job.current_stage == "渲染中" def test_update_progress_zero(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.update_progress(0.0) assert job.progress == 0.0 - def test_update_progress_100(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_update_progress_hundred(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) job.update_progress(100.0) assert job.progress == 100.0 - def test_update_progress_negative_raises(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_update_progress_negative(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) with pytest.raises(ValueError, match="进度必须在 0~100 之间"): job.update_progress(-1.0) - def test_update_progress_over_100_raises(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) + def test_update_progress_over_100(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) with pytest.raises(ValueError, match="进度必须在 0~100 之间"): job.update_progress(101.0) - def test_update_progress_without_stage(self): - """不传 stage 时不修改 current_stage""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.current_stage = "初始阶段" - job.update_progress(30.0) + def test_update_progress_with_stage(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.update_progress(30.0, stage="渲染中") assert job.progress == 30.0 - assert job.current_stage == "初始阶段" + assert job.current_stage == "渲染中" - def test_update_progress_updates_updated_at(self): - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - old_updated = job.updated_at - import time - - time.sleep(0.001) + def test_update_progress_without_stage_unchanged(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.current_stage = "原阶段" job.update_progress(50.0) - assert job.updated_at >= old_updated + assert job.current_stage == "原阶段" + + def test_update_progress_updates_timestamp(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + old_time = job.updated_at + job.update_progress(25.0) + assert job.updated_at >= old_time -class TestJobRetry: - """重试逻辑测试""" - - def test_is_retryable_failed_within_limit(self): - """失败且未超过重试上限时可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) - job.mark_running() - job.mark_failed("错误") - assert job.is_retryable is True - - def test_is_retryable_failed_at_limit(self): - """达到重试上限时不可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=1) - job.mark_running() - job.mark_failed("错误") - job.retry_count = 1 - assert job.is_retryable is False - - def test_is_retryable_pending_false(self): - """pending 状态不可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - assert job.is_retryable is False - - def test_is_retryable_success_false(self): - """成功状态不可重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running() - job.mark_success() - assert job.is_retryable is False - - def test_prepare_retry(self): - """准备重试""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) +class TestPrepareRetry: + def test_prepare_retry_success(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3) job.mark_running() job.mark_failed("网络错误") - job.celery_task_id = "task-123" job.prepare_retry() @@ -437,38 +443,41 @@ class TestJobRetry: assert job.completed_at is None assert job.celery_task_id == "" - def test_prepare_retry_not_retryable_raises(self): - """不可重试时抛 ValueError""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=0) + def test_prepare_retry_increments_count(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=5) job.mark_running() - job.mark_failed("错误") - with pytest.raises(ValueError, match="任务不可重试"): - job.prepare_retry() + job.mark_failed("err") - def test_prepare_retry_increments_correctly(self): - """多次重试计数正确""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3) - job.mark_running() - job.mark_failed("错误1") job.prepare_retry() assert job.retry_count == 1 + # 再次失败重试 job.mark_running() - job.mark_failed("错误2") + job.mark_failed("err2") job.prepare_retry() assert job.retry_count == 2 + def test_prepare_retry_not_retryable_raises(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=0) + job.mark_running() + job.mark_failed("err") + with pytest.raises(ValueError, match="任务不可重试"): + job.prepare_retry() -class TestJobToDict: - """to_dict 序列化测试""" + def test_prepare_retry_wrong_status_raises(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE) + with pytest.raises(ValueError, match="任务不可重试"): + job.prepare_retry() - def test_to_dict_contains_all_fields(self): + +class TestToDict: + def test_to_dict_structure(self): job = Job.create( - project_id="p1", - job_type=JobType.VIDEO_COMPOSE, - payload={"key": "value"}, - source_id="src-1", - created_by_user_id="user-1", + "p1", + JobType.VIDEO_COMPOSE, + payload={"key": "val"}, + source_id="src1", + created_by_user_id="u1", ) d = job.to_dict() assert d["id"] == job.id @@ -476,33 +485,39 @@ class TestJobToDict: assert d["job_type"] == "video_compose" assert d["status"] == "pending" assert d["progress"] == 0.0 - assert d["payload"] == {"key": "value"} - assert d["source_id"] == "src-1" - assert d["created_by_user_id"] == "user-1" + assert d["current_stage"] == "" + assert d["payload"] == {"key": "val"} + assert d["result"] == {} + assert d["error_message"] == "" + assert d["retry_count"] == 0 + assert d["max_retries"] == 3 + assert d["celery_task_id"] == "" + assert d["source_id"] == "src1" + assert d["created_by_user_id"] == "u1" assert d["is_retryable"] is False - - def test_to_dict_datetime_fields_are_strings(self): - """时间字段序列化为 ISO 字符串""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - d = job.to_dict() - assert isinstance(d["created_at"], str) - assert isinstance(d["updated_at"], str) - - def test_to_dict_none_datetime_fields(self): - """未设置的时间字段为 None""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - d = job.to_dict() assert d["started_at"] is None assert d["completed_at"] is None + assert d["created_at"] is not None + assert d["updated_at"] is not None def test_to_dict_after_success(self): - """成功后 to_dict 状态正确""" - job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE) - job.mark_running() - job.mark_success({"url": "http://..."}) + job = Job.create("p1", JobType.VIDEO_COMPOSE) + job.mark_running("渲染") + job.mark_success({"url": "https://..."}) d = job.to_dict() assert d["status"] == "success" assert d["progress"] == 100.0 - assert d["result"] == {"url": "http://..."} + assert d["is_retryable"] is False assert d["started_at"] is not None assert d["completed_at"] is not None + assert isinstance(d["started_at"], str) + assert isinstance(d["completed_at"], str) + + def test_to_dict_after_failed(self): + job = Job.create("p1", JobType.VIDEO_COMPOSE, max_retries=3) + job.mark_running() + job.mark_failed("timeout") + d = job.to_dict() + assert d["status"] == "failed" + assert d["error_message"] == "timeout" + assert d["is_retryable"] is True diff --git a/tests/unit/test_media_validation.py b/tests/unit/test_media_validation.py new file mode 100755 index 000000000..f1fd8b083 --- /dev/null +++ b/tests/unit/test_media_validation.py @@ -0,0 +1,299 @@ +"""media_validation 领域模块单元测试。""" + +import pytest + +from packages.domain.media_validation import ( + MIN_AUDIO_FILE_SIZE, + MIN_IMAGE_FILE_SIZE, + MIN_VIDEO_FILE_SIZE, + SUPPORTED_VIDEO_CODECS, + is_valid_media, + safe_parse_fps, +) + + +class TestSafeParseFpsBasic: + def test_integer_fps(self): + assert safe_parse_fps("30") == 30.0 + + def test_decimal_fps(self): + assert safe_parse_fps("29.97") == pytest.approx(29.97) + + def test_fraction_simple(self): + assert safe_parse_fps("30/1") == 30.0 + + def test_fraction_ntsc(self): + assert safe_parse_fps("30000/1001") == pytest.approx(29.97002997) + + def test_fraction_pal(self): + assert safe_parse_fps("25/1") == 25.0 + + def test_fraction_24fps_cine(self): + assert safe_parse_fps("24000/1001") == pytest.approx(23.976023976) + + def test_zero_fps(self): + assert safe_parse_fps("0") == 0.0 + + def test_zero_fraction(self): + assert safe_parse_fps("0/1") == 0.0 + + +class TestSafeParseFpsEdgeCases: + def test_zero_denominator(self): + assert safe_parse_fps("30/0") == 0.0 + + def test_empty_string(self): + assert safe_parse_fps("") == 0.0 + + def test_garbage_string(self): + assert safe_parse_fps("not_a_number") == 0.0 + + def test_multiple_slashes(self): + # split("/", 1) 只切第一个，后面的作为 den 的一部分会解析失败 + assert safe_parse_fps("30/1/2") == 0.0 + + def test_negative_fps(self): + assert safe_parse_fps("-30") == -30.0 + + def test_negative_fraction(self): + assert safe_parse_fps("-30/1") == -30.0 + + def test_very_high_fps(self): + assert safe_parse_fps("240/1") == 240.0 + + def test_fraction_float_num(self): + assert safe_parse_fps("29.97/1") == pytest.approx(29.97) + + def test_fraction_float_den(self): + assert safe_parse_fps("30/1.001") == pytest.approx(29.97002997) + + def test_whitespace_in_string(self): + # float(" 30 ") 能解析，所以应该返回 30.0 + assert safe_parse_fps(" 30 ") == 30.0 + + +class TestMinFileSizeConstants: + def test_min_video_size_is_1kb(self): + assert MIN_VIDEO_FILE_SIZE == 1024 + + def test_min_audio_size(self): + assert MIN_AUDIO_FILE_SIZE == 100 + + def test_min_image_size(self): + assert MIN_IMAGE_FILE_SIZE == 100 + + +class TestSupportedVideoCodecs: + def test_h264_family_present(self): + assert "h264" in SUPPORTED_VIDEO_CODECS + assert "avc1" in SUPPORTED_VIDEO_CODECS + assert "avc" in SUPPORTED_VIDEO_CODECS + + def test_h265_family_present(self): + assert "hevc" in SUPPORTED_VIDEO_CODECS + assert "h265" in SUPPORTED_VIDEO_CODECS + assert "hev1" in SUPPORTED_VIDEO_CODECS + assert "hvc1" in SUPPORTED_VIDEO_CODECS + + def test_vp9_av1_present(self): + assert "vp9" in SUPPORTED_VIDEO_CODECS + assert "vp09" in SUPPORTED_VIDEO_CODECS + assert "av1" in SUPPORTED_VIDEO_CODECS + assert "av01" in SUPPORTED_VIDEO_CODECS + + def test_vp8_present(self): + assert "vp8" in SUPPORTED_VIDEO_CODECS + assert "vp08" in SUPPORTED_VIDEO_CODECS + + def test_mpeg_family_present(self): + assert "mpeg4" in SUPPORTED_VIDEO_CODECS + assert "mp4v" in SUPPORTED_VIDEO_CODECS + assert "mpeg2video" in SUPPORTED_VIDEO_CODECS + + def test_prores_family_present(self): + assert "prores" in SUPPORTED_VIDEO_CODECS + assert "apcn" in SUPPORTED_VIDEO_CODECS + assert "apch" in SUPPORTED_VIDEO_CODECS + + def test_unknown_codec_not_present(self): + assert "unknown_codec_xyz" not in SUPPORTED_VIDEO_CODECS + + def test_codecs_count_reasonable(self): + # 白名单应该有足够多的编码格式 + assert len(SUPPORTED_VIDEO_CODECS) >= 30 + + +class TestIsValidMediaVideo: + def test_valid_video(self): + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": "h264"} + assert is_valid_media(metadata, "video") is True + + def test_video_too_small(self): + metadata = {"size_bytes": 500, "duration": 10.0, "codec": "h264"} + assert is_valid_media(metadata, "video") is False + + def test_video_exact_min_size(self): + metadata = {"size_bytes": 1024, "duration": 10.0, "codec": "h264"} + assert is_valid_media(metadata, "video") is True + + def test_video_zero_duration(self): + metadata = {"size_bytes": 5000, "duration": 0, "codec": "h264"} + assert is_valid_media(metadata, "video") is False + + def test_video_negative_duration(self): + metadata = {"size_bytes": 5000, "duration": -1.0, "codec": "h264"} + assert is_valid_media(metadata, "video") is False + + def test_video_missing_size_default_zero(self): + metadata = {"duration": 10.0, "codec": "h264"} + assert is_valid_media(metadata, "video") is False + + def test_video_missing_duration_default_zero(self): + metadata = {"size_bytes": 5000, "codec": "h264"} + assert is_valid_media(metadata, "video") is False + + def test_video_unknown_codec_still_valid(self): + # 非白名单编码仍允许通过（渲染层统一转码） + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": "some_unknown_codec"} + assert is_valid_media(metadata, "video") is True + + def test_video_missing_codec_still_valid(self): + metadata = {"size_bytes": 5000, "duration": 10.0} + assert is_valid_media(metadata, "video") is True + + def test_video_codec_case_insensitive(self): + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": "H264"} + assert is_valid_media(metadata, "video") is True + + def test_video_empty_codec(self): + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": ""} + assert is_valid_media(metadata, "video") is True + + def test_video_hevc_codec(self): + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": "hevc"} + assert is_valid_media(metadata, "video") is True + + def test_video_vp9_codec(self): + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": "vp9"} + assert is_valid_media(metadata, "video") is True + + def test_video_av1_codec(self): + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": "av1"} + assert is_valid_media(metadata, "video") is True + + def test_video_prores_codec(self): + metadata = {"size_bytes": 5000, "duration": 10.0, "codec": "prores"} + assert is_valid_media(metadata, "video") is True + + def test_video_empty_metadata(self): + assert is_valid_media({}, "video") is False + + +class TestIsValidMediaAudio: + def test_valid_audio(self): + metadata = {"size_bytes": 5000, "duration": 30.0} + assert is_valid_media(metadata, "audio") is True + + def test_audio_too_small(self): + metadata = {"size_bytes": 50, "duration": 30.0} + assert is_valid_media(metadata, "audio") is False + + def test_audio_exact_min_size(self): + metadata = {"size_bytes": 100, "duration": 10.0} + assert is_valid_media(metadata, "audio") is True + + def test_audio_zero_duration(self): + metadata = {"size_bytes": 5000, "duration": 0} + assert is_valid_media(metadata, "audio") is False + + def test_audio_negative_duration(self): + metadata = {"size_bytes": 5000, "duration": -1.0} + assert is_valid_media(metadata, "audio") is False + + def test_audio_missing_size(self): + metadata = {"duration": 10.0} + assert is_valid_media(metadata, "audio") is False + + def test_audio_missing_duration(self): + metadata = {"size_bytes": 5000} + assert is_valid_media(metadata, "audio") is False + + def test_audio_with_codec_info(self): + metadata = {"size_bytes": 5000, "duration": 30.0, "codec": "aac"} + assert is_valid_media(metadata, "audio") is True + + def test_audio_empty_metadata(self): + assert is_valid_media({}, "audio") is False + + +class TestIsValidMediaImage: + def test_valid_image(self): + metadata = {"size_bytes": 5000, "width": 1920, "height": 1080} + assert is_valid_media(metadata, "image") is True + + def test_image_too_small(self): + metadata = {"size_bytes": 50, "width": 1920, "height": 1080} + assert is_valid_media(metadata, "image") is False + + def test_image_exact_min_size(self): + metadata = {"size_bytes": 100, "width": 100, "height": 100} + assert is_valid_media(metadata, "image") is True + + def test_image_zero_width(self): + metadata = {"size_bytes": 5000, "width": 0, "height": 1080} + assert is_valid_media(metadata, "image") is False + + def test_image_zero_height(self): + metadata = {"size_bytes": 5000, "width": 1920, "height": 0} + assert is_valid_media(metadata, "image") is False + + def test_image_negative_dimensions(self): + metadata = {"size_bytes": 5000, "width": -1, "height": 1080} + assert is_valid_media(metadata, "image") is False + + def test_image_missing_width(self): + metadata = {"size_bytes": 5000, "height": 1080} + assert is_valid_media(metadata, "image") is False + + def test_image_missing_height(self): + metadata = {"size_bytes": 5000, "width": 1920} + assert is_valid_media(metadata, "image") is False + + def test_image_missing_size(self): + metadata = {"width": 1920, "height": 1080} + assert is_valid_media(metadata, "image") is False + + def test_image_small_but_valid(self): + metadata = {"size_bytes": 100, "width": 1, "height": 1} + assert is_valid_media(metadata, "image") is True + + def test_image_empty_metadata(self): + assert is_valid_media({}, "image") is False + + +class TestIsValidMediaUnknownType: + def test_unknown_type_returns_false(self): + metadata = {"size_bytes": 5000, "duration": 10.0} + assert is_valid_media(metadata, "unknown") is False + + def test_empty_type_returns_false(self): + metadata = {"size_bytes": 5000, "duration": 10.0} + assert is_valid_media(metadata, "") is False + + def test_text_type_returns_false(self): + metadata = {"size_bytes": 5000} + assert is_valid_media(metadata, "text") is False + + +class TestIsValidMediaSizeTypes: + def test_size_as_string(self): + # int("5000") 能解析 + metadata = {"size_bytes": "5000", "duration": 10.0, "codec": "h264"} + assert is_valid_media(metadata, "video") is True + + def test_size_as_none(self): + # int(None) 会 TypeError，但 metadata.get 返回 0 默认值 + metadata = {"size_bytes": None, "duration": 10.0, "codec": "h264"} + # int(None) 会抛 TypeError + with pytest.raises(TypeError): + is_valid_media(metadata, "video") diff --git a/tests/unit/test_preset_bgm.py b/tests/unit/test_preset_bgm.py new file mode 100755 index 000000000..2120fcf7d --- /dev/null +++ b/tests/unit/test_preset_bgm.py @@ -0,0 +1,201 @@ +"""Preset BGM 预设背景音乐单元测试。""" + +import pytest + +from packages.domain.preset_bgm import ( + BGM_STYLES, + PRESET_BGM_LIBRARY, + PresetBGM, + get_preset_bgm, + list_preset_bgm_by_style, + search_preset_bgm, +) + + +class TestPresetBGMDataclass: + def test_creation_required_fields(self): + bgm = PresetBGM(id="test_001", name="Test BGM", style="upbeat", duration=120.0) + assert bgm.id == "test_001" + assert bgm.name == "Test BGM" + assert bgm.style == "upbeat" + assert bgm.duration == 120.0 + assert bgm.artist == "" + assert bgm.description == "" + assert bgm.tags == [] + assert bgm.audio_url == "" + + def test_creation_all_fields(self): + bgm = PresetBGM( + id="test_002", + name="Full BGM", + style="relax", + duration=180.5, + artist="Artist Name", + description="A test description", + tags=["tag1", "tag2"], + audio_url="https://cdn/test.mp3", + ) + assert bgm.artist == "Artist Name" + assert bgm.description == "A test description" + assert bgm.tags == ["tag1", "tag2"] + assert bgm.audio_url == "https://cdn/test.mp3" + + def test_frozen_immutable(self): + bgm = PresetBGM(id="t1", name="T", style="upbeat", duration=60.0) + with pytest.raises(Exception): # FrozenInstanceError + bgm.name = "new name" + + def test_equality(self): + bgm1 = PresetBGM(id="same", name="N", style="upbeat", duration=60.0) + bgm2 = PresetBGM(id="same", name="N", style="upbeat", duration=60.0) + assert bgm1 == bgm2 + + def test_inequality(self): + bgm1 = PresetBGM(id="a", name="A", style="upbeat", duration=60.0) + bgm2 = PresetBGM(id="b", name="B", style="upbeat", duration=60.0) + assert bgm1 != bgm2 + + def test_frozen_with_list_field_not_hashable(self): + # 包含 list 字段的 frozen dataclass 仍然不可哈希（list 不可哈希） + bgm = PresetBGM(id="h1", name="H", style="upbeat", duration=60.0, tags=["a"]) + with pytest.raises(TypeError, match="unhashable"): + hash(bgm) + + +class TestPresetBGMLibrary: + def test_library_not_empty(self): + assert len(PRESET_BGM_LIBRARY) > 0 + + def test_library_has_entries(self): + assert len(PRESET_BGM_LIBRARY) >= 10 + + def test_all_have_unique_ids(self): + ids = [bgm.id for bgm in PRESET_BGM_LIBRARY] + assert len(ids) == len(set(ids)) + + def test_all_have_valid_styles(self): + for bgm in PRESET_BGM_LIBRARY: + assert bgm.style in BGM_STYLES + + def test_all_have_positive_duration(self): + for bgm in PRESET_BGM_LIBRARY: + assert bgm.duration > 0 + + def test_all_have_non_empty_name(self): + for bgm in PRESET_BGM_LIBRARY: + assert bgm.name.strip() != "" + + +class TestBGMStyles: + def test_styles_dict_keys(self): + assert "upbeat" in BGM_STYLES + assert "relax" in BGM_STYLES + assert "tech" in BGM_STYLES + assert "commerce" in BGM_STYLES + assert "emotional" in BGM_STYLES + assert "cinematic" in BGM_STYLES + + def test_styles_have_chinese_names(self): + for key, value in BGM_STYLES.items(): + assert isinstance(value, str) + assert len(value) > 0 + + +class TestGetPresetBGM: + def test_get_existing(self): + bgm = get_preset_bgm("bgm_upbeat_001") + assert bgm is not None + assert bgm.id == "bgm_upbeat_001" + assert bgm.name == "阳光清晨" + assert bgm.style == "upbeat" + + def test_get_nonexistent(self): + assert get_preset_bgm("nonexistent_id") is None + + def test_get_empty_string(self): + assert get_preset_bgm("") is None + + def test_get_returns_same_object(self): + bgm1 = get_preset_bgm("bgm_relax_001") + bgm2 = get_preset_bgm("bgm_relax_001") + assert bgm1 is bgm2 # 同一实例（引用同一列表中的对象） + + +class TestListPresetBGMByStyle: + def test_list_upbeat(self): + results = list_preset_bgm_by_style("upbeat") + assert len(results) >= 3 + for bgm in results: + assert bgm.style == "upbeat" + + def test_list_relax(self): + results = list_preset_bgm_by_style("relax") + assert len(results) >= 3 + for bgm in results: + assert bgm.style == "relax" + + def test_list_tech(self): + results = list_preset_bgm_by_style("tech") + assert len(results) >= 2 + for bgm in results: + assert bgm.style == "tech" + + def test_list_commerce(self): + results = list_preset_bgm_by_style("commerce") + assert len(results) >= 2 + for bgm in results: + assert bgm.style == "commerce" + + def test_list_empty_style(self): + results = list_preset_bgm_by_style("nonexistent_style") + assert results == [] + + def test_list_preserves_order(self): + results = list_preset_bgm_by_style("upbeat") + ids = [b.id for b in results] + # 应该按照在列表中的出现顺序排列 + assert ids == sorted(ids, key=lambda x: PRESET_BGM_LIBRARY.index(get_preset_bgm(x))) + + +class TestSearchPresetBGM: + def test_search_by_name(self): + results = search_preset_bgm("阳光") + assert len(results) >= 1 + assert any("阳光" in b.name for b in results) + + def test_search_by_description(self): + results = search_preset_bgm("钢琴") + assert len(results) >= 1 + # 钢琴出现在名称或描述或标签中 + found = False + for b in results: + if "钢琴" in b.description or "钢琴" in b.name or "钢琴" in b.tags: + found = True + break + assert found + + def test_search_by_tag(self): + results = search_preset_bgm("科技") + assert len(results) >= 1 + found_tech = any(b.style == "tech" for b in results) + assert found_tech + + def test_search_case_insensitive(self): + results1 = search_preset_bgm("Tech") + results2 = search_preset_bgm("tech") + assert len(results1) == len(results2) + + def test_search_no_match(self): + results = search_preset_bgm("zzzzzzzzzzz_nonexistent_keyword") + assert results == [] + + def test_search_empty_keyword(self): + # 空字符串应该匹配所有（因为空字符串 in 任何字符串都是 True） + results = search_preset_bgm("") + assert len(results) == len(PRESET_BGM_LIBRARY) + + def test_search_no_duplicates(self): + # 确保同一个 BGM 不会出现多次 + results = search_preset_bgm("电子") + ids = [b.id for b in results] + assert len(ids) == len(set(ids)) diff --git a/tests/unit/test_quota_domain.py b/tests/unit/test_quota_domain.py index c00023f17..3d11e1ec7 100755 --- a/tests/unit/test_quota_domain.py +++ b/tests/unit/test_quota_domain.py @@ -1,13 +1,11 @@ -"""Quota 领域层单元测试 - quota.py""" - -import math +"""Quota 配额系统单元测试。""" import pytest from packages.domain.quota import ( QUOTA_TIERS, - QuotaChecker, QuotaCheckResult, + QuotaChecker, QuotaDimension, QuotaRegistry, QuotaTier, @@ -19,103 +17,114 @@ from packages.domain.quota import ( class TestQuotaDimension: - """QuotaDimension 枚举测试""" + def test_core_dimensions_exist(self): + assert QuotaDimension.STORAGE_GB.value == "storage_gb" + assert QuotaDimension.VIDEOS_PER_MONTH.value == "videos_per_month" + assert QuotaDimension.MAX_CONCURRENT.value == "max_concurrent" + assert QuotaDimension.MAX_TEMPLATES.value == "max_templates" + assert QuotaDimension.MAX_TITLES.value == "max_titles" + assert QuotaDimension.MAX_VOICEOVERS.value == "max_voiceovers" + assert QuotaDimension.AI_VOICE_ENABLED.value == "ai_voice_enabled" - def test_all_dimensions_have_values(self): - """所有枚举成员都有字符串值""" + def test_extended_dimensions_exist(self): + assert QuotaDimension.AI_VOICE_CREDITS.value == "ai_voice_credits" + assert QuotaDimension.BATCH_EXPORT_ENABLED.value == "batch_export_enabled" + assert QuotaDimension.MULTI_PLATFORM_ENABLED.value == "multi_platform_enabled" + assert QuotaDimension.DEDUP_REPORT_ENABLED.value == "dedup_report_enabled" + + def test_all_dimensions_are_strings(self): for dim in QuotaDimension: assert isinstance(dim.value, str) - assert dim.value - - def test_dimension_count(self): - """配额维度数量 >= 内置维度""" - # 至少有 storage_gb, videos_per_month, max_concurrent, max_templates 等 - assert len(QuotaDimension) >= 7 - - def test_str_enum_behavior(self): - """是 str 枚举，可直接当字符串用""" - assert QuotaDimension.STORAGE_GB == "storage_gb" - assert isinstance(QuotaDimension.STORAGE_GB, str) class TestQuotaTier: - """QuotaTier 测试""" - def test_get_limit_defined(self): - """已定义的维度返回正确值""" - tier = QuotaTier(name="test", limits={"storage": 10, "videos": 5}) - assert tier.get_limit("storage") == 10 + tier = QuotaTier(name="test", limits={"storage_gb": 10, "videos": 5}) + assert tier.get_limit("storage_gb") == 10 assert tier.get_limit("videos") == 5 def test_get_limit_undefined_returns_zero(self): - """未定义的维度返回 0""" - tier = QuotaTier(name="test", limits={"storage": 10}) - assert tier.get_limit("unknown") == 0 + tier = QuotaTier(name="test", limits={"storage_gb": 10}) + assert tier.get_limit("unknown_dim") == 0 - def test_is_unlimited_true(self): - """不限量判断 - inf""" + def test_is_unlimited_false_for_finite(self): + tier = QuotaTier(name="test", limits={"storage_gb": 10}) + assert tier.is_unlimited("storage_gb") is False + + def test_is_unlimited_true_for_inf(self): tier = QuotaTier(name="test", limits={"templates": float("inf")}) assert tier.is_unlimited("templates") is True - def test_is_unlimited_false(self): - """限量判断""" - tier = QuotaTier(name="test", limits={"storage": 10}) - assert tier.is_unlimited("storage") is False - - def test_is_unlimited_undefined_returns_true(self): - """未定义的维度默认 inf，is_unlimited 返回 True""" + def test_is_unlimited_undefined(self): tier = QuotaTier(name="test", limits={}) - # get_limit 用 dict.get 默认 0，但 is_unlimited 用 dict.get 默认 inf + # 未定义的维度，limits.get 返回默认 inf，所以 is_unlimited 返回 True assert tier.is_unlimited("unknown") is True + def test_empty_limits(self): + tier = QuotaTier(name="empty") + assert tier.limits == {} + assert tier.name == "empty" + class TestQuotaTiers: - """内置套餐配额测试""" - def test_three_tiers_exist(self): - """三个套餐等级都存在""" assert "free" in QUOTA_TIERS assert "basic" in QUOTA_TIERS assert "premium" in QUOTA_TIERS - def test_free_tier_storage(self): - """free 套餐 2GB 存储""" - assert QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB) == 2 + def test_free_tier_limits(self): + free = QUOTA_TIERS["free"] + assert free.get_limit("storage_gb") == 2 + assert free.get_limit("videos_per_month") == 5 + assert free.get_limit("max_concurrent") == 3 + assert free.get_limit("max_templates") == 3 + assert free.get_limit("max_titles") == 50 + assert free.get_limit("max_voiceovers") == 10 + assert free.get_limit("ai_voice_enabled") == 0 - def test_basic_tier_storage(self): - """basic 套餐 20GB 存储""" - assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB) == 20 + def test_basic_tier_limits(self): + basic = QUOTA_TIERS["basic"] + assert basic.get_limit("storage_gb") == 20 + assert basic.get_limit("videos_per_month") == 30 + assert basic.get_limit("max_concurrent") == 10 + assert basic.get_limit("max_templates") == 15 + assert basic.get_limit("max_titles") == 500 + assert basic.get_limit("max_voiceovers") == 100 + assert basic.get_limit("ai_voice_enabled") == 1 + assert basic.get_limit("ai_voice_credits") == 100 + assert basic.get_limit("batch_export_enabled") == 1 - def test_premium_tier_storage(self): - """premium 套餐 100GB 存储""" - assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB) == 100 + def test_premium_tier_limits(self): + premium = QUOTA_TIERS["premium"] + assert premium.get_limit("storage_gb") == 100 + assert premium.get_limit("videos_per_month") == 100 + assert premium.get_limit("max_concurrent") == 20 + assert premium.is_unlimited("max_templates") is True + assert premium.get_limit("ai_voice_enabled") == 1 + assert premium.get_limit("ai_voice_credits") == 500 + assert premium.get_limit("batch_export_enabled") == 1 + assert premium.get_limit("multi_platform_enabled") == 1 + assert premium.get_limit("dedup_report_enabled") == 1 - def test_free_no_ai_voice(self): - """free 套餐没有 AI 配音""" - assert QUOTA_TIERS["free"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 0 - - def test_basic_has_ai_voice(self): - """basic 套餐有 AI 配音""" - assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 1 - - def test_premium_templates_unlimited(self): - """premium 套餐模板不限量""" - assert QUOTA_TIERS["premium"].is_unlimited(QuotaDimension.MAX_TEMPLATES) is True - - def test_free_videos_per_month(self): - """free 每月 5 个视频""" - assert QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 5 - - def test_premium_multi_platform_enabled(self): - """premium 支持多平台发布""" - assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.MULTI_PLATFORM_ENABLED) == 1 + def test_tier_increase_monotonic(self): + free = QUOTA_TIERS["free"] + basic = QUOTA_TIERS["basic"] + premium = QUOTA_TIERS["premium"] + # 高级套餐应该 >= 低级套餐的所有限制 + for dim in [ + "storage_gb", + "videos_per_month", + "max_concurrent", + "max_titles", + "max_voiceovers", + "ai_voice_credits", + ]: + assert basic.get_limit(dim) >= free.get_limit(dim) + assert premium.get_limit(dim) >= basic.get_limit(dim) class TestQuotaWarningLevel: - """告警级别常量测试""" - - def test_level_values(self): - """四个告警级别都有定义""" + def test_levels_exist(self): assert QuotaWarningLevel.NORMAL == "normal" assert QuotaWarningLevel.WARNING == "warning" assert QuotaWarningLevel.CRITICAL == "critical" @@ -123,302 +132,231 @@ class TestQuotaWarningLevel: class TestQuotaCheckResult: - """QuotaCheckResult 测试""" - def test_usage_percent_normal(self): - """正常使用百分比计算""" result = QuotaCheckResult( allowed=True, - dimension="storage", + dimension="storage_gb", limit=100, - used=30, - remaining=70, - warning_level=QuotaWarningLevel.NORMAL, + used=50, + remaining=50, + warning_level="normal", ) - assert result.usage_percent == 30.0 + assert result.usage_percent == 50.0 - def test_usage_percent_capped_at_100(self): - """超过 100% 时截断为 100%""" + def test_usage_percent_exceeded(self): result = QuotaCheckResult( - allowed=False, - dimension="storage", - limit=100, - used=150, - remaining=0, - warning_level=QuotaWarningLevel.EXCEEDED, + allowed=False, dimension="d", limit=100, used=150, remaining=0, warning_level="exceeded" ) - assert result.usage_percent == 100.0 + assert result.usage_percent == 100.0 # min(100, 150%) + + def test_usage_percent_zero_used(self): + result = QuotaCheckResult(allowed=True, dimension="d", limit=100, used=0, remaining=100, warning_level="normal") + assert result.usage_percent == 0.0 def test_usage_percent_zero_limit_with_usage(self): - """limit=0 但有使用量，返回 100%""" - result = QuotaCheckResult( - allowed=False, - dimension="storage", - limit=0, - used=5, - remaining=0, - warning_level=QuotaWarningLevel.EXCEEDED, - ) + result = QuotaCheckResult(allowed=False, dimension="d", limit=0, used=10, remaining=0, warning_level="exceeded") assert result.usage_percent == 100.0 def test_usage_percent_zero_limit_no_usage(self): - """limit=0 且无使用量，返回 0%""" - result = QuotaCheckResult( - allowed=True, - dimension="storage", - limit=0, - used=0, - remaining=0, - warning_level=QuotaWarningLevel.NORMAL, - ) + result = QuotaCheckResult(allowed=True, dimension="d", limit=0, used=0, remaining=0, warning_level="normal") assert result.usage_percent == 0.0 def test_usage_percent_unlimited(self): - """不限量时使用百分比为 0""" result = QuotaCheckResult( allowed=True, - dimension="templates", + dimension="d", limit=float("inf"), - used=50, + used=1000, remaining=float("inf"), - warning_level=QuotaWarningLevel.NORMAL, + warning_level="normal", ) assert result.usage_percent == 0.0 class TestQuotaRegistry: - """QuotaRegistry 测试""" - def test_initial_dimensions(self): - """初始化时内置维度已注册""" - registry = QuotaRegistry() - dims = registry.list_dimensions() - assert QuotaDimension.STORAGE_GB in dims - assert QuotaDimension.VIDEOS_PER_MONTH in dims + reg = QuotaRegistry() + dims = reg.list_dimensions() + assert "storage_gb" in dims + assert "videos_per_month" in dims + assert len(dims) == len(QuotaDimension) - def test_initial_tiers(self): - """初始化时三个套餐已注册""" - registry = QuotaRegistry() - tiers = registry.list_tiers() + def test_list_tiers(self): + reg = QuotaRegistry() + tiers = reg.list_tiers() assert "free" in tiers assert "basic" in tiers assert "premium" in tiers - - def test_register_new_dimension(self): - """注册新的配额维度""" - registry = QuotaRegistry() - registry.register_dimension("custom_dim", "自定义维度") - dims = registry.list_dimensions() - assert "custom_dim" in dims - assert dims["custom_dim"] == "自定义维度" - - def test_register_dimension_idempotent(self): - """重复注册是幂等的""" - registry = QuotaRegistry() - registry.register_dimension("custom", "描述1") - registry.register_dimension("custom", "描述2") - # 保留第一次注册的描述 - assert registry.list_dimensions()["custom"] == "描述1" - - def test_register_with_default_limits(self): - """注册时指定各套餐的默认限制""" - registry = QuotaRegistry() - registry.register_dimension( - "custom", - "自定义", - default_limits={"free": 1, "basic": 10, "premium": 100}, - ) - assert registry.get_limit("free", "custom") == 1 - assert registry.get_limit("basic", "custom") == 10 - assert registry.get_limit("premium", "custom") == 100 - - def test_register_without_default_limits_defaults_to_zero(self): - """不指定默认限制时各套餐该维度为 0""" - registry = QuotaRegistry() - registry.register_dimension("custom_no_limit", "自定义") - assert registry.get_limit("free", "custom_no_limit") == 0 - assert registry.get_limit("basic", "custom_no_limit") == 0 - - def test_register_default_limits_ignores_unknown_plan(self): - """默认限制中未知的套餐名被忽略""" - registry = QuotaRegistry() - registry.register_dimension( - "custom", - "自定义", - default_limits={"nonexistent": 999}, - ) - # 不报错，但也不会创建新套餐 - assert registry.get_tier("nonexistent") is None + assert len(tiers) == 3 def test_get_tier_existing(self): - """获取存在的套餐""" - registry = QuotaRegistry() - tier = registry.get_tier("free") + reg = QuotaRegistry() + tier = reg.get_tier("free") assert tier is not None assert tier.name == "free" - def test_get_tier_nonexistent(self): - """获取不存在的套餐返回 None""" - registry = QuotaRegistry() - assert registry.get_tier("enterprise") is None + def test_get_tier_unknown(self): + reg = QuotaRegistry() + assert reg.get_tier("unknown_plan") is None - def test_get_limit_existing(self): - """获取存在的套餐和维度的限制""" - registry = QuotaRegistry() - assert registry.get_limit("free", QuotaDimension.STORAGE_GB) == 2 + def test_get_limit_known(self): + reg = QuotaRegistry() + assert reg.get_limit("free", "storage_gb") == 2 + assert reg.get_limit("premium", "storage_gb") == 100 - def test_get_limit_nonexistent_plan(self): - """不存在的套餐返回 0""" - registry = QuotaRegistry() - assert registry.get_limit("unknown", QuotaDimension.STORAGE_GB) == 0 + def test_get_limit_unknown_plan(self): + reg = QuotaRegistry() + assert reg.get_limit("unknown", "storage_gb") == 0 - def test_list_dimensions_returns_copy(self): - """list_dimensions 返回副本，修改不影响内部""" - registry = QuotaRegistry() - dims = registry.list_dimensions() - dims["fake"] = "fake" - assert "fake" not in registry.list_dimensions() + def test_register_new_dimension(self): + reg = QuotaRegistry() + reg.register_dimension("new_feature", "新功能", default_limits={"free": 0, "basic": 1, "premium": 5}) + dims = reg.list_dimensions() + assert "new_feature" in dims + assert dims["new_feature"] == "新功能" + assert reg.get_limit("free", "new_feature") == 0 + assert reg.get_limit("basic", "new_feature") == 1 + assert reg.get_limit("premium", "new_feature") == 5 - def test_list_tiers_returns_all_three(self): - """列出所有套餐""" - registry = QuotaRegistry() - tiers = registry.list_tiers() - assert len(tiers) == 3 - assert set(tiers) == {"free", "basic", "premium"} + def test_register_dimension_idempotent(self): + reg = QuotaRegistry() + reg.register_dimension("storage_gb", "should not change", default_limits={"free": 999}) + # 已经存在的不覆盖 + assert reg.get_limit("free", "storage_gb") == 2 + + def test_register_without_defaults(self): + reg = QuotaRegistry() + reg.register_dimension("new_dim", "描述") + assert reg.get_limit("free", "new_dim") == 0 + assert reg.get_limit("basic", "new_dim") == 0 + assert reg.get_limit("premium", "new_dim") == 0 + + def test_register_partial_limits(self): + reg = QuotaRegistry() + reg.register_dimension("partial", "partial", default_limits={"premium": 42}) + assert reg.get_limit("free", "partial") == 0 # 未设置的保持 0 + assert reg.get_limit("premium", "partial") == 42 class TestQuotaChecker: - """QuotaChecker 测试""" - - def test_check_under_limit_allowed(self): - """使用量低于限制，允许""" + def test_check_within_limit(self): checker = QuotaChecker() - result = checker.check("free", QuotaDimension.STORAGE_GB, 1.0) + result = checker.check("free", "storage_gb", 1) assert result.allowed is True - assert result.remaining == 1.0 - assert result.warning_level == QuotaWarningLevel.NORMAL + assert result.limit == 2 + assert result.used == 1 + assert result.remaining == 1 + assert result.dimension == "storage_gb" - def test_check_at_limit_not_allowed(self): - """使用量等于限制，不允许（used < limit 判定）""" + def test_check_exceeded(self): checker = QuotaChecker() - result = checker.check("free", QuotaDimension.STORAGE_GB, 2.0) + result = checker.check("free", "storage_gb", 3) assert result.allowed is False assert result.remaining == 0 - assert result.warning_level == QuotaWarningLevel.EXCEEDED + assert result.warning_level == "exceeded" - def test_check_over_limit(self): - """使用量超过限制""" + def test_check_exact_limit_not_allowed(self): + # used < limit 才 allowed，等于不算 checker = QuotaChecker() - result = checker.check("free", QuotaDimension.STORAGE_GB, 3.0) + result = checker.check("free", "storage_gb", 2) assert result.allowed is False assert result.remaining == 0 - assert result.warning_level == QuotaWarningLevel.EXCEEDED - def test_check_warning_level_80_percent(self): - """80% 触发 WARNING""" + def test_check_unlimited(self): checker = QuotaChecker() - # 100GB 的 80% = 80GB - result = checker.check("premium", QuotaDimension.STORAGE_GB, 80.0) - assert result.warning_level == QuotaWarningLevel.WARNING + result = checker.check("premium", "max_templates", 999999) + assert result.allowed is True + assert result.remaining == float("inf") + assert result.warning_level == "normal" - def test_check_warning_level_95_percent(self): - """95% 触发 CRITICAL""" + def test_check_warning_level_normal(self): checker = QuotaChecker() - result = checker.check("premium", QuotaDimension.STORAGE_GB, 95.0) - assert result.warning_level == QuotaWarningLevel.CRITICAL + result = checker.check("free", "storage_gb", 1) # 50% + assert result.warning_level == "normal" + + def test_check_warning_level_warning(self): + checker = QuotaChecker() + # 80% <= used < 95% + result = checker.check("free", "max_templates", 2.5) # 2.5/3 = 83% + assert result.warning_level == "warning" + + def test_check_warning_level_critical(self): + checker = QuotaChecker() + # 95% <= used < 100% + result = checker.check("free", "max_templates", 2.9) # 2.9/3 = 97% + assert result.warning_level == "critical" def test_check_warning_level_exceeded(self): - """100% 及以上触发 EXCEEDED""" checker = QuotaChecker() - result = checker.check("premium", QuotaDimension.STORAGE_GB, 100.0) - assert result.warning_level == QuotaWarningLevel.EXCEEDED - - def test_check_unlimited_always_allowed(self): - """不限量的维度始终允许""" - checker = QuotaChecker() - result = checker.check("premium", QuotaDimension.MAX_TEMPLATES, 9999) - assert result.allowed is True - assert math.isinf(result.remaining) - assert result.warning_level == QuotaWarningLevel.NORMAL - - def test_check_unknown_plan_zero_limit(self): - """未知套餐限制为 0，used=0 时不允许（0 < 0 为 False）""" - checker = QuotaChecker() - result = checker.check("unknown", QuotaDimension.STORAGE_GB, 0) - assert result.limit == 0 - assert result.allowed is False + result = checker.check("free", "storage_gb", 5) # 250% + assert result.warning_level == "exceeded" def test_check_multiple(self): - """批量检查多个维度""" checker = QuotaChecker() results = checker.check_multiple( "free", - { - QuotaDimension.STORAGE_GB: 1.0, - QuotaDimension.VIDEOS_PER_MONTH: 3, - }, + {"storage_gb": 1, "max_templates": 2, "max_titles": 10}, ) - assert len(results) == 2 + assert len(results) == 3 + assert results[0].dimension == "storage_gb" + assert results[1].dimension == "max_templates" + assert results[2].dimension == "max_titles" assert all(r.allowed for r in results) - dims = {r.dimension for r in results} - assert QuotaDimension.STORAGE_GB in dims - assert QuotaDimension.VIDEOS_PER_MONTH in dims - def test_check_with_custom_registry(self): - """使用自定义注册表""" - registry = QuotaRegistry() - registry.register_dimension("custom", "自定义", default_limits={"free": 5}) - checker = QuotaChecker(registry) - result = checker.check("free", "custom", 3) - assert result.allowed is True - assert result.limit == 5 + def test_check_zero_limit(self): + checker = QuotaChecker() + result = checker.check("free", "ai_voice_enabled", 0) + # limit=0, used=0: used < limit 为 False → allowed=False + assert result.allowed is False + assert result.remaining == 0 + assert result.warning_level == "normal" - def test_compute_warning_level_zero_limit_no_usage(self): - """limit=0, used=0 → NORMAL""" - level = QuotaChecker._compute_warning_level(0, 0) - assert level == QuotaWarningLevel.NORMAL + def test_compute_warning_level_normal(self): + assert QuotaChecker._compute_warning_level(50, 100) == "normal" + assert QuotaChecker._compute_warning_level(79, 100) == "normal" + + def test_compute_warning_level_warning_boundary(self): + assert QuotaChecker._compute_warning_level(80, 100) == "warning" + assert QuotaChecker._compute_warning_level(94, 100) == "warning" + + def test_compute_warning_level_critical_boundary(self): + assert QuotaChecker._compute_warning_level(95, 100) == "critical" + assert QuotaChecker._compute_warning_level(99, 100) == "critical" + + def test_compute_warning_level_exceeded(self): + assert QuotaChecker._compute_warning_level(100, 100) == "exceeded" + assert QuotaChecker._compute_warning_level(150, 100) == "exceeded" + + def test_compute_warning_level_unlimited(self): + assert QuotaChecker._compute_warning_level(9999, float("inf")) == "normal" def test_compute_warning_level_zero_limit_with_usage(self): - """limit=0, used>0 → EXCEEDED""" - level = QuotaChecker._compute_warning_level(1, 0) - assert level == QuotaWarningLevel.EXCEEDED + assert QuotaChecker._compute_warning_level(1, 0) == "exceeded" + + def test_compute_warning_level_zero_limit_no_usage(self): + assert QuotaChecker._compute_warning_level(0, 0) == "normal" def test_compute_warning_level_negative_limit(self): - """limit<0 视同 0 处理""" - level = QuotaChecker._compute_warning_level(1, -1) - assert level == QuotaWarningLevel.EXCEEDED + # limit <= 0 且 used=0 → NORMAL + assert QuotaChecker._compute_warning_level(0, -1) == "normal" class TestGetWarningLevel: - """get_warning_level 便捷函数测试""" - - def test_normal(self): - assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL - - def test_warning(self): - assert get_warning_level(85, 100) == QuotaWarningLevel.WARNING - - def test_critical(self): - assert get_warning_level(97, 100) == QuotaWarningLevel.CRITICAL - - def test_exceeded(self): - assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED - - def test_unlimited(self): - assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL + def test_convenience_function(self): + assert get_warning_level(50, 100) == "normal" + assert get_warning_level(99, 100) == "critical" + assert get_warning_level(100, 100) == "exceeded" + assert get_warning_level(0, 0) == "normal" + assert get_warning_level(1, 0) == "exceeded" class TestGlobalSingletons: - """全局单例测试""" - def test_quota_registry_is_instance(self): assert isinstance(quota_registry, QuotaRegistry) def test_quota_checker_is_instance(self): assert isinstance(quota_checker, QuotaChecker) - def test_global_checker_uses_global_registry(self): - """全局 checker 使用全局 registry""" - # 验证能正常工作 - result = quota_checker.check("free", QuotaDimension.STORAGE_GB, 1.0) + def test_global_checker_works(self): + result = quota_checker.check("free", "storage_gb", 1) assert result.allowed is True