Coverage for tests/test_microservice_websocket/test_routes/test_route_user.py: 100%

1import pytest 

2from fastapi.testclient import TestClient 

3 

4from microservice_websocket.app.services.database import User, user_manager 

5 

6 

7class TestGetUserList: 

8 endpoint = "/api/users" 

9 

10 @pytest.mark.asyncio 

11 async def test_get_user_list(self, app_client: TestClient, auth_header): 

12 await user_manager.create_user("foo", "bar", "John", "Doe") 

13 await user_manager.create_user("baz", "qux", "Janie", "Doe") 

14 

15 response = app_client.get(self.endpoint, headers=auth_header) 

16 

17 assert ( 

18 response.status_code == 200 and len(response.json()["users"]) == 3 

19 ), "Invalid response when trying to get user list" 

20 

21 

22class TestGetUserInfo: 

23 endpoint = "/api/user/" 

24 

25 def test_get_non_existing_user(self, app_client: TestClient, auth_header, obj_id): 

26 response = app_client.get(self.endpoint + obj_id, headers=auth_header) 

27 

28 assert ( 

29 response.status_code == 404 

30 ), "Invalid response code when trying to get non existing user info" 

31 

32 @pytest.mark.asyncio 

33 async def test_get_user_info(self, app_client: TestClient, auth_header): 

34 await user_manager.create_user("bar", "baz", "John", "Doe") 

35 user = await user_manager.get_user_from_mail("bar") 

36 assert user 

37 

38 response = app_client.get(self.endpoint + str(user.id), headers=auth_header) 

39 

40 assert response.status_code == 200 and response.json() == { 

41 "id": str(user.id), 

42 "email": user.email, 

43 "first_name": user.first_name, 

44 "last_name": user.last_name, 

45 "role": user.role, 

46 }, "Invalid response when trying to get user info" 

47 

48 

49class TestCreateUser: 

50 endpoint = "/api/user" 

51 

52 # Test creating user with invalid payloa 

53 def test_create_user_invalid_payload(self, app_client: TestClient, auth_header): 

54 response = app_client.post(self.endpoint, json={}, headers=auth_header) 

55 

56 assert ( 

57 response.status_code == 422 

58 ), "Invalid response code when trying to create user\ 

59 with invalid payload" 

60 

61 @pytest.mark.asyncio 

62 async def test_create_user_already_existing( 

63 self, app_client: TestClient, auth_header 

64 ): 

65 await user_manager.create_user("foo", "qux", "John", "Doe") 

66 

67 response = app_client.post( 

68 self.endpoint, 

69 json={ 

70 "email": "foo", 

71 "password": "bar", 

72 "role": "standard", 

73 "first_name": "John", 

74 "last_name": "Doe", 

75 }, 

76 headers=auth_header, 

77 ) 

78 

79 assert ( 

80 response.status_code == 400 

81 ), "Invalid response when trying to create already existing user" 

82 

83 @pytest.mark.asyncio 

84 async def test_create_user(self, app_client: TestClient, auth_header): 

85 response = app_client.post( 

86 self.endpoint, 

87 json={ 

88 "email": "pippo", 

89 "password": "pluto", 

90 "role": "standard", 

91 "first_name": "John", 

92 "last_name": "Doe", 

93 }, 

94 headers=auth_header, 

95 ) 

96 

97 assert ( 

98 response.status_code == 200 

99 ), "Invalid response code when trying to create user with valid payload" 

100 

101 assert ( 

102 len(await User.find(User.email == "pippo").to_list()) == 1 

103 ), "Invalid number of User with email='pippo'" 

104 

105 

106class TestUpdateUser: 

107 endpoint = "/api/user/" 

108 

109 def test_update_non_existing_user( 

110 self, app_client: TestClient, auth_header, obj_id 

111 ): 

112 response = app_client.put( 

113 self.endpoint + obj_id, 

114 json={ 

115 "email": "foo", 

116 "newPassword": "bar", 

117 "oldPassword": "baz", 

118 "role": "admin", 

119 }, 

120 headers=auth_header, 

121 ) 

122 

123 assert ( 

124 response.status_code == 404 

125 ), "Invalid response code when trying to update non existing user" 

126 

127 @pytest.mark.asyncio 

128 async def test_update_user(self, app_client: TestClient, auth_header): 

129 await user_manager.create_user( 

130 "foo", "baz", role="standard", first_name="John", last_name="Doe" 

131 ) 

132 user = await user_manager.get_user_from_mail("foo") 

133 assert user 

134 

135 response = app_client.put( 

136 self.endpoint + str(user.id), 

137 json={ 

138 "email": "foo", 

139 "new_password": "bar", 

140 "old_password": "baz", 

141 "role": "admin", 

142 }, 

143 headers=auth_header, 

144 ) 

145 

146 assert ( 

147 response.status_code == 200 

148 ), "Invalid response code when trying to update user" 

149 

150 assert ( 

151 user := await User.find(User.email == "foo").first_or_none() 

152 ) and user_manager.verify_password( 

153 "bar", user.hashed_password 

154 ), "Password didn't change" 

155 

156 assert ( 

157 user := await User.find(User.email == "foo").first_or_none() 

158 ) and user.role == "admin", "Role didn't change" 

159 

160 

161class TestDeleteUser: 

162 endpoint = "/api/user/" 

163 

164 def test_delete_non_existing_user( 

165 self, app_client: TestClient, auth_header, obj_id 

166 ): 

167 response = app_client.delete(self.endpoint + obj_id, headers=auth_header) 

168 

169 assert ( 

170 response.status_code == 404 

171 ), "Invalid response code when trying to delete non existing user" 

172 

173 @pytest.mark.asyncio 

174 async def test_delete_user(self, app_client: TestClient, auth_header): 

175 await user_manager.create_user("foo", "bar", first_name="John", last_name="Doe") 

176 user = await user_manager.get_user_from_mail("foo") 

177 assert user 

178 

179 response = app_client.delete(self.endpoint + str(user.id), headers=auth_header) 

180 

181 assert response.status_code == 200, "Invalid response code when deleting user" 

182 assert ( 

183 len(await User.find_all().to_list()) == 1 

184 ), "Invalid number of users in db"