194 lines
7.6 KiB
Python
194 lines
7.6 KiB
Python
from __future__ import annotations
|
|
import unittest
|
|
from os import chdir
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from src.pyrestresource import (
|
|
RestField,
|
|
register_rest_rootpoint,
|
|
RestResourceBase,
|
|
rsrc_verb,
|
|
RestRequestParams_GET,
|
|
RestRequestParams_POST,
|
|
RestRequestParams_Dict_GET,
|
|
RestRequestParams_PUT,
|
|
RestRequestParams_Dict_elem_GET,
|
|
RestRequestParams_Dict_elem_PUT,
|
|
RestRequestParams_Dict_DELETE,
|
|
RestRequestParams_Dict_POST,
|
|
T_SupportedRESTFields,
|
|
ResourcePlugin_dict_default,
|
|
)
|
|
|
|
testdir_path = Path(__file__).parent.resolve()
|
|
chdir(testdir_path.parent.resolve())
|
|
|
|
|
|
# to allow mock-ing, all the tested classes are in a function
|
|
def init_classes():
|
|
class Test_Record(RestResourceBase):
|
|
test_str: str
|
|
test_int: int
|
|
|
|
class ResourcePlugin_dict_Test_Record(ResourcePlugin_dict_default[str, Test_Record]):
|
|
static_Test_Record_active: bool = True
|
|
static_Test_Record = Test_Record(test_str="mytest", test_int=84)
|
|
|
|
def handle_dict_get_keys(
|
|
self,
|
|
resource_dict: dict[str, Test_Record],
|
|
params: RestRequestParams_Dict_GET,
|
|
) -> list[str]:
|
|
result = super().handle_dict_get_keys(resource_dict, params)
|
|
if self.static_Test_Record_active is True:
|
|
result.append("static_elem")
|
|
return result
|
|
|
|
def handle_dict_elem_get(
|
|
self,
|
|
resource_dict: dict[str, Test_Record],
|
|
key: str,
|
|
params: RestRequestParams_Dict_elem_GET,
|
|
) -> Test_Record:
|
|
if key == "static_elem":
|
|
if self.static_Test_Record_active is True:
|
|
return self.static_Test_Record
|
|
else:
|
|
raise RuntimeError("Key Not Found")
|
|
return super().handle_dict_elem_get(resource_dict, key, params)
|
|
|
|
def handle_dict_delete(
|
|
self,
|
|
resource_dict: dict[str, Test_Record],
|
|
key: str,
|
|
params: RestRequestParams_Dict_DELETE[str],
|
|
) -> None:
|
|
if key == "static_elem":
|
|
self.static_Test_Record_active = False
|
|
else:
|
|
super().handle_dict_delete(resource_dict, key, params)
|
|
|
|
def handle_dict_delete_all(
|
|
self,
|
|
resource_dict: dict[str, Test_Record],
|
|
params: RestRequestParams_Dict_DELETE[str],
|
|
) -> None:
|
|
self.static_Test_Record_active = False
|
|
super().handle_dict_delete_all(resource_dict, params)
|
|
|
|
def handle_dict_elem_put(
|
|
self,
|
|
resource_dict: dict[str, Test_Record],
|
|
key: str,
|
|
resource: Test_Record,
|
|
params: RestRequestParams_Dict_elem_PUT,
|
|
) -> None:
|
|
if key == "static_elem":
|
|
if self.static_Test_Record_active is True:
|
|
self.static_Test_Record = resource
|
|
else:
|
|
raise RuntimeError("Key Not Found")
|
|
else:
|
|
super().handle_dict_elem_put(resource_dict, key, resource, params)
|
|
|
|
def handle_dict_post(
|
|
self,
|
|
resource_dict: dict[str, Test_Record],
|
|
key: str,
|
|
resource: Test_Record,
|
|
params: RestRequestParams_Dict_POST[str],
|
|
) -> Optional[str]:
|
|
resource.test_int = resource.test_int + 1
|
|
return super().handle_dict_post(resource_dict, key, resource, params)
|
|
|
|
@register_rest_rootpoint
|
|
class RootApp(RestResourceBase):
|
|
str_dict_Test_Record: dict[str, Test_Record] = RestField(
|
|
default={"test": Test_Record(test_str="Hi", test_int=42)},
|
|
plugin=ResourcePlugin_dict_Test_Record,
|
|
)
|
|
|
|
# this add the classes to globals to allow using them later on
|
|
# => this is only for uinit-testing purpose and is not needed in real use
|
|
|
|
globals()[Test_Record.__name__] = Test_Record
|
|
globals()[RootApp.__name__] = RootApp
|
|
|
|
|
|
class Test_RestAPI_Plugin_Dict(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
chdir(testdir_path.parent.resolve())
|
|
init_classes()
|
|
self.testapp = RootApp()
|
|
|
|
def test_get_root(self):
|
|
result = self.testapp.process_request("/", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), "{}")
|
|
# print(result.get_result())
|
|
|
|
def test_get_dict_keys(self):
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '["test", "static_elem"]')
|
|
|
|
def test_get_dict_elems(self):
|
|
result = self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '{"test_str": "Hi", "test_int": 42}')
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record/static_elem", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '{"test_str": "mytest", "test_int": 84}')
|
|
|
|
def test_delete_dict_elems(self):
|
|
self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.DELETE)
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '["static_elem"]')
|
|
|
|
self.testapp.process_request("/str_dict_Test_Record/static_elem", rsrc_verb.DELETE)
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), "[]")
|
|
|
|
def test_delete_all_dict_elems(self):
|
|
self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.DELETE)
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '["static_elem"]')
|
|
|
|
self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.DELETE)
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), "[]")
|
|
|
|
def test_delete_dict_elems_API_key(self):
|
|
self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.DELETE)
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '["static_elem"]')
|
|
|
|
self.testapp.process_request("/str_dict_Test_Record?API_key=static_elem", rsrc_verb.DELETE)
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), "[]")
|
|
|
|
def test_put_dict_elem(self):
|
|
self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.PUT, '{"test_str": "Hi", "test_int": 43}')
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record/test", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '{"test_str": "Hi", "test_int": 43}')
|
|
|
|
self.testapp.process_request("/str_dict_Test_Record/static_elem", rsrc_verb.PUT, '{"test_str": "Hi you", "test_int": 7}')
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record/static_elem", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '{"test_str": "Hi you", "test_int": 7}')
|
|
|
|
def test_post_dict_elem(self):
|
|
result = self.testapp.process_request("/str_dict_Test_Record?API_key=newval", rsrc_verb.POST, '{"test_str": "Hi2", "test_int": 77}')
|
|
self.assertEqual(result.get_result(), '"newval"')
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record/newval", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '{"test_str": "Hi2", "test_int": 78}')
|
|
|
|
result = self.testapp.process_request("/str_dict_Test_Record", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '["test", "newval", "static_elem"]')
|