From b1d4f466e45f4717dffef1d7708ef86c923b9ab9 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 08:13:44 +0800 Subject: [PATCH 1/2] test: wave215 add 77 unit tests for video_share (domain + use cases) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Domain层 (43测): - 密码哈希 (加盐SHA-256) - Token生成 (去歧义字符、长度、URL友好) - VideoShare.create 校验 (空ID、空白、去重、过期时间校验) - has_password / is_expired / is_accessible 属性 - verify_password 密码验证 - 浏览/下载计数 - revoke 撤销分享 Use case层 (34测): - CreateShareUseCase (视频存在性/所有权校验) - GetShareByTokenUseCase (过期/撤销拦截) - AccessShareUseCase (密码校验、浏览量+1、视频查询) - ListSharesByVideoUseCase / ListSharesByUserUseCase (分页) - UpdateShareUseCase (改密码/清密码/改过期时间) - RevokeShareUseCase - RecordShareDownloadUseCase (密码校验、下载量+1) --- tests/unit/domain/test_video_share.py | 402 +++++++------ tests/unit/test_video_share_use_cases.py | 690 +++++++++++------------ 2 files changed, 530 insertions(+), 562 deletions(-) diff --git a/tests/unit/domain/test_video_share.py b/tests/unit/domain/test_video_share.py index 3105d10c1..40064a6cb 100755 --- a/tests/unit/domain/test_video_share.py +++ b/tests/unit/domain/test_video_share.py @@ -1,58 +1,55 @@ -"""video_share 视频分享领域实体单测.""" +"""视频分享领域模型单元测试 — wave215""" +from __future__ import annotations + +import re from datetime import datetime, timedelta, timezone import pytest -from domain.video_share import ( + +from packages.domain.video_share import ( VideoShare, _hash_password, generate_share_token, ) -# ── _hash_password ─────────────────────────────────────────────────────────── +# ── 密码哈希 ───────────────────────────────────────────────────────────────── class TestHashPassword: - """_hash_password 函数""" - def test_empty_password_returns_empty(self): assert _hash_password("") == "" - def test_none_password_returns_empty(self): - assert _hash_password(None) == "" - def test_same_password_same_hash(self): - h1 = _hash_password("mypassword") - h2 = _hash_password("mypassword") + h1 = _hash_password("secret123") + h2 = _hash_password("secret123") assert h1 == h2 + assert h1 != "" - def test_different_passwords_different_hashes(self): - h1 = _hash_password("password1") - h2 = _hash_password("password2") + def test_different_password_different_hash(self): + h1 = _hash_password("pass1") + h2 = _hash_password("pass2") assert h1 != h2 - def test_hash_is_hex_string(self): + def test_hash_is_sha256_hex(self): h = _hash_password("test") - assert isinstance(h, str) - assert len(h) == 64 # SHA-256 hex - int(h, 16) # 应该能被解析为16进制 + assert len(h) == 64 + assert re.match(r"^[0-9a-f]{64}$", h) def test_hash_contains_salt(self): - # 直接的 SHA-256(password) 应该不等于加盐后的 - from hashlib import sha256 + # 直接SHA-256("test") vs 加盐后的结果应该不同 + import hashlib - raw = sha256("mypass".encode()).hexdigest() - salted = _hash_password("mypass") - assert raw != salted + direct = hashlib.sha256(b"test").hexdigest() + salted = _hash_password("test") + assert direct != salted -# ── generate_share_token ───────────────────────────────────────────────────── +# ── Token 生成 ────────────────────────────────────────────────────────────── class TestGenerateShareToken: - """generate_share_token 函数""" - - def test_default_length(self): + def test_default_length_12(self): token = generate_share_token() assert len(token) == 12 @@ -60,231 +57,228 @@ class TestGenerateShareToken: token = generate_share_token(20) assert len(token) == 20 - def test_short_token(self): - token = generate_share_token(6) - assert len(token) == 6 - - def test_url_friendly_chars(self): + def test_url_friendly_no_ambiguous_chars(self): + # 不应包含容易混淆的字符：i, l, o, I, L, O, 0, 1 token = generate_share_token(100) - # 不应该有容易混淆的字符 i,l,o,0,1 - assert "i" not in token - assert "l" not in token - assert "o" not in token - assert "0" not in token - assert "1" not in token + for ch in "ilO01": + assert ch not in token - def test_unique_tokens(self): - tokens = {generate_share_token() for _ in range(100)} - assert len(tokens) == 100 # 应该都是唯一的 - - def test_alphanumeric(self): + def test_alphanumeric_only(self): token = generate_share_token(50) assert token.isalnum() + def test_two_tokens_different(self): + # 随机生成的两个token应该不同 + t1 = generate_share_token() + t2 = generate_share_token() + assert t1 != t2 + # ── VideoShare.create ─────────────────────────────────────────────────────── class TestVideoShareCreate: - """VideoShare.create 工厂方法""" + def test_basic_create(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.id is not None + assert share.video_id == "v1" + assert share.user_id == "u1" + assert share.share_token is not None + assert len(share.share_token) == 12 + assert share.password_hash is None + assert share.expires_at is None + assert share.view_count == 0 + assert share.download_count == 0 + assert share.is_active is True + assert share.created_at is not None + assert share.updated_at is not None - def test_minimal_create(self): - s = VideoShare.create(video_id="vid_001", user_id="user_001") - assert s.id is not None - assert len(s.id) == 32 # uuid4 hex - assert s.video_id == "vid_001" - assert s.user_id == "user_001" - assert s.share_token is not None - assert len(s.share_token) == 12 - assert s.password_hash is None - assert s.expires_at is None - assert s.view_count == 0 - assert s.download_count == 0 - assert s.is_active is True + def test_create_with_password(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="secret") + assert share.password_hash is not None + assert share.password_hash != "secret" + assert len(share.password_hash) == 64 - def test_with_password(self): - s = VideoShare.create(video_id="v1", user_id="u1", password="secret123") - assert s.password_hash is not None - assert s.password_hash != "secret123" # 不是明文 - assert len(s.password_hash) == 64 # SHA-256 + def test_create_with_empty_password_no_hash(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="") + assert share.password_hash is None - def test_with_expiry(self): + def test_create_with_expires_at(self): future = datetime.now(timezone.utc) + timedelta(days=7) - s = VideoShare.create(video_id="v1", user_id="u1", expires_at=future) - assert s.expires_at == future + share = VideoShare.create(video_id="v1", user_id="u1", expires_at=future) + assert share.expires_at == future - def test_empty_video_id_raises(self): - with pytest.raises(ValueError, match="video_id"): - VideoShare.create(video_id="", user_id="u1") - - def test_whitespace_video_id_raises(self): - with pytest.raises(ValueError): - VideoShare.create(video_id=" ", user_id="u1") - - def test_empty_user_id_raises(self): - with pytest.raises(ValueError, match="user_id"): - VideoShare.create(video_id="v1", user_id="") - - def test_past_expiry_raises(self): - past = datetime.now(timezone.utc) - timedelta(hours=1) - with pytest.raises(ValueError, match="past"): + def test_create_past_expires_at_raises(self): + past = datetime.now(timezone.utc) - timedelta(days=1) + with pytest.raises(ValueError, match="expires_at cannot be in the past"): VideoShare.create(video_id="v1", user_id="u1", expires_at=past) - def test_video_id_stripped(self): - s = VideoShare.create(video_id=" vid_123 ", user_id="u1") - assert s.video_id == "vid_123" + def test_create_empty_video_id_raises(self): + with pytest.raises(ValueError, match="video_id cannot be empty"): + VideoShare.create(video_id="", user_id="u1") - def test_user_id_stripped(self): - s = VideoShare.create(video_id="v1", user_id=" user_456 ") - assert s.user_id == "user_456" + def test_create_whitespace_video_id_raises(self): + with pytest.raises(ValueError, match="video_id cannot be empty"): + VideoShare.create(video_id=" ", user_id="u1") - def test_unique_ids(self): + def test_create_empty_user_id_raises(self): + with pytest.raises(ValueError, match="user_id cannot be empty"): + VideoShare.create(video_id="v1", user_id="") + + def test_create_strips_whitespace(self): + share = VideoShare.create(video_id=" v1 ", user_id=" u1 ") + assert share.video_id == "v1" + assert share.user_id == "u1" + + def test_create_unique_id_each_time(self): s1 = VideoShare.create(video_id="v1", user_id="u1") s2 = VideoShare.create(video_id="v1", user_id="u1") assert s1.id != s2.id - def test_unique_tokens(self): + def test_create_unique_token_each_time(self): s1 = VideoShare.create(video_id="v1", user_id="u1") s2 = VideoShare.create(video_id="v1", user_id="u1") assert s1.share_token != s2.share_token - def test_timestamps_set(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.created_at.tzinfo is not None - assert s.updated_at.tzinfo is not None + +# ── has_password ──────────────────────────────────────────────────────────── -# ── VideoShare 属性方法 ───────────────────────────────────────────────────── +class TestVideoShareHasPassword: + def test_no_password(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.has_password is False + + def test_with_password(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="pass") + assert share.has_password is True + + def test_empty_password_none(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="") + assert share.has_password is False -class TestVideoShareProperties: - """VideoShare 属性方法""" +# ── is_expired ────────────────────────────────────────────────────────────── - def test_has_password_true(self): - s = VideoShare.create(video_id="v1", user_id="u1", password="pass") - assert s.has_password is True - def test_has_password_false(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.has_password is False +class TestVideoShareIsExpired: + def test_no_expiry_never_expired(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.is_expired is False - def test_is_expired_false_no_expiry(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.is_expired is False - - def test_is_expired_false_future_expiry(self): + def test_future_expiry_not_expired(self): future = datetime.now(timezone.utc) + timedelta(hours=1) - s = VideoShare.create(video_id="v1", user_id="u1", expires_at=future) - assert s.is_expired is False + share = VideoShare.create(video_id="v1", user_id="u1", expires_at=future) + assert share.is_expired is False - def test_is_expired_true_past_expiry(self): - # 直接构造一个已过期的 - past = datetime.now(timezone.utc) - timedelta(hours=1) - s = VideoShare( - id="test", - video_id="v1", - user_id="u1", - share_token="abc", - expires_at=past, - ) - assert s.is_expired is True - - def test_is_accessible_true(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.is_accessible is True - - def test_is_accessible_false_inactive(self): - s = VideoShare.create(video_id="v1", user_id="u1") - s.is_active = False - assert s.is_accessible is False - - def test_is_accessible_false_expired(self): - past = datetime.now(timezone.utc) - timedelta(hours=1) - s = VideoShare( - id="test", - video_id="v1", - user_id="u1", - share_token="abc", - expires_at=past, - ) - assert s.is_accessible is False + def test_past_expiry_is_expired(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1) + assert share.is_expired is True -# ── VideoShare 方法 ───────────────────────────────────────────────────────── +# ── is_accessible ─────────────────────────────────────────────────────────── -class TestVideoShareMethods: - """VideoShare 方法""" +class TestVideoShareIsAccessible: + def test_active_no_expiry_accessible(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.is_accessible is True - def test_verify_password_no_password_true(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.verify_password("anything") is True - assert s.verify_password("") is True + def test_revoked_not_accessible(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.is_active = False + assert share.is_accessible is False - def test_verify_password_correct(self): - s = VideoShare.create(video_id="v1", user_id="u1", password="mypass") - assert s.verify_password("mypass") is True + def test_expired_not_accessible(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + assert share.is_accessible is False - def test_verify_password_wrong(self): - s = VideoShare.create(video_id="v1", user_id="u1", password="mypass") - assert s.verify_password("wrongpass") is False + def test_revoked_and_expired_not_accessible(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.is_active = False + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + assert share.is_accessible is False - def test_verify_password_empty_false(self): - s = VideoShare.create(video_id="v1", user_id="u1", password="mypass") - assert s.verify_password("") is False - def test_increment_view_count(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.view_count == 0 - s.increment_view_count() - assert s.view_count == 1 - s.increment_view_count() - assert s.view_count == 2 +# ── verify_password ───────────────────────────────────────────────────────── - def test_increment_download_count(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.download_count == 0 - s.increment_download_count() - assert s.download_count == 1 - s.increment_download_count() - assert s.download_count == 2 - def test_revoke(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.is_active is True - s.revoke() - assert s.is_active is False +class TestVideoShareVerifyPassword: + def test_no_password_any_pass_ok(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.verify_password("anything") is True + assert share.verify_password("") is True + + def test_no_password_none_ok(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.verify_password("") is True + + def test_correct_password(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="mysecret") + assert share.verify_password("mysecret") is True + + def test_wrong_password(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="mysecret") + assert share.verify_password("wrong") is False + + def test_empty_password_with_protection(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="mysecret") + assert share.verify_password("") is False + + def test_password_case_sensitive(self): + share = VideoShare.create(video_id="v1", user_id="u1", password="Secret") + assert share.verify_password("secret") is False + assert share.verify_password("Secret") is True + + +# ── 计数方法 ──────────────────────────────────────────────────────────────── + + +class TestVideoShareCounters: + def test_increment_view(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.view_count == 0 + share.increment_view_count() + assert share.view_count == 1 + share.increment_view_count() + assert share.view_count == 2 + + def test_increment_download(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.download_count == 0 + share.increment_download_count() + assert share.download_count == 1 + share.increment_download_count() + assert share.download_count == 2 + + def test_counters_independent(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.increment_view_count() + share.increment_view_count() + share.increment_download_count() + assert share.view_count == 2 + assert share.download_count == 1 + + +# ── revoke ────────────────────────────────────────────────────────────────── + + +class TestVideoShareRevoke: + def test_revoke_sets_inactive(self): + share = VideoShare.create(video_id="v1", user_id="u1") + assert share.is_active is True + share.revoke() + assert share.is_active is False def test_revoke_makes_inaccessible(self): - s = VideoShare.create(video_id="v1", user_id="u1") - assert s.is_accessible is True - s.revoke() - assert s.is_accessible is False + share = VideoShare.create(video_id="v1", user_id="u1") + share.revoke() + assert share.is_accessible is False - -# ── dataclass 基础特性 ─────────────────────────────────────────────────────── - - -class TestVideoShareBasics: - """VideoShare 基础特性""" - - def test_slots_no_extra_attrs(self): - s = VideoShare.create(video_id="v1", user_id="u1") - with pytest.raises(AttributeError): - s.nonexistent = "value" - - def test_direct_construction(self): - s = VideoShare( - id="custom_id", - video_id="v1", - user_id="u1", - share_token="abc123", - ) - assert s.id == "custom_id" - assert s.share_token == "abc123" - - def test_equality_same_id(self): - now = datetime.now(timezone.utc) - s1 = VideoShare(id="same", video_id="v1", user_id="u1", share_token="t", created_at=now, updated_at=now) - s2 = VideoShare(id="same", video_id="v1", user_id="u1", share_token="t", created_at=now, updated_at=now) - assert s1 == s2 + def test_revoke_idempotent(self): + share = VideoShare.create(video_id="v1", user_id="u1") + share.revoke() + share.revoke() # 第二次也不报错 + assert share.is_active is False diff --git a/tests/unit/test_video_share_use_cases.py b/tests/unit/test_video_share_use_cases.py index 2d4152826..ebd10f78a 100755 --- a/tests/unit/test_video_share_use_cases.py +++ b/tests/unit/test_video_share_use_cases.py @@ -1,4 +1,4 @@ -"""视频分享 UseCase 单元测试.""" +"""视频分享 Use Cases 单元测试 — wave215""" from __future__ import annotations @@ -22,6 +22,7 @@ from packages.application.video_share.use_cases import ( PasswordRequiredError, RecordShareDownloadUseCase, RevokeShareUseCase, + ShareAccessResult, ShareExpiredError, UpdateShareUseCase, VideoNotFoundError, @@ -29,486 +30,459 @@ from packages.application.video_share.use_cases import ( from packages.domain.generated_video import GeneratedVideo from packages.domain.video_share import VideoShare - -@pytest.fixture -def mock_share_repo(): - return MagicMock() +# ── helpers ────────────────────────────────────────────────────────────────── -@pytest.fixture -def mock_video_repo(): - return MagicMock() - - -@pytest.fixture -def sample_video(): - video = MagicMock(spec=GeneratedVideo) - video.id = "video_001" - video.user_id = "user_001" - return video - - -@pytest.fixture -def sample_share(): +def _make_share( + video_id="v1", + user_id="u1", + password=None, + expires_at=None, + is_active=True, + view_count=0, + download_count=0, +): share = VideoShare.create( - video_id="video_001", - user_id="user_001", + video_id=video_id, + user_id=user_id, + password=password, + expires_at=expires_at, ) + share.is_active = is_active + share.view_count = view_count + share.download_count = download_count return share -@pytest.fixture -def sample_share_with_password(): - share = VideoShare.create( - video_id="video_001", - user_id="user_001", - password="secret123", +def _make_video(video_id="v1", user_id="u1", name="test.mp4", file_url="http://x/v.mp4"): + return GeneratedVideo( + id=video_id, + project_id="p1", + generation_task_id="t1", + name=name, + file_url=file_url, + file_size=1024, + duration=10.0, + width=1920, + height=1080, + fps=30.0, + user_id=user_id, ) - return share -@pytest.fixture -def sample_share_expired(): - # 直接构造已过期的分享（不经过create方法的校验） - share = VideoShare( - id="share_expired_001", - video_id="video_001", - user_id="user_001", - share_token="expiredtoken123", - expires_at=datetime.now(timezone.utc) - timedelta(hours=1), - ) - return share +# ── CreateShareUseCase ────────────────────────────────────────────────────── class TestCreateShareUseCase: - """CreateShareUseCase 测试""" + def test_create_success(self): + video = _make_video() + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = video + share_repo.create.side_effect = lambda s: s - def test_create_share_success(self, mock_share_repo, mock_video_repo, sample_video): - """正常创建分享链接""" - mock_video_repo.get.return_value = sample_video - mock_share_repo.create.side_effect = lambda s: s + uc = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="v1", user_id="u1") + result = uc.execute(cmd) - use_case = CreateShareUseCase(mock_share_repo, mock_video_repo) - command = CreateShareCommand(video_id="video_001", user_id="user_001") - result = use_case.execute(command) + assert result.video_id == "v1" + assert result.user_id == "u1" + video_repo.get.assert_called_once_with("v1") + share_repo.create.assert_called_once() - assert result.video_id == "video_001" - assert result.user_id == "user_001" - assert result.share_token is not None - assert result.has_password is False - mock_share_repo.create.assert_called_once() + def test_create_with_password(self): + video = _make_video() + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = video + share_repo.create.side_effect = lambda s: s - def test_create_share_with_password(self, mock_share_repo, mock_video_repo, sample_video): - """创建带密码的分享""" - mock_video_repo.get.return_value = sample_video - mock_share_repo.create.side_effect = lambda s: s - - use_case = CreateShareUseCase(mock_share_repo, mock_video_repo) - command = CreateShareCommand( - video_id="video_001", - user_id="user_001", - password="mypassword", - ) - result = use_case.execute(command) + uc = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="v1", user_id="u1", password="secret") + result = uc.execute(cmd) assert result.has_password is True - assert result.password_hash is not None - def test_create_share_with_expiry(self, mock_share_repo, mock_video_repo, sample_video): - """创建带有效期的分享""" - mock_video_repo.get.return_value = sample_video - mock_share_repo.create.side_effect = lambda s: s + def test_video_not_found_raises(self): + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = None - future = datetime.now(timezone.utc) + timedelta(days=7) - use_case = CreateShareUseCase(mock_share_repo, mock_video_repo) - command = CreateShareCommand( - video_id="video_001", - user_id="user_001", - expires_at=future, - ) - result = use_case.execute(command) - - assert result.expires_at == future - - def test_create_share_video_not_found(self, mock_share_repo, mock_video_repo): - """视频不存在时抛出 VideoNotFoundError""" - mock_video_repo.get.return_value = None - - use_case = CreateShareUseCase(mock_share_repo, mock_video_repo) - command = CreateShareCommand(video_id="nonexistent", user_id="user_001") + uc = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="v999", user_id="u1") with pytest.raises(VideoNotFoundError): - use_case.execute(command) + uc.execute(cmd) - mock_share_repo.create.assert_not_called() + def test_wrong_user_video_not_found(self): + video = _make_video(user_id="u2") + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = video - def test_create_share_wrong_user(self, mock_share_repo, mock_video_repo, sample_video): - """非视频所有者创建分享失败""" - sample_video.user_id = "user_other" - mock_video_repo.get.return_value = sample_video - - use_case = CreateShareUseCase(mock_share_repo, mock_video_repo) - command = CreateShareCommand(video_id="video_001", user_id="user_001") + uc = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="v1", user_id="u1") with pytest.raises(VideoNotFoundError): - use_case.execute(command) + uc.execute(cmd) - mock_share_repo.create.assert_not_called() + def test_video_without_user_id_attribute(self): + # 视频没有user_id字段的情况 + class SimpleVideo: + pass + + video = SimpleVideo() + video.id = "v1" + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = video + share_repo.create.side_effect = lambda s: s + + uc = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="v1", user_id="u1") + result = uc.execute(cmd) + assert result is not None + + +# ── GetShareByTokenUseCase ────────────────────────────────────────────────── class TestGetShareByTokenUseCase: - """GetShareByTokenUseCase 测试""" + def test_get_success(self): + share = _make_share() + repo = MagicMock() + repo.get_by_token.return_value = share - def test_get_share_success(self, mock_share_repo, sample_share): - """通过 token 正常获取分享信息""" - mock_share_repo.get_by_token.return_value = sample_share + uc = GetShareByTokenUseCase(repo) + result = uc.execute(share.share_token) + assert result.id == share.id - use_case = GetShareByTokenUseCase(mock_share_repo) - result = use_case.execute(sample_share.share_token) - - assert result.id == sample_share.id - mock_share_repo.get_by_token.assert_called_once_with(sample_share.share_token) - - def test_get_share_not_found(self, mock_share_repo): - """token 不存在时抛出 NotFoundError""" - mock_share_repo.get_by_token.return_value = None - - use_case = GetShareByTokenUseCase(mock_share_repo) + def test_not_found_raises(self): + repo = MagicMock() + repo.get_by_token.return_value = None + uc = GetShareByTokenUseCase(repo) with pytest.raises(NotFoundError): - use_case.execute("invalid_token") + uc.execute("nonexistent") - def test_get_share_expired_raises(self, mock_share_repo, sample_share_expired): - """已过期的分享不可访问""" - mock_share_repo.get_by_token.return_value = sample_share_expired - - use_case = GetShareByTokenUseCase(mock_share_repo) + def test_expired_share_raises(self): + share = _make_share() + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + repo = MagicMock() + repo.get_by_token.return_value = share + uc = GetShareByTokenUseCase(repo) with pytest.raises(ShareExpiredError): - use_case.execute(sample_share_expired.share_token) + uc.execute(share.share_token) + + def test_revoked_share_raises(self): + share = _make_share(is_active=False) + repo = MagicMock() + repo.get_by_token.return_value = share + + uc = GetShareByTokenUseCase(repo) + with pytest.raises(ShareExpiredError): + uc.execute(share.share_token) + + +# ── AccessShareUseCase ────────────────────────────────────────────────────── class TestAccessShareUseCase: - """AccessShareUseCase 测试""" + def test_access_no_password(self): + share = _make_share() + video = _make_video() + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = share + video_repo.get.return_value = video - def test_access_without_password(self, mock_share_repo, mock_video_repo, sample_share, sample_video): - """无密码分享直接访问成功""" - mock_share_repo.get_by_token.return_value = sample_share - mock_video_repo.get.return_value = sample_video + uc = AccessShareUseCase(share_repo, video_repo) + result = uc.execute(share.share_token) - use_case = AccessShareUseCase(mock_share_repo, mock_video_repo) - result = use_case.execute(sample_share.share_token) - - assert result.share.id == sample_share.id - assert result.video.id == "video_001" + assert isinstance(result, ShareAccessResult) + assert result.share.id == share.id + assert result.video.id == video.id assert result.password_verified is True - mock_share_repo.increment_view.assert_called_once_with(sample_share.id) - assert sample_share.view_count == 1 + assert share.view_count == 1 + share_repo.increment_view.assert_called_once_with(share.id) - def test_access_with_correct_password( - self, mock_share_repo, mock_video_repo, sample_share_with_password, sample_video - ): - """带密码分享输入正确密码访问成功""" - mock_share_repo.get_by_token.return_value = sample_share_with_password - mock_video_repo.get.return_value = sample_video - - use_case = AccessShareUseCase(mock_share_repo, mock_video_repo) - result = use_case.execute(sample_share_with_password.share_token, password="secret123") + def test_access_with_correct_password(self): + share = _make_share(password="secret") + video = _make_video() + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = share + video_repo.get.return_value = video + uc = AccessShareUseCase(share_repo, video_repo) + result = uc.execute(share.share_token, password="secret") assert result.password_verified is True - mock_share_repo.increment_view.assert_called_once() - def test_access_password_required_but_not_provided( - self, mock_share_repo, mock_video_repo, sample_share_with_password - ): - """带密码分享不输入密码抛出 PasswordRequiredError""" - mock_share_repo.get_by_token.return_value = sample_share_with_password - - use_case = AccessShareUseCase(mock_share_repo, mock_video_repo) + def test_access_password_required_but_not_provided(self): + share = _make_share(password="secret") + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = share + uc = AccessShareUseCase(share_repo, video_repo) with pytest.raises(PasswordRequiredError): - use_case.execute(sample_share_with_password.share_token) + uc.execute(share.share_token) - mock_share_repo.increment_view.assert_not_called() - - def test_access_wrong_password(self, mock_share_repo, mock_video_repo, sample_share_with_password): - """密码错误抛出 InvalidPasswordError""" - mock_share_repo.get_by_token.return_value = sample_share_with_password - - use_case = AccessShareUseCase(mock_share_repo, mock_video_repo) + def test_access_wrong_password(self): + share = _make_share(password="secret") + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = share + uc = AccessShareUseCase(share_repo, video_repo) with pytest.raises(InvalidPasswordError): - use_case.execute(sample_share_with_password.share_token, password="wrongpass") + uc.execute(share.share_token, password="wrong") - mock_share_repo.increment_view.assert_not_called() - - def test_access_share_not_found(self, mock_share_repo, mock_video_repo): - """分享不存在抛出 NotFoundError""" - mock_share_repo.get_by_token.return_value = None - - use_case = AccessShareUseCase(mock_share_repo, mock_video_repo) - - with pytest.raises(NotFoundError): - use_case.execute("invalid_token") - - def test_access_expired_share(self, mock_share_repo, mock_video_repo, sample_share_expired): - """已过期分享不可访问""" - mock_share_repo.get_by_token.return_value = sample_share_expired - - use_case = AccessShareUseCase(mock_share_repo, mock_video_repo) + def test_access_expired_share(self): + share = _make_share() + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + share_repo = MagicMock() + share_repo.get_by_token.return_value = share + uc = AccessShareUseCase(share_repo, MagicMock()) with pytest.raises(ShareExpiredError): - use_case.execute(sample_share_expired.share_token) + uc.execute(share.share_token) - mock_share_repo.increment_view.assert_not_called() + def test_access_share_not_found(self): + share_repo = MagicMock() + share_repo.get_by_token.return_value = None - def test_access_video_not_found(self, mock_share_repo, mock_video_repo, sample_share): - """分享存在但视频不存在""" - mock_share_repo.get_by_token.return_value = sample_share - mock_video_repo.get.return_value = None + uc = AccessShareUseCase(share_repo, MagicMock()) + with pytest.raises(NotFoundError): + uc.execute("nonexistent") - use_case = AccessShareUseCase(mock_share_repo, mock_video_repo) + def test_access_video_not_found(self): + share = _make_share() + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = share + video_repo.get.return_value = None + uc = AccessShareUseCase(share_repo, video_repo) with pytest.raises(VideoNotFoundError): - use_case.execute(sample_share.share_token) + uc.execute(share.share_token) + + +# ── ListSharesByVideoUseCase ──────────────────────────────────────────────── class TestListSharesByVideoUseCase: - """ListSharesByVideoUseCase 测试""" + def test_list_success(self): + shares = [_make_share(), _make_share()] + repo = MagicMock() + repo.list_by_video.return_value = shares - def test_list_by_video(self, mock_share_repo, sample_share): - """列出某个视频的所有分享""" - mock_share_repo.list_by_video.return_value = [sample_share] + uc = ListSharesByVideoUseCase(repo) + result = uc.execute("v1", "u1") - use_case = ListSharesByVideoUseCase(mock_share_repo) - result = use_case.execute("video_001", "user_001") + assert len(result) == 2 + repo.list_by_video.assert_called_once_with("v1", "u1") - assert len(result) == 1 - mock_share_repo.list_by_video.assert_called_once_with("video_001", "user_001") - - def test_list_by_video_empty(self, mock_share_repo): - """视频没有分享记录时返回空列表""" - mock_share_repo.list_by_video.return_value = [] - - use_case = ListSharesByVideoUseCase(mock_share_repo) - result = use_case.execute("video_001", "user_001") + def test_list_empty(self): + repo = MagicMock() + repo.list_by_video.return_value = [] + uc = ListSharesByVideoUseCase(repo) + result = uc.execute("v1", "u1") assert result == [] +# ── ListSharesByUserUseCase ───────────────────────────────────────────────── + + class TestListSharesByUserUseCase: - """ListSharesByUserUseCase 测试""" + def test_list_with_pagination(self): + shares = [_make_share() for _ in range(5)] + repo = MagicMock() + repo.list_by_user.return_value = shares + repo.count_by_user.return_value = 20 - def test_list_by_user(self, mock_share_repo, sample_share): - """列出用户的所有分享""" - mock_share_repo.list_by_user.return_value = [sample_share] - mock_share_repo.count_by_user.return_value = 1 + uc = ListSharesByUserUseCase(repo) + items, total = uc.execute("u1", skip=0, limit=5) - use_case = ListSharesByUserUseCase(mock_share_repo) - items, total = use_case.execute("user_001") + assert len(items) == 5 + assert total == 20 + repo.list_by_user.assert_called_once_with("u1", skip=0, limit=5) + repo.count_by_user.assert_called_once_with("u1") - assert len(items) == 1 - assert total == 1 - mock_share_repo.list_by_user.assert_called_once_with("user_001", skip=0, limit=20) + def test_list_default_params(self): + repo = MagicMock() + repo.list_by_user.return_value = [] + repo.count_by_user.return_value = 0 - def test_list_by_user_with_pagination(self, mock_share_repo): - """带分页参数查询""" - mock_share_repo.list_by_user.return_value = [] - mock_share_repo.count_by_user.return_value = 50 + uc = ListSharesByUserUseCase(repo) + uc.execute("u1") - use_case = ListSharesByUserUseCase(mock_share_repo) - items, total = use_case.execute("user_001", skip=10, limit=5) + repo.list_by_user.assert_called_once_with("u1", skip=0, limit=20) - assert total == 50 - mock_share_repo.list_by_user.assert_called_once_with("user_001", skip=10, limit=5) - def test_list_by_user_empty(self, mock_share_repo): - """用户没有分享记录""" - mock_share_repo.list_by_user.return_value = [] - mock_share_repo.count_by_user.return_value = 0 - - use_case = ListSharesByUserUseCase(mock_share_repo) - items, total = use_case.execute("user_001") - - assert items == [] - assert total == 0 +# ── UpdateShareUseCase ────────────────────────────────────────────────────── class TestUpdateShareUseCase: - """UpdateShareUseCase 测试""" + def test_update_password(self): + share = _make_share(password="oldpass") + repo = MagicMock() + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s - def test_update_password(self, mock_share_repo, sample_share): - """更新分享密码""" - mock_share_repo.get_by_id.return_value = sample_share - mock_share_repo.update.side_effect = lambda s: s + uc = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id=share.id, user_id="u1", password="newpass") + result = uc.execute(cmd) - use_case = UpdateShareUseCase(mock_share_repo) - command = UpdateShareCommand( - share_id=sample_share.id, - user_id="user_001", - password="newpassword", - ) - result = use_case.execute(command) + assert result is not None + assert share.verify_password("newpass") is True + assert share.verify_password("oldpass") is False + repo.update.assert_called_once() - assert result.has_password is True - mock_share_repo.update.assert_called_once() + def test_clear_password(self): + share = _make_share(password="oldpass") + repo = MagicMock() + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s - def test_clear_password(self, mock_share_repo, sample_share_with_password): - """清除分享密码（空字符串）""" - mock_share_repo.get_by_id.return_value = sample_share_with_password - mock_share_repo.update.side_effect = lambda s: s - - use_case = UpdateShareUseCase(mock_share_repo) - command = UpdateShareCommand( - share_id=sample_share_with_password.id, - user_id="user_001", - password="", # 空字符串表示清除 - ) - result = use_case.execute(command) + uc = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id=share.id, user_id="u1", password="") + result = uc.execute(cmd) assert result.has_password is False - assert result.password_hash is None - def test_update_password_none_no_change(self, mock_share_repo, sample_share_with_password): - """password=None 不修改密码""" - original_hash = sample_share_with_password.password_hash - mock_share_repo.get_by_id.return_value = sample_share_with_password - mock_share_repo.update.side_effect = lambda s: s + def test_update_password_none_no_change(self): + share = _make_share(password="oldpass") + repo = MagicMock() + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s - use_case = UpdateShareUseCase(mock_share_repo) - command = UpdateShareCommand( - share_id=sample_share_with_password.id, - user_id="user_001", - password=None, # None表示不修改 - ) - result = use_case.execute(command) + uc = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id=share.id, user_id="u1", password=None) + result = uc.execute(cmd) - assert result.password_hash == original_hash + # password=None 表示不修改 + assert result.has_password is True + assert share.verify_password("oldpass") is True - def test_update_expires_at(self, mock_share_repo, sample_share): - """更新有效期""" - mock_share_repo.get_by_id.return_value = sample_share - mock_share_repo.update.side_effect = lambda s: s + def test_update_expires_at(self): + share = _make_share() + new_expiry = datetime.now(timezone.utc) + timedelta(days=30) + repo = MagicMock() + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s - future = datetime.now(timezone.utc) + timedelta(days=3) - use_case = UpdateShareUseCase(mock_share_repo) - command = UpdateShareCommand( - share_id=sample_share.id, - user_id="user_001", - expires_at=future, - ) - result = use_case.execute(command) + uc = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id=share.id, user_id="u1", expires_at=new_expiry) + result = uc.execute(cmd) - assert result.expires_at == future + assert result.expires_at == new_expiry - def test_update_expires_at_past_raises(self, mock_share_repo, sample_share): - """设置过去的有效期抛出 ValueError""" - mock_share_repo.get_by_id.return_value = sample_share + def test_update_expires_at_past_raises(self): + share = _make_share() + past = datetime.now(timezone.utc) - timedelta(days=1) + repo = MagicMock() + repo.get_by_id.return_value = share - past = datetime.now(timezone.utc) - timedelta(hours=1) - use_case = UpdateShareUseCase(mock_share_repo) - command = UpdateShareCommand( - share_id=sample_share.id, - user_id="user_001", - expires_at=past, - ) + uc = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id=share.id, user_id="u1", expires_at=past) with pytest.raises(ValueError, match="expires_at cannot be in the past"): - use_case.execute(command) + uc.execute(cmd) - mock_share_repo.update.assert_not_called() + def test_update_not_found_raises(self): + repo = MagicMock() + repo.get_by_id.return_value = None - def test_update_share_not_found(self, mock_share_repo): - """分享不存在抛出 NotFoundError""" - mock_share_repo.get_by_id.return_value = None - - use_case = UpdateShareUseCase(mock_share_repo) - command = UpdateShareCommand( - share_id="nonexistent", - user_id="user_001", - password="newpass", - ) + uc = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id="nonexistent", user_id="u1") with pytest.raises(NotFoundError): - use_case.execute(command) + uc.execute(cmd) - mock_share_repo.update.assert_not_called() + def test_update_wrong_user_not_found(self): + share = _make_share(user_id="u2") + repo = MagicMock() + repo.get_by_id.return_value = None # 仓储层已经按user_id过滤了 + + uc = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id=share.id, user_id="u1") + + with pytest.raises(NotFoundError): + uc.execute(cmd) + + +# ── RevokeShareUseCase ────────────────────────────────────────────────────── class TestRevokeShareUseCase: - """RevokeShareUseCase 测试""" - - def test_revoke_success(self, mock_share_repo, sample_share): - """撤销分享成功""" - mock_share_repo.get_by_id.return_value = sample_share - mock_share_repo.delete.return_value = True - - use_case = RevokeShareUseCase(mock_share_repo) - result = use_case.execute(sample_share.id, "user_001") + def test_revoke_success(self): + repo = MagicMock() + repo.get_by_id.return_value = MagicMock() + repo.delete.return_value = True + uc = RevokeShareUseCase(repo) + result = uc.execute("s1", "u1") assert result is True - mock_share_repo.delete.assert_called_once_with(sample_share.id, "user_001") + repo.delete.assert_called_once_with("s1", "u1") - def test_revoke_not_found(self, mock_share_repo): - """分享不存在抛出 NotFoundError""" - mock_share_repo.get_by_id.return_value = None - - use_case = RevokeShareUseCase(mock_share_repo) + def test_revoke_not_found_raises(self): + repo = MagicMock() + repo.get_by_id.return_value = None + uc = RevokeShareUseCase(repo) with pytest.raises(NotFoundError): - use_case.execute("nonexistent", "user_001") + uc.execute("s1", "u1") - mock_share_repo.delete.assert_not_called() + +# ── RecordShareDownloadUseCase ────────────────────────────────────────────── class TestRecordShareDownloadUseCase: - """RecordShareDownloadUseCase 测试""" + def test_record_download_success(self): + share = _make_share(download_count=3) + repo = MagicMock() + repo.get_by_token.return_value = share - def test_record_download_no_password(self, mock_share_repo, sample_share): - """无密码分享记录下载""" - mock_share_repo.get_by_token.return_value = sample_share + uc = RecordShareDownloadUseCase(repo) + uc.execute(share.share_token) - use_case = RecordShareDownloadUseCase(mock_share_repo) - use_case.execute(sample_share.share_token) + repo.increment_download.assert_called_once_with(share.id) - mock_share_repo.increment_download.assert_called_once_with(sample_share.id) + def test_record_download_with_password(self): + share = _make_share(password="secret") + repo = MagicMock() + repo.get_by_token.return_value = share - def test_record_download_with_password(self, mock_share_repo, sample_share_with_password): - """带密码分享正确密码记录下载""" - mock_share_repo.get_by_token.return_value = sample_share_with_password + uc = RecordShareDownloadUseCase(repo) + uc.execute(share.share_token, password="secret") + repo.increment_download.assert_called_once() - use_case = RecordShareDownloadUseCase(mock_share_repo) - use_case.execute(sample_share_with_password.share_token, password="secret123") - - mock_share_repo.increment_download.assert_called_once() - - def test_record_download_wrong_password(self, mock_share_repo, sample_share_with_password): - """密码错误不记录下载""" - mock_share_repo.get_by_token.return_value = sample_share_with_password - - use_case = RecordShareDownloadUseCase(mock_share_repo) + def test_record_download_wrong_password_raises(self): + share = _make_share(password="secret") + repo = MagicMock() + repo.get_by_token.return_value = share + uc = RecordShareDownloadUseCase(repo) with pytest.raises(InvalidPasswordError): - use_case.execute(sample_share_with_password.share_token, password="wrong") + uc.execute(share.share_token, password="wrong") - mock_share_repo.increment_download.assert_not_called() - - def test_record_download_not_found(self, mock_share_repo): - """分享不存在抛出 NotFoundError""" - mock_share_repo.get_by_token.return_value = None - - use_case = RecordShareDownloadUseCase(mock_share_repo) - - with pytest.raises(NotFoundError): - use_case.execute("invalid_token") - - def test_record_download_expired(self, mock_share_repo, sample_share_expired): - """已过期分享不能下载""" - mock_share_repo.get_by_token.return_value = sample_share_expired - - use_case = RecordShareDownloadUseCase(mock_share_repo) + def test_record_download_expired_raises(self): + share = _make_share() + share.expires_at = datetime.now(timezone.utc) - timedelta(days=1) + repo = MagicMock() + repo.get_by_token.return_value = share + uc = RecordShareDownloadUseCase(repo) with pytest.raises(ShareExpiredError): - use_case.execute(sample_share_expired.share_token) + uc.execute(share.share_token) - mock_share_repo.increment_download.assert_not_called() + def test_record_download_not_found_raises(self): + repo = MagicMock() + repo.get_by_token.return_value = None + + uc = RecordShareDownloadUseCase(repo) + with pytest.raises(NotFoundError): + uc.execute("nonexistent") -- 2.54.0 From 1f2f1d5d5ba6276217ce6772ef5eb0d07acda5fa Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 08:18:16 +0800 Subject: [PATCH 2/2] =?UTF-8?q?test:=20wave216=20add=2046=20unit=20tests?= =?UTF-8?q?=20for=20template=20use=20cases=20(12=E4=B8=AAUC=E5=85=A8?= =?UTF-8?q?=E9=87=8F=E8=A6=86=E7=9B=96)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 覆盖的 Use Cases: - CreateTemplateUseCase (创建+片段创建，模式校验) - ListTemplatesUseCase (无过滤/带过滤/分页) - CountTemplatesUseCase (无过滤/带过滤) - GetTemplateUseCase (存在/不存在) - UpdateTemplateUseCase (改名字/模式/片段/NONE不修改，模式校验) - DeleteTemplateUseCase (成功/失败) - CopyTemplateUseCase (成功/不存在/空名/空白名/strip) - ValidateTemplateUseCase (核心业务校验): - one_take模式必须恰好1个片段 - voice_over模式每个片段必须有有效material_type - 配音时长±30%偏差警告 - 边界值(0.7/1.3精确边界) - zero/None voiceover_duration跳过 - warnings详情结构验证 - CreateCategoryUseCase / ListCategoriesUseCase / DeleteCategoryUseCase - ListTagsUseCase - GetTemplateUsageUseCase --- tests/unit/test_template_use_cases.py | 815 ++++++++++++++------------ 1 file changed, 444 insertions(+), 371 deletions(-) diff --git a/tests/unit/test_template_use_cases.py b/tests/unit/test_template_use_cases.py index 96de98d88..7dc2eec95 100755 --- a/tests/unit/test_template_use_cases.py +++ b/tests/unit/test_template_use_cases.py @@ -1,8 +1,7 @@ -"""Template use cases 单元测试.""" +"""模板 Use Cases 单元测试 — wave216""" from __future__ import annotations -from typing import List, Optional from unittest.mock import MagicMock import pytest @@ -29,610 +28,684 @@ from packages.application.template.use_cases import ( ListCategoriesUseCase, ListTagsUseCase, ListTemplatesUseCase, - NotFoundError, UpdateTemplateUseCase, ValidateResult, ValidateTemplateUseCase, - ValidationError, ) from packages.domain.editing_mode import EditingMode +from packages.domain.exceptions import NotFoundError, ValidationError from packages.domain.template import Template, TemplateCategory, TemplateSegment +# ── helpers ────────────────────────────────────────────────────────────────── + def _make_template( - template_id: str = "tpl_001", - user_id: str = "user_001", - name: str = "测试模板", - mode: str = "one_take", - segments: Optional[List[TemplateSegment]] = None, - estimated_duration: float = 60.0, -) -> Template: - tpl = Template( + template_id="t1", + user_id="u1", + name="测试模板", + mode=EditingMode.ONE_TAKE.value, + category="", + estimated_duration=30.0, + segments=None, +): + if segments is None: + segments = [ + TemplateSegment( + id="s1", + template_id=template_id, + segment_order=1, + duration_min=5.0, + duration_max=10.0, + ) + ] + return Template( id=template_id, user_id=user_id, name=name, mode=mode, - category="测试分类", - tags=["tag1", "tag2"], - title_config={"enabled": True}, - subtitle_config={"enabled": False}, - bgm_config={"enabled": True}, + category=category, estimated_duration=estimated_duration, + segments=segments, ) - if segments is not None: - tpl.segments = segments - return tpl -def _make_segments(count: int = 1, material_type: Optional[str] = None) -> List[TemplateSegment]: +def _make_segments(n, *, start_order=1, material_type=None): return [ TemplateSegment( - id=f"seg_{i}", - template_id="tpl_001", - segment_order=i, - duration_min=3.0, - duration_max=8.0, + id=f"s{i}", + template_id="t1", + segment_order=start_order + i - 1, + duration_min=5.0, + duration_max=10.0, material_type=material_type, ) - for i in range(count) + for i in range(1, n + 1) ] +# ── CreateTemplateUseCase ──────────────────────────────────────────────────── + + class TestCreateTemplateUseCase: - def test_creates_template_with_segments(self) -> None: + def test_create_success(self): repo = MagicMock() - repo.create.side_effect = lambda t: t # 返回传入的template + repo.create.side_effect = lambda t: t repo.create_segments.return_value = None - use_case = CreateTemplateUseCase(repo) + uc = CreateTemplateUseCase(repo) cmd = CreateTemplateCommand( - user_id="user_001", - name="新模板", + user_id="u1", + name="我的模板", mode=EditingMode.ONE_TAKE.value, - category="分类A", - tags=["t1", "t2"], - segments=[ - SegmentCommand(segment_order=0, duration_min=2.0, duration_max=5.0), - SegmentCommand(segment_order=1, duration_min=3.0, duration_max=6.0), - ], + segments=[SegmentCommand(segment_order=1, duration_min=5, duration_max=10)], ) + result = uc.execute(cmd) - result = use_case.execute(cmd) - - assert result.name == "新模板" + assert result.name == "我的模板" assert result.mode == EditingMode.ONE_TAKE.value - assert len(result.segments) == 2 - assert result.segments[0].segment_order == 0 - assert result.segments[1].segment_order == 1 + assert result.user_id == "u1" + assert len(result.segments) == 1 repo.create.assert_called_once() repo.create_segments.assert_called_once() - def test_invalid_mode_raises_validation_error(self) -> None: + def test_create_invalid_mode_raises(self): repo = MagicMock() - use_case = CreateTemplateUseCase(repo) + uc = CreateTemplateUseCase(repo) cmd = CreateTemplateCommand( - user_id="user_001", - name="测试", + user_id="u1", + name="test", mode="invalid_mode", - segments=[], ) - with pytest.raises(ValidationError, match="无效的剪辑模式"): - use_case.execute(cmd) + uc.execute(cmd) - def test_creates_without_segments(self) -> None: + def test_create_with_multiple_segments(self): repo = MagicMock() repo.create.side_effect = lambda t: t repo.create_segments.return_value = None - use_case = CreateTemplateUseCase(repo) + uc = CreateTemplateUseCase(repo) cmd = CreateTemplateCommand( - user_id="user_001", - name="空片段模板", - mode=EditingMode.ONE_TAKE.value, - segments=[], + user_id="u1", + name="多片段模板", + mode=EditingMode.VOICE_OVER.value, + segments=[ + SegmentCommand(segment_order=1, duration_min=3, duration_max=5, material_type="人物"), + SegmentCommand(segment_order=2, duration_min=5, duration_max=8, material_type="场景"), + ], ) + result = uc.execute(cmd) + assert len(result.segments) == 2 + assert result.segments[0].segment_order == 1 + assert result.segments[1].segment_order == 2 - result = use_case.execute(cmd) + def test_create_with_empty_segments(self): + repo = MagicMock() + repo.create.side_effect = lambda t: t + repo.create_segments.return_value = None + + uc = CreateTemplateUseCase(repo) + cmd = CreateTemplateCommand( + user_id="u1", + name="无片段模板", + mode=EditingMode.PIP.value, + ) + result = uc.execute(cmd) assert len(result.segments) == 0 repo.create_segments.assert_called_once_with([]) - def test_generates_uuid_for_template_and_segments(self) -> None: - repo = MagicMock() - repo.create.side_effect = lambda t: t - repo.create_segments.return_value = None - use_case = CreateTemplateUseCase(repo) - cmd = CreateTemplateCommand( - user_id="user_001", - name="UUID测试", - mode=EditingMode.VOICE_OVER.value, - segments=[ - SegmentCommand(segment_order=0, duration_min=1.0, duration_max=3.0, material_type="人物"), - ], - ) - - result = use_case.execute(cmd) - assert len(result.id) == 32 # uuid hex - assert len(result.segments[0].id) == 32 - assert result.segments[0].template_id == result.id +# ── ListTemplatesUseCase ──────────────────────────────────────────────────── class TestListTemplatesUseCase: - def test_list_without_filter(self) -> None: + def test_list_no_filter(self): + templates = [_make_template("t1"), _make_template("t2")] repo = MagicMock() - expected = [_make_template("t1"), _make_template("t2")] - repo.list_by_user.return_value = expected + repo.list_by_user.return_value = templates - use_case = ListTemplatesUseCase(repo) - result = use_case.execute("user_001", skip=0, limit=10) + uc = ListTemplatesUseCase(repo) + result = uc.execute("u1", skip=0, limit=10) assert len(result) == 2 - repo.list_by_user.assert_called_once_with("user_001", skip=0, limit=10) + repo.list_by_user.assert_called_once_with("u1", skip=0, limit=10) - def test_list_with_filter(self) -> None: + def test_list_with_filter(self): + templates = [_make_template("t1")] repo = MagicMock() - expected = [_make_template("t1")] - repo.list_by_user.return_value = expected + repo.list_by_user.return_value = templates - use_case = ListTemplatesUseCase(repo) - f = ListTemplatesFilter(category="分类A", tag="t1", keyword="测试", mode="one_take") - result = use_case.execute("user_001", skip=0, limit=10, filter=f) + uc = ListTemplatesUseCase(repo) + f = ListTemplatesFilter(category="cat1", tag="tag1", keyword="test", mode="one_take") + result = uc.execute("u1", filter=f) assert len(result) == 1 repo.list_by_user.assert_called_once_with( - "user_001", + "u1", skip=0, - limit=10, - category="分类A", - tag="t1", - keyword="测试", + limit=50, + category="cat1", + tag="tag1", + keyword="test", mode="one_take", ) + def test_list_pagination(self): + repo = MagicMock() + repo.list_by_user.return_value = [] + + uc = ListTemplatesUseCase(repo) + uc.execute("u1", skip=20, limit=10) + repo.list_by_user.assert_called_once_with("u1", skip=20, limit=10) + + +# ── CountTemplatesUseCase ─────────────────────────────────────────────────── + class TestCountTemplatesUseCase: - def test_count_without_filter(self) -> None: + def test_count_no_filter(self): repo = MagicMock() repo.count_by_user.return_value = 42 - use_case = CountTemplatesUseCase(repo) - result = use_case.execute("user_001") - + uc = CountTemplatesUseCase(repo) + result = uc.execute("u1") assert result == 42 - repo.count_by_user.assert_called_once_with("user_001") + repo.count_by_user.assert_called_once_with("u1") - def test_count_with_filter(self) -> None: + def test_count_with_filter(self): repo = MagicMock() repo.count_by_user.return_value = 5 - use_case = CountTemplatesUseCase(repo) - f = ListTemplatesFilter(category="分类A") - result = use_case.execute("user_001", filter=f) - + uc = CountTemplatesUseCase(repo) + f = ListTemplatesFilter(category="cat1", tag="tag1", keyword="kw", mode="pip") + result = uc.execute("u1", filter=f) assert result == 5 repo.count_by_user.assert_called_once_with( - "user_001", - category="分类A", - tag=None, - keyword=None, - mode=None, + "u1", + category="cat1", + tag="tag1", + keyword="kw", + mode="pip", ) +# ── GetTemplateUseCase ────────────────────────────────────────────────────── + + class TestGetTemplateUseCase: - def test_returns_template_when_found(self) -> None: + def test_get_found(self): + template = _make_template() repo = MagicMock() - expected = _make_template() - repo.get.return_value = expected + repo.get.return_value = template - use_case = GetTemplateUseCase(repo) - result = use_case.execute("tpl_001", "user_001") + uc = GetTemplateUseCase(repo) + result = uc.execute("t1", "u1") + assert result.id == "t1" + repo.get.assert_called_once_with("t1", "u1") - assert result is expected - repo.get.assert_called_once_with("tpl_001", "user_001") - - def test_returns_none_when_not_found(self) -> None: + def test_get_not_found(self): repo = MagicMock() repo.get.return_value = None - use_case = GetTemplateUseCase(repo) - result = use_case.execute("nonexistent", "user_001") - + uc = GetTemplateUseCase(repo) + result = uc.execute("nonexistent", "u1") assert result is None +# ── UpdateTemplateUseCase ─────────────────────────────────────────────────── + + class TestUpdateTemplateUseCase: - def test_updates_name_and_tags(self) -> None: + def test_update_name(self): + existing = _make_template(name="old") repo = MagicMock() - existing = _make_template() - existing.segments = _make_segments(2) repo.get.return_value = existing - repo.update.side_effect = lambda t: t + repo.update.return_value = existing repo.list_segments.return_value = existing.segments - use_case = UpdateTemplateUseCase(repo) - cmd = UpdateTemplateCommand( - template_id="tpl_001", - user_id="user_001", - name="新名字", - tags=["new_tag"], - ) + uc = UpdateTemplateUseCase(repo) + cmd = UpdateTemplateCommand(template_id="t1", user_id="u1", name="new") + result = uc.execute(cmd) - result = use_case.execute(cmd) - assert result.name == "新名字" - assert result.tags == ["new_tag"] - # mode没变 - assert result.mode == EditingMode.ONE_TAKE.value + assert result.name == "new" repo.update.assert_called_once() - def test_not_found_raises(self) -> None: + def test_update_mode(self): + existing = _make_template(mode=EditingMode.ONE_TAKE.value) + repo = MagicMock() + repo.get.return_value = existing + repo.update.return_value = existing + repo.list_segments.return_value = existing.segments + + uc = UpdateTemplateUseCase(repo) + cmd = UpdateTemplateCommand(template_id="t1", user_id="u1", mode=EditingMode.PIP.value) + result = uc.execute(cmd) + assert result.mode == EditingMode.PIP.value + + def test_update_invalid_mode_raises(self): + existing = _make_template() + repo = MagicMock() + repo.get.return_value = existing + + uc = UpdateTemplateUseCase(repo) + cmd = UpdateTemplateCommand(template_id="t1", user_id="u1", mode="bad") + with pytest.raises(ValidationError, match="无效的剪辑模式"): + uc.execute(cmd) + + def test_update_not_found_raises(self): repo = MagicMock() repo.get.return_value = None - use_case = UpdateTemplateUseCase(repo) - cmd = UpdateTemplateCommand(template_id="nonexistent", user_id="user_001", name="x") - + uc = UpdateTemplateUseCase(repo) + cmd = UpdateTemplateCommand(template_id="t999", user_id="u1", name="x") with pytest.raises(NotFoundError): - use_case.execute(cmd) + uc.execute(cmd) - def test_invalid_mode_raises(self) -> None: + def test_update_segments(self): + existing = _make_template(segments=_make_segments(1)) repo = MagicMock() - repo.get.return_value = _make_template() - - use_case = UpdateTemplateUseCase(repo) - cmd = UpdateTemplateCommand( - template_id="tpl_001", - user_id="user_001", - mode="invalid", - ) - - with pytest.raises(ValidationError, match="无效的剪辑模式"): - use_case.execute(cmd) - - def test_replaces_segments_when_provided(self) -> None: - repo = MagicMock() - existing = _make_template() - existing.segments = _make_segments(2) repo.get.return_value = existing - repo.update.side_effect = lambda t: t + repo.update.return_value = existing repo.delete_segments_by_template.return_value = None repo.create_segments.return_value = None - use_case = UpdateTemplateUseCase(repo) + uc = UpdateTemplateUseCase(repo) cmd = UpdateTemplateCommand( - template_id="tpl_001", - user_id="user_001", + template_id="t1", + user_id="u1", segments=[ - SegmentCommand(segment_order=0, duration_min=1.0, duration_max=2.0), - SegmentCommand(segment_order=1, duration_min=3.0, duration_max=4.0), - SegmentCommand(segment_order=2, duration_min=5.0, duration_max=6.0), + SegmentCommand(segment_order=1, duration_min=2, duration_max=5), + SegmentCommand(segment_order=2, duration_min=3, duration_max=6), ], ) + result = uc.execute(cmd) - result = use_case.execute(cmd) - assert len(result.segments) == 3 - repo.delete_segments_by_template.assert_called_once_with("tpl_001") + repo.delete_segments_by_template.assert_called_once_with("t1") repo.create_segments.assert_called_once() + assert len(result.segments) == 2 - def test_no_segments_keeps_existing(self) -> None: + def test_update_none_fields_not_modified(self): + existing = _make_template(name="keep_name", category="keep_cat") repo = MagicMock() - existing = _make_template() - existing.segments = _make_segments(3) repo.get.return_value = existing - repo.update.side_effect = lambda t: t + repo.update.return_value = existing repo.list_segments.return_value = existing.segments - use_case = UpdateTemplateUseCase(repo) - cmd = UpdateTemplateCommand( - template_id="tpl_001", - user_id="user_001", - name="只改名字", - ) + uc = UpdateTemplateUseCase(repo) + # 只传 name=None, category=None 表示不修改 + cmd = UpdateTemplateCommand(template_id="t1", user_id="u1") + result = uc.execute(cmd) - result = use_case.execute(cmd) - assert len(result.segments) == 3 - repo.delete_segments_by_template.assert_not_called() - repo.create_segments.assert_not_called() - repo.list_segments.assert_called_once_with("tpl_001") + assert result.name == "keep_name" + assert result.category == "keep_cat" + + +# ── DeleteTemplateUseCase ─────────────────────────────────────────────────── class TestDeleteTemplateUseCase: - def test_delete_success(self) -> None: + def test_delete_success(self): repo = MagicMock() repo.delete.return_value = True - use_case = DeleteTemplateUseCase(repo) - result = use_case.execute("tpl_001", "user_001") - + uc = DeleteTemplateUseCase(repo) + result = uc.execute("t1", "u1") assert result is True - repo.delete.assert_called_once_with("tpl_001", "user_001") + repo.delete.assert_called_once_with("t1", "u1") - def test_delete_not_found(self) -> None: + def test_delete_not_found(self): repo = MagicMock() repo.delete.return_value = False - use_case = DeleteTemplateUseCase(repo) - result = use_case.execute("nonexistent", "user_001") - + uc = DeleteTemplateUseCase(repo) + result = uc.execute("t999", "u1") assert result is False +# ── CopyTemplateUseCase ───────────────────────────────────────────────────── + + class TestCopyTemplateUseCase: - def test_copy_success(self) -> None: + def test_copy_success(self): + copied = _make_template("t2", name="副本") repo = MagicMock() - original = _make_template(name="原模板") - repo.get.return_value = original - copied = _make_template(template_id="copied_001", name="原模板 副本") + repo.get.return_value = _make_template("t1") repo.copy_template.return_value = copied - use_case = CopyTemplateUseCase(repo) - cmd = CopyTemplateCommand( - template_id="tpl_001", - user_id="user_001", - new_name="原模板 副本", - ) + uc = CopyTemplateUseCase(repo) + cmd = CopyTemplateCommand(template_id="t1", user_id="u1", new_name="副本") + result = uc.execute(cmd) - result = use_case.execute(cmd) - assert result.name == "原模板 副本" - repo.copy_template.assert_called_once_with("tpl_001", "user_001", "原模板 副本") + assert result.name == "副本" + repo.copy_template.assert_called_once_with("t1", "u1", "副本") - def test_not_found_raises(self) -> None: + def test_copy_not_found_raises(self): repo = MagicMock() repo.get.return_value = None - use_case = CopyTemplateUseCase(repo) - cmd = CopyTemplateCommand(template_id="no", user_id="u1", new_name="x") - + uc = CopyTemplateUseCase(repo) + cmd = CopyTemplateCommand(template_id="t999", user_id="u1", new_name="副本") with pytest.raises(NotFoundError): - use_case.execute(cmd) + uc.execute(cmd) - def test_empty_name_raises(self) -> None: + def test_copy_empty_name_raises(self): repo = MagicMock() repo.get.return_value = _make_template() - use_case = CopyTemplateUseCase(repo) + uc = CopyTemplateUseCase(repo) + cmd = CopyTemplateCommand(template_id="t1", user_id="u1", new_name="") + with pytest.raises(ValidationError, match="新模板名称不能为空"): + uc.execute(cmd) + + def test_copy_whitespace_name_raises(self): + repo = MagicMock() + repo.get.return_value = _make_template() + + uc = CopyTemplateUseCase(repo) cmd = CopyTemplateCommand(template_id="t1", user_id="u1", new_name=" ") + with pytest.raises(ValidationError, match="新模板名称不能为空"): + uc.execute(cmd) - with pytest.raises(ValidationError, match="名称不能为空"): - use_case.execute(cmd) - - def test_name_stripped(self) -> None: + def test_copy_name_stripped(self): + copied = _make_template("t2", name="副本") repo = MagicMock() - repo.get.return_value = _make_template() - repo.copy_template.return_value = _make_template(name="新名字") + repo.get.return_value = _make_template("t1") + repo.copy_template.return_value = copied - use_case = CopyTemplateUseCase(repo) - cmd = CopyTemplateCommand(template_id="t1", user_id="u1", new_name=" 新名字 ") + uc = CopyTemplateUseCase(repo) + cmd = CopyTemplateCommand(template_id="t1", user_id="u1", new_name=" 副本 ") + result = uc.execute(cmd) + # 会被strip后传给repository + repo.copy_template.assert_called_once_with("t1", "u1", "副本") - use_case.execute(cmd) - repo.copy_template.assert_called_once_with("t1", "u1", "新名字") + +# ── ValidateTemplateUseCase ───────────────────────────────────────────────── class TestValidateTemplateUseCase: - def test_one_take_with_one_segment_passes(self) -> None: + def test_one_take_one_segment_ok(self): + template = _make_template(mode=EditingMode.ONE_TAKE.value, segments=_make_segments(1)) repo = MagicMock() - tpl = _make_template(mode=EditingMode.ONE_TAKE.value) - tpl.segments = _make_segments(1) - repo.get.return_value = tpl + repo.get.return_value = template - use_case = ValidateTemplateUseCase(repo) - cmd = ValidateTemplateCommand(template_id="tpl_001", user_id="user_001") - result = use_case.execute(cmd) + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") + result = uc.execute(cmd) assert isinstance(result, ValidateResult) - assert result.template is tpl + assert result.template.id == "t1" assert len(result.warnings) == 0 - def test_one_take_with_multiple_segments_raises(self) -> None: + def test_one_take_zero_segments_raises(self): + template = _make_template(mode=EditingMode.ONE_TAKE.value, segments=[]) repo = MagicMock() - tpl = _make_template(mode=EditingMode.ONE_TAKE.value) - tpl.segments = _make_segments(3) - repo.get.return_value = tpl + repo.get.return_value = template - use_case = ValidateTemplateUseCase(repo) - cmd = ValidateTemplateCommand(template_id="tpl_001", user_id="user_001") + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") + with pytest.raises(ValidationError, match="一镜到底模式必须恰好有 1 个片段"): + uc.execute(cmd) - with pytest.raises(ValidationError, match="恰好有 1 个片段"): - use_case.execute(cmd) - - def test_voice_over_with_valid_material_types_passes(self) -> None: + def test_one_take_multiple_segments_raises(self): + template = _make_template(mode=EditingMode.ONE_TAKE.value, segments=_make_segments(3)) repo = MagicMock() - tpl = _make_template(mode=EditingMode.VOICE_OVER.value, estimated_duration=30.0) - tpl.segments = [ + repo.get.return_value = template + + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") + with pytest.raises(ValidationError): + uc.execute(cmd) + + def test_voice_over_valid_material_types(self): + segments = [ TemplateSegment( - id="s1", template_id="t1", segment_order=0, duration_min=3, duration_max=5, material_type="人物" + id="s1", template_id="t1", segment_order=1, duration_min=3, duration_max=5, material_type="人物" ), TemplateSegment( - id="s2", template_id="t1", segment_order=1, duration_min=3, duration_max=5, material_type="场景" + id="s2", template_id="t1", segment_order=2, duration_min=5, duration_max=8, material_type="场景" ), ] - repo.get.return_value = tpl - - use_case = ValidateTemplateUseCase(repo) - cmd = ValidateTemplateCommand(template_id="tpl_001", user_id="user_001") - result = use_case.execute(cmd) + template = _make_template(mode=EditingMode.VOICE_OVER.value, segments=segments) + repo = MagicMock() + repo.get.return_value = template + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") + result = uc.execute(cmd) assert len(result.warnings) == 0 - def test_voice_over_missing_material_type_raises(self) -> None: - repo = MagicMock() - tpl = _make_template(mode=EditingMode.VOICE_OVER.value) - tpl.segments = [ + def test_voice_over_missing_material_type_raises(self): + segments = [ TemplateSegment( - id="s1", template_id="t1", segment_order=0, duration_min=3, duration_max=5, material_type=None + id="s1", template_id="t1", segment_order=1, duration_min=3, duration_max=5, material_type=None ), ] - repo.get.return_value = tpl - - use_case = ValidateTemplateUseCase(repo) - cmd = ValidateTemplateCommand(template_id="tpl_001", user_id="user_001") + template = _make_template(mode=EditingMode.VOICE_OVER.value, segments=segments) + repo = MagicMock() + repo.get.return_value = template + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") with pytest.raises(ValidationError, match="material_type"): - use_case.execute(cmd) + uc.execute(cmd) - def test_voice_over_invalid_material_type_raises(self) -> None: - repo = MagicMock() - tpl = _make_template(mode=EditingMode.VOICE_OVER.value) - tpl.segments = [ + def test_voice_over_invalid_material_type_raises(self): + segments = [ TemplateSegment( - id="s1", template_id="t1", segment_order=0, duration_min=3, duration_max=5, material_type="动物" + id="s1", template_id="t1", segment_order=1, duration_min=3, duration_max=5, material_type="动物" ), ] - repo.get.return_value = tpl - - use_case = ValidateTemplateUseCase(repo) - cmd = ValidateTemplateCommand(template_id="tpl_001", user_id="user_001") + template = _make_template(mode=EditingMode.VOICE_OVER.value, segments=segments) + repo = MagicMock() + repo.get.return_value = template + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") with pytest.raises(ValidationError, match="material_type"): - use_case.execute(cmd) + uc.execute(cmd) - def test_voiceover_duration_within_range_no_warning(self) -> None: + def test_voice_over_second_segment_invalid(self): + segments = [ + TemplateSegment( + id="s1", template_id="t1", segment_order=1, duration_min=3, duration_max=5, material_type="人物" + ), + TemplateSegment( + id="s2", template_id="t1", segment_order=2, duration_min=5, duration_max=8, material_type="bad" + ), + ] + template = _make_template(mode=EditingMode.VOICE_OVER.value, segments=segments) repo = MagicMock() - tpl = _make_template(mode=EditingMode.ONE_TAKE.value, estimated_duration=60.0) - tpl.segments = _make_segments(1) - repo.get.return_value = tpl + repo.get.return_value = template - use_case = ValidateTemplateUseCase(repo) - # 65s vs 60s = 1.08 ratio，在±30%内 - cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=65.0) - result = use_case.execute(cmd) + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") + with pytest.raises(ValidationError) as exc_info: + uc.execute(cmd) + # 报错应该提到片段2 + assert "2" in str(exc_info.value) - assert len(result.warnings) == 0 - - def test_voiceover_duration_too_short_warns(self) -> None: + def test_voice_duration_mismatch_warning(self): + template = _make_template( + mode=EditingMode.VOICE_OVER.value, + estimated_duration=100.0, + segments=_make_segments(2, material_type="人物"), + ) repo = MagicMock() - tpl = _make_template(mode=EditingMode.ONE_TAKE.value, estimated_duration=60.0) - tpl.segments = _make_segments(1) - repo.get.return_value = tpl + repo.get.return_value = template - use_case = ValidateTemplateUseCase(repo) - # 20s vs 60s = 0.33 ratio，超过±30% - cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=20.0) - result = use_case.execute(cmd) - - assert len(result.warnings) == 1 - assert result.warnings[0].code == "voiceover_duration_mismatch" - assert "偏差超过" in result.warnings[0].message - assert result.warnings[0].details["ratio"] < 0.7 - - def test_voiceover_duration_too_long_warns(self) -> None: - repo = MagicMock() - tpl = _make_template(mode=EditingMode.ONE_TAKE.value, estimated_duration=60.0) - tpl.segments = _make_segments(1) - repo.get.return_value = tpl - - use_case = ValidateTemplateUseCase(repo) - # 100s vs 60s = 1.67 ratio，超过±30% - cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=100.0) - result = use_case.execute(cmd) - - assert len(result.warnings) == 1 - assert result.warnings[0].code == "voiceover_duration_mismatch" - assert result.warnings[0].details["ratio"] > 1.3 - - def test_zero_estimated_duration_no_warning(self) -> None: - repo = MagicMock() - tpl = _make_template(mode=EditingMode.ONE_TAKE.value, estimated_duration=0.0) - tpl.segments = _make_segments(1) - repo.get.return_value = tpl - - use_case = ValidateTemplateUseCase(repo) + uc = ValidateTemplateUseCase(repo) + # 配音时长只有50s，预估100s，偏差50% > 30% cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=50.0) - result = use_case.execute(cmd) + result = uc.execute(cmd) - # estimated_duration=0不做偏差检查 - assert len(result.warnings) == 0 + assert len(result.warnings) == 1 + w = result.warnings[0] + assert isinstance(w, GenerateWarning) + assert w.code == "voiceover_duration_mismatch" + assert "偏差超过" in w.message - def test_no_voiceover_duration_no_warning(self) -> None: + def test_voice_duration_match_no_warning(self): + template = _make_template( + mode=EditingMode.VOICE_OVER.value, + estimated_duration=100.0, + segments=_make_segments(1, material_type="人物"), + ) repo = MagicMock() - tpl = _make_template(estimated_duration=60.0) - tpl.segments = _make_segments(1) - repo.get.return_value = tpl - - use_case = ValidateTemplateUseCase(repo) - cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") # 不传voiceover_duration - result = use_case.execute(cmd) + repo.get.return_value = template + uc = ValidateTemplateUseCase(repo) + # 配音时长95s，预估100s，偏差5% < 30% + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=95.0) + result = uc.execute(cmd) assert len(result.warnings) == 0 - def test_not_found_raises(self) -> None: + def test_voice_duration_at_30_percent_boundary_lower(self): + # 恰好 0.7 边界不触发 + template = _make_template(mode=EditingMode.PIP.value, estimated_duration=100.0) + repo = MagicMock() + repo.get.return_value = template + + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=70.0) + result = uc.execute(cmd) + # 恰好 0.7，不算 < 0.7，应该不触发 + assert len(result.warnings) == 0 + + def test_voice_duration_below_70_percent_triggers(self): + template = _make_template(mode=EditingMode.PIP.value, estimated_duration=100.0) + repo = MagicMock() + repo.get.return_value = template + + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=69.0) + result = uc.execute(cmd) + assert len(result.warnings) == 1 + + def test_voice_duration_above_130_percent_triggers(self): + template = _make_template(mode=EditingMode.PIP.value, estimated_duration=100.0) + repo = MagicMock() + repo.get.return_value = template + + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=131.0) + result = uc.execute(cmd) + assert len(result.warnings) == 1 + + def test_voice_duration_zero_estimated_skip(self): + # estimated_duration = 0 不会做比例计算 + template = _make_template(estimated_duration=0.0) + repo = MagicMock() + repo.get.return_value = template + + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=10.0) + result = uc.execute(cmd) + assert len(result.warnings) == 0 + + def test_voiceover_duration_none_no_warning(self): + template = _make_template() + repo = MagicMock() + repo.get.return_value = template + + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1") + result = uc.execute(cmd) + assert len(result.warnings) == 0 + + def test_validate_not_found_raises(self): repo = MagicMock() repo.get.return_value = None - use_case = ValidateTemplateUseCase(repo) - cmd = ValidateTemplateCommand(template_id="no", user_id="u1") - + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t999", user_id="u1") with pytest.raises(NotFoundError): - use_case.execute(cmd) + uc.execute(cmd) + + def test_warning_details_structure(self): + template = _make_template(estimated_duration=100.0) + repo = MagicMock() + repo.get.return_value = template + + uc = ValidateTemplateUseCase(repo) + cmd = ValidateTemplateCommand(template_id="t1", user_id="u1", voiceover_duration=200.0) + result = uc.execute(cmd) + + assert len(result.warnings) == 1 + details = result.warnings[0].details + assert "voiceover_duration" in details + assert "estimated_duration" in details + assert "ratio" in details + assert details["voiceover_duration"] == 200.0 + assert details["estimated_duration"] == 100.0 + assert details["ratio"] == 2.0 + + +# ── Category Use Cases ────────────────────────────────────────────────────── class TestCategoryUseCases: - def test_create_category(self) -> None: + def test_create_category(self): + cat = TemplateCategory(id="c1", user_id="u1", name="分类A") repo = MagicMock() - cat = TemplateCategory(id="cat_001", user_id="u1", name="新分类") repo.create_category.return_value = cat - use_case = CreateCategoryUseCase(repo) - cmd = CreateCategoryCommand(user_id="u1", name="新分类") - result = use_case.execute(cmd) - - assert result.name == "新分类" + uc = CreateCategoryUseCase(repo) + cmd = CreateCategoryCommand(user_id="u1", name="分类A") + result = uc.execute(cmd) + assert result.name == "分类A" repo.create_category.assert_called_once() - def test_list_categories(self) -> None: + def test_list_categories(self): + cats = [TemplateCategory(id="c1", user_id="u1", name="A"), TemplateCategory(id="c2", user_id="u1", name="B")] repo = MagicMock() - expected = [TemplateCategory(id="c1", user_id="u1", name="A")] - repo.list_categories.return_value = expected + repo.list_categories.return_value = cats - use_case = ListCategoriesUseCase(repo) - result = use_case.execute("u1") - - assert result == expected + uc = ListCategoriesUseCase(repo) + result = uc.execute("u1") + assert len(result) == 2 repo.list_categories.assert_called_once_with("u1") - def test_delete_category(self) -> None: + def test_delete_category(self): repo = MagicMock() repo.delete_category.return_value = True - use_case = DeleteCategoryUseCase(repo) - result = use_case.execute("cat_001", "u1") - + uc = DeleteCategoryUseCase(repo) + result = uc.execute("c1", "u1") assert result is True - repo.delete_category.assert_called_once_with("cat_001", "u1") + repo.delete_category.assert_called_once_with("c1", "u1") + + +# ── Tags Use Case ─────────────────────────────────────────────────────────── class TestListTagsUseCase: - def test_returns_tags_list(self) -> None: + def test_list_tags(self): repo = MagicMock() repo.list_tags.return_value = ["tag1", "tag2", "tag3"] - use_case = ListTagsUseCase(repo) - result = use_case.execute("u1") - + uc = ListTagsUseCase(repo) + result = uc.execute("u1") assert result == ["tag1", "tag2", "tag3"] repo.list_tags.assert_called_once_with("u1") +# ── Usage Stats Use Case ──────────────────────────────────────────────────── + + class TestGetTemplateUsageUseCase: - def test_returns_usage_count(self) -> None: + def test_get_usage(self): repo = MagicMock() - repo.get_usage_count.return_value = 15 + repo.get_usage_count.return_value = 5 - use_case = GetTemplateUsageUseCase(repo) - result = use_case.execute("tpl_001") + uc = GetTemplateUsageUseCase(repo) + result = uc.execute("t1") + assert result == 5 + repo.get_usage_count.assert_called_once_with("t1") - assert result == 15 - repo.get_usage_count.assert_called_once_with("tpl_001") + def test_get_usage_zero(self): + repo = MagicMock() + repo.get_usage_count.return_value = 0 - -class TestGenerateWarning: - def test_warning_default_details(self) -> None: - w = GenerateWarning(code="test_code", message="test message") - assert w.code == "test_code" - assert w.message == "test message" - assert w.details == {} - - def test_warning_with_details(self) -> None: - w = GenerateWarning(code="test", message="msg", details={"key": "value"}) - assert w.details == {"key": "value"} + uc = GetTemplateUsageUseCase(repo) + result = uc.execute("t999") + assert result == 0 -- 2.54.0