xiaoxia
ebe68429bc
CI/CD Pipeline / Check if frontend-only change (push) Has been cancelled
CI/CD Pipeline / Validate - Code Quality (push) Has been cancelled
CI/CD Pipeline / Validate - Type Check (mypy) (push) Has been cancelled
CI/CD Pipeline / Validate - Migration (alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Lint (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / PR Build API Image (push) Has been cancelled
CI/CD Pipeline / PR Build Web Image (push) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
489 lines
16 KiB
Python
Executable File
489 lines
16 KiB
Python
Executable File
"""视频分享 Use Cases 单元测试 — wave215"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from packages.application.video_share.commands import (
|
|
CreateShareCommand,
|
|
UpdateShareCommand,
|
|
)
|
|
from packages.application.video_share.use_cases import (
|
|
AccessShareUseCase,
|
|
CreateShareUseCase,
|
|
GetShareByTokenUseCase,
|
|
InvalidPasswordError,
|
|
ListSharesByUserUseCase,
|
|
ListSharesByVideoUseCase,
|
|
NotFoundError,
|
|
PasswordRequiredError,
|
|
RecordShareDownloadUseCase,
|
|
RevokeShareUseCase,
|
|
ShareAccessResult,
|
|
ShareExpiredError,
|
|
UpdateShareUseCase,
|
|
VideoNotFoundError,
|
|
)
|
|
from packages.domain.generated_video import GeneratedVideo
|
|
from packages.domain.video_share import VideoShare
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
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_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
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
# ── CreateShareUseCase ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestCreateShareUseCase:
|
|
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
|
|
|
|
uc = CreateShareUseCase(share_repo, video_repo)
|
|
cmd = CreateShareCommand(video_id="v1", user_id="u1")
|
|
result = uc.execute(cmd)
|
|
|
|
assert result.video_id == "v1"
|
|
assert result.user_id == "u1"
|
|
video_repo.get.assert_called_once_with("v1")
|
|
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
|
|
|
|
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
|
|
|
|
def test_video_not_found_raises(self):
|
|
share_repo = MagicMock()
|
|
video_repo = MagicMock()
|
|
video_repo.get.return_value = None
|
|
|
|
uc = CreateShareUseCase(share_repo, video_repo)
|
|
cmd = CreateShareCommand(video_id="v999", user_id="u1")
|
|
|
|
with pytest.raises(VideoNotFoundError):
|
|
uc.execute(cmd)
|
|
|
|
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
|
|
|
|
uc = CreateShareUseCase(share_repo, video_repo)
|
|
cmd = CreateShareCommand(video_id="v1", user_id="u1")
|
|
|
|
with pytest.raises(VideoNotFoundError):
|
|
uc.execute(cmd)
|
|
|
|
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:
|
|
def test_get_success(self):
|
|
share = _make_share()
|
|
repo = MagicMock()
|
|
repo.get_by_token.return_value = share
|
|
|
|
uc = GetShareByTokenUseCase(repo)
|
|
result = uc.execute(share.share_token)
|
|
assert result.id == share.id
|
|
|
|
def test_not_found_raises(self):
|
|
repo = MagicMock()
|
|
repo.get_by_token.return_value = None
|
|
|
|
uc = GetShareByTokenUseCase(repo)
|
|
with pytest.raises(NotFoundError):
|
|
uc.execute("nonexistent")
|
|
|
|
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):
|
|
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:
|
|
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
|
|
|
|
uc = AccessShareUseCase(share_repo, video_repo)
|
|
result = uc.execute(share.share_token)
|
|
|
|
assert isinstance(result, ShareAccessResult)
|
|
assert result.share.id == share.id
|
|
assert result.video.id == video.id
|
|
assert result.password_verified is True
|
|
assert share.view_count == 1
|
|
share_repo.increment_view.assert_called_once_with(share.id)
|
|
|
|
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
|
|
|
|
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):
|
|
uc.execute(share.share_token)
|
|
|
|
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):
|
|
uc.execute(share.share_token, password="wrong")
|
|
|
|
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):
|
|
uc.execute(share.share_token)
|
|
|
|
def test_access_share_not_found(self):
|
|
share_repo = MagicMock()
|
|
share_repo.get_by_token.return_value = None
|
|
|
|
uc = AccessShareUseCase(share_repo, MagicMock())
|
|
with pytest.raises(NotFoundError):
|
|
uc.execute("nonexistent")
|
|
|
|
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):
|
|
uc.execute(share.share_token)
|
|
|
|
|
|
# ── ListSharesByVideoUseCase ────────────────────────────────────────────────
|
|
|
|
|
|
class TestListSharesByVideoUseCase:
|
|
def test_list_success(self):
|
|
shares = [_make_share(), _make_share()]
|
|
repo = MagicMock()
|
|
repo.list_by_video.return_value = shares
|
|
|
|
uc = ListSharesByVideoUseCase(repo)
|
|
result = uc.execute("v1", "u1")
|
|
|
|
assert len(result) == 2
|
|
repo.list_by_video.assert_called_once_with("v1", "u1")
|
|
|
|
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:
|
|
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
|
|
|
|
uc = ListSharesByUserUseCase(repo)
|
|
items, total = uc.execute("u1", skip=0, limit=5)
|
|
|
|
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")
|
|
|
|
def test_list_default_params(self):
|
|
repo = MagicMock()
|
|
repo.list_by_user.return_value = []
|
|
repo.count_by_user.return_value = 0
|
|
|
|
uc = ListSharesByUserUseCase(repo)
|
|
uc.execute("u1")
|
|
|
|
repo.list_by_user.assert_called_once_with("u1", skip=0, limit=20)
|
|
|
|
|
|
# ── UpdateShareUseCase ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestUpdateShareUseCase:
|
|
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
|
|
|
|
uc = UpdateShareUseCase(repo)
|
|
cmd = UpdateShareCommand(share_id=share.id, user_id="u1", password="newpass")
|
|
result = uc.execute(cmd)
|
|
|
|
assert result is not None
|
|
assert share.verify_password("newpass") is True
|
|
assert share.verify_password("oldpass") is False
|
|
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
|
|
|
|
uc = UpdateShareUseCase(repo)
|
|
cmd = UpdateShareCommand(share_id=share.id, user_id="u1", password="")
|
|
result = uc.execute(cmd)
|
|
|
|
assert result.has_password is False
|
|
|
|
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
|
|
|
|
uc = UpdateShareUseCase(repo)
|
|
cmd = UpdateShareCommand(share_id=share.id, user_id="u1", password=None)
|
|
result = uc.execute(cmd)
|
|
|
|
# password=None 表示不修改
|
|
assert result.has_password is True
|
|
assert share.verify_password("oldpass") is True
|
|
|
|
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
|
|
|
|
uc = UpdateShareUseCase(repo)
|
|
cmd = UpdateShareCommand(share_id=share.id, user_id="u1", expires_at=new_expiry)
|
|
result = uc.execute(cmd)
|
|
|
|
assert result.expires_at == new_expiry
|
|
|
|
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
|
|
|
|
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"):
|
|
uc.execute(cmd)
|
|
|
|
def test_update_not_found_raises(self):
|
|
repo = MagicMock()
|
|
repo.get_by_id.return_value = None
|
|
|
|
uc = UpdateShareUseCase(repo)
|
|
cmd = UpdateShareCommand(share_id="nonexistent", user_id="u1")
|
|
|
|
with pytest.raises(NotFoundError):
|
|
uc.execute(cmd)
|
|
|
|
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:
|
|
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
|
|
repo.delete.assert_called_once_with("s1", "u1")
|
|
|
|
def test_revoke_not_found_raises(self):
|
|
repo = MagicMock()
|
|
repo.get_by_id.return_value = None
|
|
|
|
uc = RevokeShareUseCase(repo)
|
|
with pytest.raises(NotFoundError):
|
|
uc.execute("s1", "u1")
|
|
|
|
|
|
# ── RecordShareDownloadUseCase ──────────────────────────────────────────────
|
|
|
|
|
|
class TestRecordShareDownloadUseCase:
|
|
def test_record_download_success(self):
|
|
share = _make_share(download_count=3)
|
|
repo = MagicMock()
|
|
repo.get_by_token.return_value = share
|
|
|
|
uc = RecordShareDownloadUseCase(repo)
|
|
uc.execute(share.share_token)
|
|
|
|
repo.increment_download.assert_called_once_with(share.id)
|
|
|
|
def test_record_download_with_password(self):
|
|
share = _make_share(password="secret")
|
|
repo = MagicMock()
|
|
repo.get_by_token.return_value = share
|
|
|
|
uc = RecordShareDownloadUseCase(repo)
|
|
uc.execute(share.share_token, password="secret")
|
|
repo.increment_download.assert_called_once()
|
|
|
|
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):
|
|
uc.execute(share.share_token, password="wrong")
|
|
|
|
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):
|
|
uc.execute(share.share_token)
|
|
|
|
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")
|