Coverage for apps/kwai-api/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 

6from kwai_bc_identity.enact_user import EnactUser, EnactUserCommand 

7from kwai_bc_identity.revoke_user import RevokeUser, RevokeUserCommand 

8from kwai_bc_identity.tokens.user_token_db_repository import UserTokenDbRepository 

9from kwai_bc_identity.users.user import UserEntity 

10from kwai_bc_identity.users.user_account_db_repository import UserAccountDbRepository 

11from kwai_core.db.database import Database 

12from kwai_core.db.uow import UnitOfWork 

13 

14from kwai_api.dependencies import create_database, get_current_user 

15from kwai_api.v1.auth.presenters import JsonApiRevokedUserPresenter 

16from kwai_api.v1.auth.schemas.revoked_user import RevokedUserDocument 

17 

18 

19router = APIRouter() 

20 

21 

22@router.post( 

23 "/revoked_users", 

24 summary="Revoke a user", 

25 status_code=status.HTTP_201_CREATED, 

26 responses={ 

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

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

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

30 }, 

31) 

32async def post( 

33 document: RevokedUserDocument, 

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

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

36) -> RevokedUserDocument: 

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

38 if document.data.id is None: 

39 raise HTTPException( 

40 status_code=status.HTTP_400_BAD_REQUEST, 

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

42 ) 

43 presenter = JsonApiRevokedUserPresenter() 

44 if document.data.attributes.revoked: 

45 async with UnitOfWork(database): 

46 await RevokeUser( 

47 UserAccountDbRepository(database), 

48 UserTokenDbRepository(database), 

49 presenter, 

50 ).execute(RevokeUserCommand(uuid=document.data.id)) 

51 else: 

52 async with UnitOfWork(database): 

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

54 EnactUserCommand(uuid=document.data.id) 

55 ) 

56 return presenter.get_document() 

57 

58 

59@router.delete( 

60 "/revoked_users/{id}", 

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

62 status_code=status.HTTP_200_OK, 

63 responses={ 

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

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

66 }, 

67) 

68async def delete( 

69 id: str, 

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

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

72): 

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

74 presenter = JsonApiRevokedUserPresenter() 

75 async with UnitOfWork(database): 

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

77 EnactUserCommand(uuid=id) 

78 )