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

1from datetime import datetime 

2 

3import pytest 

4from fastapi.testclient import TestClient 

5from mock import patch 

6 

7from microservice_websocket.app.services import database as db 

8from microservice_websocket.app.utils.enums import NodeState 

9 

10 

11class TestAlertHandle: 

12 endpoint = "/api/alert/" 

13 

14 # Handle not existing alert 

15 @pytest.mark.asyncio 

16 async def test_handle_not_existing(self, app_client: TestClient, auth_header): 

17 org = db.Organization(organizationName="foo") 

18 await org.save() 

19 app = db.Application(applicationName="bar", organization=org.id) 

20 await app.save() 

21 node = db.Node( 

22 nodeID=123, 

23 nodeName="nodeName", 

24 application=app.id, 

25 state=NodeState.ALERT_READY, 

26 lastSeenAt=datetime.now(), 

27 ) 

28 await node.save() 

29 reading = db.Reading( 

30 node=node.id, 

31 canID=1, 

32 sensorNumber=2, 

33 readingID=32704, 

34 sessionID=12892, 

35 publishedAt=datetime.now(), 

36 ) 

37 await reading.save() 

38 # Done setup 

39 

40 response = app_client.post( 

41 self.endpoint + "63186eab0ca2d54a5c258384", 

42 json={ 

43 "isConfirmed": True, 

44 "handleNote": "foo", 

45 }, 

46 headers=auth_header, 

47 ) 

48 

49 assert ( 

50 response.status_code == 404 

51 ), "Invalid response code when trying to handle non-existing alert" 

52 

53 @pytest.mark.asyncio 

54 async def test_handle(self, app_client: TestClient, auth_header): 

55 org = db.Organization(organizationName="foo") 

56 await org.save() 

57 app = db.Application(applicationName="bar", organization=org.id) 

58 await app.save() 

59 node = db.Node( 

60 nodeID=123, 

61 nodeName="nodeName", 

62 application=app.id, 

63 state=NodeState.ALERT_READY, 

64 lastSeenAt=datetime.now(), 

65 ) 

66 await node.save() 

67 reading = db.Reading( 

68 node=node.id, 

69 canID=1, 

70 sensorNumber=2, 

71 readingID=32704, 

72 sessionID=12892, 

73 publishedAt=datetime.now(), 

74 ) 

75 await reading.save() 

76 # Done setup 

77 

78 reading = await db.Reading.find_one() 

79 assert reading 

80 node = await db.Node.find_one() 

81 assert node 

82 

83 # Manually create Alerts 

84 alert = db.Alert( 

85 reading=reading.id, 

86 node=node.id, 

87 sessionID=reading.sessionID, 

88 isHandled=False, 

89 raisedAt=datetime.now(), 

90 ) 

91 await alert.save() 

92 alert2 = db.Alert( 

93 reading=reading.id, 

94 node=node.id, 

95 sessionID=reading.sessionID, 

96 isHandled=False, 

97 raisedAt=datetime.now(), 

98 ) 

99 await alert2.save() 

100 

101 # Try to handle newly created alert 

102 with patch("socketio.Client.emit", return_value=None): 

103 response = app_client.post( 

104 self.endpoint + str(alert.id), 

105 json={ 

106 "isConfirmed": True, 

107 "handleNote": "foo", 

108 }, 

109 headers=auth_header, 

110 ) 

111 

112 alert = await db.Alert.find_one() 

113 

114 assert ( 

115 response.status_code == 200 

116 and alert 

117 and alert.isConfirmed 

118 and alert.handleNote == "foo" 

119 and alert.handledBy 

120 and (user := await db.User.get(alert.handledBy)) 

121 and user.email == "foo@bar.com" 

122 ), "Invalid response code when trying to handle existing alert" 

123 

124 node = await db.Node.find_one() 

125 

126 assert ( 

127 node and node.state == NodeState.ALERT_READY 

128 ), "Invalid state when handling 1/2 alert" 

129 

130 # Handle leftover alert 

131 with patch("socketio.Client.emit", return_value=None): 

132 response = app_client.post( 

133 self.endpoint + str(alert2.id), 

134 json={ 

135 "isConfirmed": True, 

136 "handleNote": "foo", 

137 }, 

138 headers=auth_header, 

139 ) 

140 

141 assert ( 

142 node := await db.Node.find_one() 

143 ) and node.state == NodeState.READY, "Invalid state when handling all alert" 

144 

145 

146class TestAlertInfo: 

147 endpoint = "/api/alert/" 

148 

149 # Test get info of non-existing alert 

150 def test_get_info_non_existing_alert(self, app_client: TestClient, auth_header): 

151 response = app_client.get( 

152 self.endpoint + "63186eab0ca2d54a5c258384", 

153 headers=auth_header, 

154 ) 

155 

156 assert ( 

157 response.status_code == 404 

158 ), "Invalid response code when querying infos of non-existing alert" 

159 

160 # Test get info 

161 @pytest.mark.asyncio 

162 async def test_get_info_alert(self, app_client: TestClient, auth_header): 

163 org = db.Organization(organizationName="foo") 

164 await org.save() 

165 app = db.Application(applicationName="bar", organization=org.id) 

166 await app.save() 

167 node = db.Node( 

168 nodeID=123, 

169 nodeName="nodeName", 

170 application=app.id, 

171 state=NodeState.ALERT_READY, 

172 lastSeenAt=datetime.now(), 

173 ) 

174 await node.save() 

175 reading = db.Reading( 

176 node=node.id, 

177 canID=1, 

178 sensorNumber=2, 

179 readingID=32704, 

180 sessionID=12892, 

181 publishedAt=datetime.now(), 

182 ) 

183 await reading.save() 

184 # Done setup 

185 

186 reading = await db.Reading.find_one() 

187 assert reading 

188 node = await db.Node.get(reading.node) 

189 assert node 

190 

191 # Manually create Alerts 

192 alert = db.Alert( 

193 reading=reading.id, 

194 node=node.id, 

195 sessionID=reading.sessionID, 

196 isHandled=False, 

197 raisedAt=datetime.now(), 

198 ) 

199 await alert.save() 

200 

201 response = app_client.get(self.endpoint + str(alert.id), headers=auth_header) 

202 

203 assert ( 

204 response.status_code == 200 

205 ), "Invalid response code when querying infos of extisting alert" 

206 

207 for key in [ 

208 "canID", 

209 "readingID", 

210 "nodeID", 

211 "id", 

212 "sessionID", 

213 "raisedAt", 

214 ]: 

215 assert key in response.json(), "Invalid response structure"