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

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