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

1from datetime import datetime 

2 

3import pytest 

4from fastapi.testclient import TestClient 

5 

6from microservice_websocket.app.services import database as db 

7from microservice_websocket.app.services.database.models import NodeState 

8 

9 

10class TestGetSessions: 

11 endpoint = "/api/session" 

12 

13 # Get without args 

14 def test_get_no_args(self, app_client: TestClient, auth_header): 

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

16 

17 assert ( 

18 response.status_code == 422 

19 ), "Invalid response code when getting sessions with no nodeID" 

20 

21 # Get session of non existing node 

22 def test_get_non_existing_node(self, app_client: TestClient, auth_header): 

23 response = app_client.get(self.endpoint + "?nodeID=100", headers=auth_header) 

24 

25 assert ( 

26 response.status_code == 404 

27 ), "Invalid response code when trying to get session of non existing node" 

28 

29 # Get whene there's no session 

30 @pytest.mark.asyncio 

31 async def test_get_no_session(self, app_client: TestClient, auth_header): 

32 o = db.Organization(organizationName="orgName") 

33 await o.save() 

34 a = db.Application(applicationName="appName", organization=o.id) 

35 await a.save() 

36 node = db.Node( 

37 nodeID=123, 

38 nodeName="nodeName", 

39 application=a.id, 

40 state=NodeState.READY, 

41 lastSeenAt=datetime.now(), 

42 ) 

43 await node.save() 

44 # Done setup 

45 

46 response = app_client.get(self.endpoint + "?nodeID=123", headers=auth_header) 

47 

48 assert ( 

49 response.status_code == 200 

50 ), "Invalid response code when getting session from valid node" 

51 

52 assert len(response.json()["readings"]) == 0, "Invalid readings length" 

53 

54 # Get the latest session 

55 @pytest.mark.asyncio 

56 async def test_get_latest_session(self, app_client: TestClient, auth_header): 

57 o = db.Organization(organizationName="orgName") 

58 await o.save() 

59 a = db.Application(applicationName="appName", organization=o.id) 

60 await a.save() 

61 node = db.Node( 

62 nodeID=123, 

63 nodeName="nodeName", 

64 application=a.id, 

65 state=NodeState.READY, 

66 lastSeenAt=datetime.now(), 

67 ) 

68 await node.save() 

69 r1 = db.Reading( 

70 node=node.id, 

71 canID=1, 

72 sensorNumber=1, 

73 readingID=1, 

74 sessionID=1, 

75 publishedAt=datetime.now(), 

76 ) 

77 await r1.save() 

78 r2 = db.Reading( 

79 node=node.id, 

80 canID=2, 

81 sensorNumber=1, 

82 readingID=1, 

83 sessionID=2, 

84 publishedAt=datetime.now(), 

85 ) 

86 await r2.save() 

87 r3 = db.Reading( 

88 node=node.id, 

89 canID=3, 

90 sensorNumber=1, 

91 readingID=1, 

92 sessionID=3, 

93 publishedAt=datetime.now(), 

94 ) 

95 await r3.save() 

96 # Done setup 

97 

98 response = app_client.get( 

99 self.endpoint + f"?nodeID={str(node.nodeID)}", 

100 headers=auth_header, 

101 ) 

102 

103 assert ( 

104 response.status_code == 200 

105 ), "Invalid response code when querying latest session" 

106 

107 assert ( 

108 response.json()["readings"][0]["canID"] == 3 

109 ), "Invalid response when querying latest session" 

110 

111 @pytest.mark.asyncio 

112 async def test_get_specific_session(self, app_client: TestClient, auth_header): 

113 o = db.Organization(organizationName="orgName") 

114 await o.save() 

115 a = db.Application(applicationName="appName", organization=o.id) 

116 await a.save() 

117 node = db.Node( 

118 nodeID=123, 

119 nodeName="nodeName", 

120 application=a.id, 

121 state=NodeState.READY, 

122 lastSeenAt=datetime.now(), 

123 ) 

124 await node.save() 

125 r1 = db.Reading( 

126 node=node.id, 

127 canID=1, 

128 sensorNumber=1, 

129 readingID=1, 

130 sessionID=1, 

131 publishedAt=datetime.now(), 

132 ) 

