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

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

2 

3from typing import Annotated 

4 

5from fastapi import APIRouter, Depends, 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 401: {"description": "Not authorized"}, 

31 }, 

32) 

33async def post( 

34 resource: RevokedUserDocument, 

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

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

37) -> RevokedUserDocument: 

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

39 presenter = JsonApiRevokedUserPresenter() 

40 if resource.data.attributes.revoked: 

41 async with UnitOfWork(database): 

42 await RevokeUser( 

43 UserAccountDbRepository(database), 

44 UserTokenDbRepository(database), 

45 presenter, 

46 ).execute(RevokeUserCommand(uuid=resource.data.id)) 

47 else: 

48 async with UnitOfWork(database): 

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

50 EnactUserCommand(uuid=resource.data.id) 

51 ) 

52 return presenter.get_document() 

53 

54 

55@router.delete( 

56 "/revoked_users/{id}", 

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

58 status_code=status.HTTP_200_OK, 

59 responses={ 

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

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

62 }, 

63) 

64async def delete( 

65 id: str, 

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

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

68): 

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

70 presenter = JsonApiRevokedUserPresenter() 

71 async with UnitOfWork(database): 

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

73 EnactUserCommand(uuid=id) 

74 )