phase 14
This commit is contained in:
277
backend/app/api/sharing.py
Normal file
277
backend/app/api/sharing.py
Normal file
@@ -0,0 +1,277 @@
|
||||
"""Board sharing API endpoints."""
|
||||
|
||||
from datetime import datetime
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.boards.schemas import (
|
||||
BoardDetail,
|
||||
CommentCreate,
|
||||
CommentResponse,
|
||||
ShareLinkCreate,
|
||||
ShareLinkResponse,
|
||||
)
|
||||
from app.boards.sharing import generate_secure_token
|
||||
from app.core.deps import get_current_user, get_db
|
||||
from app.database.models.board import Board
|
||||
from app.database.models.comment import Comment
|
||||
from app.database.models.share_link import ShareLink
|
||||
from app.database.models.user import User
|
||||
|
||||
router = APIRouter(tags=["sharing"])
|
||||
|
||||
|
||||
def validate_share_link(token: str, db: Session, required_permission: str = "view-only") -> ShareLink:
|
||||
"""
|
||||
Validate share link token and check permissions.
|
||||
|
||||
Args:
|
||||
token: Share link token
|
||||
db: Database session
|
||||
required_permission: Required permission level
|
||||
|
||||
Returns:
|
||||
ShareLink if valid
|
||||
|
||||
Raises:
|
||||
HTTPException: 403 if invalid or insufficient permissions
|
||||
"""
|
||||
share_link = (
|
||||
db.query(ShareLink)
|
||||
.filter(
|
||||
ShareLink.token == token,
|
||||
ShareLink.is_revoked == False, # noqa: E712
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if share_link is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Invalid or revoked share link",
|
||||
)
|
||||
|
||||
# Check expiration
|
||||
if share_link.expires_at and share_link.expires_at < datetime.utcnow():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Share link has expired",
|
||||
)
|
||||
|
||||
# Check permission level
|
||||
if required_permission == "view-comment" and share_link.permission_level != "view-comment":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Insufficient permissions - commenting not allowed",
|
||||
)
|
||||
|
||||
# Update access tracking
|
||||
share_link.access_count += 1
|
||||
share_link.last_accessed_at = datetime.utcnow()
|
||||
db.commit()
|
||||
|
||||
return share_link
|
||||
|
||||
|
||||
@router.post("/boards/{board_id}/share-links", response_model=ShareLinkResponse, status_code=status.HTTP_201_CREATED)
|
||||
def create_share_link(
|
||||
board_id: UUID,
|
||||
share_link_data: ShareLinkCreate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> ShareLinkResponse:
|
||||
"""
|
||||
Create a new share link for a board.
|
||||
|
||||
Only the board owner can create share links.
|
||||
"""
|
||||
# Verify board exists and user owns it
|
||||
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
||||
|
||||
if board is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Board not found or access denied",
|
||||
)
|
||||
|
||||
# Generate unique token
|
||||
token = generate_secure_token()
|
||||
|
||||
# Create share link
|
||||
share_link = ShareLink(
|
||||
board_id=board_id,
|
||||
token=token,
|
||||
permission_level=share_link_data.permission_level,
|
||||
expires_at=share_link_data.expires_at,
|
||||
)
|
||||
db.add(share_link)
|
||||
db.commit()
|
||||
db.refresh(share_link)
|
||||
|
||||
return ShareLinkResponse.model_validate(share_link)
|
||||
|
||||
|
||||
@router.get("/boards/{board_id}/share-links", response_model=list[ShareLinkResponse])
|
||||
def list_share_links(
|
||||
board_id: UUID,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> list[ShareLinkResponse]:
|
||||
"""
|
||||
List all share links for a board.
|
||||
|
||||
Only the board owner can list share links.
|
||||
"""
|
||||
# Verify board exists and user owns it
|
||||
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
||||
|
||||
if board is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Board not found or access denied",
|
||||
)
|
||||
|
||||
# Get all share links for board
|
||||
share_links = db.query(ShareLink).filter(ShareLink.board_id == board_id).order_by(ShareLink.created_at.desc()).all()
|
||||
|
||||
return [ShareLinkResponse.model_validate(link) for link in share_links]
|
||||
|
||||
|
||||
@router.delete("/boards/{board_id}/share-links/{link_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def revoke_share_link(
|
||||
board_id: UUID,
|
||||
link_id: UUID,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> None:
|
||||
"""
|
||||
Revoke (soft delete) a share link.
|
||||
|
||||
Only the board owner can revoke share links.
|
||||
"""
|
||||
# Verify board exists and user owns it
|
||||
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
||||
|
||||
if board is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Board not found or access denied",
|
||||
)
|
||||
|
||||
# Get and revoke share link
|
||||
share_link = db.query(ShareLink).filter(ShareLink.id == link_id, ShareLink.board_id == board_id).first()
|
||||
|
||||
if share_link is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Share link not found",
|
||||
)
|
||||
|
||||
share_link.is_revoked = True
|
||||
db.commit()
|
||||
|
||||
|
||||
@router.get("/shared/{token}", response_model=BoardDetail)
|
||||
def get_shared_board(
|
||||
token: str,
|
||||
db: Session = Depends(get_db),
|
||||
) -> BoardDetail:
|
||||
"""
|
||||
Access a shared board via token.
|
||||
|
||||
No authentication required - access controlled by share link token.
|
||||
"""
|
||||
# Validate share link
|
||||
share_link = validate_share_link(token, db, required_permission="view-only")
|
||||
|
||||
# Get board details
|
||||
board = db.query(Board).filter(Board.id == share_link.board_id).first()
|
||||
|
||||
if board is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Board not found",
|
||||
)
|
||||
|
||||
return BoardDetail.model_validate(board)
|
||||
|
||||
|
||||
@router.post("/shared/{token}/comments", response_model=CommentResponse, status_code=status.HTTP_201_CREATED)
|
||||
def create_comment(
|
||||
token: str,
|
||||
comment_data: CommentCreate,
|
||||
db: Session = Depends(get_db),
|
||||
) -> CommentResponse:
|
||||
"""
|
||||
Create a comment on a shared board.
|
||||
|
||||
Requires view-comment permission level.
|
||||
"""
|
||||
# Validate share link with comment permission
|
||||
share_link = validate_share_link(token, db, required_permission="view-comment")
|
||||
|
||||
# Create comment
|
||||
comment = Comment(
|
||||
board_id=share_link.board_id,
|
||||
share_link_id=share_link.id,
|
||||
author_name=comment_data.author_name,
|
||||
content=comment_data.content,
|
||||
position=comment_data.position,
|
||||
)
|
||||
db.add(comment)
|
||||
db.commit()
|
||||
db.refresh(comment)
|
||||
|
||||
return CommentResponse.model_validate(comment)
|
||||
|
||||
|
||||
@router.get("/shared/{token}/comments", response_model=list[CommentResponse])
|
||||
def list_comments(
|
||||
token: str,
|
||||
db: Session = Depends(get_db),
|
||||
) -> list[CommentResponse]:
|
||||
"""
|
||||
List all comments on a shared board.
|
||||
|
||||
Requires view-only or view-comment permission.
|
||||
"""
|
||||
# Validate share link
|
||||
share_link = validate_share_link(token, db, required_permission="view-only")
|
||||
|
||||
# Get all comments for board (non-deleted)
|
||||
comments = (
|
||||
db.query(Comment)
|
||||
.filter(Comment.board_id == share_link.board_id, Comment.is_deleted == False) # noqa: E712
|
||||
.order_by(Comment.created_at.desc())
|
||||
.all()
|
||||
)
|
||||
|
||||
return [CommentResponse.model_validate(comment) for comment in comments]
|
||||
|
||||
|
||||
@router.get("/boards/{board_id}/comments", response_model=list[CommentResponse])
|
||||
def list_board_comments(
|
||||
board_id: UUID,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> list[CommentResponse]:
|
||||
"""
|
||||
List all comments on a board (owner view).
|
||||
|
||||
Only the board owner can access this endpoint.
|
||||
"""
|
||||
# Verify board exists and user owns it
|
||||
board = db.query(Board).filter(Board.id == board_id, Board.user_id == current_user.id).first()
|
||||
|
||||
if board is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Board not found or access denied",
|
||||
)
|
||||
|
||||
# Get all comments for board (including deleted for owner)
|
||||
comments = db.query(Comment).filter(Comment.board_id == board_id).order_by(Comment.created_at.desc()).all()
|
||||
|
||||
return [CommentResponse.model_validate(comment) for comment in comments]
|
||||
@@ -106,3 +106,49 @@ class GroupResponse(BaseModel):
|
||||
member_count: int = Field(default=0, description="Number of images in group")
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
|
||||
class ShareLinkCreate(BaseModel):
|
||||
"""Schema for creating a new share link."""
|
||||
|
||||
permission_level: str = Field(..., pattern=r"^(view-only|view-comment)$", description="Permission level")
|
||||
expires_at: datetime | None = Field(None, description="Optional expiration datetime")
|
||||
|
||||
|
||||
class ShareLinkResponse(BaseModel):
|
||||
"""Response schema for share link."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: UUID
|
||||
board_id: UUID
|
||||
token: str
|
||||
permission_level: str
|
||||
created_at: datetime
|
||||
expires_at: datetime | None = None
|
||||
last_accessed_at: datetime | None = None
|
||||
access_count: int = 0
|
||||
is_revoked: bool = False
|
||||
|
||||
|
||||
class CommentCreate(BaseModel):
|
||||
"""Schema for creating a new comment."""
|
||||
|
||||
author_name: str = Field(..., min_length=1, max_length=100, description="Commenter name")
|
||||
content: str = Field(..., min_length=1, max_length=5000, description="Comment text")
|
||||
position: dict | None = Field(None, description="Optional canvas position {x, y}")
|
||||
|
||||
|
||||
class CommentResponse(BaseModel):
|
||||
"""Response schema for comment."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: UUID
|
||||
board_id: UUID
|
||||
share_link_id: UUID | None = None
|
||||
author_name: str
|
||||
content: str
|
||||
position: dict | None = None
|
||||
created_at: datetime
|
||||
is_deleted: bool = False
|
||||
|
||||
84
backend/app/boards/sharing.py
Normal file
84
backend/app/boards/sharing.py
Normal file
@@ -0,0 +1,84 @@
|
||||
"""Board sharing functionality."""
|
||||
|
||||
import secrets
|
||||
import string
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database.models.share_link import ShareLink
|
||||
|
||||
|
||||
def generate_secure_token(length: int = 64) -> str:
|
||||
"""
|
||||
Generate a cryptographically secure random token for share links.
|
||||
|
||||
Args:
|
||||
length: Length of the token (default 64 characters)
|
||||
|
||||
Returns:
|
||||
URL-safe random string
|
||||
"""
|
||||
# Use URL-safe characters (alphanumeric + - and _)
|
||||
alphabet = string.ascii_letters + string.digits + "-_"
|
||||
return "".join(secrets.choice(alphabet) for _ in range(length))
|
||||
|
||||
|
||||
def validate_share_link_token(token: str, db: Session) -> ShareLink | None:
|
||||
"""
|
||||
Validate a share link token and return the share link if valid.
|
||||
|
||||
A share link is valid if:
|
||||
- Token exists
|
||||
- Not revoked
|
||||
- Not expired (if expires_at is set)
|
||||
|
||||
Args:
|
||||
token: The share link token
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
ShareLink if valid, None otherwise
|
||||
"""
|
||||
share_link = (
|
||||
db.query(ShareLink)
|
||||
.filter(
|
||||
ShareLink.token == token,
|
||||
ShareLink.is_revoked == False, # noqa: E712
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if share_link is None:
|
||||
return None
|
||||
|
||||
# Check expiration
|
||||
if share_link.expires_at and share_link.expires_at < datetime.utcnow():
|
||||
return None
|
||||
|
||||
# Update access tracking
|
||||
share_link.access_count += 1
|
||||
share_link.last_accessed_at = datetime.utcnow()
|
||||
db.commit()
|
||||
|
||||
return share_link
|
||||
|
||||
|
||||
def check_permission(share_link: ShareLink, required_permission: str) -> bool:
|
||||
"""
|
||||
Check if a share link has the required permission level.
|
||||
|
||||
Args:
|
||||
share_link: The share link to check
|
||||
required_permission: Required permission level ('view-only' or 'view-comment')
|
||||
|
||||
Returns:
|
||||
True if permission granted, False otherwise
|
||||
"""
|
||||
if required_permission == "view-only":
|
||||
# Both view-only and view-comment can view
|
||||
return share_link.permission_level in ("view-only", "view-comment")
|
||||
elif required_permission == "view-comment":
|
||||
# Only view-comment can comment
|
||||
return share_link.permission_level == "view-comment"
|
||||
return False
|
||||
@@ -13,6 +13,7 @@ from app.database.base import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from app.database.models.board_image import BoardImage
|
||||
from app.database.models.comment import Comment
|
||||
from app.database.models.group import Group
|
||||
from app.database.models.share_link import ShareLink
|
||||
from app.database.models.user import User
|
||||
@@ -56,6 +57,7 @@ class Board(Base):
|
||||
share_links: Mapped[list["ShareLink"]] = relationship(
|
||||
"ShareLink", back_populates="board", cascade="all, delete-orphan"
|
||||
)
|
||||
comments: Mapped[list["Comment"]] = relationship("Comment", back_populates="board", cascade="all, delete-orphan")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""String representation of Board."""
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Comment model for board comments."""
|
||||
"""Comment model for board annotations."""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
@@ -11,18 +11,16 @@ from app.database.base import Base
|
||||
|
||||
|
||||
class Comment(Base):
|
||||
"""Comment model for viewer comments on shared boards."""
|
||||
"""Comment model representing viewer comments on shared boards."""
|
||||
|
||||
__tablename__ = "comments"
|
||||
|
||||
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
||||
board_id = Column(UUID(as_uuid=True), ForeignKey("boards.id", ondelete="CASCADE"), nullable=False, index=True)
|
||||
share_link_id = Column(
|
||||
UUID(as_uuid=True), ForeignKey("share_links.id", ondelete="SET NULL"), nullable=True, index=True
|
||||
)
|
||||
board_id = Column(UUID(as_uuid=True), ForeignKey("boards.id", ondelete="CASCADE"), nullable=False)
|
||||
share_link_id = Column(UUID(as_uuid=True), ForeignKey("share_links.id", ondelete="SET NULL"), nullable=True)
|
||||
author_name = Column(String(100), nullable=False)
|
||||
content = Column(Text, nullable=False)
|
||||
position = Column(JSONB, nullable=True) # Optional canvas position
|
||||
position = Column(JSONB, nullable=True) # Optional canvas position reference
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
is_deleted = Column(Boolean, nullable=False, default=False)
|
||||
|
||||
@@ -31,4 +29,4 @@ class Comment(Base):
|
||||
share_link = relationship("ShareLink", back_populates="comments")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Comment(id={self.id}, author={self.author_name})>"
|
||||
return f"<Comment(id={self.id}, board_id={self.board_id}, author={self.author_name})>"
|
||||
|
||||
@@ -1,45 +1,33 @@
|
||||
"""ShareLink database model."""
|
||||
"""ShareLink model for board sharing functionality."""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String
|
||||
from sqlalchemy.dialects.postgresql import UUID as PGUUID
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String
|
||||
from sqlalchemy.dialects.postgresql import UUID
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from app.database.base import Base
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from app.database.models.board import Board
|
||||
|
||||
|
||||
class ShareLink(Base):
|
||||
"""
|
||||
ShareLink model for sharing boards with configurable permissions.
|
||||
|
||||
Share links allow users to share boards with others without requiring
|
||||
authentication, with permission levels controlling what actions are allowed.
|
||||
"""
|
||||
"""ShareLink model representing shareable board links with permissions."""
|
||||
|
||||
__tablename__ = "share_links"
|
||||
|
||||
id: Mapped[UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=uuid4)
|
||||
board_id: Mapped[UUID] = mapped_column(
|
||||
PGUUID(as_uuid=True), ForeignKey("boards.id", ondelete="CASCADE"), nullable=False
|
||||
)
|
||||
token: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True)
|
||||
permission_level: Mapped[str] = mapped_column(String(20), nullable=False) # 'view-only' or 'view-comment'
|
||||
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
expires_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
|
||||
last_accessed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
|
||||
access_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
is_revoked: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
||||
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
||||
board_id = Column(UUID(as_uuid=True), ForeignKey("boards.id", ondelete="CASCADE"), nullable=False)
|
||||
token = Column(String(64), unique=True, nullable=False, index=True)
|
||||
permission_level = Column(String(20), nullable=False) # 'view-only' or 'view-comment'
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
expires_at = Column(DateTime, nullable=True)
|
||||
last_accessed_at = Column(DateTime, nullable=True)
|
||||
access_count = Column(Integer, nullable=False, default=0)
|
||||
is_revoked = Column(Boolean, nullable=False, default=False)
|
||||
|
||||
# Relationships
|
||||
board: Mapped["Board"] = relationship("Board", back_populates="share_links")
|
||||
board = relationship("Board", back_populates="share_links")
|
||||
comments = relationship("Comment", back_populates="share_link", cascade="all, delete-orphan")
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""String representation of ShareLink."""
|
||||
return f"<ShareLink(id={self.id}, token={self.token[:8]}..., board_id={self.board_id})>"
|
||||
return f"<ShareLink(id={self.id}, board_id={self.board_id}, permission={self.permission_level})>"
|
||||
|
||||
@@ -5,7 +5,7 @@ import logging
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from app.api import auth, boards, groups, images
|
||||
from app.api import auth, boards, groups, images, sharing
|
||||
from app.core.config import settings
|
||||
from app.core.errors import WebRefException
|
||||
from app.core.logging import setup_logging
|
||||
@@ -86,6 +86,7 @@ app.include_router(auth.router, prefix=f"{settings.API_V1_PREFIX}")
|
||||
app.include_router(boards.router, prefix=f"{settings.API_V1_PREFIX}")
|
||||
app.include_router(groups.router, prefix=f"{settings.API_V1_PREFIX}")
|
||||
app.include_router(images.router, prefix=f"{settings.API_V1_PREFIX}")
|
||||
app.include_router(sharing.router, prefix=f"{settings.API_V1_PREFIX}")
|
||||
|
||||
|
||||
@app.on_event("startup")
|
||||
|
||||
302
backend/tests/api/test_sharing.py
Normal file
302
backend/tests/api/test_sharing.py
Normal file
@@ -0,0 +1,302 @@
|
||||
"""Tests for board sharing endpoints."""
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from fastapi import status
|
||||
|
||||
|
||||
def test_create_share_link_view_only(client, auth_headers, test_board):
|
||||
"""Test creating a view-only share link."""
|
||||
response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
data = response.json()
|
||||
assert data["permission_level"] == "view-only"
|
||||
assert data["board_id"] == str(test_board.id)
|
||||
assert data["token"] is not None
|
||||
assert len(data["token"]) == 64
|
||||
assert data["is_revoked"] == False # noqa: E712
|
||||
assert data["access_count"] == 0
|
||||
|
||||
|
||||
def test_create_share_link_view_comment(client, auth_headers, test_board):
|
||||
"""Test creating a view-comment share link."""
|
||||
response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-comment"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
data = response.json()
|
||||
assert data["permission_level"] == "view-comment"
|
||||
|
||||
|
||||
def test_create_share_link_with_expiration(client, auth_headers, test_board):
|
||||
"""Test creating a share link with expiration."""
|
||||
expires_at = (datetime.utcnow() + timedelta(days=7)).isoformat()
|
||||
response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only", "expires_at": expires_at},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
data = response.json()
|
||||
assert data["expires_at"] is not None
|
||||
|
||||
|
||||
def test_create_share_link_invalid_permission(client, auth_headers, test_board):
|
||||
"""Test creating share link with invalid permission level."""
|
||||
response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "invalid-permission"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
|
||||
|
||||
|
||||
def test_create_share_link_unauthorized(client, test_board):
|
||||
"""Test creating share link without authentication."""
|
||||
response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
def test_create_share_link_not_owner(client, other_auth_headers, test_board):
|
||||
"""Test creating share link for board user doesn't own."""
|
||||
response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=other_auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
def test_list_share_links(client, auth_headers, test_board):
|
||||
"""Test listing all share links for a board."""
|
||||
# Create multiple share links
|
||||
client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-comment"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
response = client.get(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) >= 2
|
||||
assert all("token" in link for link in data)
|
||||
|
||||
|
||||
def test_list_share_links_unauthorized(client, test_board):
|
||||
"""Test listing share links without authentication."""
|
||||
response = client.get(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
def test_revoke_share_link(client, auth_headers, test_board):
|
||||
"""Test revoking a share link."""
|
||||
# Create a share link
|
||||
create_response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
link_id = create_response.json()["id"]
|
||||
|
||||
# Revoke it
|
||||
response = client.delete(
|
||||
f"/api/boards/{test_board.id}/share-links/{link_id}",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
# Verify it's revoked by listing
|
||||
list_response = client.get(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
headers=auth_headers,
|
||||
)
|
||||
revoked_link = next((link for link in list_response.json() if link["id"] == link_id), None)
|
||||
assert revoked_link is not None
|
||||
assert revoked_link["is_revoked"] == True # noqa: E712
|
||||
|
||||
|
||||
def test_revoke_share_link_not_found(client, auth_headers, test_board):
|
||||
"""Test revoking non-existent share link."""
|
||||
import uuid
|
||||
|
||||
fake_id = uuid.uuid4()
|
||||
response = client.delete(
|
||||
f"/api/boards/{test_board.id}/share-links/{fake_id}",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
def test_access_shared_board(client, auth_headers, test_board):
|
||||
"""Test accessing a board via share link."""
|
||||
# Create share link
|
||||
create_response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
token = create_response.json()["token"]
|
||||
|
||||
# Access shared board (no auth required)
|
||||
response = client.get(f"/api/shared/{token}")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["id"] == str(test_board.id)
|
||||
assert data["title"] == test_board.title
|
||||
|
||||
|
||||
def test_access_shared_board_invalid_token(client):
|
||||
"""Test accessing board with invalid token."""
|
||||
response = client.get("/api/shared/invalid-token-12345")
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
def test_access_shared_board_revoked_token(client, auth_headers, test_board):
|
||||
"""Test accessing board with revoked token."""
|
||||
# Create and revoke share link
|
||||
create_response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
data = create_response.json()
|
||||
token = data["token"]
|
||||
link_id = data["id"]
|
||||
|
||||
client.delete(
|
||||
f"/api/boards/{test_board.id}/share-links/{link_id}",
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Try to access with revoked token
|
||||
response = client.get(f"/api/shared/{token}")
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
def test_create_comment_on_shared_board(client, auth_headers, test_board):
|
||||
"""Test creating a comment via share link with view-comment permission."""
|
||||
# Create view-comment share link
|
||||
create_response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-comment"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
token = create_response.json()["token"]
|
||||
|
||||
# Create comment (no auth required, just token)
|
||||
comment_data = {
|
||||
"author_name": "Test Viewer",
|
||||
"content": "This is a test comment",
|
||||
"position": {"x": 100, "y": 200},
|
||||
}
|
||||
response = client.post(f"/api/shared/{token}/comments", json=comment_data)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
data = response.json()
|
||||
assert data["author_name"] == "Test Viewer"
|
||||
assert data["content"] == "This is a test comment"
|
||||
assert data["position"]["x"] == 100
|
||||
|
||||
|
||||
def test_create_comment_view_only_permission_denied(client, auth_headers, test_board):
|
||||
"""Test creating comment with view-only permission fails."""
|
||||
# Create view-only share link
|
||||
create_response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
token = create_response.json()["token"]
|
||||
|
||||
# Try to create comment (should fail)
|
||||
comment_data = {
|
||||
"author_name": "Test Viewer",
|
||||
"content": "This should fail",
|
||||
}
|
||||
response = client.post(f"/api/shared/{token}/comments", json=comment_data)
|
||||
assert response.status_code == status.HTTP_403_FORBIDDEN
|
||||
|
||||
|
||||
def test_list_comments_on_shared_board(client, auth_headers, test_board):
|
||||
"""Test listing comments via share link."""
|
||||
# Create view-comment share link
|
||||
create_response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-comment"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
token = create_response.json()["token"]
|
||||
|
||||
# Create a comment
|
||||
client.post(
|
||||
f"/api/shared/{token}/comments",
|
||||
json={"author_name": "Viewer 1", "content": "Comment 1"},
|
||||
)
|
||||
|
||||
# List comments
|
||||
response = client.get(f"/api/shared/{token}/comments")
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) >= 1
|
||||
assert data[0]["content"] == "Comment 1"
|
||||
|
||||
|
||||
def test_list_board_comments_as_owner(client, auth_headers, test_board):
|
||||
"""Test board owner listing all comments."""
|
||||
# Create share link and comment
|
||||
create_response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-comment"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
token = create_response.json()["token"]
|
||||
client.post(
|
||||
f"/api/shared/{token}/comments",
|
||||
json={"author_name": "Viewer", "content": "Test comment"},
|
||||
)
|
||||
|
||||
# Owner lists comments
|
||||
response = client.get(
|
||||
f"/api/boards/{test_board.id}/comments",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) >= 1
|
||||
|
||||
|
||||
def test_token_uniqueness(client, auth_headers, test_board):
|
||||
"""Test that generated tokens are unique."""
|
||||
tokens = set()
|
||||
for _ in range(10):
|
||||
response = client.post(
|
||||
f"/api/boards/{test_board.id}/share-links",
|
||||
json={"permission_level": "view-only"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
token = response.json()["token"]
|
||||
tokens.add(token)
|
||||
|
||||
# All tokens should be unique
|
||||
assert len(tokens) == 10
|
||||
|
||||
@@ -104,3 +104,106 @@ def test_user_data_no_uppercase() -> dict:
|
||||
"""
|
||||
return {"email": "test@example.com", "password": "testpassword123"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_user(client: TestClient, test_user_data: dict):
|
||||
"""
|
||||
Create and return a test user.
|
||||
|
||||
Args:
|
||||
client: Test client
|
||||
test_user_data: User credentials
|
||||
|
||||
Returns:
|
||||
User object
|
||||
"""
|
||||
from app.database.models.user import User
|
||||
|
||||
response = client.post("/api/v1/auth/register", json=test_user_data)
|
||||
user_id = response.json()["id"]
|
||||
|
||||
# Get user from database (use same db session)
|
||||
from app.core.deps import get_db
|
||||
|
||||
db_gen = next(app.dependency_overrides[get_db]())
|
||||
user = db_gen.query(User).filter(User.id == user_id).first()
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_headers(client: TestClient, test_user_data: dict) -> dict:
|
||||
"""
|
||||
Create authenticated headers with JWT token.
|
||||
|
||||
Args:
|
||||
client: Test client
|
||||
test_user_data: User credentials
|
||||
|
||||
Returns:
|
||||
Dictionary with Authorization header
|
||||
"""
|
||||
# Register and login
|
||||
client.post("/api/v1/auth/register", json=test_user_data)
|
||||
login_response = client.post("/api/v1/auth/login", json=test_user_data)
|
||||
token = login_response.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def other_user_data() -> dict:
|
||||
"""
|
||||
Data for a second test user.
|
||||
|
||||
Returns:
|
||||
Dictionary with test user credentials
|
||||
"""
|
||||
return {"email": "other@example.com", "password": "OtherPassword123"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def other_auth_headers(client: TestClient, other_user_data: dict) -> dict:
|
||||
"""
|
||||
Create authenticated headers for a second user.
|
||||
|
||||
Args:
|
||||
client: Test client
|
||||
other_user_data: Other user credentials
|
||||
|
||||
Returns:
|
||||
Dictionary with Authorization header
|
||||
"""
|
||||
# Register and login
|
||||
client.post("/api/v1/auth/register", json=other_user_data)
|
||||
login_response = client.post("/api/v1/auth/login", json=other_user_data)
|
||||
token = login_response.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_board(client: TestClient, auth_headers: dict):
|
||||
"""
|
||||
Create a test board.
|
||||
|
||||
Args:
|
||||
client: Test client
|
||||
auth_headers: Authentication headers
|
||||
|
||||
Returns:
|
||||
Board object
|
||||
"""
|
||||
from app.database.models.board import Board
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/boards",
|
||||
json={"title": "Test Board", "description": "Test description"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
board_id = response.json()["id"]
|
||||
|
||||
# Get board from database
|
||||
from app.core.deps import get_db
|
||||
|
||||
db_gen = next(app.dependency_overrides[get_db]())
|
||||
board = db_gen.query(Board).filter(Board.id == board_id).first()
|
||||
return board
|
||||
|
||||
|
||||
Reference in New Issue
Block a user