133 await r1.save() 

134 r2 = db.Reading( 

135 node=node.id, 

136 canID=2, 

137 sensorNumber=1, 

138 readingID=1, 

139 sessionID=2, 

140 publishedAt=datetime.now(), 

141 ) 

142 await r2.save() 

143 r3 = db.Reading( 

144 node=node.id, 

145 canID=3, 

146 sensorNumber=1, 

147 readingID=1, 

148 sessionID=3, 

149 publishedAt=datetime.now(), 

150 ) 

151 await r3.save() 

152 # Done setup 

153 

154 response = app_client.get( 

155 self.endpoint + f"/2?nodeID={str(node.nodeID)}", 

156 headers=auth_header, 

157 ) 

158 

159 assert ( 

160 response.status_code == 200 

161 ), "Invalid response code when querying specific session" 

162 

163 assert ( 

164 response.json()["readings"][0]["canID"] == 2 

165 ), "Invalid response when querying specific session" 

166 

167 

168class TestGetSessionID: 

169 endpoint = "/api/sessions" 

170 

171 # Get sessionIDs without args 

172 def test_get_no_args(self, app_client: TestClient, auth_header): 

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

174 

175 assert ( 

176 response.status_code == 422 

177 ), "Invalid response code when submitting no args" 

178 

179 # Get sessionIDs of non existings node 

180 def test_get_non_existing_node(self, app_client: TestClient, auth_header): 

181 response = app_client.get(self.endpoint + "?nodeID=100", headers=auth_header) 

182 

183 assert ( 

184 response.status_code == 404 

185 ), "Invalid response code when getting sessionIDs of non existring Node" 

186 

187 # Get sessionIDs with no readings 

188 @pytest.mark.asyncio 

189 async def test_get_no_ids(self, app_client: TestClient, auth_header): 

190 o = db.Organization(organizationName="orgName") 

191 await o.save() 

192 a = db.Application(applicationName="appName", organization=o.id) 

193 await a.save() 

194 node = db.Node( 

195 nodeID=123, 

196 nodeName="nodeName", 

197 application=a.id, 

198 state=NodeState.READY, 

199 lastSeenAt=datetime.now(), 

200 ) 

201 await node.save() 

202 # Done setup 

203 

204 response = app_client.get( 

205 self.endpoint + f"?nodeID={str(node.nodeID)}", 

206 headers=auth_header, 

207 ) 

208 

209 assert ( 

210 response.status_code == 200 

211 ), "Invalid response code when getting sessionIDs from valid Node" 

212 

213 assert ( 

214 len(response.json()["IDs"]) == 0 

215 ), "Invalid response when getting sessionIDs from valid Node" 

216 

217 # Get sessionIDs 

218 @pytest.mark.asyncio 

219 async def test_get(self, app_client: TestClient, auth_header): 

220 o = db.Organization(organizationName="orgName") 

221 await o.save() 

222 a = db.Application(applicationName="appName", organization=o.id) 

223 await a.save() 

224 node = db.Node( 

225 nodeID=123, 

226 nodeName="nodeName", 

227 application=a.id, 

228 state=NodeState.READY, 

229 lastSeenAt=datetime.now(), 

230 ) 

231 await node.save() 

232 r1 = db.Reading( 

233 node=node.id, 

234 canID=1, 

235 sensorNumber=1, 

236 readingID=1, 

237 sessionID=1, 

238 publishedAt=datetime.now(), 

239 ) 

240 await r1.save() 

241 r2 = db.Reading( 

242 node=node.id, 

243 canID=2, 

244 sensorNumber=1, 

245 readingID=1, 

246 sessionID=2, 

247 publishedAt=datetime.now(), 

248 ) 

249 await r2.save() 

250 # Done setup 

251 

252 response = app_client.get( 

253 self.endpoint + f"?nodeID={str(node.nodeID)}", 

254 headers=auth_header, 

255 ) 

256 

257 assert ( 

258 response.status_code == 200 

259 ), "Invalid response code when getting sessionIDs from valid Node" 

260 

261 assert ( 

262 len(response.json()["IDs"]) == 2 

263 ), "Invalid response when getting sessionIDs from valid Node"