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

finish 1st login version + clean code + fix unittest regressions

Browse Source
This commit is contained in:
cclecle
2023-11-05 17:44:15 +00:00
parent f00cf7b4b2
commit cffa209c9a
16 changed files with 463 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files

251
test/test_rest_login.py
View File

@@ -12,6 +12,7 @@ import socket
import requests
from contextlib import closing
from multiprocessing import Process
from requests.adapters import HTTPAdapter
print(__name__)
print(__package__)
@@ -43,20 +44,30 @@ chdir(testdir_path.parent.resolve())
# to allow mock-ing, all the tested classes are in a function
def init_classes():
user_CHACHA = UserLogin(username="chacha", secret="123456")
user_test = UserLogin(username="TestUser", secret="123456")
class TestResource(RestResourceBase):
test_field: Optional[str] = Field("ORIGIN_VALUE")
class TestResourceACL(RestResourceBase):
test_field: Optional[str] = Field(
"ORIGIN_VALUE",
ACL=[
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_user(name="chacha"), rule=ACL_rule.ALLOW),
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_user.from_user_login(user_test), rule=ACL_rule.ALLOW),
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
],
)
@register_rest_rootpoint
class RootApp(RestResourceBaseLogin):
_ar_user_login: ClassVar[list[UserLogin]] = [user_CHACHA]
_ar_user_login: ClassVar[list[UserLogin]] = [user_test]
test_resourceACL: TestResource = Field(
TestResource(),
ACL=[
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_user(name=user_test.username), rule=ACL_rule.ALLOW),
ACL_record(verbs=[rsrc_verb.PUT], target=ACL_target_group_Any(), rule=ACL_rule.DENY),
],
)
test_resource: TestResourceACL = TestResourceACL()
# this add the classes to globals to allow using them later on
@@ -73,21 +84,16 @@ def find_free_port():
def launch_server(ip, port):
print(f"port2={port}")
init_classes()
uvicorn.run(f"{__loader__.name}:RootApp", port=port, host="0.0.0.0", log_level="warning", factory=True)
class Test_RestAPI_LOGIN(unittest.TestCase):
class Test_RestAPI_LOGIN_Web(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
init_classes()
self.testapp = RootApp()
def test_access(self):
def test_login(self):
ip, port = find_free_port()
print(f"ip1={ip}")
print(f"port1={port}")
proc = Process(
target=launch_server,
args=(
@@ -98,6 +104,139 @@ class Test_RestAPI_LOGIN(unittest.TestCase):
proc.start()
sleep(1)
s = requests.Session()
s.mount("http://", HTTPAdapter(max_retries=0))
try:
# read full login resource
response = s.get(
f"http://{ip}:{port}/login",
)
self.assertEqual(response.status_code, 200)
self.assertDictEqual(response.json(), {"username": "__ANNONYMOUS__"})
# read login username field
response = s.get(
f"http://{ip}:{port}/login/username",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "__ANNONYMOUS__")
# login
response = s.put(
f"http://{ip}:{port}/login",
json={"username": "TestUser", "secret": "123456"},
)
self.assertEqual(response.status_code, 201)
# read full login resource
response = s.get(
f"http://{ip}:{port}/login",
)
self.assertEqual(response.status_code, 200)
self.assertDictEqual(response.json(), {"username": "TestUser"})
# read login username field
response = s.get(
f"http://{ip}:{port}/login/username",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "TestUser")
finally:
proc.terminate()
s.close()
def test_access_resourceACL(self):
ip, port = find_free_port()
proc = Process(
target=launch_server,
args=(
ip,
port,
),
)
proc.start()
sleep(1)
s = requests.Session()
s.mount("http://", HTTPAdapter(max_retries=0))
try:
# before modification read
response = s.get(
f"http://{ip}:{port}/test_resourceACL/test_field",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "ORIGIN_VALUE")
# try unauthenticated write (to field)
response = s.put(f"http://{ip}:{port}/test_resourceACL/test_field", json="TEST SET VALUE")
self.assertEqual(response.status_code, 500)
# check not modified
response = s.get(
f"http://{ip}:{port}/test_resourceACL/test_field",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "ORIGIN_VALUE")
# try unauthenticated write (to resource)
response = s.put(f"http://{ip}:{port}/test_resourceACL", json={"test_field": "TEST SET VALUE"})
self.assertEqual(response.status_code, 500)
# check not modified
response = s.get(
f"http://{ip}:{port}/test_resourceACL/test_field",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "ORIGIN_VALUE")
# login
response = s.put(
f"http://{ip}:{port}/login",
json={"username": "TestUser", "secret": "123456"},
)
self.assertEqual(response.status_code, 201)
# authenticated write (to field)
response = s.put(f"http://{ip}:{port}/test_resourceACL/test_field", json="TEST SET VALUE")
self.assertEqual(response.status_code, 201)
# modified
response = s.get(
f"http://{ip}:{port}/test_resourceACL/test_field",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "TEST SET VALUE")
# authenticated write (to resource)
response = s.put(f"http://{ip}:{port}/test_resourceACL", json={"test_field": "TEST SET VALUE 2"})
self.assertEqual(response.status_code, 201)
# modified
response = s.get(
f"http://{ip}:{port}/test_resourceACL/test_field",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "TEST SET VALUE 2")
finally:
proc.terminate()
s.close()
def test_access_fieldACL(self):
ip, port = find_free_port()
proc = Process(
target=launch_server,
args=(
ip,
port,
),
)
proc.start()
sleep(1)
s = requests.Session()
s.mount("http://", HTTPAdapter(max_retries=0))
try:
# before modification read
response = s.get(
@@ -106,8 +245,19 @@ class Test_RestAPI_LOGIN(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "ORIGIN_VALUE")
# try unauthenticated write
response = s.put(f"http://{ip}:{port}/test_resource/test_field", json='"TEST SET VALUE"')
# try unauthenticated write (to field)
response = s.put(f"http://{ip}:{port}/test_resource/test_field", json="TEST SET VALUE")
self.assertEqual(response.status_code, 500)
# check not modified
response = s.get(
f"http://{ip}:{port}/test_resource/test_field",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "ORIGIN_VALUE")
# try unauthenticated write (to resource)
response = s.put(f"http://{ip}:{port}/test_resource", json={"test_field": "TEST SET VALUE"})
self.assertEqual(response.status_code, 500)
# check not modified
@@ -120,11 +270,11 @@ class Test_RestAPI_LOGIN(unittest.TestCase):
# login
response = s.put(
f"http://{ip}:{port}/login",
json={"username": "chacha", "secret": "123456"},
json={"username": "TestUser", "secret": "123456"},
)
self.assertEqual(response.status_code, 201)
# authenticated write
# authenticated write (to field)
response = s.put(f"http://{ip}:{port}/test_resource/test_field", json="TEST SET VALUE")
self.assertEqual(response.status_code, 201)
@@ -135,73 +285,16 @@ class Test_RestAPI_LOGIN(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "TEST SET VALUE")
finally:
proc.terminate()
s.close()
def test_login(self):
result = self.testapp.process_request("/login", rsrc_verb.GET)
print("*****************")
print(result.get_result())
result = self.testapp.process_request("/login/username", rsrc_verb.GET)
print("*****************")
print(result.get_result())
# 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())
result = self.testapp.process_request("/login/username", rsrc_verb.GET)
print("*****************")
print(result.get_result())
# result = self.testapp.process_request("/login/secret", rsrc_verb.GET)
# print("*****************")
# print(result.get_result())
class Test_RestAPI_LOGIN_Web(unittest.TestCase):
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
def test_login(self):
ip, port = find_free_port()
print(f"ip1={ip}")
print(f"port1={port}")
proc = Process(
target=launch_server,
args=(
ip,
port,
),
)
proc.start()
sleep(1)
s = requests.Session()
try:
# Login in
response = s.put(
f"http://{ip}:{port}/login",
json={"username": "chacha", "secret": "123456"},
)
print(response)
print("??????")
print(response.headers)
# authenticated write (to resource)
response = s.put(f"http://{ip}:{port}/test_resource", json={"test_field": "TEST SET VALUE 2"})
self.assertEqual(response.status_code, 201)
response = s.get(f"http://{ip}:{port}/login")
response = s.get(f"http://{ip}:{port}/")
# modified
response = s.get(
f"http://{ip}:{port}/test_resource/test_field",
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), "TEST SET VALUE 2")
finally:
proc.terminate()

44
test/test_rest_resource_plugins.py
View File

@@ -16,6 +16,7 @@ from src.pyrestresource import (
T_SupportedRESTFields,
ResourcePlugin_field_default,
ResourcePlugin_RestResourceBase_default,
RestResourcePluginException_InvalidPluginSignature,
)
testdir_path = Path(__file__).parent.resolve()
@@ -34,6 +35,7 @@ def init_classes():
class ResourcePlugin_Info(ResourcePlugin_RestResourceBase_default):
def handle_resource_get(self, resource: Info_get, params: RestRequestParams_GET) -> Info_get:
print("HOOK GET !!")
return Info_get(version="65.45", api_version="98.321")
class Info_get(RestResourceBase):
@@ -69,41 +71,9 @@ def init_classes():
def init_bad_plugin1():
# plugin with missing handle_resource_put() method
# plugin not inheriting from the right base type
class ResourcePlugin_TestResource:
def handle_field_get(self, resource: TestResource, params: RestRequestParams_GET) -> TestResource:
return resource
class TestResource(RestResourceBase):
tetvaluestr: Annotated[str, Field(plugin=ResourcePlugin_TestResource)]
@register_rest_rootpoint
class RootApp2(RestResourceBase):
test: TestResource = Field(default=TestResource(tetvaluestr="testvalue"))
RootApp2()
def init_bad_plugin2():
# plugin with missing handle_resource_get() method
class ResourcePlugin_TestResource:
def handle_field_put(self, resource: TestResource, params: RestRequestParams_PUT) -> TestResource:
return resource
class TestResource(RestResourceBase):
tetvaluestr: Annotated[str, Field(plugin=ResourcePlugin_TestResource)]
@register_rest_rootpoint
class RootApp2(RestResourceBase):
test: TestResource = Field(default=TestResource(tetvaluestr="testvalue"))
RootApp2()
def init_bad_plugin3():
# wrong plugin
class ResourcePlugin_TestResource(ResourcePlugin_RestResourceBase_default):
pass
...
class TestResource(RestResourceBase):
tetvaluestr: Annotated[str, Field(plugin=ResourcePlugin_TestResource)]
@@ -200,9 +170,5 @@ class Test_RestAPI_Plugin_GET(unittest.TestCase):
self.assertEqual(result.get_result(), '"1.5.6"')
def test_defect_plugin_field(self):
with self.assertRaises(RuntimeError):
with self.assertRaises(RestResourcePluginException_InvalidPluginSignature):
init_bad_plugin1()
with self.assertRaises(RuntimeError):
init_bad_plugin2()
with self.assertRaises(RuntimeError):
init_bad_plugin3()

11
test/test_rest_webserver.py
View File

@@ -13,7 +13,7 @@ import socket
import requests
from contextlib import closing
from multiprocessing import Process
from requests.adapters import HTTPAdapter
print(__name__)
print(__package__)
@@ -121,7 +121,6 @@ def find_free_port():
def launch_server(ip, port):
print(f"port2={port}")
init_classes()
uvicorn.run(f"{__loader__.name}:RootApp", port=port, host="0.0.0.0", log_level="warning", factory=True)
@@ -132,8 +131,6 @@ class Test_RestAPI_WebServer(unittest.TestCase):
def test_nomal_AllCmd_games(self):
ip, port = find_free_port()
print(f"ip1={ip}")
print(f"port1={port}")
proc = Process(
target=launch_server,
args=(
@@ -144,6 +141,8 @@ class Test_RestAPI_WebServer(unittest.TestCase):
proc.start()
sleep(1)
s = requests.Session()
s.mount("http://", HTTPAdapter(max_retries=0))
try:
# Fetching games
response = s.get(f"http://{ip}:{port}/games")
@@ -291,8 +290,6 @@ class Test_RestAPI_WebServer(unittest.TestCase):
n_loop = 10000
ip, port = find_free_port()
print(f"ip1={ip}")
print(f"port1={port}")
proc = Process(
target=launch_server,
args=(
@@ -303,6 +300,8 @@ class Test_RestAPI_WebServer(unittest.TestCase):
proc.start()
sleep(1)
s = requests.Session()
s.mount("http://", HTTPAdapter(max_retries=0))
try:
start = time()
for _ in range(n_loop):