diff --git a/alembic/versions/050_video_shares.py b/alembic/versions/050_video_shares.py new file mode 100755 index 000000000..0bd35d32b --- /dev/null +++ b/alembic/versions/050_video_shares.py @@ -0,0 +1,48 @@ +"""#P3-2 - 视频分享表 video_shares + +Revision ID: 050 +Revises: 049 +Create Date: 2026-07-22 + +Changes: +1. 新建 video_shares 表，支持视频匿名分享链接 +2. share_token 唯一索引，用于公开分享URL +3. 支持密码保护、有效期、浏览/下载计数 +""" + +import sqlalchemy as sa + +from alembic import context, op + +revision = "050_video_shares" +down_revision = "049_wechat_login_phone" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + conn = op.get_bind() + if context.get_context().dialect.name == "postgresql": + # 检查表是否已存在（幂等） + result = conn.execute(sa.text("SELECT to_regclass('public.video_shares')")) + if result.scalar() is not None: + return + + op.create_table( + "video_shares", + sa.Column("id", sa.String(32), primary_key=True), + sa.Column("video_id", sa.String(32), nullable=False, index=True), + sa.Column("user_id", sa.String(32), nullable=False, index=True), + sa.Column("share_token", sa.String(16), nullable=False, unique=True), + sa.Column("password_hash", sa.String(255), nullable=True), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("view_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("download_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.true()), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + ) + + +def downgrade() -> None: + op.drop_table("video_shares") diff --git a/apps/api/app/api/router.py b/apps/api/app/api/router.py index a06c0a86c..15cb50fe2 100755 --- a/apps/api/app/api/router.py +++ b/apps/api/app/api/router.py @@ -11,6 +11,7 @@ from app.api.routes.health import router as health_check_router from app.api.routes.ingest_jobs import router as ingest_jobs_router from app.api.routes.internal_render import router as internal_render_router from app.api.routes.projects import router as projects_router +from app.api.routes.share import router as share_router from app.api.routes.subscription import router as subscription_router from app.api.routes.tags import router as tags_router from app.api.routes.task_center import router as task_center_router @@ -104,6 +105,10 @@ api_router.include_router( videos_router, tags=["VideoCenter"], ) +api_router.include_router( + share_router, + tags=["Share"], +) api_router.include_router( duplication_router, prefix="/duplication", diff --git a/apps/api/app/api/routes/share.py b/apps/api/app/api/routes/share.py new file mode 100755 index 000000000..f180da38f --- /dev/null +++ b/apps/api/app/api/routes/share.py @@ -0,0 +1,298 @@ +"""视频分享 API 路由.""" + +from __future__ import annotations + +import logging +from typing import Optional + +from app.api.routes._helpers import format_utc_datetime +from app.auth import AuthenticatedUser, get_current_user +from app.core.storage import OSSStorageService, get_storage_service +from app.dependencies import get_db_session, get_generated_video_repository +from app.schemas.video_share import ( + CreateShareRequest, + ShareAccessResponse, + ShareListResponse, + ShareMetaResponse, + ShareResponse, + UpdateShareRequest, + VerifySharePasswordRequest, +) +from fastapi import APIRouter, Depends, HTTPException, Query, Response, status +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.video_share_repository import ( + SQLAlchemyVideoShareRepository, +) +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, + ShareExpiredError, + UpdateShareUseCase, + VideoNotFoundError, +) +from packages.ports.generated_video_repository import GeneratedVideoRepository + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +def _get_share_repository( + session: Session = Depends(get_db_session), +) -> SQLAlchemyVideoShareRepository: + return SQLAlchemyVideoShareRepository(session) + + +def _to_share_response(share) -> ShareResponse: + return ShareResponse( + id=share.id, + video_id=share.video_id, + share_token=share.share_token, + has_password=share.has_password, + expires_at=share.expires_at, + view_count=share.view_count, + download_count=share.download_count, + is_active=share.is_active, + created_at=format_utc_datetime(share.created_at), + updated_at=format_utc_datetime(share.updated_at), + ) + + +# ── 用户侧：创建/管理分享 ────────────────────────────────── + + +@router.post("/videos/{video_id}/share", response_model=ShareResponse) +def create_share( + video_id: str, + request: CreateShareRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), + video_repo: GeneratedVideoRepository = Depends(get_generated_video_repository), +) -> ShareResponse: + """为视频创建分享链接.""" + use_case = CreateShareUseCase(share_repo, video_repo) + try: + share = use_case.execute( + CreateShareCommand( + video_id=video_id, + user_id=authenticated_user.user.id, + password=request.password, + expires_at=request.expires_at, + ) + ) + except VideoNotFoundError as e: + raise HTTPException(status_code=404, detail=str(e)) from e + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + logger.info( + "Share created: video_id=%s share_id=%s token=%s user=%s", + video_id, + share.id, + share.share_token, + authenticated_user.user.id, + ) + return _to_share_response(share) + + +@router.get("/videos/{video_id}/shares", response_model=ShareListResponse) +def list_video_shares( + video_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), +) -> ShareListResponse: + """获取某个视频的所有分享记录.""" + use_case = ListSharesByVideoUseCase(share_repo) + items = use_case.execute(video_id, authenticated_user.user.id) + return ShareListResponse( + items=[_to_share_response(s) for s in items], + total=len(items), + skip=0, + limit=len(items), + ) + + +@router.get("/shares", response_model=ShareListResponse) +def list_user_shares( + skip: int = Query(0, ge=0), + limit: int = Query(20, ge=1, le=100), + authenticated_user: AuthenticatedUser = Depends(get_current_user), + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), +) -> ShareListResponse: + """获取用户创建的所有分享记录.""" + use_case = ListSharesByUserUseCase(share_repo) + items, total = use_case.execute(authenticated_user.user.id, skip=skip, limit=limit) + return ShareListResponse( + items=[_to_share_response(s) for s in items], + total=total, + skip=skip, + limit=limit, + ) + + +@router.patch("/shares/{share_id}", response_model=ShareResponse) +def update_share( + share_id: str, + request: UpdateShareRequest, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), +) -> ShareResponse: + """更新分享配置（密码、有效期等）.""" + use_case = UpdateShareUseCase(share_repo) + try: + share = use_case.execute( + UpdateShareCommand( + share_id=share_id, + user_id=authenticated_user.user.id, + password=request.password, + expires_at=request.expires_at, + ) + ) + except NotFoundError as e: + raise HTTPException(status_code=404, detail=str(e)) from e + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + return _to_share_response(share) + + +@router.delete( + "/shares/{share_id}", + status_code=status.HTTP_204_NO_CONTENT, + response_model=None, + response_class=Response, +) +def revoke_share( + share_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), +) -> Response: + """撤销/删除分享链接.""" + use_case = RevokeShareUseCase(share_repo) + try: + use_case.execute(share_id, authenticated_user.user.id) + except NotFoundError as e: + raise HTTPException(status_code=404, detail=str(e)) from e + return Response(status_code=status.HTTP_204_NO_CONTENT) + + +# ── 公开侧：访问分享内容（无需登录） ────────────────────── + + +@router.get("/share/{token}/meta", response_model=ShareMetaResponse) +def get_share_meta( + token: str, + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), + video_repo: GeneratedVideoRepository = Depends(get_generated_video_repository), + storage: OSSStorageService = Depends(get_storage_service), +) -> ShareMetaResponse: + """获取分享元信息（不需要密码，用于分享页加载前判断）。""" + use_case = GetShareByTokenUseCase(share_repo) + try: + share = use_case.execute(token) + except (NotFoundError, ShareExpiredError) as e: + raise HTTPException(status_code=404, detail="分享链接不存在或已失效") from e + + video = video_repo.get(share.video_id) + video_name = video.name if video else "" + video_duration = video.duration if video else 0.0 + thumbnail_url = None + if video and video.thumbnail_url: + try: + thumbnail_url = storage.get_download_url(video.thumbnail_url) + except Exception: + thumbnail_url = video.thumbnail_url + + return ShareMetaResponse( + share_token=share.share_token, + has_password=share.has_password, + is_expired=share.is_expired, + is_active=share.is_active, + video_name=video_name, + video_duration=video_duration, + thumbnail_url=thumbnail_url, + created_at=format_utc_datetime(share.created_at), + ) + + +@router.post("/share/{token}/access", response_model=ShareAccessResponse) +def access_share( + token: str, + request: Optional[VerifySharePasswordRequest] = None, + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), + video_repo: GeneratedVideoRepository = Depends(get_generated_video_repository), + storage: OSSStorageService = Depends(get_storage_service), +) -> ShareAccessResponse: + """访问分享内容（验证密码后返回视频信息+播放/下载地址）。""" + use_case = AccessShareUseCase(share_repo, video_repo) + password = request.password if request else None + try: + result = use_case.execute(token, password=password) + except NotFoundError as e: + raise HTTPException(status_code=404, detail="分享链接不存在或已失效") from e + except ShareExpiredError as e: + raise HTTPException(status_code=410, detail="分享链接已过期或已撤销") from e + except PasswordRequiredError as e: + raise HTTPException(status_code=403, detail="需要访问密码") from e + except InvalidPasswordError as e: + raise HTTPException(status_code=403, detail="密码错误") from e + except VideoNotFoundError as e: + raise HTTPException(status_code=404, detail="视频不存在") from e + + # 生成下载URL + download_url = None + if result.video.file_url: + try: + download_url = storage.get_download_url(result.video.file_url) + except Exception: + download_url = result.video.file_url + + # 缩略图URL + thumbnail_url = None + if result.video.thumbnail_url: + try: + thumbnail_url = storage.get_download_url(result.video.thumbnail_url) + except Exception: + thumbnail_url = result.video.thumbnail_url + + return ShareAccessResponse( + share=_to_share_response(result.share), + video_name=result.video.name, + video_duration=result.video.duration, + video_size=result.video.file_size, + thumbnail_url=thumbnail_url, + download_url=download_url, + password_verified=result.password_verified, + ) + + +@router.post("/share/{token}/download") +def record_share_download( + token: str, + request: Optional[VerifySharePasswordRequest] = None, + share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository), +) -> dict: + """记录分享下载（下载计数+1）。""" + use_case = RecordShareDownloadUseCase(share_repo) + password = request.password if request else None + try: + use_case.execute(token, password=password) + except NotFoundError as e: + raise HTTPException(status_code=404, detail="分享链接不存在或已失效") from e + except ShareExpiredError as e: + raise HTTPException(status_code=410, detail="分享链接已过期或已撤销") from e + except InvalidPasswordError as e: + raise HTTPException(status_code=403, detail="密码错误") from e + return {"success": True} diff --git a/apps/api/app/schemas/video_share.py b/apps/api/app/schemas/video_share.py new file mode 100755 index 000000000..df0590422 --- /dev/null +++ b/apps/api/app/schemas/video_share.py @@ -0,0 +1,92 @@ +"""视频分享相关 schemas.""" + +from __future__ import annotations + +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, Field + + +class CreateShareRequest(BaseModel): + """创建分享请求.""" + + password: Optional[str] = Field( + None, + description="访问密码（可选，不设置则无需密码）", + min_length=0, + max_length=50, + ) + expires_at: Optional[datetime] = Field( + None, + description="过期时间（可选，不设置则永久有效）", + ) + + +class UpdateShareRequest(BaseModel): + """更新分享配置请求.""" + + password: Optional[str] = Field( + None, + description="新密码（传空字符串清除密码，不传则不修改）", + max_length=50, + ) + expires_at: Optional[datetime] = Field( + None, + description="新的过期时间（不传则不修改）", + ) + + +class VerifySharePasswordRequest(BaseModel): + """验证分享密码请求.""" + + password: str = Field(..., description="访问密码") + + +class ShareResponse(BaseModel): + """分享记录响应.""" + + id: str + video_id: str + share_token: str + has_password: bool = False + expires_at: Optional[datetime] = None + view_count: int = 0 + download_count: int = 0 + is_active: bool = True + created_at: Optional[datetime] = None + updated_at: Optional[datetime] = None + + +class ShareListResponse(BaseModel): + """分享列表响应.""" + + items: List[ShareResponse] + total: int = 0 + skip: int = 0 + limit: int = 20 + + +class ShareAccessResponse(BaseModel): + """分享访问成功响应（含视频信息）.""" + + share: ShareResponse + video_name: str + video_duration: float = 0.0 + video_size: int = 0 + thumbnail_url: Optional[str] = None + download_url: Optional[str] = None + password_verified: bool = True + + +class ShareMetaResponse(BaseModel): + """分享元信息响应（访问前获取，用于判断是否需要密码）。""" + + share_token: str + has_password: bool = False + is_expired: bool = False + is_active: bool = True + video_name: str = "" + video_duration: float = 0.0 + thumbnail_url: Optional[str] = None + created_at: Optional[datetime] = None diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index a3ee20e74..8ed664673 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -574,3 +574,21 @@ class VerificationCodeModel(Base): used_at = Column(DateTime, nullable=True) attempts = Column(Integer, nullable=False, default=0) created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) + + +class VideoShareModel(Base): + """视频分享记录.""" + + __tablename__ = "video_shares" + + id = Column(String(32), primary_key=True) + video_id = Column(String(32), nullable=False, index=True) + user_id = Column(String(32), nullable=False, index=True) + share_token = Column(String(16), nullable=False, unique=True) + password_hash = Column(String(255), nullable=True) + expires_at = Column(DateTime(timezone=True), nullable=True) + view_count = Column(Integer, nullable=False, default=0) + download_count = Column(Integer, nullable=False, default=0) + is_active = Column(Boolean, nullable=False, default=True) + created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) + updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) diff --git a/packages/adapters/sqlalchemy_impl/video_share_repository.py b/packages/adapters/sqlalchemy_impl/video_share_repository.py new file mode 100755 index 000000000..fe15555f2 --- /dev/null +++ b/packages/adapters/sqlalchemy_impl/video_share_repository.py @@ -0,0 +1,147 @@ +"""视频分享 SQLAlchemy Repository 实现.""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import List, Optional + +from sqlalchemy.orm import Session + +from packages.adapters.sqlalchemy_impl.models import VideoShareModel +from packages.domain.video_share import VideoShare +from packages.ports.video_share_repository import VideoShareRepositoryPort + + +def _model_to_domain(model: VideoShareModel) -> VideoShare: + return VideoShare( + id=model.id, + video_id=model.video_id, + user_id=model.user_id, + share_token=model.share_token, + password_hash=model.password_hash, + expires_at=model.expires_at, + view_count=model.view_count or 0, + download_count=model.download_count or 0, + is_active=model.is_active if model.is_active is not None else True, + created_at=model.created_at, + updated_at=model.updated_at, + ) + + +class SQLAlchemyVideoShareRepository(VideoShareRepositoryPort): + def __init__(self, session: Session): + self.session = session + + def create(self, share: VideoShare) -> VideoShare: + model = VideoShareModel( + id=share.id, + video_id=share.video_id, + user_id=share.user_id, + share_token=share.share_token, + password_hash=share.password_hash, + expires_at=share.expires_at, + view_count=share.view_count, + download_count=share.download_count, + is_active=share.is_active, + created_at=share.created_at, + updated_at=share.updated_at, + ) + self.session.add(model) + self.session.commit() + return share + + def get_by_token(self, token: str) -> Optional[VideoShare]: + model = self.session.query(VideoShareModel).filter(VideoShareModel.share_token == token).first() + if model is None: + return None + return _model_to_domain(model) + + def get_by_id(self, share_id: str, user_id: str) -> Optional[VideoShare]: + model = ( + self.session.query(VideoShareModel) + .filter( + VideoShareModel.id == share_id, + VideoShareModel.user_id == user_id, + ) + .first() + ) + if model is None: + return None + return _model_to_domain(model) + + def list_by_video(self, video_id: str, user_id: str) -> List[VideoShare]: + models = ( + self.session.query(VideoShareModel) + .filter( + VideoShareModel.video_id == video_id, + VideoShareModel.user_id == user_id, + ) + .order_by(VideoShareModel.created_at.desc()) + .all() + ) + return [_model_to_domain(m) for m in models] + + def list_by_user(self, user_id: str, skip: int = 0, limit: int = 20) -> List[VideoShare]: + models = ( + self.session.query(VideoShareModel) + .filter(VideoShareModel.user_id == user_id) + .order_by(VideoShareModel.created_at.desc()) + .offset(skip) + .limit(limit) + .all() + ) + return [_model_to_domain(m) for m in models] + + def count_by_user(self, user_id: str) -> int: + return self.session.query(VideoShareModel).filter(VideoShareModel.user_id == user_id).count() + + def update(self, share: VideoShare) -> VideoShare: + model = self.session.query(VideoShareModel).filter(VideoShareModel.id == share.id).first() + if model is None: + return share + model.password_hash = share.password_hash + model.expires_at = share.expires_at + model.is_active = share.is_active + model.view_count = share.view_count + model.download_count = share.download_count + model.updated_at = datetime.now(timezone.utc) + self.session.add(model) + self.session.commit() + return share + + def delete(self, share_id: str, user_id: str) -> bool: + model = ( + self.session.query(VideoShareModel) + .filter( + VideoShareModel.id == share_id, + VideoShareModel.user_id == user_id, + ) + .first() + ) + if model is None: + return False + model.is_active = False + model.updated_at = datetime.now(timezone.utc) + self.session.add(model) + self.session.commit() + return True + + def increment_view(self, share_id: str) -> None: + self.session.query(VideoShareModel).filter(VideoShareModel.id == share_id).update( + { + "view_count": VideoShareModel.view_count + 1, + "updated_at": datetime.now(timezone.utc), + }, + synchronize_session=False, + ) + self.session.commit() + + def increment_download(self, share_id: str) -> None: + self.session.query(VideoShareModel).filter(VideoShareModel.id == share_id).update( + { + "download_count": VideoShareModel.download_count + 1, + "updated_at": datetime.now(timezone.utc), + }, + synchronize_session=False, + ) + self.session.commit() diff --git a/packages/application/video_share/commands.py b/packages/application/video_share/commands.py new file mode 100755 index 000000000..b30b0ad85 --- /dev/null +++ b/packages/application/video_share/commands.py @@ -0,0 +1,35 @@ +"""视频分享 Commands.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Optional + + +@dataclass +class CreateShareCommand: + """创建分享链接命令.""" + + video_id: str + user_id: str + password: Optional[str] = None + expires_at: Optional[datetime] = None # None表示永久有效 + + +@dataclass +class VerifySharePasswordCommand: + """验证分享密码命令.""" + + share_token: str + password: str + + +@dataclass +class UpdateShareCommand: + """更新分享配置命令.""" + + share_id: str + user_id: str + password: Optional[str] = None # None表示不修改，空字符串表示清除密码 + expires_at: Optional[datetime] = None # None表示不修改 diff --git a/packages/application/video_share/use_cases.py b/packages/application/video_share/use_cases.py new file mode 100755 index 000000000..2290b8a28 --- /dev/null +++ b/packages/application/video_share/use_cases.py @@ -0,0 +1,227 @@ +"""视频分享 Use cases.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import List, Optional + +from packages.application.video_share.commands import ( + CreateShareCommand, + UpdateShareCommand, +) +from packages.domain.generated_video import GeneratedVideo +from packages.domain.video_share import VideoShare +from packages.ports.generated_video_repository import GeneratedVideoRepository +from packages.ports.video_share_repository import VideoShareRepositoryPort + + +class NotFoundError(Exception): + """分享记录不存在.""" + + pass + + +class VideoNotFoundError(Exception): + """视频不存在.""" + + pass + + +class ShareExpiredError(Exception): + """分享已过期或已撤销.""" + + pass + + +class PasswordRequiredError(Exception): + """需要访问密码.""" + + pass + + +class InvalidPasswordError(Exception): + """密码错误.""" + + pass + + +@dataclass +class ShareAccessResult: + """分享访问结果（验证通过后返回视频信息+分享记录）.""" + + share: VideoShare + video: GeneratedVideo + password_verified: bool = True + + +class CreateShareUseCase: + """创建视频分享链接.""" + + def __init__( + self, + share_repository: VideoShareRepositoryPort, + video_repository: GeneratedVideoRepository, + ) -> None: + self.share_repo = share_repository + self.video_repo = video_repository + + def execute(self, command: CreateShareCommand) -> VideoShare: + # 校验视频存在且属于该用户 + video = self.video_repo.get(command.video_id) + if video is None: + raise VideoNotFoundError(f"Video {command.video_id} not found") + + # 用 user_id 校验（视频的user_id需要匹配） + if hasattr(video, "user_id") and video.user_id and video.user_id != command.user_id: + raise VideoNotFoundError("Video not found") + + share = VideoShare.create( + video_id=command.video_id, + user_id=command.user_id, + password=command.password, + expires_at=command.expires_at, + ) + return self.share_repo.create(share) + + +class GetShareByTokenUseCase: + """通过token获取分享信息（不带视频内容，仅元信息）。 + + 用于分享页加载前判断：是否需要密码、是否过期等。 + """ + + def __init__(self, share_repository: VideoShareRepositoryPort) -> None: + self.share_repo = share_repository + + def execute(self, token: str) -> VideoShare: + share = self.share_repo.get_by_token(token) + if share is None: + raise NotFoundError(f"Share not found: {token}") + if not share.is_accessible: + raise ShareExpiredError("Share is not accessible") + return share + + +class AccessShareUseCase: + """访问分享内容（验证密码+返回视频信息+计数浏览量）。""" + + def __init__( + self, + share_repository: VideoShareRepositoryPort, + video_repository: GeneratedVideoRepository, + ) -> None: + self.share_repo = share_repository + self.video_repo = video_repository + + def execute(self, token: str, password: Optional[str] = None) -> ShareAccessResult: + share = self.share_repo.get_by_token(token) + if share is None: + raise NotFoundError(f"Share not found: {token}") + if not share.is_accessible: + raise ShareExpiredError("Share is not accessible") + + # 密码校验 + password_verified = True + if share.has_password: + if not password: + raise PasswordRequiredError("Password required") + if not share.verify_password(password): + raise InvalidPasswordError("Invalid password") + password_verified = True + + # 获取视频信息 + video = self.video_repo.get(share.video_id) + if video is None: + raise VideoNotFoundError("Video not found") + + # 浏览量+1 + self.share_repo.increment_view(share.id) + share.view_count += 1 + + return ShareAccessResult(share=share, video=video, password_verified=password_verified) + + +class ListSharesByVideoUseCase: + """列出某个视频的所有分享记录.""" + + def __init__(self, share_repository: VideoShareRepositoryPort) -> None: + self.share_repo = share_repository + + def execute(self, video_id: str, user_id: str) -> List[VideoShare]: + return self.share_repo.list_by_video(video_id, user_id) + + +class ListSharesByUserUseCase: + """列出用户创建的所有分享记录.""" + + def __init__(self, share_repository: VideoShareRepositoryPort) -> None: + self.share_repo = share_repository + + def execute(self, user_id: str, skip: int = 0, limit: int = 20) -> tuple[List[VideoShare], int]: + items = self.share_repo.list_by_user(user_id, skip=skip, limit=limit) + total = self.share_repo.count_by_user(user_id) + return items, total + + +class UpdateShareUseCase: + """更新分享配置（密码、有效期等）.""" + + def __init__(self, share_repository: VideoShareRepositoryPort) -> None: + self.share_repo = share_repository + + def execute(self, command: UpdateShareCommand) -> VideoShare: + share = self.share_repo.get_by_id(command.share_id, command.user_id) + if share is None: + raise NotFoundError(f"Share {command.share_id} not found") + + # password=None表示不修改；空字符串表示清除密码 + if command.password is not None: + from packages.domain.video_share import _hash_password + + if command.password == "": + share.password_hash = None + else: + share.password_hash = _hash_password(command.password) + + # expires_at=None表示不修改 + if command.expires_at is not None: + if command.expires_at < datetime.now(timezone.utc): + raise ValueError("expires_at cannot be in the past") + share.expires_at = command.expires_at + + return self.share_repo.update(share) + + +class RevokeShareUseCase: + """撤销/删除分享.""" + + def __init__(self, share_repository: VideoShareRepositoryPort) -> None: + self.share_repo = share_repository + + def execute(self, share_id: str, user_id: str) -> bool: + share = self.share_repo.get_by_id(share_id, user_id) + if share is None: + raise NotFoundError(f"Share {share_id} not found") + return self.share_repo.delete(share_id, user_id) + + +class RecordShareDownloadUseCase: + """记录分享下载（下载量+1）.""" + + def __init__(self, share_repository: VideoShareRepositoryPort) -> None: + self.share_repo = share_repository + + def execute(self, token: str, password: Optional[str] = None) -> None: + share = self.share_repo.get_by_token(token) + if share is None: + raise NotFoundError(f"Share not found: {token}") + if not share.is_accessible: + raise ShareExpiredError("Share is not accessible") + + # 密码校验 + if share.has_password: + if not password or not share.verify_password(password): + raise InvalidPasswordError("Invalid password") + + self.share_repo.increment_download(share.id) diff --git a/packages/domain/video_share.py b/packages/domain/video_share.py new file mode 100755 index 000000000..b3b28a1ee --- /dev/null +++ b/packages/domain/video_share.py @@ -0,0 +1,108 @@ +"""视频分享领域实体.""" + +from __future__ import annotations + +import secrets +from dataclasses import dataclass, field +from datetime import datetime, timezone +from hashlib import sha256 +from typing import Optional +from uuid import uuid4 + + +def _hash_password(password: str) -> str: + """简单密码哈希（SHA-256 + salt）。 + + 分享链接的密码保护安全级别要求不高， + 使用简单的加盐哈希即可，避免引入bcrypt等重依赖。 + """ + if not password: + return "" + salt = "xiaoxia_share_salt" + return sha256(f"{salt}:{password}".encode()).hexdigest() + + +def generate_share_token(length: int = 12) -> str: + """生成URL友好的分享token.""" + # 使用urlsafe的base64，但去掉可能引起歧义的字符 + alphabet = "abcdefghjkmnpqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789" + return "".join(secrets.choice(alphabet) for _ in range(length)) + + +@dataclass(slots=True) +class VideoShare: + """视频分享记录.""" + + id: str + video_id: str + user_id: str + share_token: str + password_hash: Optional[str] = None + expires_at: Optional[datetime] = None + view_count: int = 0 + download_count: int = 0 + is_active: bool = True + created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + @classmethod + def create( + cls, + video_id: str, + user_id: str, + *, + password: Optional[str] = None, + expires_at: Optional[datetime] = None, + ) -> "VideoShare": + if not video_id.strip(): + raise ValueError("video_id cannot be empty") + if not user_id.strip(): + raise ValueError("user_id cannot be empty") + if expires_at and expires_at < datetime.now(timezone.utc): + raise ValueError("expires_at cannot be in the past") + + return cls( + id=uuid4().hex, + video_id=video_id.strip(), + user_id=user_id.strip(), + share_token=generate_share_token(), + password_hash=_hash_password(password) if password else None, + expires_at=expires_at, + ) + + @property + def has_password(self) -> bool: + """是否设置了访问密码.""" + return bool(self.password_hash) + + @property + def is_expired(self) -> bool: + """是否已过期.""" + if not self.expires_at: + return False + return datetime.now(timezone.utc) > self.expires_at + + @property + def is_accessible(self) -> bool: + """是否可以访问（活跃且未过期）.""" + return self.is_active and not self.is_expired + + def verify_password(self, password: str) -> bool: + """验证访问密码.""" + if not self.password_hash: + return True # 没有密码直接通过 + if not password: + return False + return _hash_password(password) == self.password_hash + + def increment_view_count(self) -> None: + """浏览次数+1.""" + self.view_count += 1 + + def increment_download_count(self) -> None: + """下载次数+1.""" + self.download_count += 1 + + def revoke(self) -> None: + """撤销分享.""" + self.is_active = False diff --git a/packages/ports/video_share_repository.py b/packages/ports/video_share_repository.py new file mode 100755 index 000000000..96f137d16 --- /dev/null +++ b/packages/ports/video_share_repository.py @@ -0,0 +1,62 @@ +"""视频分享 Repository 端口.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import List, Optional + +from packages.domain.video_share import VideoShare + + +class VideoShareRepositoryPort(ABC): + """视频分享 Repository 接口.""" + + @abstractmethod + def create(self, share: VideoShare) -> VideoShare: + """创建分享记录.""" + ... + + @abstractmethod + def get_by_token(self, token: str) -> Optional[VideoShare]: + """通过分享token获取分享记录.""" + ... + + @abstractmethod + def get_by_id(self, share_id: str, user_id: str) -> Optional[VideoShare]: + """通过ID获取分享记录（带用户校验）.""" + ... + + @abstractmethod + def list_by_video(self, video_id: str, user_id: str) -> List[VideoShare]: + """列出某个视频的所有分享记录.""" + ... + + @abstractmethod + def list_by_user(self, user_id: str, skip: int = 0, limit: int = 20) -> List[VideoShare]: + """列出用户创建的所有分享记录.""" + ... + + @abstractmethod + def count_by_user(self, user_id: str) -> int: + """统计用户创建的分享数量.""" + ... + + @abstractmethod + def update(self, share: VideoShare) -> VideoShare: + """更新分享记录.""" + ... + + @abstractmethod + def delete(self, share_id: str, user_id: str) -> bool: + """删除分享记录（软删除：is_active=False）.""" + ... + + @abstractmethod + def increment_view(self, share_id: str) -> None: + """浏览次数+1.""" + ... + + @abstractmethod + def increment_download(self, share_id: str) -> None: + """下载次数+1.""" + ... diff --git a/tests/unit/test_video_share.py b/tests/unit/test_video_share.py new file mode 100755 index 000000000..6ba21920e --- /dev/null +++ b/tests/unit/test_video_share.py @@ -0,0 +1,603 @@ +"""视频分享 - 领域实体 + Use cases 单元测试.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +import pytest + +from packages.application.video_share.commands import ( + CreateShareCommand, + UpdateShareCommand, +) +from packages.application.video_share.use_cases import ( + AccessShareUseCase, + CreateShareUseCase, + GetShareByTokenUseCase, + InvalidPasswordError, + ListSharesByUserUseCase, + ListSharesByVideoUseCase, + NotFoundError, + PasswordRequiredError, + RecordShareDownloadUseCase, + RevokeShareUseCase, + ShareAccessResult, + ShareExpiredError, + UpdateShareUseCase, + VideoNotFoundError, +) +from packages.domain.generated_video import GeneratedVideo +from packages.domain.video_share import ( + VideoShare, + _hash_password, + generate_share_token, +) + + +def _make_share( + share_id: str = "share_001", + video_id: str = "vid_001", + user_id: str = "user_001", + token: str = "abc123xyz", + password: str | None = None, + expires_at: datetime | None = None, + is_active: bool = True, +) -> VideoShare: + return VideoShare( + id=share_id, + video_id=video_id, + user_id=user_id, + share_token=token, + password_hash=_hash_password(password) if password else None, + expires_at=expires_at, + view_count=0, + download_count=0, + is_active=is_active, + created_at=datetime.now(timezone.utc), + updated_at=datetime.now(timezone.utc), + ) + + +def _make_video(video_id: str = "vid_001", user_id: str = "user_001") -> GeneratedVideo: + return GeneratedVideo( + id=video_id, + project_id="proj_001", + generation_task_id="task_001", + name="测试视频", + file_url="oss://bucket/video.mp4", + file_size=1024000, + duration=30.5, + width=1080, + height=1920, + fps=30.0, + user_id=user_id, + ) + + +class TestVideoShareDomain: + def test_generate_token_length(self) -> None: + token = generate_share_token(12) + assert len(token) == 12 + + def test_generate_token_url_safe(self) -> None: + token = generate_share_token(16) + # 只包含字母数字，没有特殊字符 + assert token.isalnum() + + def test_hash_password_consistent(self) -> None: + h1 = _hash_password("mypassword") + h2 = _hash_password("mypassword") + assert h1 == h2 + assert len(h1) == 64 # sha256 hex + + def test_hash_password_different_for_different_passwords(self) -> None: + h1 = _hash_password("password1") + h2 = _hash_password("password2") + assert h1 != h2 + + def test_hash_empty_password(self) -> None: + assert _hash_password("") == "" + + def test_create_share_success(self) -> None: + share = VideoShare.create( + video_id="vid_001", + user_id="user_001", + ) + assert share.video_id == "vid_001" + assert share.user_id == "user_001" + assert len(share.id) == 32 + assert len(share.share_token) == 12 + assert share.password_hash is None + assert share.expires_at is None + assert share.is_active is True + assert share.view_count == 0 + assert share.download_count == 0 + + def test_create_share_with_password(self) -> None: + share = VideoShare.create( + video_id="vid_001", + user_id="user_001", + password="secret123", + ) + assert share.has_password is True + assert share.verify_password("secret123") is True + assert share.verify_password("wrong") is False + + def test_create_share_with_expiry(self) -> None: + future = datetime.now(timezone.utc) + timedelta(days=7) + share = VideoShare.create( + video_id="vid_001", + user_id="user_001", + expires_at=future, + ) + assert share.expires_at == future + assert share.is_expired is False + + def test_create_share_past_expiry_raises(self) -> None: + past = datetime.now(timezone.utc) - timedelta(days=1) + with pytest.raises(ValueError, match="past"): + VideoShare.create( + video_id="vid_001", + user_id="user_001", + expires_at=past, + ) + + def test_create_share_empty_video_id_raises(self) -> None: + with pytest.raises(ValueError, match="video_id"): + VideoShare.create(video_id="", user_id="user_001") + + def test_create_share_empty_user_id_raises(self) -> None: + with pytest.raises(ValueError, match="user_id"): + VideoShare.create(video_id="vid_001", user_id=" ") + + def test_is_expired_false_when_no_expiry(self) -> None: + share = _make_share() + assert share.is_expired is False + + def test_is_expired_true_when_past(self) -> None: + past = datetime.now(timezone.utc) - timedelta(hours=1) + share = _make_share(expires_at=past) + assert share.is_expired is True + + def test_is_accessible_active_not_expired(self) -> None: + share = _make_share() + assert share.is_accessible is True + + def test_is_accessible_inactive(self) -> None: + share = _make_share(is_active=False) + assert share.is_accessible is False + + def test_is_accessible_expired(self) -> None: + past = datetime.now(timezone.utc) - timedelta(hours=1) + share = _make_share(expires_at=past) + assert share.is_accessible is False + + def test_has_password_false_when_no_password(self) -> None: + share = _make_share() + assert share.has_password is False + + def test_has_password_true_when_password_set(self) -> None: + share = _make_share(password="pass123") + assert share.has_password is True + + def test_verify_no_password_always_true(self) -> None: + share = _make_share() # 没有密码 + assert share.verify_password("") is True + assert share.verify_password("anything") is True + + def test_verify_correct_password(self) -> None: + share = _make_share(password="mysecret") + assert share.verify_password("mysecret") is True + + def test_verify_wrong_password(self) -> None: + share = _make_share(password="mysecret") + assert share.verify_password("wrong") is False + + def test_verify_empty_password_with_password_set(self) -> None: + share = _make_share(password="mysecret") + assert share.verify_password("") is False + + def test_increment_view_count(self) -> None: + share = _make_share() + share.increment_view_count() + assert share.view_count == 1 + share.increment_view_count() + assert share.view_count == 2 + + def test_increment_download_count(self) -> None: + share = _make_share() + share.increment_download_count() + assert share.download_count == 1 + + def test_revoke_sets_inactive(self) -> None: + share = _make_share() + share.revoke() + assert share.is_active is False + assert share.is_accessible is False + + +class TestCreateShareUseCase: + def test_create_success(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = _make_video() + share_repo.create.side_effect = lambda s: s + + use_case = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="vid_001", user_id="user_001") + result = use_case.execute(cmd) + + assert result.video_id == "vid_001" + assert result.user_id == "user_001" + share_repo.create.assert_called_once() + + def test_create_with_password(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = _make_video() + share_repo.create.side_effect = lambda s: s + + use_case = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand( + video_id="vid_001", + user_id="user_001", + password="secret", + ) + result = use_case.execute(cmd) + + assert result.has_password is True + + def test_create_with_expiry(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = _make_video() + share_repo.create.side_effect = lambda s: s + + use_case = CreateShareUseCase(share_repo, video_repo) + future = datetime.now(timezone.utc) + timedelta(days=1) + cmd = CreateShareCommand( + video_id="vid_001", + user_id="user_001", + expires_at=future, + ) + result = use_case.execute(cmd) + + assert result.expires_at == future + + def test_video_not_found_raises(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = None + + use_case = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="nonexistent", user_id="user_001") + + with pytest.raises(VideoNotFoundError): + use_case.execute(cmd) + + def test_wrong_user_cannot_share(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + video_repo.get.return_value = _make_video(user_id="other_user") + + use_case = CreateShareUseCase(share_repo, video_repo) + cmd = CreateShareCommand(video_id="vid_001", user_id="user_001") + + with pytest.raises(VideoNotFoundError): + use_case.execute(cmd) + + +class TestGetShareByTokenUseCase: + def test_found_active_share(self) -> None: + repo = MagicMock() + repo.get_by_token.return_value = _make_share() + + use_case = GetShareByTokenUseCase(repo) + result = use_case.execute("abc123xyz") + assert result.share_token == "abc123xyz" + + def test_not_found_raises(self) -> None: + repo = MagicMock() + repo.get_by_token.return_value = None + + use_case = GetShareByTokenUseCase(repo) + with pytest.raises(NotFoundError): + use_case.execute("nonexistent") + + def test_inactive_share_raises_expired(self) -> None: + repo = MagicMock() + repo.get_by_token.return_value = _make_share(is_active=False) + + use_case = GetShareByTokenUseCase(repo) + with pytest.raises(ShareExpiredError): + use_case.execute("token") + + def test_expired_share_raises(self) -> None: + repo = MagicMock() + past = datetime.now(timezone.utc) - timedelta(days=1) + repo.get_by_token.return_value = _make_share(expires_at=past) + + use_case = GetShareByTokenUseCase(repo) + with pytest.raises(ShareExpiredError): + use_case.execute("token") + + +class TestAccessShareUseCase: + def test_access_no_password(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = _make_share() + video_repo.get.return_value = _make_video() + share_repo.increment_view.return_value = None + + use_case = AccessShareUseCase(share_repo, video_repo) + result = use_case.execute("abc123xyz") + + assert isinstance(result, ShareAccessResult) + assert result.video.id == "vid_001" + assert result.password_verified is True + assert result.share.view_count == 1 # 浏览量+1 + share_repo.increment_view.assert_called_once() + + def test_access_with_correct_password(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = _make_share(password="mypass") + video_repo.get.return_value = _make_video() + share_repo.increment_view.return_value = None + + use_case = AccessShareUseCase(share_repo, video_repo) + result = use_case.execute("token", password="mypass") + + assert result.password_verified is True + + def test_access_password_required_but_not_provided(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = _make_share(password="secret") + video_repo.get.return_value = _make_video() + + use_case = AccessShareUseCase(share_repo, video_repo) + + with pytest.raises(PasswordRequiredError): + use_case.execute("token", password=None) + + def test_access_wrong_password(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = _make_share(password="correct") + video_repo.get.return_value = _make_video() + + use_case = AccessShareUseCase(share_repo, video_repo) + + with pytest.raises(InvalidPasswordError): + use_case.execute("token", password="wrong") + + def test_access_share_not_found(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = None + + use_case = AccessShareUseCase(share_repo, video_repo) + + with pytest.raises(NotFoundError): + use_case.execute("nonexistent") + + def test_access_share_expired(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + past = datetime.now(timezone.utc) - timedelta(days=1) + share_repo.get_by_token.return_value = _make_share(expires_at=past) + video_repo.get.return_value = _make_video() + + use_case = AccessShareUseCase(share_repo, video_repo) + + with pytest.raises(ShareExpiredError): + use_case.execute("token") + + def test_access_video_not_found(self) -> None: + share_repo = MagicMock() + video_repo = MagicMock() + share_repo.get_by_token.return_value = _make_share() + video_repo.get.return_value = None + + use_case = AccessShareUseCase(share_repo, video_repo) + + with pytest.raises(VideoNotFoundError): + use_case.execute("token") + + +class TestListSharesByVideoUseCase: + def test_lists_shares(self) -> None: + repo = MagicMock() + expected = [_make_share(), _make_share(share_id="share_002", token="tok2")] + repo.list_by_video.return_value = expected + + use_case = ListSharesByVideoUseCase(repo) + result = use_case.execute("vid_001", "user_001") + + assert len(result) == 2 + repo.list_by_video.assert_called_once_with("vid_001", "user_001") + + +class TestListSharesByUserUseCase: + def test_lists_with_total(self) -> None: + repo = MagicMock() + items = [_make_share(), _make_share(share_id="s2", token="t2")] + repo.list_by_user.return_value = items + repo.count_by_user.return_value = 10 + + use_case = ListSharesByUserUseCase(repo) + result_items, total = use_case.execute("user_001", skip=0, limit=2) + + assert len(result_items) == 2 + assert total == 10 + repo.list_by_user.assert_called_once_with("user_001", skip=0, limit=2) + + +class TestUpdateShareUseCase: + def test_update_password(self) -> None: + repo = MagicMock() + share = _make_share() + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s + + use_case = UpdateShareUseCase(repo) + cmd = UpdateShareCommand( + share_id="share_001", + user_id="user_001", + password="newpass", + ) + result = use_case.execute(cmd) + + assert result.has_password is True + assert result.verify_password("newpass") is True + repo.update.assert_called_once() + + def test_clear_password(self) -> None: + repo = MagicMock() + share = _make_share(password="oldpass") + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s + + use_case = UpdateShareUseCase(repo) + cmd = UpdateShareCommand( + share_id="share_001", + user_id="user_001", + password="", # 空字符串=清除密码 + ) + result = use_case.execute(cmd) + + assert result.has_password is False + assert result.password_hash is None + + def test_password_none_does_not_change(self) -> None: + repo = MagicMock() + share = _make_share(password="existing") + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s + + use_case = UpdateShareUseCase(repo) + cmd = UpdateShareCommand( + share_id="share_001", + user_id="user_001", + password=None, # None=不修改 + ) + result = use_case.execute(cmd) + + assert result.verify_password("existing") is True + + def test_update_expires_at(self) -> None: + repo = MagicMock() + share = _make_share() + repo.get_by_id.return_value = share + repo.update.side_effect = lambda s: s + + use_case = UpdateShareUseCase(repo) + future = datetime.now(timezone.utc) + timedelta(days=3) + cmd = UpdateShareCommand( + share_id="share_001", + user_id="user_001", + expires_at=future, + ) + result = use_case.execute(cmd) + + assert result.expires_at == future + + def test_not_found_raises(self) -> None: + repo = MagicMock() + repo.get_by_id.return_value = None + + use_case = UpdateShareUseCase(repo) + cmd = UpdateShareCommand(share_id="no", user_id="u1", password="x") + + with pytest.raises(NotFoundError): + use_case.execute(cmd) + + def test_past_expiry_raises(self) -> None: + repo = MagicMock() + repo.get_by_id.return_value = _make_share() + + use_case = UpdateShareUseCase(repo) + past = datetime.now(timezone.utc) - timedelta(days=1) + cmd = UpdateShareCommand( + share_id="share_001", + user_id="user_001", + expires_at=past, + ) + + with pytest.raises(ValueError, match="past"): + use_case.execute(cmd) + + +class TestRevokeShareUseCase: + def test_revoke_success(self) -> None: + repo = MagicMock() + repo.get_by_id.return_value = _make_share() + repo.delete.return_value = True + + use_case = RevokeShareUseCase(repo) + result = use_case.execute("share_001", "user_001") + + assert result is True + repo.delete.assert_called_once_with("share_001", "user_001") + + def test_revoke_not_found_raises(self) -> None: + repo = MagicMock() + repo.get_by_id.return_value = None + + use_case = RevokeShareUseCase(repo) + + with pytest.raises(NotFoundError): + use_case.execute("nonexistent", "user_001") + + +class TestRecordShareDownloadUseCase: + def test_record_success(self) -> None: + repo = MagicMock() + repo.get_by_token.return_value = _make_share() + repo.increment_download.return_value = None + + use_case = RecordShareDownloadUseCase(repo) + use_case.execute("token") + + repo.increment_download.assert_called_once() + + def test_record_with_password(self) -> None: + repo = MagicMock() + repo.get_by_token.return_value = _make_share(password="pass") + repo.increment_download.return_value = None + + use_case = RecordShareDownloadUseCase(repo) + use_case.execute("token", password="pass") + + repo.increment_download.assert_called_once() + + def test_record_wrong_password_raises(self) -> None: + repo = MagicMock() + repo.get_by_token.return_value = _make_share(password="correct") + + use_case = RecordShareDownloadUseCase(repo) + + with pytest.raises(InvalidPasswordError): + use_case.execute("token", password="wrong") + + def test_record_share_not_found(self) -> None: + repo = MagicMock() + repo.get_by_token.return_value = None + + use_case = RecordShareDownloadUseCase(repo) + + with pytest.raises(NotFoundError): + use_case.execute("nonexistent") + + def test_record_expired_share(self) -> None: + repo = MagicMock() + past = datetime.now(timezone.utc) - timedelta(days=1) + repo.get_by_token.return_value = _make_share(expires_at=past) + + use_case = RecordShareDownloadUseCase(repo) + + with pytest.raises(ShareExpiredError): + use_case.execute("token")