Coverage for src / mysingle / auth / router / oauth_management.py: 0%

1"""OAuth 계정 관리 엔드포인트""" 

2 

3from beanie import PydanticObjectId 

4from fastapi import APIRouter, Request, status 

5 

6from ..deps import admin_only, get_current_user, verified_only 

7from ..exceptions import UserNotExists 

8from ..user_manager import UserManager 

9 

10user_manager = UserManager() 

11 

12 

13def get_oauth_management_router() -> APIRouter: 

14 """OAuth 계정 관리를 위한 라우터 생성""" 

15 router = APIRouter() 

16 

17 @router.get( 

18 "/me/oauth-accounts", 

19 response_model=dict, 

20 ) 

21 @verified_only 

22 async def get_my_oauth_accounts( 

23 request: Request, 

24 ) -> dict: 

25 """현재 사용자의 연결된 OAuth 계정 목록을 조회합니다.""" 

26 current_user = get_current_user(request) 

27 

28 oauth_accounts = [] 

29 for oauth_account in current_user.oauth_accounts: 

30 oauth_accounts.append( 

31 { 

32 "oauth_name": oauth_account.oauth_name, 

33 "account_id": oauth_account.account_id, 

34 "account_email": oauth_account.account_email, 

35 "expires_at": oauth_account.expires_at, 

36 "created_at": getattr(oauth_account, "created_at", None), 

37 } 

38 ) 

39 

40 return {"oauth_accounts": oauth_accounts, "total_count": len(oauth_accounts)} 

41 

42 @router.delete( 

43 "/me/oauth-accounts/{oauth_name}/{account_id}", 

44 status_code=status.HTTP_204_NO_CONTENT, 

45 ) 

46 @verified_only 

47 async def remove_oauth_account( 

48 request: Request, 

49 oauth_name: str, 

50 account_id: str, 

51 ) -> None: 

52 """특정 OAuth 계정 연결을 해제합니다.""" 

53 current_user = get_current_user(request) 

54 

55 await user_manager.remove_oauth_account(current_user, oauth_name, account_id) 

56 await user_manager.on_after_update( 

57 current_user, {"oauth_accounts": "removed"}, request 

58 ) 

59 

60 @router.get( 

61 "/{user_id}/oauth-accounts", 

62 response_model=dict, 

63 ) 

64 @admin_only 

65 async def get_user_oauth_accounts( 

66 request: Request, 

67 user_id: PydanticObjectId, 

68 ) -> dict: 

69 """특정 사용자의 OAuth 계정 목록을 조회합니다. (관리자 전용)""" 

70 

71 user = await user_manager.get(user_id) 

72 if user is None: 

73 raise UserNotExists(identifier=str(user_id), identifier_type="user") 

74 

75 oauth_accounts = [] 

76 for oauth_account in user.oauth_accounts: 

77 oauth_accounts.append( 

78 { 

79 "oauth_name": oauth_account.oauth_name, 

80 "account_id": oauth_account.account_id, 

81 "account_email": oauth_account.account_email, 

82 "expires_at": oauth_account.expires_at, 

83 "created_at": getattr(oauth_account, "created_at", None), 

84 } 

85 ) 

86 

87 return { 

88 "user_id": str(user.id), 

89 "user_email": user.email, 

90 "oauth_accounts": oauth_accounts, 

91 "total_count": len(oauth_accounts), 

92 } 

93 

94 return router