Coverage for src/tests/api/v1/auth/endpoints/test_user_invitations.py: 92%

1"""Test the endpoint users.""" 

2 

3import json 

4 

5import pytest 

6 

7from fastapi import status 

8from fastapi.testclient import TestClient 

9from kwai_api.v1.auth.schemas.user_invitation import ( 

10 UserInvitationAttributes, 

11 UserInvitationDocument, 

12 UserInvitationResource, 

13) 

14from kwai_bc_identity.user_invitations.user_invitation_db_repository import ( 

15 UserInvitationDbRepository, 

16) 

17from kwai_bc_identity.user_invitations.user_invitation_repository import ( 

18 UserInvitationRepository, 

19) 

20from kwai_core.db.database import Database 

21 

22 

23pytestmark = [pytest.mark.api, pytest.mark.db] 

24 

25 

26@pytest.fixture(scope="module") 

27def repo(database: Database) -> UserInvitationRepository: 

28 """Create the invitation repository.""" 

29 return UserInvitationDbRepository(database) 

30 

31 

32@pytest.fixture(scope="module") 

33def invitation_data() -> list[str]: 

34 """Create array to capture the uuid created in test_create_user_invitation.""" 

35 return [] 

36 

37 

38@pytest.mark.mail 

39def test_create_user_invitation(secure_client: TestClient, invitation_data: list[str]): 

40 """Test POST users/invitations.""" 

41 data = { 

42 "data": { 

43 "type": "user_invitations", 

44 "attributes": { 

45 "email": "ichiro.abe@kwai.com", 

46 "first_name": "Ichiro", 

47 "last_name": "Abe", 

48 "remark": "Invitation created in test_create_user_invitation", 

49 "revoked": False, 

50 }, 

51 } 

52 } 

53 response = secure_client.post( 

54 "api/v1/auth/users/invitations", content=json.dumps(data) 

55 ) 

56 assert response.status_code == status.HTTP_201_CREATED, response.content 

57 

58 invitation_data.append(response.json()["data"]["id"]) 

59 

60 

61def test_get_user_invitations(secure_client: TestClient): 

62 """Test GET users/invitations.""" 

63 response = secure_client.get("api/v1/auth/users/invitations") 

64 assert response.status_code == status.HTTP_200_OK 

65 

66 

67def test_get_user_invitation( 

68 secure_client: TestClient, 

69 invitation_data: list[str], 

70): 

71 """Test GET users/invitations/<uuid>.""" 

72 if len(invitation_data) > 0: 

73 response = secure_client.get( 

74 f"api/v1/auth/users/invitations/{invitation_data[0]}" 

75 ) 

76 assert response.status_code == status.HTTP_200_OK 

77 else: 

78 pytest.skip("Test skipped because there is no user invitation") 

79 

80 

81def test_delete_user_invitation(secure_client: TestClient, invitation_data: list[str]): 

82 """Test DELETE users/invitations/<uuid>.""" 

83 if len(invitation_data) > 0: 

84 response = secure_client.delete( 

85 f"api/v1/auth/users/invitations/{invitation_data[0]}" 

86 ) 

87 assert response.status_code == status.HTTP_200_OK 

88 else: 

89 pytest.skip("Test skipped because there is no user invitation") 

90 

91 

92async def test_update_user_invitation( 

93 secure_client: TestClient, make_user_invitation_in_db 

94): 

95 """Test PATCH users/invitations+/<uuid>.""" 

96 user_invitation = await make_user_invitation_in_db() 

97 

98 user_invitation_document = UserInvitationDocument( 

99 data=UserInvitationResource( 

100 id=str(user_invitation.uuid), 

101 attributes=UserInvitationAttributes( 

102 email=str(user_invitation.email), 

103 first_name=user_invitation.name.first_name, 

104 last_name=user_invitation.name.last_name, 

105 remark="Updated with a test", 

106 revoked=True, 

107 ), 

108 ) 

109 ) 

110 

111 response = secure_client.patch( 

112 f"api/v1/auth/users/invitations/{user_invitation.uuid}", 

113 content=user_invitation_document.model_dump_json(), 

114 ) 

115 assert response.status_code == status.HTTP_200_OK