Coverage for tests/test_microservice_websocket/test_routes/test_route_nodes.py: 100%
76 statements
« prev ^ index » next coverage.py v7.0.0, created at 2022-12-20 14:31 +0000
1from datetime import datetime
3import pytest
4from fastapi.testclient import TestClient
6from microservice_websocket.app.models.node_settings import (
7 DetectorSettings,
8 SensorSettings,
9)
10from microservice_websocket.app.services import database as db
11from microservice_websocket.app.utils.enums import NodeState
14class TestGetNodes:
15 endpoint = "/api/nodes/"
16 name = "nodeName"
18 # Getting nodes with no args
19 def test_get_no_args(self, app_client: TestClient, auth_header):
20 response = app_client.get(self.endpoint, headers=auth_header)
22 assert (
23 response.status_code == 422
24 ), "Invalid response code when getting node with no args"
26 @pytest.mark.asyncio
27 async def test_get(self, app_client: TestClient, auth_header):
28 o = db.Organization(organizationName="foo")
29 await o.save()
30 a = db.Application(applicationName="bar", organization=o.id)
31 await a.save()
32 # Done setup
34 # Manually create node
35 n = db.Node(
36 nodeID=123,
37 nodeName=self.name,
38 application=a.id,
39 state=NodeState.READY,
40 lastSeenAt=datetime.now(),
41 )
42 await n.save()
44 # Getting newly created node
45 response = app_client.get(
46 self.endpoint + f"?applicationID={str(a.id)}",
47 headers=auth_header,
48 )
50 assert response.status_code == 200, "Invalid response code when getting nodes"
52 assert response.json()["nodes"][0]["nodeName"] == self.name
55class TestGetSettings:
56 def endpoint(self, nodeID: int):
57 return f"/api/node/{nodeID}/settings"
59 @pytest.mark.asyncio
60 async def test_get(self, app_client: TestClient, auth_header):
61 o = db.Organization(organizationName="foo")
62 await o.save()
63 a = db.Application(applicationName="bar", organization=o.id)
64 await a.save()
65 n = db.Node(
66 nodeID=123,
67 nodeName="nodeName",
68 application=a.id,
69 state=NodeState.READY,
70 lastSeenAt=datetime.now(),
71 )
72 await n.save()
73 settings = db.NodeSettings(
74 node=n.id,
75 d1=DetectorSettings(
76 s1=SensorSettings(
77 hv=1,
78 w1_low=1,
79 w1_high=2,
80 w2_low=3,
81 w2_high=4,
82 w3_low=5,
83 w3_high=6,
84 ),
85 s2=SensorSettings(
86 hv=1,
87 w1_low=6,
88 w1_high=5,
89 w2_low=4,
90 w2_high=3,
91 w3_low=2,
92 w3_high=1,
93 ),
94 ),
95 d2=DetectorSettings(
96 s1=SensorSettings(
97 hv=2,
98 w1_low=2,
99 w1_high=4,
100 w2_low=6,
101 w2_high=8,
102 w3_low=10,
103 w3_high=12,
104 ),
105 s2=SensorSettings(
106 hv=2,
107 w1_low=12,
108 w1_high=10,
109 w2_low=8,
110 w2_high=6,
111 w3_low=4,
112 w3_high=2,
113 ),
114 ),
115 d3=DetectorSettings(
116 s1=SensorSettings(
117 hv=3,
118 w1_low=3,
119 w1_high=6,
120 w2_low=9,
121 w2_high=12,
122 w3_low=15,
123 w3_high=18,
124 ),
125 s2=SensorSettings(
126 hv=3,
127 w1_low=18,
128 w1_high=15,
129 w2_low=12,
130 w2_high=9,
131 w3_low=6,
132 w3_high=3,
133 ),
134 ),
135 d4=DetectorSettings(
136 s1=SensorSettings(
137 hv=4,
138 w1_low=4,
139 w1_high=8,
140 w2_low=12,
141 w2_high=16,
142 w3_low=20,
143 w3_high=24,
144 ),
145 s2=SensorSettings(
146 hv=4,
147 w1_low=24,
148 w1_high=20,
149 w2_low=16,
150 w2_high=12,
151 w3_low=8,
152 w3_high=4,
153 ),
154 ),
155 )
156 await settings.save()
157 # Done setup
159 response = app_client.get(self.endpoint(n.nodeID), headers=auth_header)
161 assert (
162 response.status_code == 200
163 ), "Invalid response code when gettings node settings"
165 assert response.json() == settings.serialize()
168class TestUpdateSettings:
169 def endpoint(self, nodeID: int):
170 return f"/api/node/{nodeID}/settings"
172 def test_update_non_existing_node(self, app_client: TestClient, auth_header):
173 payload = {
174 "d4": {
175 "s1": {
176 "hv": 1,
177 "w1_low": 1,
178 "w1_high": 2,
179 "w2_low": 3,
180 "w2_high": 4,
181 "w3_low": 5,
182 "w3_high": 6,
183 },
184 "s2": {
185 "hv": 1,
186 "w1_low": 6,
187 "w1_high": 5,
188 "w2_low": 4,
189 "w2_high": 3,
190 "w3_low": 2,
191 "w3_high": 1,
192 },
193 },
194 "d3": {
195 "s1": {
196 "hv": 2,
197 "w1_low": 2,
198 "w1_high": 4,
199 "w2_low": 6,
200 "w2_high": 8,
201 "w3_low": 10,
202 "w3_high": 12,
203 },
204 "s2": {
205 "hv": 2,
206 "w1_low": 12,
207 "w1_high": 10,
208 "w2_low": 8,
209 "w2_high": 6,
210 "w3_low": 4,
211 "w3_high": 2,
212 },
213 },
214 "d2": {
215 "s1": {
216 "hv": 3,
217 "w1_low": 3,
218 "w1_high": 6,
219 "w2_low": 9,
220 "w2_high": 12,
221 "w3_low": 15,
222 "w3_high": 18,
223 },
224 "s2": {
225 "hv": 3,
226 "w1_low": 18,
227 "w1_high": 15,
228 "w2_low": 12,
229 "w2_high": 9,
230 "w3_low": 6,
231 "w3_high": 3,
232 },
233 },
234 "d1": {
235 "s1": {
236 "hv": 4,
237 "w1_low": 4,
238 "w1_high": 8,
239 "w2_low": 12,
240 "w2_high": 16,
241 "w3_low": 20,
242 "w3_high": 24,
243 },
244 "s2": {
245 "hv": 4,
246 "w1_low": 24,
247 "w1_high": 20,
248 "w2_low": 16,
249 "w2_high": 12,
250 "w3_low": 8,
251 "w3_high": 4,
252 },
253 },
254 }
256 response = app_client.put(self.endpoint(123), json=payload, headers=auth_header)
258 assert (
259 response.status_code == 404
260 ), "Invalid response code when updating settings of non-extisting node"
262 @pytest.mark.asyncio
263 async def test_update_non_existing_settings(
264 self, app_client: TestClient, auth_header
265 ):
266 o = db.Organization(organizationName="foo")
267 await o.save()
268 a = db.Application(applicationName="bar", organization=o.id)
269 await a.save()
270 n = db.Node(
271 nodeID=123,
272 nodeName="nodeName",
273 application=a.id,
274 state=NodeState.READY,
275 lastSeenAt=datetime.now(),
276 )
277 await n.save()
278 # Done setup
280 payload = {
281 "d1": {
282 "s1": {
283 "hv": 1,
284 "w1_low": 1,
285 "w1_high": 2,
286 "w2_low": 3,
287 "w2_high": 4,
288 "w3_low": 5,
289 "w3_high": 6,
290 },
291 "s2": {
292 "hv": 1,
293 "w1_low": 6,
294 "w1_high": 5,
295 "w2_low": 4,
296 "w2_high": 3,
297 "w3_low": 2,
298 "w3_high": 1,
299 },
300 },
301 "d2": {
302 "s1": {
303 "hv": 2,
304 "w1_low": 2,
305 "w1_high": 4,
306 "w2_low": 6,
307 "w2_high": 8,
308 "w3_low": 10,
309 "w3_high": 12,
310 },
311 "s2": {
312 "hv": 2,
313 "w1_low": 12,
314 "w1_high": 10,
315 "w2_low": 8,
316 "w2_high": 6,
317 "w3_low": 4,
318 "w3_high": 2,
319 },
320 },
321 "d3": {
322 "s1": {
323 "hv": 3,
324 "w1_low": 3,
325 "w1_high": 6,
326 "w2_low": 9,
327 "w2_high": 12,
328 "w3_low": 15,
329 "w3_high": 18,
330 },
331 "s2": {
332 "hv": 3,
333 "w1_low": 18,
334 "w1_high": 15,
335 "w2_low": 12,
336 "w2_high": 9,
337 "w3_low": 6,
338 "w3_high": 3,
339 },
340 },
341 "d4": {
342 "s1": {
343 "hv": 4,
344 "w1_low": 4,
345 "w1_high": 8,
346 "w2_low": 12,
347 "w2_high": 16,
348 "w3_low": 20,
349 "w3_high": 24,
350 },
351 "s2": {
352 "hv": 4,
353 "w1_low": 24,
354 "w1_high": 20,
355 "w2_low": 16,
356 "w2_high": 12,
357 "w3_low": 8,
358 "w3_high": 4,
359 },
360 },
361 }
363 response = app_client.put(
364 self.endpoint(n.nodeID), json=payload, headers=auth_header
365 )
367 assert (
368 response.status_code == 200
369 ), "Invalid response code when trying to update non-existing settings"
371 assert (
372 len(await db.NodeSettings.find_all().to_list()) == 1
373 ), "Invalid number of settings"
375 assert (
376 settings := await db.NodeSettings.find_one()
377 ) and settings.serialize() == payload, "Invalid settings in database"
379 @pytest.mark.asyncio
380 async def test_update_settings(self, app_client: TestClient, auth_header):
381 o = db.Organization(organizationName="foo")
382 await o.save()
383 a = db.Application(applicationName="bar", organization=o.id)
384 await a.save()
385 n = db.Node(
386 nodeID=123,
387 nodeName="nodeName",
388 application=a.id,
389 state=NodeState.READY,
390 lastSeenAt=datetime.now(),
391 )
392 await n.save()
393 settings = db.NodeSettings(
394 node=n.id,
395 d1=DetectorSettings(
396 s1=SensorSettings(
397 hv=1,
398 w1_low=1,
399 w1_high=2,
400 w2_low=3,
401 w2_high=4,
402 w3_low=5,
403 w3_high=6,
404 ),
405 s2=SensorSettings(
406 hv=1,
407 w1_low=6,
408 w1_high=5,
409 w2_low=4,
410 w2_high=3,
411 w3_low=2,
412 w3_high=1,
413 ),
414 ),
415 d2=DetectorSettings(
416 s1=SensorSettings(
417 hv=2,
418 w1_low=2,
419 w1_high=4,
420 w2_low=6,
421 w2_high=8,
422 w3_low=10,
423 w3_high=12,
424 ),
425 s2=SensorSettings(
426 hv=2,
427 w1_low=12,
428 w1_high=10,
429 w2_low=8,
430 w2_high=6,
431 w3_low=4,
432 w3_high=2,
433 ),
434 ),
435 d3=DetectorSettings(
436 s1=SensorSettings(
437 hv=3,
438 w1_low=3,
439 w1_high=6,
440 w2_low=9,
441 w2_high=12,
442 w3_low=15,
443 w3_high=18,
444 ),
445 s2=SensorSettings(
446 hv=3,
447 w1_low=18,
448 w1_high=15,
449 w2_low=12,
450 w2_high=9,
451 w3_low=6,
452 w3_high=3,
453 ),
454 ),
455 d4=DetectorSettings(
456 s1=SensorSettings(
457 hv=4,
458 w1_low=4,
459 w1_high=8,
460 w2_low=12,
461 w2_high=16,
462 w3_low=20,
463 w3_high=24,
464 ),
465 s2=SensorSettings(
466 hv=4,
467 w1_low=24,
468 w1_high=20,
469 w2_low=16,
470 w2_high=12,
471 w3_low=8,
472 w3_high=4,
473 ),
474 ),
475 )
476 await settings.save()
477 # Done setup
479 payload = {
480 "d4": {
481 "s1": {
482 "hv": 1,
483 "w1_low": 1,
484 "w1_high": 2,
485 "w2_low": 3,
486 "w2_high": 4,
487 "w3_low": 5,
488 "w3_high": 6,
489 },
490 "s2": {
491 "hv": 1,
492 "w1_low": 6,
493 "w1_high": 5,
494 "w2_low": 4,
495 "w2_high": 3,
496 "w3_low": 2,
497 "w3_high": 1,
498 },
499 },
500 "d3": {
501 "s1": {
502 "hv": 2,
503 "w1_low": 2,
504 "w1_high": 4,
505 "w2_low": 6,
506 "w2_high": 8,
507 "w3_low": 10,
508 "w3_high": 12,
509 },
510 "s2": {
511 "hv": 2,
512 "w1_low": 12,
513 "w1_high": 10,
514 "w2_low": 8,
515 "w2_high": 6,
516 "w3_low": 4,
517 "w3_high": 2,
518 },
519 },
520 "d2": {
521 "s1": {
522 "hv": 3,
523 "w1_low": 3,
524 "w1_high": 6,
525 "w2_low": 9,
526 "w2_high": 12,
527 "w3_low": 15,
528 "w3_high": 18,
529 },
530 "s2": {
531 "hv": 3,
532 "w1_low": 18,
533 "w1_high": 15,
534 "w2_low": 12,
535 "w2_high": 9,
536 "w3_low": 6,
537 "w3_high": 3,
538 },
539 },
540 "d1": {
541 "s1": {
542 "hv": 4,
543 "w1_low": 4,
544 "w1_high": 8,
545 "w2_low": 12,
546 "w2_high": 16,
547 "w3_low": 20,
548 "w3_high": 24,
549 },
550 "s2": {
551 "hv": 4,
552 "w1_low": 24,
553 "w1_high": 20,
554 "w2_low": 16,
555 "w2_high": 12,
556 "w3_low": 8,
557 "w3_high": 4,
558 },
559 },
560 }
562 from mock import patch
564 with patch("microservice_websocket.app.config.TESTING", True):
565 response = app_client.put(
566 self.endpoint(n.nodeID), json=payload, headers=auth_header
567 )
569 assert (
570 response.status_code == 200
571 ), "Invalid response code when updating node settings"
573 assert (
574 len(await db.NodeSettings.find_all().to_list()) == 1
575 ), "Invalid number of NodeSettings"
577 assert (
578 settings := await db.NodeSettings.find_one()
579 ) and settings.serialize() == payload, "Invalid settings in database"