chacha/pyrestresource
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 21 Wiki Activity

fix ACL + cleaning

Browse Source
This commit is contained in:
cclecle
2023-11-03 17:42:34 +00:00
parent c3ff00e877
commit 346ff649ec
11 changed files with 332 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/pyrestresource/rest_request.py
View File

@@ -10,8 +10,9 @@ from re import sub
from urllib.parse import urlparse, parse_qs
from pydantic import BaseModel, Field
from typeguard import check_type
from .rest_types import rsrc_verb, T_SupportedRESTFields
from .rest_types import rsrc_verb, T_SupportedRESTFields, T_AllSupportedFields
from .rest_request_opt import (
RestRequestParams_POST,
@@ -129,6 +130,10 @@ class RestRequest(Generic[_T_RestRequestParams]):
raise RuntimeError("url and verb and data must be set")
self.url = url
self.verb = verb
if data != {} and not check_type(data, T_AllSupportedFields):
raise RuntimeError(f"Wrong data type received: {data}")
self.data = data
# parse_qs returns list[] for all keys, the command convert list to single items so pydantic can eat them :)

39
src/pyrestresource/rest_resource.py
View File

@@ -142,8 +142,8 @@ class RestResourceWalker_Sub_T_Dict__tree_init(RestResourceWalker_Sub_T_Dict):
if "ACL" in self.resource.json_schema_extra:
if isinstance(self.resource.json_schema_extra["ACL"], list):
print(f"found ACL (Dict): {self.resource.json_schema_extra['ACL']}")
self.parent.annotation._ACL_record_[self.resource_name] = self.resource.json_schema_extra["ACL"]
# print(f"found ACL (Dict): {self.resource.json_schema_extra['ACL']}")
self.parent.annotation._ACL_record_[self.resource_name] += self.resource.json_schema_extra["ACL"]
else:
raise RuntimeError("ACL must be a list()")
@@ -174,6 +174,9 @@ class RestResourceWalker_Sub_RestFields__tree_init(RestResourceWalker_Sub_RestFi
if self.parent.annotation._primary_key_ is not None:
raise RuntimeError(f"Only one primary key is allowed {self.parent.resource_name}.{self.resource_name}")
self.parent.annotation._primary_key_ = self.resource_name
self.parent.annotation._ACL_record_[self.resource_name] = [
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.DENY)
]
if "plugin" in self.resource.json_schema_extra:
plugin_field: ResourcePlugin_field = self.resource.json_schema_extra["plugin"]
@@ -184,8 +187,8 @@ class RestResourceWalker_Sub_RestFields__tree_init(RestResourceWalker_Sub_RestFi
if "ACL" in self.resource.json_schema_extra:
if isinstance(self.resource.json_schema_extra["ACL"], list):
print(f"found ACL (Field): {self.resource.json_schema_extra['ACL']}")
self.parent.annotation._ACL_record_[self.resource_name] = self.resource.json_schema_extra["ACL"]
# print(f"found ACL (Field): {self.resource.json_schema_extra['ACL']}")
self.parent.annotation._ACL_record_[self.resource_name] += self.resource.json_schema_extra["ACL"]
else:
raise RuntimeError("ACL must be a list()")
@@ -223,8 +226,8 @@ class RestResourceWalker_Sub_RestResourceBase__tree_init(RestResourceWalker_Sub_
if "ACL" in self.resource.json_schema_extra:
if isinstance(self.resource.json_schema_extra["ACL"], list):
print(f"found ACL (Resource): {self.resource.json_schema_extra['ACL']}")
self.parent.annotation._ACL_record_[self.resource_name] = self.resource.json_schema_extra["ACL"]
# print(f"found ACL (Resource): {self.resource.json_schema_extra['ACL']}")
self.parent.annotation._ACL_record_[self.resource_name] += self.resource.json_schema_extra["ACL"]
else:
raise RuntimeError("ACL must be a list()")
@@ -261,35 +264,35 @@ class RestResourceBase(ABC, BaseModel, validate_assignment=True):
]
] = {}
def _check_acl(self, user: ACL_target_user, group: ACL_target_group, verb: rsrc_verb, field: str):
print(f"evaluate self ACLs rule: {self._ACL_record_}")
if verb is rsrc_verb.GET and self.model_fields[field].exclude is True:
print("ALLOWED (excluded field)")
def _check_acl(self, user: ACL_target_user, group: ACL_target_group, verb: rsrc_verb, field: str, is_self: bool = True):
# print(f"evaluate self ACLs rule: {self._ACL_record_}")
if is_self and verb is rsrc_verb.GET and self.model_fields[field].exclude is True:
# print("ALLOWED (excluded field)")
return
for acl in self._ACL_record_[field]:
print(f"evaluate ACL rule: {acl}")
# print(f"evaluate ACL rule: {acl}")
if verb in acl.verbs:
if isinstance(acl.target, ACL_target_user):
if user == acl.target:
if acl.rule is ACL_rule.ALLOW:
print("ALLOWED (user)")
# print("ALLOWED (user)")
return
raise RuntimeError(f"Not allowed access detected: {field}")
elif isinstance(acl.target, ACL_target_group):
if group == acl.target or acl.target == ACL_target_group_Any():
if acl.rule is ACL_rule.ALLOW:
print("ALLOWED (group)")
# print("ALLOWED (group)")
return
raise RuntimeError(f"Not allowed access detected: {field}")
else:
raise RuntimeError(f"Wrong ACL target type: {field}")
print("ALLOWED (Default)")
# print("ALLOWED (Default)")
def check_acl_access(self, request: RestRequest) -> None:
def check_acl_field(self, request: RestRequest, req_index: int = 0) -> None:
"""Check ACL on requested field access"""
self._check_acl(request.user, request.group, request.get_verb(), request.get_resource_origin(0))
self._check_acl(request.user, request.group, request.get_verb(), request.get_resource_origin(req_index), False)
def check_acl_operation(self, request: RestRequest, new_data: Optional[dict[str, _T_SupportedRESTFields]]) -> None:
def check_acl_self(self, request: RestRequest, new_data: Optional[dict[str, _T_SupportedRESTFields]]) -> None:
"""Check ACL on requested field operation (involving checking sub-fields)"""
if request.get_verb() is rsrc_verb.GET:
for key in self.model_fields.keys():
@@ -401,5 +404,7 @@ class RestResourceBase(ABC, BaseModel, validate_assignment=True):
request.set_result(json.dumps(result.model_dump(mode="json")))
elif result is not None:
request.set_result(json.dumps(result, cls=_JSONEncoder))
else:
request.set_result("null")
return request

24
src/pyrestresource/rest_resource_handler.py
View File

@@ -183,7 +183,7 @@ class ResourceHandler(
# reveal_type(next_resource)
_next_resource = cast(_T_Resource, next_resource)
# reveal_type(_next_resource)
print(f"[DEBUG] next_resource = {type(next_resource).__name__}")
# print(f"[DEBUG] next_resource = {type(next_resource).__name__}")
if (
isinstance(_next_resource, RestResourceBase)
@@ -464,13 +464,13 @@ class ResourceHandler_RestResourceBase(
def _check_access_rights(self) -> None:
super()._check_access_rights()
print(f"{type(self).__name__}->_check_access_rights()")
# print(f"{type(self).__name__}->_check_access_rights()")
if self.req.get_resource_origin(0) == "/":
return
print("==================")
print(self.req.get_resource_origin(0))
# print("==================")
# print(self.req.get_resource_origin(0))
# print(len(self.req.get_url_stack()))
# print(self.resource._model_dump_excluded_)
# print(type(self.resource))
@@ -479,7 +479,7 @@ class ResourceHandler_RestResourceBase(
if self.req.get_resource_origin(0) not in self.resource.model_fields:
raise RuntimeError(f"Unknown field access detected: {self.req.get_url_stack()}")
self.resource.check_acl_access(self.req)
self.resource.check_acl_field(self.req)
if len(self.req.get_url_stack()) == 0: # destination reached
if self.resource.model_fields[self.req.get_resource_origin(0)].exclude is True and self.req.get_verb() is rsrc_verb.GET:
@@ -500,7 +500,7 @@ class ResourceHandler_RestResourceBase(
# CASE 1: no more item in url_stack => we reached the endpoint (operation)
# So we are in a RestResourceBase instance and must return the content
if len(self.req.get_url_stack()) == 0:
self.resource.check_acl_operation(self.req)
self.resource.check_acl_self(self.req, None)
for key, attr in self.resource.model_fields.items():
if key in self.resource._plugins_:
if isinstance(self.resource._plugins_[key], ResourcePlugin_field):
@@ -522,6 +522,7 @@ class ResourceHandler_RestResourceBase(
return self.resource
# CASE 3: in between (access)
self.resource.check_acl_field(self.req)
value = getattr(self.resource, self.req.get_resource_origin(0))
key = self.req.get_resource_origin(0)
@@ -546,7 +547,7 @@ class ResourceHandler_RestResourceBase(
# print(f"{type(self).__name__}->_process_put()")
# print(f"{type(self).__name__}->resource = {type(self.resource).__name__}")
self.resource.check_acl_operation(self.req, self.req.get_data())
self.resource.check_acl_self(self.req, self.req.get_data())
# creating a copy of the current resource
_new_resrc = self.resource.copy()
@@ -564,9 +565,8 @@ class ResourceHandler_RestResourceBase(
# applying plugins (from parent element)
if self.prev_handler is not None:
# element is within a dict
if (
isinstance(self.prev_handler.resource, dict)
isinstance(self.prev_handler.resource, dict) # element is within a dict
and self.prev_handler.prev_handler is not None
and isinstance(self.prev_handler.prev_handler.resource, RestResourceBase)
):
@@ -587,7 +587,7 @@ class ResourceHandler_RestResourceBase(
)
_new_resrc = plugin_rsrc.handle_resource_put(_new_resrc, params)
self.resource.update(**_new_resrc.dict())
self.resource.update(**_new_resrc.__dict__)
return
def _handle_process_delete(self, params) -> None:
@@ -629,6 +629,8 @@ class ResourceHandler_simple(
assert self.prev_handler is not None
assert isinstance(self.prev_handler.resource, RestResourceBase)
self.prev_handler.resource.check_acl_field(self.req, 1)
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:
plugin_simple: ResourcePlugin_field = cast(
ResourcePlugin_field,
@@ -645,6 +647,8 @@ class ResourceHandler_simple(
assert self.prev_handler is not None
assert isinstance(self.prev_handler.resource, RestResourceBase)
self.prev_handler.resource.check_acl_field(self.req, 1)
value = self.req.get_data()
if self.req.get_resource_origin(1) in self.prev_handler.resource._plugins_:

1
src/pyrestresource/rest_types.py
View File

@@ -98,5 +98,4 @@ T_Dict = dict[T_DictKey, T_DictValues]
_T_Dict = dict[_T_DictKey, _T_DictValues]
T_AllSupportedFields = T_Dict | T_FieldValue
T_AllSupportedFiels = T_Dict | T_FieldValue
T_AllSupportedContainers = Union[T_Dict, "RestResourceBase"]

180
test/test_ACL.py Normal file
View File

@@ -0,0 +1,180 @@
from __future__ import annotations
import unittest
from os import chdir
from pathlib import Path
from typing import Optional
from pydantic import Field
print(__name__)
print(__package__)
from src.pyrestresource import (
register_rest_rootpoint,
RestResourceBase,
rsrc_verb,
RestRequestParams_GET,
RestRequestParams_POST,
RestRequestParams_Dict_GET,
RestRequestParams_PUT,
T_SupportedRESTFields,
ResourcePlugin_field_default,
ResourcePlugin_RestResourceBase_default,
ACL_target_group_Any,
ACL_record,
ACL_rule,
)
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 TestResource(RestResourceBase):
username: Optional[str] = Field(None)
secret: Optional[str] = Field(
None,
exclude=True,
ACL=[
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.ALLOW),
ACL_record(verbs=[rsrc_verb.GET], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
],
)
class TestResource2(RestResourceBase):
version_ro: Optional[str] = Field(
"1.2.3",
ACL=[
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
],
)
version: Optional[str] = Field("3.2.1")
@register_rest_rootpoint
class RootApp(RestResourceBase):
resource_with_secret: TestResource = Field(default=TestResource())
resource_with_secret_ACL: TestResource = Field(
default=TestResource(), ACL=[ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.DENY)]
)
resource2: TestResource2 = Field(TestResource2())
# 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()[TestResource.__name__] = TestResource
globals()[RootApp.__name__] = RootApp
class Test_RestAPI_ACL(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_subresource_readonly(self):
result = self.testapp.process_request("/", rsrc_verb.GET)
self.assertEqual(result.get_result(), "{}")
result = self.testapp.process_request("/resource2", rsrc_verb.GET)
self.assertEqual(result.get_result(), '{"version_ro": "1.2.3", "version": "3.2.1"}')
self.testapp.process_request("/resource2/version", rsrc_verb.PUT, '"6.6.6"')
result = self.testapp.process_request("/resource2", rsrc_verb.GET)
self.assertEqual(result.get_result(), '{"version_ro": "1.2.3", "version": "6.6.6"}')
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource2/version_ro", rsrc_verb.PUT, '"6.6.6"')
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource2", rsrc_verb.PUT, '{"version_ro": "6.6.1", "version": "6.6.2"}')
result = self.testapp.process_request("/resource2", rsrc_verb.GET)
self.assertEqual(result.get_result(), '{"version_ro": "1.2.3", "version": "6.6.6"}')
def test_subresource(self):
result = self.testapp.process_request("/", rsrc_verb.GET)
self.assertEqual(result.get_result(), "{}")
result = self.testapp.process_request("/resource_with_secret", rsrc_verb.GET)
self.assertEqual(result.get_result(), '{"username": null}')
result = self.testapp.process_request("/resource_with_secret/username", rsrc_verb.GET)
self.assertEqual(result.get_result(), "null")
self.assertEqual(self.testapp.resource_with_secret.username, None)
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret/secret", rsrc_verb.GET)
self.assertEqual(self.testapp.resource_with_secret.secret, None)
result = self.testapp.process_request("/resource_with_secret", rsrc_verb.PUT, '{"username":"chacha","secret":"123456"}')
self.assertEqual(result.get_result(), "null")
result = self.testapp.process_request("/resource_with_secret", rsrc_verb.GET)
self.assertEqual(result.get_result(), '{"username": "chacha"}')
result = self.testapp.process_request("/resource_with_secret/username", rsrc_verb.GET)
self.assertEqual(result.get_result(), '"chacha"')
self.assertEqual(self.testapp.resource_with_secret.username, "chacha")
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret/secret", rsrc_verb.GET)
self.assertEqual(self.testapp.resource_with_secret.secret, "123456")
def test_subresource_field(self):
result = self.testapp.process_request("/resource_with_secret/username", rsrc_verb.PUT, '"chacha"')
self.assertEqual(result.get_result(), "null")
result = self.testapp.process_request("/resource_with_secret", rsrc_verb.GET)
self.assertEqual(result.get_result(), '{"username": "chacha"}')
result = self.testapp.process_request("/resource_with_secret/username", rsrc_verb.GET)
self.assertEqual(result.get_result(), '"chacha"')
self.assertEqual(self.testapp.resource_with_secret.username, "chacha")
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret/secret", rsrc_verb.GET)
result = self.testapp.process_request("/resource_with_secret/secret", rsrc_verb.PUT, '"123456"')
self.assertEqual(result.get_result(), "null")
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret/secret", rsrc_verb.GET)
self.assertEqual(self.testapp.resource_with_secret.secret, "123456")
def test_subresource_ACL(self):
result = self.testapp.process_request("/", rsrc_verb.GET)
self.assertEqual(result.get_result(), "{}")
result = self.testapp.process_request("/resource_with_secret_ACL", rsrc_verb.GET)
self.assertEqual(result.get_result(), '{"username": null}')
result = self.testapp.process_request("/resource_with_secret_ACL/username", rsrc_verb.GET)
self.assertEqual(result.get_result(), "null")
self.assertEqual(self.testapp.resource_with_secret_ACL.username, None)
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret_ACL/secret", rsrc_verb.GET)
self.assertEqual(self.testapp.resource_with_secret_ACL.secret, None)
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret_ACL", rsrc_verb.PUT, '{"username":"chacha","secret":"123456"}')
self.assertEqual(self.testapp.resource_with_secret_ACL.username, None)
self.assertEqual(self.testapp.resource_with_secret_ACL.secret, None)
def test_subresource_ACL_field(self):
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret_ACL/username", rsrc_verb.PUT, '"chacha"')
self.assertEqual(self.testapp.resource_with_secret_ACL.username, None)
self.assertEqual(self.testapp.resource_with_secret_ACL.secret, None)
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request("/resource_with_secret_ACL/secret", rsrc_verb.PUT, '"123456"')
self.assertEqual(self.testapp.resource_with_secret_ACL.username, None)
self.assertEqual(self.testapp.resource_with_secret_ACL.secret, None)

16
test/test_rest_login.py
View File

@@ -7,8 +7,6 @@ from typing import Optional, Annotated
from pydantic import Field
from uuid import UUID, uuid4
from time import time, sleep
from time import time
import json
import uvicorn
import socket
import requests
@@ -32,9 +30,11 @@ from src.pyrestresource import (
T_SupportedRESTFields,
ResourcePlugin_field_default,
ResourcePlugin_RestResourceBase_default,
ACL_target_group_Any,
ACL_record,
ACL_rule,
)
from src.pyrestresource import ACL_target_user, ACL_target_group, ACL_target_group_Any, ACL_record, ACL_rule
from pprint import pprint
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
@@ -83,7 +83,7 @@ def init_classes():
exclude=True,
ACL=[
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.ALLOW),
ACL_record(verbs=[rsrc_verb.GET, rsrc_verb.DELETE, rsrc_verb.POST], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
ACL_record(verbs=[rsrc_verb.GET], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
],
)
@@ -101,8 +101,6 @@ def find_free_port():
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(("", 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
hostname = socket.gethostname()
IPAddr = socket.gethostbyname(hostname)
return "localhost", s.getsockname()[1]
@@ -119,7 +117,6 @@ class Test_RestAPI_LOGIN(unittest.TestCase):
self.testapp = RootApp()
def test_login(self):
"""
result = self.testapp.process_request("/login", rsrc_verb.GET)
print("*****************")
print(result.get_result())
@@ -131,13 +128,11 @@ class Test_RestAPI_LOGIN(unittest.TestCase):
# result = self.testapp.process_request("/login/secret", rsrc_verb.GET)
# print("*****************")
# print(result.get_result())
"""
result = self.testapp.process_request("/login", rsrc_verb.PUT, '{"username":"chacha","secret":"123456"}')
print("*****************")
print(result.get_result())
"""
result = self.testapp.process_request("/login", rsrc_verb.GET)
print("*****************")
print(result.get_result())
@@ -149,7 +144,6 @@ class Test_RestAPI_LOGIN(unittest.TestCase):
# result = self.testapp.process_request("/login/secret", rsrc_verb.GET)
# print("*****************")
# print(result.get_result())
"""
class Test_RestAPI_LOGIN_Web(unittest.TestCase):

147
test/test_rest_resource.py
View File

@@ -21,6 +21,9 @@ from src.pyrestresource import (
RestRequestParams_POST,
RestRequestParams_Dict_GET,
T_SupportedRESTFields,
ACL_target_group_Any,
ACL_record,
ACL_rule,
)
from pprint import pprint
@@ -58,9 +61,19 @@ def init_classes():
Patch_2 = Patch(uuid="d385a1d2-65fa-11ee-8c99-0242ac120002", shortname="testPatch2")
class User(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
uuid: UUID = Field(
default_factory=uuid4,
primary_key=True,
)
name: str
secret: str = Field(..., exclude=True)
secret: str = Field(
...,
exclude=True,
ACL=[
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.ALLOW),
ACL_record(verbs=[rsrc_verb.GET], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
],
)
User1 = User(
uuid="8da57a3c-661f-11ee-8c99-0242ac120002",
@@ -68,8 +81,6 @@ def init_classes():
secret="la blanquette est bonne",
)
ext_patchs: dict[UUID, Patch] = {}
class Patch2(RestResourceBase):
uuid: UUID = Field(default_factory=uuid4, primary_key=True)
shortname: str
@@ -117,100 +128,100 @@ class Test_RestAPI_GET(unittest.TestCase):
def test_get_root(self):
result = self.testapp.process_request("/", rsrc_verb.GET)
self.assertEqual(result, '{"testValueRoot": 3.14}')
self.assertEqual(result.get_result(), '{"testValueRoot": 3.14}')
def test_get_root__multiple_slash(self):
result = self.testapp.process_request("/////", rsrc_verb.GET)
self.assertEqual(result, '{"testValueRoot": 3.14}')
self.assertEqual(result.get_result(), '{"testValueRoot": 3.14}')
result = self.testapp.process_request("////", rsrc_verb.GET)
self.assertEqual(result, '{"testValueRoot": 3.14}')
self.assertEqual(result.get_result(), '{"testValueRoot": 3.14}')
def test_get_root__nested_value(self):
result = self.testapp.process_request("/testValueRoot", rsrc_verb.GET)
self.assertEqual(result, "3.14")
self.assertEqual(result.get_result(), "3.14")
def test_get_root__nested_value__trailing_slash(self):
result = self.testapp.process_request("/testValueRoot/", rsrc_verb.GET)
self.assertEqual(result, "3.14")
self.assertEqual(result.get_result(), "3.14")
result = self.testapp.process_request("/testValueRoot//", rsrc_verb.GET)
self.assertEqual(result, "3.14")
self.assertEqual(result.get_result(), "3.14")
result = self.testapp.process_request("/testValueRoot///", rsrc_verb.GET)
self.assertEqual(result, "3.14")
self.assertEqual(result.get_result(), "3.14")
def test_get_root__nested_value__multiple_slash(self):
result = self.testapp.process_request("//testValueRoot", rsrc_verb.GET)
self.assertEqual(result, "3.14")
self.assertEqual(result.get_result(), "3.14")
result = self.testapp.process_request("///testValueRoot", rsrc_verb.GET)
self.assertEqual(result, "3.14")
self.assertEqual(result.get_result(), "3.14")
def test_get_version(self):
result = self.testapp.process_request("/info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
self.assertEqual(result.get_result(), '{"version": "0.0.1", "api_version": "0.0.2"}')
def test_get_version__trailing_slash(self):
result = self.testapp.process_request("/info/", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
self.assertEqual(result.get_result(), '{"version": "0.0.1", "api_version": "0.0.2"}')
result = self.testapp.process_request("/info//", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
self.assertEqual(result.get_result(), '{"version": "0.0.1", "api_version": "0.0.2"}')
result = self.testapp.process_request("/info///", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
self.assertEqual(result.get_result(), '{"version": "0.0.1", "api_version": "0.0.2"}')
def test_get_version__multiple_slash(self):
result = self.testapp.process_request("//info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
self.assertEqual(result.get_result(), '{"version": "0.0.1", "api_version": "0.0.2"}')
result = self.testapp.process_request("///info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "0.0.1", "api_version": "0.0.2"}')
self.assertEqual(result.get_result(), '{"version": "0.0.1", "api_version": "0.0.2"}')
def test_get_version__nested_value(self):
result = self.testapp.process_request("/info/api_version", rsrc_verb.GET)
self.assertEqual(result, '"0.0.2"')
self.assertEqual(result.get_result(), '"0.0.2"')
result = self.testapp.process_request("/info/version", rsrc_verb.GET)
self.assertEqual(result, '"0.0.1"')
self.assertEqual(result.get_result(), '"0.0.1"')
def test_get_dict_games(self):
result = self.testapp.process_request("/games", rsrc_verb.GET)
self.assertEqual(result, '["9b0381d4-65f6-11ee-8c99-0242ac120002"]')
self.assertEqual(result.get_result(), '["9b0381d4-65f6-11ee-8c99-0242ac120002"]')
def test_get_dict_patchs(self):
result = self.testapp.process_request("/patchs", rsrc_verb.GET)
self.assertEqual(
result,
result.get_result(),
'["cee1e870-65fa-11ee-8c99-0242ac120002", "d385a1d2-65fa-11ee-8c99-0242ac120002"]',
)
def test_get_dict_patch_element(self):
result = self.testapp.process_request("/patchs/cee1e870-65fa-11ee-8c99-0242ac120002", rsrc_verb.GET)
self.assertEqual(
result,
result.get_result(),
'{"uuid": "cee1e870-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch1", "name": null, "description": null}',
)
def test_get_dict_game_element(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "9b0381d4-65f6-11ee-8c99-0242ac120002", "shortname": "testGame", "name": null, "description": null}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_get_dict_game_element__nested_value(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/shortname", rsrc_verb.GET)
expected = '"testGame"'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_get_dict_game_element__nested_value2(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/uuid", rsrc_verb.GET)
expected = '"9b0381d4-65f6-11ee-8c99-0242ac120002"'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_get_nested_dict_games_patchs(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
self.assertEqual(result, '["cee1e870-65fa-11ee-8c99-0242ac120002"]')
self.assertEqual(result.get_result(), '["cee1e870-65fa-11ee-8c99-0242ac120002"]')
def test_get_nested_dict_games_patch_element(self):
result = self.testapp.process_request(
@@ -218,28 +229,28 @@ class Test_RestAPI_GET(unittest.TestCase):
rsrc_verb.GET,
)
expected = '{"uuid": "cee1e870-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch1", "name": null, "description": null}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_get_nested_dict_games_patch_element__nested_value(self):
result = self.testapp.process_request(
"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/uuid",
rsrc_verb.GET,
)
self.assertEqual(result, '"cee1e870-65fa-11ee-8c99-0242ac120002"')
self.assertEqual(result.get_result(), '"cee1e870-65fa-11ee-8c99-0242ac120002"')
def test_get_dict_game_element__API_nested(self):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002?API_nested=True", rsrc_verb.GET)
expected = '{"uuid": "9b0381d4-65f6-11ee-8c99-0242ac120002", "shortname": "testGame", "name": null, "description": null}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_get_dict_users(self):
result = self.testapp.process_request("/users", rsrc_verb.GET)
self.assertEqual(result, '["8da57a3c-661f-11ee-8c99-0242ac120002"]')
self.assertEqual(result.get_result(), '["8da57a3c-661f-11ee-8c99-0242ac120002"]')
def test_get_dict_user_element(self):
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.GET)
self.assertEqual(
result,
result.get_result(),
'{"uuid": "8da57a3c-661f-11ee-8c99-0242ac120002", "name": "chacha"}',
"no secret seen",
)
@@ -247,14 +258,14 @@ class Test_RestAPI_GET(unittest.TestCase):
def test_get_dict_user_element2(self):
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002?API_nested=True", rsrc_verb.GET)
self.assertEqual(
result,
result.get_result(),
'{"uuid": "8da57a3c-661f-11ee-8c99-0242ac120002", "name": "chacha"}',
"no secret seen",
)
def test_get_dict_user_element__nested_value(self):
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002/name", rsrc_verb.GET)
self.assertEqual(result, '"chacha"')
self.assertEqual(result.get_result(), '"chacha"')
def test_get_dict_user_element__nested_value__forbiden(self):
with self.assertRaises(RuntimeError): # TODO: custom exception
@@ -278,7 +289,7 @@ class Test_RestAPI_PUT(unittest.TestCase):
self.testapp.process_request("/info", rsrc_verb.PUT, '{"version": "1.2.3", "api_version": "3.2.1"}')
result = self.testapp.process_request("/info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.2.3", "api_version": "3.2.1"}')
self.assertEqual(result.get_result(), '{"version": "1.2.3", "api_version": "3.2.1"}')
def test_put_dict_user_nested_value(self):
self.testapp.process_request(
@@ -288,12 +299,12 @@ class Test_RestAPI_PUT(unittest.TestCase):
)
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002/name", rsrc_verb.GET)
self.assertEqual(result, '"chacha2"')
self.assertEqual(result.get_result(), '"chacha2"')
def test_put_user_nested_value__forbiden(self):
with self.assertRaises(RuntimeError): # TODO: custom exception
self.testapp.process_request(
"/users/8da57a3c-661f-11ee-8c99-0242ac120002/secret",
"/users/8da57a3c-661f-11ee-8c99-0242ac120002/uuid",
rsrc_verb.PUT,
'"test"',
)
@@ -307,11 +318,11 @@ class Test_RestAPI_PUT(unittest.TestCase):
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
result = self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "8da57a3c-661f-11ee-8c99-0242ac120002", "name": "testUser4"}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_put_dict_patch__nested(self):
self.testapp.process_request(
@@ -325,7 +336,7 @@ class Test_RestAPI_PUT(unittest.TestCase):
rsrc_verb.GET,
)
expected = '{"uuid": "cee1e870-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch998", "name": "MyPatch", "description": "MyDescription123"}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
class Test_RestAPI_POST(unittest.TestCase):
@@ -340,15 +351,15 @@ class Test_RestAPI_POST(unittest.TestCase):
rsrc_verb.POST,
'{"name": "testUser", "secret": "test"}',
)
self.assertEqual(result, '"e5e87d32-662b-11ee-8c99-0242ac120002"')
self.assertEqual(result.get_result(), '"e5e87d32-662b-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "e5e87d32-662b-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
result = self.testapp.process_request("/users/e5e87d32-662b-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "e5e87d32-662b-11ee-8c99-0242ac120002", "name": "testUser"}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_post_dict_user__nested_key(self):
result = self.testapp.process_request(
@@ -356,15 +367,15 @@ class Test_RestAPI_POST(unittest.TestCase):
rsrc_verb.POST,
'{"name": "testUser2", "secret": "test", "uuid":"e7e86d32-662b-11ee-8c99-0242ac120002"}',
)
self.assertEqual(result, '"e7e86d32-662b-11ee-8c99-0242ac120002"')
self.assertEqual(result.get_result(), '"e7e86d32-662b-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "e7e86d32-662b-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
result = self.testapp.process_request("/users/e7e86d32-662b-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "e7e86d32-662b-11ee-8c99-0242ac120002", "name": "testUser2"}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
@patch(f"{__loader__.name }.uuid4")
def test_post_dict_user__auto_key(self, mock_uuid4):
@@ -375,15 +386,15 @@ class Test_RestAPI_POST(unittest.TestCase):
self.testapp = RootApp()
result = self.testapp.process_request("/users", rsrc_verb.POST, '{"name": "testUser3", "secret": "test"}')
self.assertEqual(result, '"5faccb2e-69aa-11ee-8c99-0242ac120002"')
self.assertEqual(result.get_result(), '"5faccb2e-69aa-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "5faccb2e-69aa-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
result = self.testapp.process_request("/users/5faccb2e-69aa-11ee-8c99-0242ac120002", rsrc_verb.GET)
expected = '{"uuid": "5faccb2e-69aa-11ee-8c99-0242ac120002", "name": "testUser3"}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_post_dict_patch__nested_API_key(self):
self.testapp.process_request(
@@ -397,7 +408,7 @@ class Test_RestAPI_POST(unittest.TestCase):
rsrc_verb.GET,
)
expected = '{"uuid": "cee1e971-65fa-11ee-8c99-0242ac120002", "shortname": "testPatch99", "name": "MyPatch", "description": "MyDescription"}'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
class Test_RestAPI_DELETE(unittest.TestCase):
@@ -411,7 +422,7 @@ class Test_RestAPI_DELETE(unittest.TestCase):
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_delete_dict_user__All(self):
result = self.testapp.process_request(
@@ -419,24 +430,24 @@ class Test_RestAPI_DELETE(unittest.TestCase):
rsrc_verb.POST,
'{"name": "testUser", "secret": "test"}',
)
self.assertEqual(result, '"e5e87d32-662b-11ee-8c99-0242ac120002"')
self.assertEqual(result.get_result(), '"e5e87d32-662b-11ee-8c99-0242ac120002"')
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = '["8da57a3c-661f-11ee-8c99-0242ac120002", "e5e87d32-662b-11ee-8c99-0242ac120002"]'
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
self.testapp.process_request("/users", rsrc_verb.DELETE)
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_delete_dict_user_element(self):
self.testapp.process_request("/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.DELETE)
result = self.testapp.process_request("/users", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_delete_nested_dict_games_patch_element(self):
self.testapp.process_request(
@@ -446,7 +457,7 @@ class Test_RestAPI_DELETE(unittest.TestCase):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_delete_nested_dict_games_patch_API_key(self):
self.testapp.process_request(
@@ -456,14 +467,14 @@ class Test_RestAPI_DELETE(unittest.TestCase):
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
def test_delete_nested_dict_games_patch_All(self):
self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.DELETE)
result = self.testapp.process_request("/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs", rsrc_verb.GET)
expected = "[]"
self.assertEqual(result, expected)
self.assertEqual(result.get_result(), expected)
class Test_RestAPI_PERFO(unittest.TestCase):
@@ -478,13 +489,13 @@ class Test_RestAPI_PERFO(unittest.TestCase):
n_loop = 10000
start = time()
for i in range(n_loop):
for _ in range(n_loop):
self.testapp.process_request(f"/users/8da57a3c-661f-11ee-8c99-0242ac120002", rsrc_verb.GET)
end = time()
print(f"GET 1st level dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
for _ in range(n_loop):
newUUID = uuid4()
self.testapp.process_request(
f"/users?API_key={newUUID}",
@@ -495,7 +506,7 @@ class Test_RestAPI_PERFO(unittest.TestCase):
print(f"POST 1st level dict (API_key): {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
for _ in range(n_loop):
newUUID = uuid4()
self.testapp.process_request(
f"/users?API_key={newUUID}",
@@ -507,14 +518,14 @@ class Test_RestAPI_PERFO(unittest.TestCase):
print(f"POST/GET 1st level dict (API_key): {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
for _ in range(n_loop):
result = self.testapp.process_request(f"/users", rsrc_verb.POST, '{"name": "testUser", "secret": "test"}')
self.testapp.process_request(f"/users/{json.loads(result)}", rsrc_verb.GET)
self.testapp.process_request(f"/users/{json.loads(result.get_result())}", rsrc_verb.GET)
end = time()
print(f"POST/GET 1st level dict (autokey): {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
for _ in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/shortname",
rsrc_verb.PUT,
@@ -525,7 +536,7 @@ class Test_RestAPI_PERFO(unittest.TestCase):
print(f"PUT/GET 1st level (value) dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
for _ in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002",
rsrc_verb.GET,
@@ -534,7 +545,7 @@ class Test_RestAPI_PERFO(unittest.TestCase):
print(f"GET 2nd level dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
for _ in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
rsrc_verb.GET,
@@ -543,7 +554,7 @@ class Test_RestAPI_PERFO(unittest.TestCase):
print(f"GET 2nd level (value) dict: {int(n_loop/(end-start))} Req/s")
start = time()
for i in range(n_loop):
for _ in range(n_loop):
self.testapp.process_request(
f"/games/9b0381d4-65f6-11ee-8c99-0242ac120002/patchs/cee1e870-65fa-11ee-8c99-0242ac120002/shortname",
rsrc_verb.PUT,

52
test/test_rest_resource_plugins.py
View File

@@ -1,16 +1,9 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from os import chdir
from pathlib import Path
from typing import Optional, Annotated
from typing import Annotated
from pydantic import Field
from uuid import UUID, uuid4
from time import time
import json
print(__name__)
print(__package__)
from src.pyrestresource import (
register_rest_rootpoint,
@@ -24,7 +17,6 @@ from src.pyrestresource import (
ResourcePlugin_field_default,
ResourcePlugin_RestResourceBase_default,
)
from pprint import pprint
testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve())
@@ -133,16 +125,16 @@ class Test_RestAPI_Plugin_PUT(unittest.TestCase):
self.testapp.process_request("/info_put/version", rsrc_verb.PUT, '"1.5.6"')
result = self.testapp.process_request("/info_put", rsrc_verb.GET)
print(result)
print(result.get_result())
result = self.testapp.process_request("/info_put/version", rsrc_verb.GET)
print(result)
self.assertEqual(result, '"42"')
print(result.get_result())
self.assertEqual(result.get_result(), '"42"')
def test_put_field_version_resourceplugin(self):
self.testapp.process_request("/info_put", rsrc_verb.PUT, '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info_put", rsrc_verb.GET)
self.assertEqual(result, '{"version": "42", "api_version": "98.321"}')
self.assertEqual(result.get_result(), '{"version": "42", "api_version": "98.321"}')
class Test_RestAPI_Plugin_GET(unittest.TestCase):
@@ -153,59 +145,59 @@ class Test_RestAPI_Plugin_GET(unittest.TestCase):
def test_get_root(self):
result = self.testapp.process_request("/", rsrc_verb.GET)
self.assertEqual(result, "{}")
self.assertEqual(result.get_result(), "{}")
def test_get_version(self):
result = self.testapp.process_request("/info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info2", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "0.0.3"}')
def test_get_version__trailing_slash(self):
result = self.testapp.process_request("/info/", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info//", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info///", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("/info2/", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "0.0.3"}')
result = self.testapp.process_request("/info2//", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "0.0.3"}')
result = self.testapp.process_request("/info2///", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "0.0.3"}')
def test_get_version__multiple_slash(self):
result = self.testapp.process_request("//info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("///info", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "98.321"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "98.321"}')
result = self.testapp.process_request("//info2", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "0.0.3"}')
result = self.testapp.process_request("///info2", rsrc_verb.GET)
self.assertEqual(result, '{"version": "1.5.6", "api_version": "0.0.3"}')
self.assertEqual(result.get_result(), '{"version": "1.5.6", "api_version": "0.0.3"}')
def test_get_version__nested_value(self):
result = self.testapp.process_request("/info/api_version", rsrc_verb.GET)
self.assertEqual(result, '"98.321"')
self.assertEqual(result.get_result(), '"98.321"')
result = self.testapp.process_request("/info/version", rsrc_verb.GET)
self.assertEqual(result, '"1.5.6"')
self.assertEqual(result.get_result(), '"1.5.6"')
result = self.testapp.process_request("/info2/api_version", rsrc_verb.GET)
self.assertEqual(result, '"0.0.3"')
self.assertEqual(result.get_result(), '"0.0.3"')
result = self.testapp.process_request("/info2/version", rsrc_verb.GET)
self.assertEqual(result, '"1.5.6"')
self.assertEqual(result.get_result(), '"1.5.6"')
def test_defect_plugin_field(self):
with self.assertRaises(RuntimeError):

4
test/test_rest_resource_walker.py
View File

@@ -1,13 +1,13 @@
from __future__ import annotations
import unittest
from typing import Optional, cast
from typing import Optional
from os import chdir
from pathlib import Path
from pydantic import Field
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr
from contextlib import redirect_stdout
print(__name__)
print(__package__)

5
test/test_rest_resource_walker_tree.py
View File

@@ -1,13 +1,12 @@
from __future__ import annotations
import unittest
from typing import Annotated, Optional
from typing import Optional
from os import chdir
from pathlib import Path
from pydantic import Field
from io import StringIO
from contextlib import redirect_stdout, redirect_stderr
print(__name__)
print(__package__)

2
test/test_rest_webserver.py
View File

@@ -117,8 +117,6 @@ def find_free_port():
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(("", 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
hostname = socket.gethostname()
IPAddr = socket.gethostbyname(hostname)
return "localhost", s.getsockname()[1]