Terarea  2
The automation project
Loading...
Searching...
No Matches
test_backend_endpoints.py
Go to the documentation of this file.
7
8"""_summary_
9This file is in charge of testing the endpoints of the character server.
10"""
11
12import os
13import sys
14import time
15import asyncio
16from typing import Dict
17from threading import Thread
18
19import pytest
20
21from status_check import QueryStatus
22from query_boilerplate import QueryEndpoint
23import constants as TCONST
24
25sys.path.append(os.getcwd())
26
27
28try:
29 from src import Server
30 from src import CONST
31except ImportError as e:
32 raise ImportError("Failed to import the src module") from e
33
34
35def pytest_configure(config: pytest.Config):
36 """_summary_
37 This function is called when the pytest configuration is being set up.
38 Args:
39 config (pytest.Config): _description_
40 """
41 config.addinivalue_line(
42 "markers",
43 "critical: mark test as critical to determine final server stop"
44 )
45 config.addinivalue_line(
46 "markers",
47 "last: mark test to run after all other tests"
48 )
49
50
51@pytest.fixture(scope="module")
52def setup_environment(request: pytest.FixtureRequest):
53 """Fixture for setting up the environment once per module."""
54
55 query: QueryEndpoint = QueryEndpoint(
56 host=TCONST.QUERY_HOST,
57 port=TCONST.PORT,
58 delay=TCONST.QUERY_DELAY
59 )
60
61 status: QueryStatus = QueryStatus()
62
63 server: Server = Server(
64 host=TCONST.SERVER_HOST,
65 port=TCONST.PORT,
66 success=TCONST.SUCCESS,
67 error=TCONST.ERROR,
68 app_name="Area",
69 debug=TCONST.DEBUG
70 )
71
72 active_thread: Thread = Thread(
73 target=server.main,
74 daemon=True, name=f"{TCONST.APP_NAME}_server_thread"
75 )
76 active_thread.start()
77
78 # Let the server start
79 print("Waiting for server to start (5 seconds delay)...")
80 time.sleep(5)
81 if active_thread.is_alive() is not True:
82 pytest.skip("(test_server) thread failed to start")
83 if server.is_running() is not True:
84 pytest.skip("(test_server) Server is not running")
85 call = server.runtime_data_initialised.database_link.get_table_names()
86 if isinstance(call, int) and call == server.error:
87 pytest.skip(
88 "(test_server) Server failed to connect to the database"
89 )
90 print("Server started.")
91
92 async def teardown():
93 if server is not None:
94 print("Shutting down server...")
95 server.stop_server()
96 print("Server stopped")
97 if active_thread.is_alive() is True:
98 print("Stopping thread (5 seconds grace) ...")
99 active_thread.join(timeout=5)
100 print("Thread stopped.")
101 else:
102 print("Thread already stopped.")
103 return TCONST.SUCCESS
104
105 def teardown_sync():
106 loop = asyncio.get_event_loop()
107 return loop.run_until_complete(teardown())
108
109 request.addfinalizer(teardown_sync)
110
111 return {
112 "query": query,
113 "status": status,
114 "server": server,
115 "success": TCONST.SUCCESS,
116 "error": TCONST.ERROR,
117 "thread": active_thread,
118 "tokens": {
119 TCONST.LAMBDA_USER_TOKEN_KEY: {
120 TCONST.PRETTY_TOKEN_KEY: {TCONST.TOKEN_AUTH_ID_STR: ""},
121 TCONST.RAW_TOKEN_KEY: ""
122 },
123 TCONST.ADMIN_USER_TOKEN_KEY: {
124 TCONST.PRETTY_TOKEN_KEY: {TCONST.TOKEN_AUTH_ID_STR: ""},
125 TCONST.RAW_TOKEN_KEY: ""
126 },
127 },
128 "accounts": {
129 "lambda_user": {
130 TCONST.USER_NORMAL_MODE: {
131 TCONST.UNODE_EMAIL_KEY: f"endpoint_{TCONST.USER_DATA_EMAIL}",
132 TCONST.UNODE_PASSWORD_KEY: f"endpoint_{TCONST.USER_DATA_PASSWORD}",
133 TCONST.UNODE_USERNAME_KEY: f"endpoint_{TCONST.USER_DATA_USERNAME}",
134 TCONST.UNODE_METHOD_KEY: f"{TCONST.USER_DATA_METHOD}",
135 TCONST.UNODE_FAVICON_KEY: f"{TCONST.USER_DATA_FAVICON}",
136 TCONST.UNODE_ADMIN_KEY: f"{TCONST.USER_DATA_ADMIN}"
137 },
138 TCONST.USER_PUT_MODE: {
139 TCONST.UNODE_EMAIL_KEY: f"endpoint_{TCONST.USER_DATA_EMAIL_REBIND}",
140 TCONST.UNODE_PASSWORD_KEY: f"endpoint_{TCONST.USER_DATA_PASSWORD_REBIND}",
141 TCONST.UNODE_USERNAME_KEY: f"endpoint_{TCONST.USER_DATA_USERNAME_REBIND}",
142 TCONST.UNODE_METHOD_KEY: f"{TCONST.USER_DATA_METHOD}",
143 TCONST.UNODE_FAVICON_KEY: f"{TCONST.USER_DATA_FAVICON}",
144 TCONST.UNODE_ADMIN_KEY: f"{TCONST.USER_DATA_ADMIN}"
145 },
146 TCONST.USER_PATCH_MODE: {
147 TCONST.UNODE_EMAIL_KEY: f"endpoint_{TCONST.USER_DATA_EMAIL_PATCH}",
148 TCONST.UNODE_PASSWORD_KEY: f"endpoint_{TCONST.USER_DATA_PASSWORD_PATCH}",
149 TCONST.UNODE_USERNAME_KEY: f"endpoint_{TCONST.USER_DATA_USERNAME_PATCH}",
150 TCONST.UNODE_METHOD_KEY: f"{TCONST.USER_DATA_METHOD}",
151 TCONST.UNODE_FAVICON_KEY: f"{TCONST.USER_DATA_FAVICON}",
152 TCONST.UNODE_ADMIN_KEY: f"{TCONST.USER_DATA_ADMIN}"
153 }
154 },
155 "admin_user": {
156 TCONST.USER_NORMAL_MODE: {
157 TCONST.UNODE_EMAIL_KEY: f"endpoint_{TCONST.ADMIN_DATA_EMAIL}",
158 TCONST.UNODE_PASSWORD_KEY: f"endpoint_{TCONST.ADMIN_DATA_PASSWORD}",
159 TCONST.UNODE_USERNAME_KEY: f"endpoint_{TCONST.ADMIN_DATA_USERNAME}",
160 TCONST.UNODE_METHOD_KEY: f"{TCONST.ADMIN_DATA_METHOD}",
161 TCONST.UNODE_FAVICON_KEY: f"{TCONST.ADMIN_DATA_FAVICON}",
162 TCONST.UNODE_ADMIN_KEY: f"{TCONST.ADMIN_DATA_ADMIN}"
163 },
164 TCONST.USER_PUT_MODE: {
165 TCONST.UNODE_EMAIL_KEY: f"endpoint_{TCONST.ADMIN_DATA_EMAIL_REBIND}",
166 TCONST.UNODE_PASSWORD_KEY: f"endpoint_{TCONST.ADMIN_DATA_PASSWORD_REBIND}",
167 TCONST.UNODE_USERNAME_KEY: f"endpoint_{TCONST.ADMIN_DATA_USERNAME_REBIND}",
168 TCONST.UNODE_METHOD_KEY: f"{TCONST.ADMIN_DATA_METHOD}",
169 TCONST.UNODE_FAVICON_KEY: f"{TCONST.ADMIN_DATA_FAVICON}",
170 TCONST.UNODE_ADMIN_KEY: f"{TCONST.ADMIN_DATA_ADMIN}"
171 },
172 TCONST.USER_PATCH_MODE: {
173 TCONST.UNODE_EMAIL_KEY: f"endpoint_{TCONST.ADMIN_DATA_EMAIL_PATCH}",
174 TCONST.UNODE_PASSWORD_KEY: f"endpoint_{TCONST.ADMIN_DATA_PASSWORD_PATCH}",
175 TCONST.UNODE_USERNAME_KEY: f"endpoint_{TCONST.ADMIN_DATA_USERNAME_PATCH}",
176 TCONST.UNODE_METHOD_KEY: f"{TCONST.ADMIN_DATA_METHOD}",
177 TCONST.UNODE_FAVICON_KEY: f"{TCONST.ADMIN_DATA_FAVICON}",
178 TCONST.UNODE_ADMIN_KEY: f"{TCONST.ADMIN_DATA_ADMIN}"
179 }
180 }
181 },
182 TCONST.RUNTIME_NODE_CRITICAL_KEY: False,
183 "teardown_func": teardown_sync
184 }
185
186
187@pytest.mark.usefixtures("setup_environment")
189 """Class for running tests on the character server endpoint."""
190
191 def check_server(self, setup_environment, critical: bool = False):
192 """Helper function to check if the server is still running."""
193 server: Server = setup_environment["server"]
194 if server is None or setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] is True:
195 if critical is True:
196 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
197 pytest.skip(
198 "(test_server) Server is not running, skipping the test."
199 )
200
201 def test_home(self, setup_environment):
202 """ Test the / endpoint of the server. """
203 self.check_server(setup_environment)
204
205 query: QueryEndpoint = setup_environment["query"]
206 status: QueryStatus = setup_environment["status"]
207 response = query.get_endpoint(TCONST.PATH_GET_HOME)
208 assert status.success(response) is True
209 assert TCONST.are_json_responses_identical(
210 response.json(),
211 TCONST.RESPONSE_GET_HOME_RESPONSE_NOT_LOGGED_IN,
212 "test_home"
213 ) is True
214
215 def test_api_home(self, setup_environment):
216 """ Test the / endpoint of the server. """
217 self.check_server(setup_environment)
218
219 query: QueryEndpoint = setup_environment["query"]
220 status: QueryStatus = setup_environment["status"]
221 response = query.get_endpoint(TCONST.PATH_GET_API_HOME)
222 assert status.success(response) is True
223 assert TCONST.are_json_responses_identical(
224 response.json(),
225 TCONST.RESPONSE_GET_HOME_API_RESPONSE_NOT_LOGGED_IN,
226 "test_api_home"
227 ) is True
228
229 @pytest.mark.critical
230 def test_post_register_lambda(self, setup_environment):
231 """_summary_
232 Test the /register endpoint of the server.
233 Args:
234 setup_environment (_type_): _description_
235 """
236 self.check_server(setup_environment, True)
237 path: str = TCONST.PATH_PUT_REGISTER
238 query: QueryEndpoint = setup_environment["query"]
239 status: QueryStatus = setup_environment["status"]
240 accounts: Dict[
241 str, any
242 ] = setup_environment["accounts"]["lambda_user"][TCONST.USER_NORMAL_MODE]
243 body = {
244 "email": accounts[TCONST.UNODE_EMAIL_KEY],
245 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
246 }
247 TCONST.IDISP.log_info(f"body = {body}")
248 response = query.post_endpoint(path, content=body)
249 TCONST.IDISP.log_info(f"response.json() = {response.json()}")
250 token_node = ""
251 if status.success(response) is True:
252 token_node = f"{response.json()['token']}"
253 else:
254 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
255 correct_node = TCONST.RESPONSE_POST_REGISTER
256 correct_node['msg'] = f"Welcome {accounts['username']}"
257 correct_node['token'] = token_node
258 assert status.success(response) is True
259 assert TCONST.are_json_responses_identical(
260 response.json(),
261 correct_node,
262 "test_post_register_lambda"
263 ) is True
264
265 @pytest.mark.critical
266 def test_post_login_lambda(self, setup_environment):
267 """_summary_
268 Test the /login endpoint of the server.
269 Args:
270 setup_environment (_type_): _description_
271 """
272 self.check_server(setup_environment, True)
273 path: str = TCONST.PATH_POST_LOGIN
274 token: Dict[str, Dict[str, str]] = setup_environment["tokens"]
275 query: QueryEndpoint = setup_environment["query"]
276 status: QueryStatus = setup_environment["status"]
277 accounts: Dict[
278 str, any
279 ] = setup_environment["accounts"]["lambda_user"][TCONST.USER_NORMAL_MODE]
280 body = {
281 "email": accounts[TCONST.UNODE_EMAIL_KEY],
282 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
283 }
284 TCONST.IDISP.log_info(f"body = {body}")
285 response = query.post_endpoint(path, content=body)
286 TCONST.IDISP.log_info(f"response.json() = {response.json()}")
287 if status.success(response) is True:
288 token_node = f"{response.json()['token']}"
289 msg = f"Bearer {token_node}"
290 token[TCONST.LAMBDA_USER_TOKEN_KEY][TCONST.PRETTY_TOKEN_KEY][TCONST.TOKEN_AUTH_ID_STR] = msg
291 token[TCONST.LAMBDA_USER_TOKEN_KEY][TCONST.RAW_TOKEN_KEY] = token_node
292 else:
293 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
294 correct_node = TCONST.RESPONSE_POST_LOGIN
295 correct_node['msg'] = f"Welcome {accounts['username']}"
296 correct_node["token"] = token_node
297 assert status.success(response) is True
298 assert TCONST.are_json_responses_identical(
299 response.json(),
300 correct_node,
301 "test_post_login_lambda"
302 ) is True
303
304 @pytest.mark.critical
305 def test_post_register_admin(self, setup_environment):
306 """_summary_
307 Test the /register endpoint of the server.
308 Args:
309 setup_environment (_type_): _description_
310 """
311 self.check_server(setup_environment, True)
312 server: Server = setup_environment["server"]
313 path = TCONST.PATH_PUT_REGISTER
314 query: QueryEndpoint = setup_environment["query"]
315 status: QueryStatus = setup_environment["status"]
316 accounts: Dict[
317 str, any
318 ] = setup_environment["accounts"]["admin_user"][TCONST.USER_NORMAL_MODE]
319 body = {
320 "email": accounts[TCONST.UNODE_EMAIL_KEY],
321 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
322 }
323 TCONST.IDISP.log_info(f"body = {body}")
324 response = query.post_endpoint(path, content=body)
325 TCONST.IDISP.log_info(f"response.json() = {response.json()}")
326 if status.success(response) is True:
327 column = server.runtime_data_initialised.database_link.get_table_column_names(
328 CONST.TAB_ACCOUNTS
329 )
330 if column == server.error or len(column) == 0:
331 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
332 assert column == server.success
333 user_node = server.runtime_data_initialised.database_link.get_data_from_table(
334 CONST.TAB_ACCOUNTS,
335 column,
336 f"email='{accounts['email']}'"
337 )
338 if user_node == server.error:
339 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
340 assert user_node == server.success
341 column.pop(0)
342 uid = user_node[0].pop("id")
343 user_node[0]['admin'] = str(TCONST.ADMIN_DATA_ADMIN)
344 user_node_list = list(user_node[0].values())
345 msg = f"user_node_list = {user_node_list}\n"
346 msg += f"user_node = {user_node}"
347 server.disp.log_debug(msg, "test_post_register_admin")
348 command_status = server.runtime_data_initialised.database_link.update_data_in_table(
349 CONST.TAB_ACCOUNTS,
350 user_node_list,
351 column,
352 f"id='{uid}'"
353 )
354 assert command_status == server.success
355 token_node = ""
356 if status.success(response) is True:
357 token_node = f"{response.json()['token']}"
358 else:
359 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
360 correct_node = TCONST.RESPONSE_POST_REGISTER
361 correct_node['msg'] = f"Welcome {accounts['username']}"
362 correct_node['token'] = token_node
363 assert status.success(response) is True
364 assert TCONST.are_json_responses_identical(
365 response.json(),
366 correct_node,
367 "test_post_register_admin"
368 ) is True
369
370 @pytest.mark.critical
371 def test_post_login_admin(self, setup_environment):
372 """_summary_
373 Test the /login endpoint of the server.
374 Args:
375 setup_environment (_type_): _description_
376 """
377 self.check_server(setup_environment, True)
378 path = TCONST.PATH_POST_LOGIN
379 token: Dict[str, Dict[str, str]] = setup_environment["tokens"]
380 query: QueryEndpoint = setup_environment["query"]
381 status: QueryStatus = setup_environment["status"]
382 accounts: Dict[
383 str, any
384 ] = setup_environment["accounts"]["admin_user"][TCONST.USER_NORMAL_MODE]
385 body = {
386 "email": accounts[TCONST.UNODE_EMAIL_KEY],
387 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
388 }
389 TCONST.IDISP.log_info(f"body = {body}")
390 response = query.post_endpoint(path, content=body)
391 TCONST.IDISP.log_info(f"response.json() = {response.json()}")
392 if status.success(response) is True:
393 token_node = f"{response.json()['token']}"
394 msg = f"Bearer {token_node}"
395 token[TCONST.ADMIN_USER_TOKEN_KEY][TCONST.PRETTY_TOKEN_KEY][TCONST.TOKEN_AUTH_ID_STR] = msg
396 token[TCONST.ADMIN_USER_TOKEN_KEY][TCONST.RAW_TOKEN_KEY] = token_node
397 else:
398 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
399 correct_node = TCONST.RESPONSE_POST_LOGIN
400 correct_node['msg'] = f"Welcome {accounts['username']}"
401 correct_node["token"] = token_node
402 assert status.success(response) is True
403 assert TCONST.are_json_responses_identical(
404 response.json(),
405 correct_node,
406 "test_post_login_admin"
407 ) is True
408
409 def test_put_user_lambda(self, setup_environment):
410 """_summary_
411 Test the /user endpoint of the server.
412 Args:
413 setup_environment (_type_): _description_
414 """
415 self.check_server(setup_environment)
416 path = TCONST.PATH_PUT_USER
417 query: QueryEndpoint = setup_environment["query"]
418 status: QueryStatus = setup_environment["status"]
419 token: Dict[
420 str, Dict[str, str]
421 ] = setup_environment["tokens"][TCONST.LAMBDA_USER_TOKEN_KEY]
422 accounts: Dict[
423 str, any
424 ] = setup_environment["accounts"]["lambda_user"][TCONST.USER_PUT_MODE]
425 body = {
426 "username": accounts[TCONST.UNODE_USERNAME_KEY],
427 "email": accounts[TCONST.UNODE_EMAIL_KEY],
428 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
429 }
430
431 response = query.put_endpoint(
432 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
433 )
434 assert status.success(response) is True
435 assert TCONST.are_json_responses_identical(
436 response.json(),
437 TCONST.RESPONSE_PUT_USER,
438 "test_put_user_lambda"
439 ) is True
440
441 def test_put_user_admin(self, setup_environment):
442 """_summary_
443 Test the /user endpoint of the server.
444 Args:
445 setup_environment (_type_): _description_
446 """
447 self.check_server(setup_environment)
448 path = TCONST.PATH_PUT_USER
449 query: QueryEndpoint = setup_environment["query"]
450 status: QueryStatus = setup_environment["status"]
451 token: Dict[
452 str, Dict[str, str]
453 ] = setup_environment["tokens"][TCONST.ADMIN_USER_TOKEN_KEY]
454 accounts: Dict[
455 str, any
456 ] = setup_environment["accounts"]["admin_user"][TCONST.USER_PUT_MODE]
457 body = {
458 "username": accounts[TCONST.UNODE_USERNAME_KEY],
459 "email": accounts[TCONST.UNODE_EMAIL_KEY],
460 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
461 }
462 response = query.put_endpoint(
463 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
464 )
465 assert status.success(response) is True
466 assert TCONST.are_json_responses_identical(
467 response.json(),
468 TCONST.RESPONSE_PUT_USER,
469 "test_put_user_admin"
470 ) is True
471
472 def test_patch_user_lambda_username(self, setup_environment):
473 """_summary_
474 Test the /user endpoint of the server.
475 Args:
476 setup_environment (_type_): _description_
477 """
478 self.check_server(setup_environment)
479 path = TCONST.PATH_PATCH_USER
480 query: QueryEndpoint = setup_environment["query"]
481 status: QueryStatus = setup_environment["status"]
482 token: Dict[
483 str, Dict[str, str]
484 ] = setup_environment["tokens"][TCONST.LAMBDA_USER_TOKEN_KEY]
485 accounts: Dict[
486 str, any
487 ] = setup_environment["accounts"]["lambda_user"][TCONST.USER_PATCH_MODE]
488 body = {
489 "username": accounts[TCONST.UNODE_USERNAME_KEY]
490 }
491 response = query.patch_endpoint(
492 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
493 )
494 assert status.success(response) is True
495 assert TCONST.are_json_responses_identical(
496 response.json(),
497 TCONST.RESPONSE_PATCH_USER,
498 "test_patch_user_lambda_username"
499 ) is True
500
501 def test_patch_user_lambda_email(self, setup_environment):
502 """_summary_
503 Test the /user endpoint of the server.
504 Args:
505 setup_environment (_type_): _description_
506 """
507 self.check_server(setup_environment)
508 path = TCONST.PATH_PATCH_USER
509 query: QueryEndpoint = setup_environment["query"]
510 status: QueryStatus = setup_environment["status"]
511 token: Dict[
512 str, Dict[str, str]
513 ] = setup_environment["tokens"][TCONST.LAMBDA_USER_TOKEN_KEY]
514 accounts: Dict[
515 str, any
516 ] = setup_environment["accounts"]["lambda_user"][TCONST.USER_PATCH_MODE]
517 body = {
518 "email": accounts[TCONST.UNODE_EMAIL_KEY]
519 }
520 response = query.patch_endpoint(
521 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
522 )
523 assert status.success(response) is True
524 assert TCONST.are_json_responses_identical(
525 response.json(),
526 TCONST.RESPONSE_PATCH_USER,
527 "test_path_user_lambda_email"
528 ) is True
529
530 def test_patch_user_lambda_password(self, setup_environment):
531 """_summary_
532 Test the /user endpoint of the server.
533 Args:
534 setup_environment (_type_): _description_
535 """
536 self.check_server(setup_environment)
537 path = TCONST.PATH_PATCH_USER
538 query: QueryEndpoint = setup_environment["query"]
539 status: QueryStatus = setup_environment["status"]
540 token: Dict[
541 str, Dict[str, str]
542 ] = setup_environment["tokens"][TCONST.LAMBDA_USER_TOKEN_KEY]
543 accounts: Dict[
544 str, any
545 ] = setup_environment["accounts"]["lambda_user"][TCONST.USER_PATCH_MODE]
546 body = {
547 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
548 }
549 response = query.patch_endpoint(
550 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
551 )
552 assert status.success(response) is True
553 assert TCONST.are_json_responses_identical(
554 response.json(),
555 TCONST.RESPONSE_PATCH_USER,
556 "test_patch_user_lambda_password"
557 ) is True
558
559 def test_patch_user_admin_username(self, setup_environment):
560 """_summary_
561 Test the /user endpoint of the server.
562 Args:
563 setup_environment (_type_): _description_
564 """
565 self.check_server(setup_environment)
566 path = TCONST.PATH_PATCH_USER
567 query: QueryEndpoint = setup_environment["query"]
568 status: QueryStatus = setup_environment["status"]
569 token: Dict[
570 str, Dict[str, str]
571 ] = setup_environment["tokens"][TCONST.ADMIN_USER_TOKEN_KEY]
572 accounts: Dict[
573 str, any
574 ] = setup_environment["accounts"]["admin_user"][TCONST.USER_PATCH_MODE]
575 body = {
576 "username": accounts[TCONST.UNODE_USERNAME_KEY]
577 }
578 response = query.patch_endpoint(
579 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
580 )
581 assert status.success(response) is True
582 assert TCONST.are_json_responses_identical(
583 response.json(),
584 TCONST.RESPONSE_PATCH_USER,
585 "test_patch_user_admin_username"
586 ) is True
587
588 def test_patch_user_admin_email(self, setup_environment):
589 """_summary_
590 Test the /user endpoint of the server.
591 Args:
592 setup_environment (_type_): _description_
593 """
594 self.check_server(setup_environment)
595 path = TCONST.PATH_PATCH_USER
596 query: QueryEndpoint = setup_environment["query"]
597 status: QueryStatus = setup_environment["status"]
598 token: Dict[
599 str, Dict[str, str]
600 ] = setup_environment["tokens"][TCONST.ADMIN_USER_TOKEN_KEY]
601 accounts: Dict[
602 str, any
603 ] = setup_environment["accounts"]["admin_user"][TCONST.USER_PATCH_MODE]
604 body = {
605 "email": accounts[TCONST.UNODE_EMAIL_KEY]
606 }
607 response = query.patch_endpoint(
608 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
609 )
610 assert status.success(response) is True
611 assert TCONST.are_json_responses_identical(
612 response.json(),
613 TCONST.RESPONSE_PATCH_USER,
614 "test_path_user_admin_email"
615 ) is True
616
617 def test_patch_user_admin_password(self, setup_environment):
618 """_summary_
619 Test the /user endpoint of the server.
620 Args:
621 setup_environment (_type_): _description_
622 """
623 self.check_server(setup_environment)
624 path = TCONST.PATH_PATCH_USER
625 query: QueryEndpoint = setup_environment["query"]
626 status: QueryStatus = setup_environment["status"]
627 token: Dict[
628 str, Dict[str, str]
629 ] = setup_environment["tokens"][TCONST.ADMIN_USER_TOKEN_KEY]
630 accounts: Dict[
631 str, any
632 ] = setup_environment["accounts"]["admin_user"][TCONST.USER_PATCH_MODE]
633 body = {
634 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
635 }
636 response = query.patch_endpoint(
637 path, content=body, header=token[TCONST.PRETTY_TOKEN_KEY]
638 )
639 assert status.success(response) is True
640 assert TCONST.are_json_responses_identical(
641 response.json(),
642 TCONST.RESPONSE_PATCH_USER,
643 "test_patch_user_admin_password"
644 ) is True
645
646 def test_get_user_lambda(self, setup_environment):
647 """_summary_
648 Test the /user endpoint of the server.
649 Args:
650 setup_environment (_type_): _description_
651 """
652 title = "test_get_user_lambda"
653 self.check_server(setup_environment)
654 server: Server = setup_environment["server"]
655 path = TCONST.PATH_GET_USER
656 query: QueryEndpoint = setup_environment["query"]
657 status: QueryStatus = setup_environment["status"]
658 token: Dict[
659 str, Dict[str, str]
660 ] = setup_environment["tokens"][TCONST.LAMBDA_USER_TOKEN_KEY]
661 response = query.get_endpoint(
662 path, header=token[TCONST.PRETTY_TOKEN_KEY]
663 )
664 response_node = TCONST.RESPONSE_GET_USER
665 if status.success(response) is True:
666 usr_id = server.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
667 title, token[TCONST.RAW_TOKEN_KEY]
668 )
669 TCONST.IDISP.log_debug(f"usr_id={usr_id}", title)
670 if isinstance(usr_id, str) is False:
671 TCONST.IDISP.log_error(f"Failed to find user: {usr_id}", title)
672 assert usr_id == server.error
673 column = server.runtime_data_initialised.database_link.get_table_column_names(
674 CONST.TAB_ACCOUNTS
675 )
676 TCONST.IDISP.log_debug(f"column = {column}", title)
677 if column == server.error or len(column) == 0:
678 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
679 assert column == server.success
680 user_node = server.runtime_data_initialised.database_link.get_data_from_table(
681 CONST.TAB_ACCOUNTS,
682 column,
683 f"id={usr_id}",
684 beautify=True
685 )
686 TCONST.IDISP.log_debug(f"user_node={user_node}", title)
687 if user_node == server.error or len(user_node) == 0:
688 assert user_node == server.success
689 new_profile = user_node[0]
690 for i in CONST.USER_INFO_BANNED:
691 if i in new_profile:
692 new_profile.pop(i)
693 if CONST.USER_INFO_ADMIN_NODE in new_profile:
694 new_profile[CONST.USER_INFO_ADMIN_NODE] = bool(
695 new_profile[CONST.USER_INFO_ADMIN_NODE]
696 )
697 response_node["msg"] = new_profile
698 TCONST.IDISP.log_debug(f"final node = {response_node}", title)
699 assert status.success(response) is True
700 assert TCONST.are_json_responses_identical(
701 response.json(),
702 response_node,
703 title
704 ) is True
705
706 def test_get_user_admin(self, setup_environment):
707 """_summary_
708 Test the /user endpoint of the server.
709 Args:
710 setup_environment (_type_): _description_
711 """
712 title = "test_get_user_admin"
713 self.check_server(setup_environment)
714 server: Server = setup_environment["server"]
715 path = TCONST.PATH_GET_USER
716 query: QueryEndpoint = setup_environment["query"]
717 status: QueryStatus = setup_environment["status"]
718 token: Dict[
719 str, Dict[str, str]
720 ] = setup_environment["tokens"][TCONST.ADMIN_USER_TOKEN_KEY]
721 response = query.get_endpoint(
722 path, header=token[TCONST.PRETTY_TOKEN_KEY]
723 )
724 response_node = TCONST.RESPONSE_GET_USER
725 if status.success(response) is True:
726 usr_id = server.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
727 title, token[TCONST.RAW_TOKEN_KEY]
728 )
729 TCONST.IDISP.log_debug(f"usr_id={usr_id}", title)
730 if isinstance(usr_id, str) is False:
731 TCONST.IDISP.log_error(f"Failed to find user: {usr_id}", title)
732 assert usr_id == server.error
733 column = server.runtime_data_initialised.database_link.get_table_column_names(
734 CONST.TAB_ACCOUNTS
735 )
736 TCONST.IDISP.log_debug(f"column = {column}", title)
737 if column == server.error or len(column) == 0:
738 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
739 assert column == server.success
740 user_node = server.runtime_data_initialised.database_link.get_data_from_table(
741 CONST.TAB_ACCOUNTS,
742 column,
743 f"id={usr_id}",
744 beautify=True
745 )
746 TCONST.IDISP.log_debug(f"user_node={user_node}", title)
747 if user_node == server.error or len(user_node) == 0:
748 assert user_node == server.success
749 new_profile = user_node[0]
750 for i in CONST.USER_INFO_BANNED:
751 if i in new_profile:
752 new_profile.pop(i)
753 if CONST.USER_INFO_ADMIN_NODE in new_profile:
754 new_profile[CONST.USER_INFO_ADMIN_NODE] = bool(
755 new_profile[CONST.USER_INFO_ADMIN_NODE]
756 )
757 response_node["msg"] = new_profile
758 TCONST.IDISP.log_debug(f"final node = {response_node}", title)
759 assert status.success(response) is True
760 assert TCONST.are_json_responses_identical(
761 response.json(),
762 response_node,
763 title
764 ) is True
765
766 def test_get_user_id_lambda(self, setup_environment):
767 """_summary_
768 Test the /user endpoint of the server.
769 Args:
770 setup_environment (_type_): _description_
771 """
772 title = "test_get_user_id_lambda"
773 self.check_server(setup_environment)
774 server: Server = setup_environment["server"]
775 path = TCONST.PATH_GET_USER_ID
776 query: QueryEndpoint = setup_environment["query"]
777 status: QueryStatus = setup_environment["status"]
778 token: Dict[
779 str, Dict[str, str]
780 ] = setup_environment["tokens"][TCONST.LAMBDA_USER_TOKEN_KEY]
781 response = query.get_endpoint(
782 path, header=token[TCONST.PRETTY_TOKEN_KEY]
783 )
784 response_node = TCONST.RESPONSE_GET_USER_ID
785 if status.success(response) is True:
786 usr_id = server.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
787 title, token[TCONST.RAW_TOKEN_KEY]
788 )
789 TCONST.IDISP.log_debug(f"usr_id={usr_id}", title)
790 if isinstance(usr_id, str) is False:
791 TCONST.IDISP.log_error(f"Failed to find user: {usr_id}", title)
792 assert usr_id == server.error
793 response_node["msg"] = f"Your id is {usr_id}"
794 response_node["resp"] = usr_id
795 TCONST.IDISP.log_debug(f"final node = {response_node}", title)
796 assert status.success(response) is True
797 assert TCONST.are_json_responses_identical(
798 response.json(),
799 response_node,
800 title
801 ) is True
802
803 def test_get_user_id_admin(self, setup_environment):
804 """_summary_
805 Test the /user endpoint of the server.
806 Args:
807 setup_environment (_type_): _description_
808 """
809 title = "test_get_user_id_admin"
810 self.check_server(setup_environment)
811 server: Server = setup_environment["server"]
812 path = TCONST.PATH_GET_USER_ID
813 query: QueryEndpoint = setup_environment["query"]
814 status: QueryStatus = setup_environment["status"]
815 token: Dict[
816 str, Dict[str, str]
817 ] = setup_environment["tokens"][TCONST.ADMIN_USER_TOKEN_KEY]
818 response = query.get_endpoint(
819 path, header=token[TCONST.PRETTY_TOKEN_KEY]
820 )
821 response_node = TCONST.RESPONSE_GET_USER_ID
822 if status.success(response) is True:
823 usr_id = server.runtime_data_initialised.boilerplate_non_http_initialised.get_user_id_from_token(
824 title, token[TCONST.RAW_TOKEN_KEY]
825 )
826 TCONST.IDISP.log_debug(f"usr_id={usr_id}", title)
827 if isinstance(usr_id, str) is False:
828 TCONST.IDISP.log_error(f"Failed to find user: {usr_id}", title)
829 assert usr_id == server.error
830 response_node["msg"] = f"Your id is {usr_id}"
831 response_node["resp"] = usr_id
832 TCONST.IDISP.log_debug(f"final node = {response_node}", title)
833 assert status.success(response) is True
834 assert TCONST.are_json_responses_identical(
835 response.json(),
836 response_node,
837 title
838 ) is True
839
840 def test_logout_user_lambda(self, setup_environment):
841 """_summary_
842 Test the /logout endpoint of the server.
843
844 Args:
845 setup_environment (_type_): _description_
846 """
847 self.check_server(setup_environment, True)
848 query: QueryEndpoint = setup_environment["query"]
849 status: QueryStatus = setup_environment["status"]
850 accounts: Dict[
851 str, any
852 ] = setup_environment["accounts"]["lambda_user"][TCONST.USER_PATCH_MODE]
853 body = {
854 "email": accounts[TCONST.UNODE_EMAIL_KEY],
855 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
856 }
857 TCONST.IDISP.log_info(f"body = {body}")
858 response = query.post_endpoint(TCONST.PATH_POST_LOGIN, content=body)
859 TCONST.IDISP.log_info(f"response.json() = {response.json()}")
860 pretty_token = ""
861 if status.success(response) is True:
862 token_node = f"{response.json()['token']}"
863 active_token = f"Bearer {token_node}"
864 pretty_token = {TCONST.TOKEN_AUTH_ID_STR: active_token}
865 else:
866 assert status.success(response) is True
867 logout_response = query.post_endpoint(
868 TCONST.PATH_POST_LOGOUT, header=pretty_token
869 )
870 correct_node = TCONST.RESPONSE_POST_LOGOUT
871 assert status.success(response) is True
872 assert status.success(logout_response) is True
873 assert TCONST.are_json_responses_identical(
874 logout_response.json(),
875 correct_node,
876 "test_post_logout_lambda"
877 ) is True
878
879 def test_logout_user_admin(self, setup_environment):
880 """_summary_
881 Test the /logout endpoint of the server.
882
883 Args:
884 setup_environment (_type_): _description_
885 """
886 self.check_server(setup_environment, True)
887 query: QueryEndpoint = setup_environment["query"]
888 status: QueryStatus = setup_environment["status"]
889 accounts: Dict[
890 str, any
891 ] = setup_environment["accounts"]["admin_user"][TCONST.USER_PATCH_MODE]
892 body = {
893 "email": accounts[TCONST.UNODE_EMAIL_KEY],
894 "password": accounts[TCONST.UNODE_PASSWORD_KEY]
895 }
896 response = query.post_endpoint(TCONST.PATH_POST_LOGIN, content=body)
897 pretty_token = ""
898 if status.success(response) is True:
899 token_node = f"{response.json()['token']}"
900 active_token = f"Bearer {token_node}"
901 pretty_token = {TCONST.TOKEN_AUTH_ID_STR: active_token}
902 else:
903 setup_environment[TCONST.RUNTIME_NODE_CRITICAL_KEY] = True
904 logout_response = query.post_endpoint(
905 TCONST.PATH_POST_LOGOUT, header=pretty_token
906 )
907 correct_node = TCONST.RESPONSE_POST_LOGOUT
908 assert status.success(response) is True
909 assert TCONST.are_json_responses_identical(
910 logout_response.json(),
911 correct_node,
912 "test_post_logout_admin"
913 ) is True
914
915 @pytest.mark.last
916 def test_delete_user_lambda(self, setup_environment):
917 """_summary_
918 Test the /user endpoint of the server.
919 Args:
920 setup_environment (_type_): _description_
921 """
922 self.check_server(setup_environment)
923 path = TCONST.PATH_DELETE_USER
924 query: QueryEndpoint = setup_environment["query"]
925 status: QueryStatus = setup_environment["status"]
926 token: Dict[str, Dict[str, str]] = setup_environment["tokens"]
927 response = query.delete_endpoint(
928 path, header=token[TCONST.LAMBDA_USER_TOKEN_KEY][TCONST.PRETTY_TOKEN_KEY]
929 )
930 assert status.success(response) is True
931 assert TCONST.are_json_responses_identical(
932 response.json(),
933 TCONST.RESPONSE_DELETE_USER,
934 "test_delete_user_lambda"
935 ) is True
936
937 @pytest.mark.last
938 def test_delete_user_admin(self, setup_environment):
939 """_summary_
940 Test the /user endpoint of the server.
941 Args:
942 setup_environment (_type_): _description_
943 """
944 self.check_server(setup_environment)
945 path = TCONST.PATH_DELETE_USER
946 query: QueryEndpoint = setup_environment["query"]
947 status: QueryStatus = setup_environment["status"]
948 token: Dict[str, Dict[str, str]] = setup_environment["tokens"]
949 response = query.delete_endpoint(
950 path, header=token[TCONST.ADMIN_USER_TOKEN_KEY][TCONST.PRETTY_TOKEN_KEY]
951 )
952 assert status.success(response) is True
953 assert TCONST.are_json_responses_identical(
954 response.json(),
955 TCONST.RESPONSE_DELETE_USER,
956 "test_delete_user_admin"
957 ) is True
958
959 # @pytest.mark.last
960 # def test_post_stop_server(self, setup_environment):
961 # """ Test the /stop endpoint of the server. """
962 # self.check_server(setup_environment)
963 # teardown_func: callable = setup_environment["teardown_func"]
964 # success: int = setup_environment["success"]
965 # status = teardown_func()
966 # assert status == success
test_post_login_lambda(self, setup_environment)
test_patch_user_admin_username(self, setup_environment)
test_post_login_admin(self, setup_environment)
test_get_user_id_lambda(self, setup_environment)
test_get_user_lambda(self, setup_environment)
test_patch_user_lambda_email(self, setup_environment)
test_post_register_admin(self, setup_environment)
test_logout_user_lambda(self, setup_environment)
test_patch_user_lambda_password(self, setup_environment)
test_logout_user_admin(self, setup_environment)
test_patch_user_lambda_username(self, setup_environment)
test_put_user_lambda(self, setup_environment)
test_get_user_admin(self, setup_environment)
test_delete_user_admin(self, setup_environment)
check_server(self, setup_environment, bool critical=False)
test_patch_user_admin_email(self, setup_environment)
test_post_register_lambda(self, setup_environment)
test_put_user_admin(self, setup_environment)
test_get_user_id_admin(self, setup_environment)
test_patch_user_admin_password(self, setup_environment)
test_delete_user_lambda(self, setup_environment)
pytest_configure(pytest.Config config)
setup_environment(pytest.FixtureRequest request)