Coverage for src/kwai/api/v1/auth/endpoints/revoked_users.py: 90%

1"""Module that implements endpoints for revoke users.""" 

2 

3from typing import Annotated 

4 

5from fastapi import APIRouter, Depends, HTTPException, status 

6 

7from kwai.api.dependencies import create_database, get_current_user 

8from kwai.api.v1.auth.presenters import JsonApiRevokedUserPresenter 

9from kwai.api.v1.auth.schemas.revoked_user import RevokedUserDocument 

10from kwai.core.db.database import Database 

11from kwai.core.db.uow import UnitOfWork 

12from kwai.modules.identity.enact_user import EnactUser, EnactUserCommand 

13from kwai.modules.identity.revoke_user import RevokeUser, RevokeUserCommand 

14from kwai.modules.identity.tokens.user_token_db_repository import UserTokenDbRepository 

15from kwai.modules.identity.users.user import UserEntity 

16from kwai.modules.identity.users.user_account_db_repository import ( 

17 UserAccountDbRepository, 

18) 

19 

20 

21router = APIRouter() 

22 

23 

24@router.post( 

25 "/revoked_users", 

26 summary="Revoke a user", 

27 status_code=status.HTTP_201_CREATED, 

28 responses={ 

29 201: {"description": "User was successfully revoked"}, 

30 400: {"description": "An invalid request was made"}, 

31 401: {"description": "Not authorized"}, 

32 }, 

33) 

34async def post( 

35 document: RevokedUserDocument, 

36 database: Annotated[Database, Depends(create_database)], 

37 user: Annotated[UserEntity, Depends(get_current_user)], 

38) -> RevokedUserDocument: 

39 """(Un)revoke a user.""" 

40 if document.resource.id is None: 

41 raise HTTPException( 

42 status_code=status.HTTP_400_BAD_REQUEST, 

43 detail="The revoked user resource must have an id", 

44 ) 

45 presenter = JsonApiRevokedUserPresenter() 

46 if document.resource.attributes.revoked: 

47 async with UnitOfWork(database): 

48 await RevokeUser( 

49 UserAccountDbRepository(database), 

50 UserTokenDbRepository(database), 

51 presenter, 

52 ).execute(RevokeUserCommand(uuid=document.resource.id)) 

53 else: 

54 async with UnitOfWork(database): 

55 await EnactUser(UserAccountDbRepository(database), presenter).execute( 

56 EnactUserCommand(uuid=document.resource.id) 

57 ) 

58 return presenter.get_document() 

59 

60 

61@router.delete( 

62 "/revoked_users/{id}", 

63 summary="Cancel the revocation of a user", 

64 status_code=status.HTTP_200_OK, 

65 responses={ 

66 200: {"description": "The revocation was successfully cancelled"}, 

67 401: {"description": "Not authorized"}, 

68 }, 

69) 

70async def delete( 

71 id: str, 

72 database: Annotated[Database, Depends(create_database)], 

73 user: Annotated[UserEntity, Depends(get_current_user)], 

74): 

75 """Cancel the revocation of the user with the given id.""" 

76 presenter = JsonApiRevokedUserPresenter() 

77 async with UnitOfWork(database): 

78 await EnactUser(UserAccountDbRepository(database), presenter).execute( 

79 EnactUserCommand(uuid=id) 

80 )