179 lines
8.1 KiB
Python
179 lines
8.1 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,
|
|
RestResourceHandlerException_Forbiden,
|
|
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] = RestField(None)
|
|
secret: Optional[str] = RestField(
|
|
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] = RestField(
|
|
"1.2.3",
|
|
ACL=[
|
|
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
|
|
],
|
|
)
|
|
version: Optional[str] = RestField("3.2.1")
|
|
|
|
@register_rest_rootpoint
|
|
class RootApp(RestResourceBase):
|
|
resource_with_secret: TestResource = RestField(default=TestResource())
|
|
resource_with_secret_ACL: TestResource = RestField(
|
|
default=TestResource(), ACL=[ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.DENY)]
|
|
)
|
|
resource_ro: TestResource2 = RestField(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("/resource_ro", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '{"version_ro": "1.2.3", "version": "3.2.1"}')
|
|
|
|
self.testapp.process_request("/resource_ro/version", rsrc_verb.PUT, '"6.6.6"')
|
|
|
|
result = self.testapp.process_request("/resource_ro", rsrc_verb.GET)
|
|
self.assertEqual(result.get_result(), '{"version_ro": "1.2.3", "version": "6.6.6"}')
|
|
|
|
with self.assertRaises(RestResourceHandlerException_Forbiden): # TODO: custom exception
|
|
self.testapp.process_request("/resource_ro/version_ro", rsrc_verb.PUT, '"6.6.6"')
|
|
self.assertEqual(self.testapp.resource_ro.version_ro, "1.2.3")
|
|
|
|
with self.assertRaises(RestResourceHandlerException_Forbiden): # TODO: custom exception
|
|
self.testapp.process_request("/resource_ro", rsrc_verb.PUT, '{"version_ro": "6.6.1", "version": "6.6.2"}')
|
|
self.assertEqual(self.testapp.resource_ro.version_ro, "1.2.3")
|
|
|
|
result = self.testapp.process_request("/resource_ro", 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(RestResourceHandlerException_Forbiden):
|
|
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(RestResourceHandlerException_Forbiden):
|
|
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(RestResourceHandlerException_Forbiden): # 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(RestResourceHandlerException_Forbiden): # 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(RestResourceHandlerException_Forbiden): # 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(RestResourceHandlerException_Forbiden): # 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(RestResourceHandlerException_Forbiden): # 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(RestResourceHandlerException_Forbiden): # 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)
|