Coverage for app / api / routes / auth.py: 100%

1from datetime import datetime, timedelta, timezone 

2from fastapi import APIRouter, Depends, HTTPException, Query 

3from sqlalchemy.ext.asyncio import AsyncSession 

4from sqlalchemy import select 

5 

6from app.db.database import get_db 

7from app.db.models import Creator, MagicLink 

8from app.schemas.auth import MagicLinkRequest, MagicLinkResponse, VerifyTokenRequest, VerifyTokenResponse 

9from app.core.security import create_magic_link_token, create_access_token 

10from app.core.config import settings 

11 

12router = APIRouter() 

13 

14 

15@router.post("/magic-link", response_model=MagicLinkResponse) 

16async def request_magic_link( 

17 request: MagicLinkRequest, 

18 db: AsyncSession = Depends(get_db), 

19): 

20 """Request a magic link for authentication.""" 

21 # For MVP, we'll just create the magic link token 

22 # In production, you'd send an email here 

23 

24 token = create_magic_link_token() 

25 expires_at = datetime.now(timezone.utc) + timedelta( 

26 minutes=settings.magic_link_expire_minutes 

27 ) 

28 

29 try: 

30 # Store magic link 

31 magic_link = MagicLink( 

32 email=request.email, 

33 token=token, 

34 expires_at=expires_at, 

35 ) 

36 

37 # Get or create creator 

38 creator_query = select(Creator).where(Creator.email == request.email) 

39 creator_result = await db.execute(creator_query) 

40 creator = creator_result.scalar_one_or_none() 

41 

42 if not creator: 

43 creator = Creator(email=request.email) 

44 db.add(creator) 

45 await db.flush() 

46 

47 db.add(magic_link) 

48 await db.commit() 

49 

50 # In production, send email with link: 

51 # https://moltfund.me/auth/verify?token={token} 

52 # For MVP, we'll just return the token (not secure, but works for dev) 

53 

54 return MagicLinkResponse( 

55 success=True, 

56 message=f"Magic link created. Token: {token} (dev only - don't expose in production)", 

57 ) 

58 except Exception as e: 

59 await db.rollback() 

60 raise HTTPException(status_code=500, detail=f"Failed to create magic link: {str(e)}") 

61 

62 

63@router.get("/verify", response_model=VerifyTokenResponse) 

64async def verify_magic_link( 

65 token: str = Query(...), 

66 db: AsyncSession = Depends(get_db), 

67): 

68 """Verify a magic link token and return JWT.""" 

69 # Find magic link 

70 magic_link_query = select(MagicLink).where( 

71 MagicLink.token == token, 

72 MagicLink.used_at.is_(None), 

73 ) 

74 magic_link_result = await db.execute(magic_link_query) 

75 magic_link = magic_link_result.scalar_one_or_none() 

76 

77 if not magic_link: 

78 raise HTTPException(status_code=401, detail="Invalid or expired token") 

79 

80 # Check expiration 

81 if magic_link.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc): 

82 raise HTTPException(status_code=401, detail="Token expired") 

83 

84 try: 

85 # Get or create creator 

86 creator_query = select(Creator).where(Creator.email == magic_link.email) 

87 creator_result = await db.execute(creator_query) 

88 creator = creator_result.scalar_one_or_none() 

89 

90 if not creator: 

91 creator = Creator(email=magic_link.email) 

92 db.add(creator) 

93 await db.flush() 

94 

95 # Mark magic link as used 

96 magic_link.used_at = datetime.now(timezone.utc) 

97 

98 # Create JWT with user info 

99 access_token = create_access_token(data={ 

100 "sub": creator.id, 

101 "email": creator.email, 

102 }) 

103 

104 await db.commit() 

105 

106 return VerifyTokenResponse( 

107 success=True, 

108 access_token=access_token, 

109 message="Authentication successful", 

110 ) 

111 except Exception as e: 

112 await db.rollback() 

113 raise HTTPException(status_code=500, detail=f"Failed to verify token: {str(e)}") 

114 

115 

116@router.post("/logout") 

117async def logout(): 

118 """Logout (client-side token removal).""" 

119 return {"success": True, "message": "Logged out"}