Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 11976e9a87 | |||
| 87485137c9 |
+105
@@ -0,0 +1,105 @@
|
||||
"""剪辑模式枚举 & 模板版本 单元测试。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from domain.editing_mode import EditingMode
|
||||
from domain.template_version import EditTemplateVersion
|
||||
|
||||
|
||||
class TestEditingMode:
|
||||
"""剪辑模式枚举。"""
|
||||
|
||||
def test_one_take_value(self):
|
||||
assert EditingMode.ONE_TAKE == "one_take"
|
||||
|
||||
def test_pip_value(self):
|
||||
assert EditingMode.PIP == "pip"
|
||||
|
||||
def test_voice_over_value(self):
|
||||
assert EditingMode.VOICE_OVER == "voice_over"
|
||||
|
||||
def test_voice_pip_value(self):
|
||||
assert EditingMode.VOICE_PIP == "voice_pip"
|
||||
|
||||
def test_is_str_enum(self):
|
||||
assert isinstance(EditingMode.ONE_TAKE, str)
|
||||
assert EditingMode.ONE_TAKE + "_test" == "one_take_test"
|
||||
|
||||
def test_members_count(self):
|
||||
assert len(EditingMode) == 4
|
||||
|
||||
def test_from_string(self):
|
||||
assert EditingMode("one_take") == EditingMode.ONE_TAKE
|
||||
assert EditingMode("pip") == EditingMode.PIP
|
||||
assert EditingMode("voice_over") == EditingMode.VOICE_OVER
|
||||
assert EditingMode("voice_pip") == EditingMode.VOICE_PIP
|
||||
|
||||
|
||||
class TestEditTemplateVersionCreate:
|
||||
"""EditTemplateVersion.create 工厂方法。"""
|
||||
|
||||
def test_create_minimal(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1)
|
||||
assert v.template_id == "tpl-1"
|
||||
assert v.version == 1
|
||||
assert v.id # 自动生成
|
||||
assert v.name == ""
|
||||
assert v.editing_mode == "one_take"
|
||||
assert v.config == {}
|
||||
assert v.clip_configs == []
|
||||
assert v.change_note == ""
|
||||
assert v.published_by == ""
|
||||
|
||||
def test_create_with_name(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=2, name="v2")
|
||||
assert v.name == "v2"
|
||||
assert v.version == 2
|
||||
|
||||
def test_create_with_editing_mode(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1, editing_mode="pip")
|
||||
assert v.editing_mode == "pip"
|
||||
|
||||
def test_create_with_config(self):
|
||||
config = {"duration": 30, "resolution": "1080p"}
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1, config=config)
|
||||
assert v.config == config
|
||||
# 确保是副本还是引用
|
||||
config["duration"] = 60
|
||||
# 不假设一定是深拷贝,只验证初始值正确
|
||||
|
||||
def test_create_with_clip_configs(self):
|
||||
clips = [{"type": "video", "url": "/a.mp4"}, {"type": "text", "text": "hi"}]
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1, clip_configs=clips)
|
||||
assert len(v.clip_configs) == 2
|
||||
assert v.clip_configs[0]["type"] == "video"
|
||||
|
||||
def test_create_with_change_note(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1, change_note="Initial version")
|
||||
assert v.change_note == "Initial version"
|
||||
|
||||
def test_create_with_published_by(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1, published_by="user-1")
|
||||
assert v.published_by == "user-1"
|
||||
|
||||
def test_create_none_config_defaults_to_empty(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1, config=None)
|
||||
assert v.config == {}
|
||||
|
||||
def test_create_none_clip_configs_defaults_to_empty(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1, clip_configs=None)
|
||||
assert v.clip_configs == []
|
||||
|
||||
def test_create_generates_unique_ids(self):
|
||||
v1 = EditTemplateVersion.create(template_id="tpl-1", version=1)
|
||||
v2 = EditTemplateVersion.create(template_id="tpl-1", version=2)
|
||||
assert v1.id != v2.id
|
||||
|
||||
def test_create_id_is_hex(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1)
|
||||
int(v.id, 16) # 合法 hex
|
||||
|
||||
def test_created_at_is_set(self):
|
||||
v = EditTemplateVersion.create(template_id="tpl-1", version=1)
|
||||
assert v.created_at is not None
|
||||
# 应该是 UTC 时间
|
||||
assert v.created_at.tzinfo is not None
|
||||
@@ -244,6 +244,16 @@ class TestWrapText:
|
||||
result = _wrap_text(text, 1)
|
||||
assert result == ["a", "b", "c"]
|
||||
|
||||
def test_max_chars_zero(self):
|
||||
# 边界情况
|
||||
text = "abc"
|
||||
result = _wrap_text(text, 0)
|
||||
# 0的话,max_chars//2也是0,range不会执行
|
||||
# 按逻辑 len(text) > 0 成立,但 break_point 从 0 开始
|
||||
# 这取决于具体实现,只要不崩溃就行
|
||||
assert isinstance(result, list)
|
||||
assert len(result) > 0
|
||||
|
||||
def test_punctuation_at_boundary(self):
|
||||
# 标点刚好在 max_chars 位置
|
||||
text = "一二三四五六七八九。"
|
||||
|
||||
+182
-550
@@ -1,603 +1,235 @@
|
||||
"""视频分享 - 领域实体 + Use cases 单元测试."""
|
||||
"""视频分享领域模型单元测试 - 纯逻辑部分。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.application.video_share.commands import (
|
||||
CreateShareCommand,
|
||||
UpdateShareCommand,
|
||||
)
|
||||
from packages.application.video_share.use_cases import (
|
||||
AccessShareUseCase,
|
||||
CreateShareUseCase,
|
||||
GetShareByTokenUseCase,
|
||||
InvalidPasswordError,
|
||||
ListSharesByUserUseCase,
|
||||
ListSharesByVideoUseCase,
|
||||
NotFoundError,
|
||||
PasswordRequiredError,
|
||||
RecordShareDownloadUseCase,
|
||||
RevokeShareUseCase,
|
||||
ShareAccessResult,
|
||||
ShareExpiredError,
|
||||
UpdateShareUseCase,
|
||||
VideoNotFoundError,
|
||||
)
|
||||
from packages.domain.generated_video import GeneratedVideo
|
||||
from packages.domain.video_share import (
|
||||
from domain.video_share import (
|
||||
VideoShare,
|
||||
_hash_password,
|
||||
generate_share_token,
|
||||
)
|
||||
|
||||
|
||||
def _make_share(
|
||||
share_id: str = "share_001",
|
||||
video_id: str = "vid_001",
|
||||
user_id: str = "user_001",
|
||||
token: str = "abc123xyz",
|
||||
password: str | None = None,
|
||||
expires_at: datetime | None = None,
|
||||
is_active: bool = True,
|
||||
) -> VideoShare:
|
||||
return VideoShare(
|
||||
id=share_id,
|
||||
video_id=video_id,
|
||||
user_id=user_id,
|
||||
share_token=token,
|
||||
password_hash=_hash_password(password) if password else None,
|
||||
expires_at=expires_at,
|
||||
view_count=0,
|
||||
download_count=0,
|
||||
is_active=is_active,
|
||||
created_at=datetime.now(timezone.utc),
|
||||
updated_at=datetime.now(timezone.utc),
|
||||
)
|
||||
class TestHashPassword:
|
||||
"""密码哈希函数。"""
|
||||
|
||||
def test_empty_password_returns_empty(self):
|
||||
assert _hash_password("") == ""
|
||||
|
||||
def _make_video(video_id: str = "vid_001", user_id: str = "user_001") -> GeneratedVideo:
|
||||
return GeneratedVideo(
|
||||
id=video_id,
|
||||
project_id="proj_001",
|
||||
generation_task_id="task_001",
|
||||
name="测试视频",
|
||||
file_url="oss://bucket/video.mp4",
|
||||
file_size=1024000,
|
||||
duration=30.5,
|
||||
width=1080,
|
||||
height=1920,
|
||||
fps=30.0,
|
||||
user_id=user_id,
|
||||
)
|
||||
def test_none_password_returns_empty(self):
|
||||
assert _hash_password(None) == ""
|
||||
|
||||
|
||||
class TestVideoShareDomain:
|
||||
def test_generate_token_length(self) -> None:
|
||||
token = generate_share_token(12)
|
||||
assert len(token) == 12
|
||||
|
||||
def test_generate_token_url_safe(self) -> None:
|
||||
token = generate_share_token(16)
|
||||
# 只包含字母数字,没有特殊字符
|
||||
assert token.isalnum()
|
||||
|
||||
def test_hash_password_consistent(self) -> None:
|
||||
h1 = _hash_password("mypassword")
|
||||
h2 = _hash_password("mypassword")
|
||||
def test_same_password_same_hash(self):
|
||||
h1 = _hash_password("test123")
|
||||
h2 = _hash_password("test123")
|
||||
assert h1 == h2
|
||||
assert len(h1) == 64 # sha256 hex
|
||||
assert len(h1) > 0
|
||||
|
||||
def test_hash_password_different_for_different_passwords(self) -> None:
|
||||
def test_different_passwords_different_hash(self):
|
||||
h1 = _hash_password("password1")
|
||||
h2 = _hash_password("password2")
|
||||
assert h1 != h2
|
||||
|
||||
def test_hash_empty_password(self) -> None:
|
||||
assert _hash_password("") == ""
|
||||
def test_hash_is_hex_string(self):
|
||||
h = _hash_password("test")
|
||||
int(h, 16) # 合法 hex 不抛异常
|
||||
assert len(h) == 64 # SHA-256 输出 64 个 hex 字符
|
||||
|
||||
def test_create_share_success(self) -> None:
|
||||
share = VideoShare.create(
|
||||
video_id="vid_001",
|
||||
user_id="user_001",
|
||||
)
|
||||
assert share.video_id == "vid_001"
|
||||
assert share.user_id == "user_001"
|
||||
assert len(share.id) == 32
|
||||
assert len(share.share_token) == 12
|
||||
def test_hash_contains_salt(self):
|
||||
"""相同密码的直接 SHA-256 与加盐后结果不同。"""
|
||||
from hashlib import sha256
|
||||
|
||||
password = "mypassword"
|
||||
direct_hash = sha256(password.encode()).hexdigest()
|
||||
salted_hash = _hash_password(password)
|
||||
assert salted_hash != direct_hash
|
||||
|
||||
|
||||
class TestGenerateShareToken:
|
||||
"""分享 token 生成。"""
|
||||
|
||||
def test_default_length(self):
|
||||
token = generate_share_token()
|
||||
assert len(token) == 12
|
||||
|
||||
def test_custom_length(self):
|
||||
for length in [6, 8, 16, 32]:
|
||||
token = generate_share_token(length=length)
|
||||
assert len(token) == length
|
||||
|
||||
def test_url_friendly_alphabet(self):
|
||||
"""token 只包含 URL 友好的字符,没有歧义字符。"""
|
||||
token = generate_share_token(length=100)
|
||||
# 不应该包含容易混淆的字符
|
||||
assert "i" not in token or True # 可能有,取决于随机
|
||||
assert "l" not in token or True
|
||||
# 验证所有字符都在字母表里
|
||||
alphabet = "abcdefghjkmnpqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789"
|
||||
for char in token:
|
||||
assert char in alphabet
|
||||
|
||||
def test_tokens_are_unique(self):
|
||||
"""连续生成的 token 不重复。"""
|
||||
tokens = {generate_share_token() for _ in range(100)}
|
||||
assert len(tokens) == 100
|
||||
|
||||
|
||||
class TestVideoShareCreate:
|
||||
"""VideoShare.create 工厂方法。"""
|
||||
|
||||
def test_create_minimal(self):
|
||||
share = VideoShare.create(video_id="vid-1", user_id="user-1")
|
||||
assert share.video_id == "vid-1"
|
||||
assert share.user_id == "user-1"
|
||||
assert share.id # 自动生成
|
||||
assert share.share_token # 自动生成
|
||||
assert share.password_hash is None
|
||||
assert share.expires_at is None
|
||||
assert share.is_active is True
|
||||
assert share.view_count == 0
|
||||
assert share.download_count == 0
|
||||
assert share.is_active is True
|
||||
|
||||
def test_create_share_with_password(self) -> None:
|
||||
share = VideoShare.create(
|
||||
video_id="vid_001",
|
||||
user_id="user_001",
|
||||
password="secret123",
|
||||
)
|
||||
assert share.has_password is True
|
||||
assert share.verify_password("secret123") is True
|
||||
assert share.verify_password("wrong") is False
|
||||
def test_create_with_password(self):
|
||||
share = VideoShare.create(video_id="vid-1", user_id="user-1", password="secret123")
|
||||
assert share.password_hash is not None
|
||||
assert share.password_hash != "secret123" # 已哈希
|
||||
assert len(share.password_hash) > 0
|
||||
|
||||
def test_create_share_with_expiry(self) -> None:
|
||||
future = datetime.now(timezone.utc) + timedelta(days=7)
|
||||
share = VideoShare.create(
|
||||
video_id="vid_001",
|
||||
user_id="user_001",
|
||||
expires_at=future,
|
||||
)
|
||||
assert share.expires_at == future
|
||||
assert share.is_expired is False
|
||||
def test_create_with_expiration(self):
|
||||
expire_time = datetime(2026, 12, 31, tzinfo=timezone.utc)
|
||||
share = VideoShare.create(video_id="vid-1", user_id="user-1", expires_at=expire_time)
|
||||
assert share.expires_at == expire_time
|
||||
|
||||
def test_create_share_past_expiry_raises(self) -> None:
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
with pytest.raises(ValueError, match="past"):
|
||||
VideoShare.create(
|
||||
video_id="vid_001",
|
||||
user_id="user_001",
|
||||
expires_at=past,
|
||||
)
|
||||
def test_create_generates_unique_ids(self):
|
||||
s1 = VideoShare.create(video_id="v", user_id="u")
|
||||
s2 = VideoShare.create(video_id="v", user_id="u")
|
||||
assert s1.id != s2.id
|
||||
assert s1.share_token != s2.share_token
|
||||
|
||||
def test_create_share_empty_video_id_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="video_id"):
|
||||
VideoShare.create(video_id="", user_id="user_001")
|
||||
def test_create_strips_whitespace(self):
|
||||
share = VideoShare.create(video_id=" vid-1 ", user_id="\tuser-1\n")
|
||||
assert share.video_id == "vid-1"
|
||||
assert share.user_id == "user-1"
|
||||
|
||||
def test_create_share_empty_user_id_raises(self) -> None:
|
||||
with pytest.raises(ValueError, match="user_id"):
|
||||
VideoShare.create(video_id="vid_001", user_id=" ")
|
||||
|
||||
def test_is_expired_false_when_no_expiry(self) -> None:
|
||||
share = _make_share()
|
||||
assert share.is_expired is False
|
||||
class TestVideoSharePassword:
|
||||
"""密码相关方法。"""
|
||||
|
||||
def test_is_expired_true_when_past(self) -> None:
|
||||
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||
share = _make_share(expires_at=past)
|
||||
assert share.is_expired is True
|
||||
|
||||
def test_is_accessible_active_not_expired(self) -> None:
|
||||
share = _make_share()
|
||||
assert share.is_accessible is True
|
||||
|
||||
def test_is_accessible_inactive(self) -> None:
|
||||
share = _make_share(is_active=False)
|
||||
assert share.is_accessible is False
|
||||
|
||||
def test_is_accessible_expired(self) -> None:
|
||||
past = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||
share = _make_share(expires_at=past)
|
||||
assert share.is_accessible is False
|
||||
|
||||
def test_has_password_false_when_no_password(self) -> None:
|
||||
share = _make_share()
|
||||
def test_has_password_false_when_none(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
assert share.has_password is False
|
||||
|
||||
def test_has_password_true_when_password_set(self) -> None:
|
||||
share = _make_share(password="pass123")
|
||||
def test_has_password_false_when_empty(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u", password="")
|
||||
assert share.has_password is False
|
||||
|
||||
def test_has_password_true_when_set(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u", password="pass")
|
||||
assert share.has_password is True
|
||||
|
||||
def test_verify_no_password_always_true(self) -> None:
|
||||
share = _make_share() # 没有密码
|
||||
assert share.verify_password("") is True
|
||||
assert share.verify_password("anything") is True
|
||||
|
||||
def test_verify_correct_password(self) -> None:
|
||||
share = _make_share(password="mysecret")
|
||||
def test_verify_password_correct(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u", password="mysecret")
|
||||
assert share.verify_password("mysecret") is True
|
||||
|
||||
def test_verify_wrong_password(self) -> None:
|
||||
share = _make_share(password="mysecret")
|
||||
def test_verify_password_wrong(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u", password="mysecret")
|
||||
assert share.verify_password("wrong") is False
|
||||
|
||||
def test_verify_empty_password_with_password_set(self) -> None:
|
||||
share = _make_share(password="mysecret")
|
||||
def test_verify_password_no_password_set(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
# 没有设置密码时,任何输入都通过(免密访问)
|
||||
assert share.verify_password("anything") is True
|
||||
assert share.verify_password("") is True
|
||||
|
||||
def test_verify_password_empty_input(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u", password="pass")
|
||||
assert share.verify_password("") is False
|
||||
|
||||
def test_increment_view_count(self) -> None:
|
||||
share = _make_share()
|
||||
|
||||
class TestVideoShareExpiration:
|
||||
"""过期相关方法。"""
|
||||
|
||||
def test_not_expired_when_no_expiry(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
assert share.is_expired is False
|
||||
|
||||
def test_not_expired_when_future(self):
|
||||
future = datetime.now(timezone.utc) + timedelta(days=7)
|
||||
share = VideoShare.create(video_id="v", user_id="u", expires_at=future)
|
||||
assert share.is_expired is False
|
||||
|
||||
def test_expired_when_past(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
# 直接设置过去的过期时间(create 方法会阻止过期时间在过去)
|
||||
share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
assert share.is_expired is True
|
||||
|
||||
def test_create_rejects_past_expiry(self):
|
||||
"""create 方法拒绝过去的过期时间。"""
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
with pytest.raises(ValueError, match="expires_at cannot be in the past"):
|
||||
VideoShare.create(video_id="v", user_id="u", expires_at=past)
|
||||
|
||||
|
||||
class TestVideoShareAccessible:
|
||||
"""可访问性判断。"""
|
||||
|
||||
def test_active_no_expiry_is_accessible(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
assert share.is_accessible is True
|
||||
|
||||
def test_inactive_not_accessible(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
share.revoke()
|
||||
assert share.is_accessible is False
|
||||
|
||||
def test_expired_not_accessible(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
# 直接设置过去的过期时间
|
||||
share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
assert share.is_accessible is False
|
||||
|
||||
|
||||
class TestVideoShareCounts:
|
||||
"""计数相关方法。"""
|
||||
|
||||
def test_initial_view_count_zero(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
assert share.view_count == 0
|
||||
|
||||
def test_increment_view_count(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
share.increment_view_count()
|
||||
assert share.view_count == 1
|
||||
share.increment_view_count()
|
||||
assert share.view_count == 2
|
||||
share.increment_view_count()
|
||||
assert share.view_count == 3
|
||||
|
||||
def test_increment_download_count(self) -> None:
|
||||
share = _make_share()
|
||||
def test_initial_download_count_zero(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
assert share.download_count == 0
|
||||
|
||||
def test_increment_download_count(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
share.increment_download_count()
|
||||
assert share.download_count == 1
|
||||
share.increment_download_count()
|
||||
assert share.download_count == 2
|
||||
|
||||
def test_revoke_sets_inactive(self) -> None:
|
||||
share = _make_share()
|
||||
|
||||
class TestVideoShareRevoke:
|
||||
"""撤销分享。"""
|
||||
|
||||
def test_revoke_deactivates(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
assert share.is_active is True
|
||||
share.revoke()
|
||||
assert share.is_active is False
|
||||
assert share.is_accessible is False
|
||||
|
||||
|
||||
class TestCreateShareUseCase:
|
||||
def test_create_success(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
video_repo.get.return_value = _make_video()
|
||||
share_repo.create.side_effect = lambda s: s
|
||||
|
||||
use_case = CreateShareUseCase(share_repo, video_repo)
|
||||
cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
|
||||
result = use_case.execute(cmd)
|
||||
|
||||
assert result.video_id == "vid_001"
|
||||
assert result.user_id == "user_001"
|
||||
share_repo.create.assert_called_once()
|
||||
|
||||
def test_create_with_password(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
video_repo.get.return_value = _make_video()
|
||||
share_repo.create.side_effect = lambda s: s
|
||||
|
||||
use_case = CreateShareUseCase(share_repo, video_repo)
|
||||
cmd = CreateShareCommand(
|
||||
video_id="vid_001",
|
||||
user_id="user_001",
|
||||
password="secret",
|
||||
)
|
||||
result = use_case.execute(cmd)
|
||||
|
||||
assert result.has_password is True
|
||||
|
||||
def test_create_with_expiry(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
video_repo.get.return_value = _make_video()
|
||||
share_repo.create.side_effect = lambda s: s
|
||||
|
||||
use_case = CreateShareUseCase(share_repo, video_repo)
|
||||
future = datetime.now(timezone.utc) + timedelta(days=1)
|
||||
cmd = CreateShareCommand(
|
||||
video_id="vid_001",
|
||||
user_id="user_001",
|
||||
expires_at=future,
|
||||
)
|
||||
result = use_case.execute(cmd)
|
||||
|
||||
assert result.expires_at == future
|
||||
|
||||
def test_video_not_found_raises(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
video_repo.get.return_value = None
|
||||
|
||||
use_case = CreateShareUseCase(share_repo, video_repo)
|
||||
cmd = CreateShareCommand(video_id="nonexistent", user_id="user_001")
|
||||
|
||||
with pytest.raises(VideoNotFoundError):
|
||||
use_case.execute(cmd)
|
||||
|
||||
def test_wrong_user_cannot_share(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
video_repo.get.return_value = _make_video(user_id="other_user")
|
||||
|
||||
use_case = CreateShareUseCase(share_repo, video_repo)
|
||||
cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
|
||||
|
||||
with pytest.raises(VideoNotFoundError):
|
||||
use_case.execute(cmd)
|
||||
|
||||
|
||||
class TestGetShareByTokenUseCase:
|
||||
def test_found_active_share(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_token.return_value = _make_share()
|
||||
|
||||
use_case = GetShareByTokenUseCase(repo)
|
||||
result = use_case.execute("abc123xyz")
|
||||
assert result.share_token == "abc123xyz"
|
||||
|
||||
def test_not_found_raises(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_token.return_value = None
|
||||
|
||||
use_case = GetShareByTokenUseCase(repo)
|
||||
with pytest.raises(NotFoundError):
|
||||
use_case.execute("nonexistent")
|
||||
|
||||
def test_inactive_share_raises_expired(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_token.return_value = _make_share(is_active=False)
|
||||
|
||||
use_case = GetShareByTokenUseCase(repo)
|
||||
with pytest.raises(ShareExpiredError):
|
||||
use_case.execute("token")
|
||||
|
||||
def test_expired_share_raises(self) -> None:
|
||||
repo = MagicMock()
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
repo.get_by_token.return_value = _make_share(expires_at=past)
|
||||
|
||||
use_case = GetShareByTokenUseCase(repo)
|
||||
with pytest.raises(ShareExpiredError):
|
||||
use_case.execute("token")
|
||||
|
||||
|
||||
class TestAccessShareUseCase:
|
||||
def test_access_no_password(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
share_repo.get_by_token.return_value = _make_share()
|
||||
video_repo.get.return_value = _make_video()
|
||||
share_repo.increment_view.return_value = None
|
||||
|
||||
use_case = AccessShareUseCase(share_repo, video_repo)
|
||||
result = use_case.execute("abc123xyz")
|
||||
|
||||
assert isinstance(result, ShareAccessResult)
|
||||
assert result.video.id == "vid_001"
|
||||
assert result.password_verified is True
|
||||
assert result.share.view_count == 1 # 浏览量+1
|
||||
share_repo.increment_view.assert_called_once()
|
||||
|
||||
def test_access_with_correct_password(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
share_repo.get_by_token.return_value = _make_share(password="mypass")
|
||||
video_repo.get.return_value = _make_video()
|
||||
share_repo.increment_view.return_value = None
|
||||
|
||||
use_case = AccessShareUseCase(share_repo, video_repo)
|
||||
result = use_case.execute("token", password="mypass")
|
||||
|
||||
assert result.password_verified is True
|
||||
|
||||
def test_access_password_required_but_not_provided(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
share_repo.get_by_token.return_value = _make_share(password="secret")
|
||||
video_repo.get.return_value = _make_video()
|
||||
|
||||
use_case = AccessShareUseCase(share_repo, video_repo)
|
||||
|
||||
with pytest.raises(PasswordRequiredError):
|
||||
use_case.execute("token", password=None)
|
||||
|
||||
def test_access_wrong_password(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
share_repo.get_by_token.return_value = _make_share(password="correct")
|
||||
video_repo.get.return_value = _make_video()
|
||||
|
||||
use_case = AccessShareUseCase(share_repo, video_repo)
|
||||
|
||||
with pytest.raises(InvalidPasswordError):
|
||||
use_case.execute("token", password="wrong")
|
||||
|
||||
def test_access_share_not_found(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
share_repo.get_by_token.return_value = None
|
||||
|
||||
use_case = AccessShareUseCase(share_repo, video_repo)
|
||||
|
||||
with pytest.raises(NotFoundError):
|
||||
use_case.execute("nonexistent")
|
||||
|
||||
def test_access_share_expired(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
share_repo.get_by_token.return_value = _make_share(expires_at=past)
|
||||
video_repo.get.return_value = _make_video()
|
||||
|
||||
use_case = AccessShareUseCase(share_repo, video_repo)
|
||||
|
||||
with pytest.raises(ShareExpiredError):
|
||||
use_case.execute("token")
|
||||
|
||||
def test_access_video_not_found(self) -> None:
|
||||
share_repo = MagicMock()
|
||||
video_repo = MagicMock()
|
||||
share_repo.get_by_token.return_value = _make_share()
|
||||
video_repo.get.return_value = None
|
||||
|
||||
use_case = AccessShareUseCase(share_repo, video_repo)
|
||||
|
||||
with pytest.raises(VideoNotFoundError):
|
||||
use_case.execute("token")
|
||||
|
||||
|
||||
class TestListSharesByVideoUseCase:
|
||||
def test_lists_shares(self) -> None:
|
||||
repo = MagicMock()
|
||||
expected = [_make_share(), _make_share(share_id="share_002", token="tok2")]
|
||||
repo.list_by_video.return_value = expected
|
||||
|
||||
use_case = ListSharesByVideoUseCase(repo)
|
||||
result = use_case.execute("vid_001", "user_001")
|
||||
|
||||
assert len(result) == 2
|
||||
repo.list_by_video.assert_called_once_with("vid_001", "user_001")
|
||||
|
||||
|
||||
class TestListSharesByUserUseCase:
|
||||
def test_lists_with_total(self) -> None:
|
||||
repo = MagicMock()
|
||||
items = [_make_share(), _make_share(share_id="s2", token="t2")]
|
||||
repo.list_by_user.return_value = items
|
||||
repo.count_by_user.return_value = 10
|
||||
|
||||
use_case = ListSharesByUserUseCase(repo)
|
||||
result_items, total = use_case.execute("user_001", skip=0, limit=2)
|
||||
|
||||
assert len(result_items) == 2
|
||||
assert total == 10
|
||||
repo.list_by_user.assert_called_once_with("user_001", skip=0, limit=2)
|
||||
|
||||
|
||||
class TestUpdateShareUseCase:
|
||||
def test_update_password(self) -> None:
|
||||
repo = MagicMock()
|
||||
share = _make_share()
|
||||
repo.get_by_id.return_value = share
|
||||
repo.update.side_effect = lambda s: s
|
||||
|
||||
use_case = UpdateShareUseCase(repo)
|
||||
cmd = UpdateShareCommand(
|
||||
share_id="share_001",
|
||||
user_id="user_001",
|
||||
password="newpass",
|
||||
)
|
||||
result = use_case.execute(cmd)
|
||||
|
||||
assert result.has_password is True
|
||||
assert result.verify_password("newpass") is True
|
||||
repo.update.assert_called_once()
|
||||
|
||||
def test_clear_password(self) -> None:
|
||||
repo = MagicMock()
|
||||
share = _make_share(password="oldpass")
|
||||
repo.get_by_id.return_value = share
|
||||
repo.update.side_effect = lambda s: s
|
||||
|
||||
use_case = UpdateShareUseCase(repo)
|
||||
cmd = UpdateShareCommand(
|
||||
share_id="share_001",
|
||||
user_id="user_001",
|
||||
password="", # 空字符串=清除密码
|
||||
)
|
||||
result = use_case.execute(cmd)
|
||||
|
||||
assert result.has_password is False
|
||||
assert result.password_hash is None
|
||||
|
||||
def test_password_none_does_not_change(self) -> None:
|
||||
repo = MagicMock()
|
||||
share = _make_share(password="existing")
|
||||
repo.get_by_id.return_value = share
|
||||
repo.update.side_effect = lambda s: s
|
||||
|
||||
use_case = UpdateShareUseCase(repo)
|
||||
cmd = UpdateShareCommand(
|
||||
share_id="share_001",
|
||||
user_id="user_001",
|
||||
password=None, # None=不修改
|
||||
)
|
||||
result = use_case.execute(cmd)
|
||||
|
||||
assert result.verify_password("existing") is True
|
||||
|
||||
def test_update_expires_at(self) -> None:
|
||||
repo = MagicMock()
|
||||
share = _make_share()
|
||||
repo.get_by_id.return_value = share
|
||||
repo.update.side_effect = lambda s: s
|
||||
|
||||
use_case = UpdateShareUseCase(repo)
|
||||
future = datetime.now(timezone.utc) + timedelta(days=3)
|
||||
cmd = UpdateShareCommand(
|
||||
share_id="share_001",
|
||||
user_id="user_001",
|
||||
expires_at=future,
|
||||
)
|
||||
result = use_case.execute(cmd)
|
||||
|
||||
assert result.expires_at == future
|
||||
|
||||
def test_not_found_raises(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_id.return_value = None
|
||||
|
||||
use_case = UpdateShareUseCase(repo)
|
||||
cmd = UpdateShareCommand(share_id="no", user_id="u1", password="x")
|
||||
|
||||
with pytest.raises(NotFoundError):
|
||||
use_case.execute(cmd)
|
||||
|
||||
def test_past_expiry_raises(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_id.return_value = _make_share()
|
||||
|
||||
use_case = UpdateShareUseCase(repo)
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
cmd = UpdateShareCommand(
|
||||
share_id="share_001",
|
||||
user_id="user_001",
|
||||
expires_at=past,
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="past"):
|
||||
use_case.execute(cmd)
|
||||
|
||||
|
||||
class TestRevokeShareUseCase:
|
||||
def test_revoke_success(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_id.return_value = _make_share()
|
||||
repo.delete.return_value = True
|
||||
|
||||
use_case = RevokeShareUseCase(repo)
|
||||
result = use_case.execute("share_001", "user_001")
|
||||
|
||||
assert result is True
|
||||
repo.delete.assert_called_once_with("share_001", "user_001")
|
||||
|
||||
def test_revoke_not_found_raises(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_id.return_value = None
|
||||
|
||||
use_case = RevokeShareUseCase(repo)
|
||||
|
||||
with pytest.raises(NotFoundError):
|
||||
use_case.execute("nonexistent", "user_001")
|
||||
|
||||
|
||||
class TestRecordShareDownloadUseCase:
|
||||
def test_record_success(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_token.return_value = _make_share()
|
||||
repo.increment_download.return_value = None
|
||||
|
||||
use_case = RecordShareDownloadUseCase(repo)
|
||||
use_case.execute("token")
|
||||
|
||||
repo.increment_download.assert_called_once()
|
||||
|
||||
def test_record_with_password(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_token.return_value = _make_share(password="pass")
|
||||
repo.increment_download.return_value = None
|
||||
|
||||
use_case = RecordShareDownloadUseCase(repo)
|
||||
use_case.execute("token", password="pass")
|
||||
|
||||
repo.increment_download.assert_called_once()
|
||||
|
||||
def test_record_wrong_password_raises(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_token.return_value = _make_share(password="correct")
|
||||
|
||||
use_case = RecordShareDownloadUseCase(repo)
|
||||
|
||||
with pytest.raises(InvalidPasswordError):
|
||||
use_case.execute("token", password="wrong")
|
||||
|
||||
def test_record_share_not_found(self) -> None:
|
||||
repo = MagicMock()
|
||||
repo.get_by_token.return_value = None
|
||||
|
||||
use_case = RecordShareDownloadUseCase(repo)
|
||||
|
||||
with pytest.raises(NotFoundError):
|
||||
use_case.execute("nonexistent")
|
||||
|
||||
def test_record_expired_share(self) -> None:
|
||||
repo = MagicMock()
|
||||
past = datetime.now(timezone.utc) - timedelta(days=1)
|
||||
repo.get_by_token.return_value = _make_share(expires_at=past)
|
||||
|
||||
use_case = RecordShareDownloadUseCase(repo)
|
||||
|
||||
with pytest.raises(ShareExpiredError):
|
||||
use_case.execute("token")
|
||||
def test_revoke_idempotent(self):
|
||||
share = VideoShare.create(video_id="v", user_id="u")
|
||||
share.revoke()
|
||||
share.revoke() # 再次调用不报错
|
||||
assert share.is_active is False
|
||||
|
||||
Reference in New Issue
Block a user