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")