From 85b1a09632c03672c6885233e68a00351435d51e Mon Sep 17 00:00:00 2001 From: Caleb Herpin Date: Thu, 22 Jul 2021 13:51:47 -0500 Subject: [PATCH] Updated tests --- .../trick/tests/civet_server/conftest.py | 79 ++++++++++ .../trick/tests/civet_server/pytest.ini | 3 + .../trick/tests/civet_server/test_http.py | 87 +++-------- .../trick/tests/civet_server/test_misc.py | 9 ++ .../trick/tests/civet_server/test_ws.py | 145 ++++++++++-------- .../pymods/trick/{parameters.py => utils.py} | 39 ++++- .../SIM_cannon_numeric/RUN_test/input.py | 2 +- 7 files changed, 234 insertions(+), 130 deletions(-) create mode 100644 share/trick/pymods/trick/tests/civet_server/conftest.py create mode 100644 share/trick/pymods/trick/tests/civet_server/pytest.ini create mode 100644 share/trick/pymods/trick/tests/civet_server/test_misc.py rename share/trick/pymods/trick/{parameters.py => utils.py} (50%) diff --git a/share/trick/pymods/trick/tests/civet_server/conftest.py b/share/trick/pymods/trick/tests/civet_server/conftest.py new file mode 100644 index 00000000..9122fbc2 --- /dev/null +++ b/share/trick/pymods/trick/tests/civet_server/conftest.py @@ -0,0 +1,79 @@ +import pytest +import sys +import os +from typing import Dict, Tuple +import subprocess +import inspect + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(inspect.getsourcefile(lambda:0))), '../..'))) +from utils import is_web_server_started, params + +# store history of failures per test class name and per index in parametrize (if parametrize used) +_test_failed_incremental: Dict[str, Dict[Tuple[int, ...], str]] = {} +web_server_status = {} + +# def pytest_runtest_makereport(item, call): +# if "incremental" in item.keywords: +# # incremental marker is used +# if call.excinfo is not None: +# # the test has failed +# # retrieve the class name of the test +# cls_name = str(item.cls) +# # retrieve the index of the test (if parametrize is used in combination with incremental) +# parametrize_index = ( +# tuple(item.callspec.indices.values()) +# if hasattr(item, "callspec") +# else () +# ) +# # retrieve the name of the test function +# test_name = item.originalname or item.name +# # store in _test_failed_incremental the original name of the failed test +# _test_failed_incremental.setdefault(cls_name, {}).setdefault( +# parametrize_index, test_name +# ) + + +def pytest_runtest_setup(item): + if "webserver" in item.keywords: + #retrieve the class name of the test + cls_name = str(item.cls) + status = web_server_status.get(cls_name, None) + if status == None: + build_sim() + web_server_status[cls_name] = is_web_server_started() + + if not web_server_status[cls_name]: + pytest.fail("web server is not started.") + +# @pytest.fixture(scope="session", autouse=True) + +def build_sim(): + with open(os.path.join(params.get_path_to_sim(), params.get_input_folder(), params.get_test_input_file()), "w") as f: + f.write( \ + f"""web.server.enable = True +web.server.debug = False +web.server.ssl_enable = {params.get_ssl_enable()} +web.server.path_to_ssl_cert = '{params.get_ssl_cert_path()}' +web.server.port = {params.get_port()} + +trick.var_server_set_port({params.get_var_server_port()}) + +exec(open("Modified_data/realtime.py").read()) +exec(open("Modified_data/cannon.dr").read())""") + + if params.get_build_sim(): + build_cmd = f"echo \"cd {params.get_path_to_sim()} && make -C {params.get_trick_home()}/trick_source/web/CivetServer && make clean && {params.get_trick_home()}/bin/trick-CP\" | /bin/bash" + print("....................Running:", build_cmd) + subprocess.run(build_cmd, shell=True) + + if params.get_start_sim(): + cmd = f'echo "cd {params.get_path_to_sim()} && ./S_main_Linux_9.3_x86_64.exe {os.path.join(params.get_input_folder(), params.get_test_input_file())} &" | /bin/bash' + print("....................Running:", cmd) + subprocess.run(cmd, shell=True) + +@pytest.fixture(scope="session", autouse=True) +def close_sim(): + yield + if params.get_start_sim(): + os.system("pgrep S_ | xargs kill -9") + os.remove(os.path.join(params.get_path_to_sim(), params.get_input_folder(), params.get_test_input_file())) \ No newline at end of file diff --git a/share/trick/pymods/trick/tests/civet_server/pytest.ini b/share/trick/pymods/trick/tests/civet_server/pytest.ini new file mode 100644 index 00000000..814f6b57 --- /dev/null +++ b/share/trick/pymods/trick/tests/civet_server/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +markers = + webserver: Tests relies on the webserver \ No newline at end of file diff --git a/share/trick/pymods/trick/tests/civet_server/test_http.py b/share/trick/pymods/trick/tests/civet_server/test_http.py index d4c31059..1a093818 100644 --- a/share/trick/pymods/trick/tests/civet_server/test_http.py +++ b/share/trick/pymods/trick/tests/civet_server/test_http.py @@ -13,73 +13,32 @@ from requests.api import get # TODO: Get rid of this and use automatic discovery when Trick requires Python 2.7 path.append("../..") -from parameters import Params -params = Params() +from utils import params, is_web_server_started -@pytest.fixture(scope="session", autouse=True) -def build_sim(): - trick_home = os.environ["TRICK_HOME"] - path_to_sim = os.path.join(trick_home, "trick_sims", "Cannon", "SIM_cannon_numeric") - input_folder = "RUN_test" - # test_input_file = f"tmp_input_for_test_{datetime.datetime.now()}.py" - test_input_file = f"tmp_input_for_test.py" - with open(os.path.join(path_to_sim, input_folder, test_input_file), "w") as f: - f.write( \ - f"""web.server.enable = True -web.server.debug = False -web.server.ssl_enable = {params.get_ssl_enable()} -web.server.path_to_ssl_cert = '{params.get_ssl_cert_path()}' -web.server.port = {params.get_port()} +@pytest.mark.webserver +class TestWebserverHttp: + def test_alloc_info(self): + url = params.get_url("api/http/alloc_info") + res = requests.get(url, verify=False) + data = res.json() + assert len(data["alloc_list"]) == 10 + assert data["chunk_size"] == 10 + assert data["chunk_start"] == 0 + assert data["alloc_total"] == 49 -trick.var_server_set_port({params.get_var_server_port()}) + def test_alloc_info_2(self): + endpoint = "api/http/alloc_info?start=0&count=10" + url = params.get_url(endpoint) + res = requests.get(url, verify=False) + assert len(res.json()["alloc_list"]) == 10 -exec(open("Modified_data/realtime.py").read()) -exec(open("Modified_data/cannon.dr").read())""") - - build_cmd = f"echo \"cd {path_to_sim} && make -C {trick_home}/trick_source/web/CivetServer && make clean && {trick_home}/bin/trick-CP\" | /bin/bash" - print("....................Running:", build_cmd) - subprocess.run(build_cmd, shell=True) - - cmd = f'echo "cd {path_to_sim} && ./S_main_Linux_9.3_x86_64.exe {os.path.join(input_folder, test_input_file)} &" | /bin/bash' - print("....................Running:", cmd) - subprocess.run(cmd, shell=True) - - while True: - p = subprocess.run(f"echo \"netstat -tulpan | grep {params.get_port()}\" | /bin/bash", capture_output=True, shell=True) - print(f"Checking for port output: {p.stdout}") - sleep(.1) #We sleep to use less recourses - if p.stdout != b"": - break - yield - os.system("pgrep S_ | xargs kill -9") - os.remove(os.path.join(path_to_sim, input_folder, test_input_file)) - -def pytest_collection_modifyitems(items): - for item in items: - item.add_marker(pytest.mark.webserver) - -def test_alloc_info(): - url = params.get_url("api/http/alloc_info") - res = requests.get(url, verify=False) - data = res.json() - assert len(data["alloc_list"]) == 10 - assert data["chunk_size"] == 10 - assert data["chunk_start"] == 0 - assert data["alloc_total"] == 49 - -def test_alloc_info_2(): - endpoint = "api/http/alloc_info?start=0&count=10" - url = params.get_url(endpoint) - res = requests.get(url, verify=False) - assert len(res.json()["alloc_list"]) == 10 - -def test_vs_connections(): - subprocess.Popen(f"nc localhost {params.get_var_server_port()}".split()) - sleep(.2) #Wait for the connection to persist. - endpoint = "api/http/vs_connections" - url = params.get_url(endpoint) - res = requests.get(url, verify=False) - assert res.json()["variable_server_connections"][0]["connection"]["client_IP_address"] == "127.0.0.1" + def test_vs_connections(self): + subprocess.Popen(f"nc localhost {params.get_var_server_port()}".split()) + sleep(.2) #Wait for the connection to persist. + endpoint = "api/http/vs_connections" + url = params.get_url(endpoint) + res = requests.get(url, verify=False) + assert res.json()["variable_server_connections"][0]["connection"]["client_IP_address"] == "127.0.0.1" diff --git a/share/trick/pymods/trick/tests/civet_server/test_misc.py b/share/trick/pymods/trick/tests/civet_server/test_misc.py new file mode 100644 index 00000000..ec2dd636 --- /dev/null +++ b/share/trick/pymods/trick/tests/civet_server/test_misc.py @@ -0,0 +1,9 @@ +import websockets +import pytest +import sys + +# sys.path.append("../..") +# from parameters import Params +# from test_ws import ssl_context + + diff --git a/share/trick/pymods/trick/tests/civet_server/test_ws.py b/share/trick/pymods/trick/tests/civet_server/test_ws.py index 43a65ffe..ca271a5e 100644 --- a/share/trick/pymods/trick/tests/civet_server/test_ws.py +++ b/share/trick/pymods/trick/tests/civet_server/test_ws.py @@ -5,77 +5,96 @@ import websockets import asyncio from time import sleep import datetime - +import sys +import os import pathlib import ssl +sys.path.append("../..") +from utils import params, is_web_server_started -from parameters import Params -params = Params() - -if params.get_ssl_enable(): - ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - # localhost_pem = pathlib.Path(__file__).with_name(params.get_ssl_cert_path()) - localhost_pem = params.get_ssl_cert_path() - ssl_context.load_verify_locations(localhost_pem) -else: - ssl_context = None - -@pytest.fixture(autouse=True) -def variable_server_path(): - return params.get_ws_url("api/ws/VariableServer") - -@pytest.fixture(autouse=True) -def time_path(): - return params.get_ws_url("api/ws/Time") - -@pytest.mark.asyncio -async def test_time(time_path): - if params.get_test_time(): - async with websockets.connect(time_path, ssl=ssl_context) as websocket: - await websocket.send("LOCAL") - count = 0 - while count < 2: - message = await websocket.recv() - test_format = "Time: %H:%M Date: %m/%d/%Y\n" #Not checking seconds. - time = datetime.datetime.strftime(datetime.datetime.strptime(message, "Time: %H:%M:%S Date: %m/%d/%Y\n"), test_format) - test_time = datetime.datetime.now().strftime(test_format) - print("Checking:", time, "=", test_time) - assert time == test_time - count += 1 +@pytest.mark.webserver +class TestWebserverWs: + if params.get_ssl_enable(): + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + # localhost_pem = pathlib.Path(__file__).with_name(params.get_ssl_cert_path()) + localhost_pem = params.get_ssl_cert_path() + ssl_context.load_verify_locations(localhost_pem) else: - raise RuntimeError("Parameter test_time is disabled.") + ssl_context = None -@pytest.mark.asyncio -async def test_variable_server_vars(variable_server_path): - msg1 = '{"cmd":"var_add","var_name":"dyn.cannon.pos[0]"}' - msg2 = '{ "cmd" : "var_send" }' - async with websockets.connect(variable_server_path, ssl=ssl_context) as websocket: - await websocket.send(msg1) - await websocket.send(msg2) - message = await websocket.recv() - vars = json.loads(message) - assert vars["msg_type"] == "values" - assert "time" in vars - assert len(vars["values"]) == 1 + @pytest.fixture(autouse=True, scope="session") + def variable_server_path(self): + return params.get_ws_url("api/ws/VariableServer") -@pytest.mark.asyncio -async def test_variable_server_sie(variable_server_path): - async with websockets.connect(variable_server_path, ssl=ssl_context) as websocket: - await websocket.send('{ "cmd" : "sie" }') - response = await websocket.recv() - assert response == '{ "msg_type": "sie", "data": ' + @pytest.fixture(autouse=True, scope="session") + def time_path(self): + return params.get_ws_url("api/ws/Time") + + @pytest.mark.asyncio + async def test_time(self, time_path): + if params.get_test_time(): + async with websockets.connect(time_path, ssl=TestWebserverWs.ssl_context) as websocket: + await websocket.send("LOCAL") + count = 0 + while count < 2: + message = await websocket.recv() + test_format = "Time: %H:%M Date: %m/%d/%Y\n" #Not checking seconds. + time = datetime.datetime.strftime(datetime.datetime.strptime(message, "Time: %H:%M:%S Date: %m/%d/%Y\n"), test_format) + test_time = datetime.datetime.now().strftime(test_format) + print("Checking:", time, "=", test_time) + assert time == test_time + count += 1 + else: + raise RuntimeError("Parameter test_time is disabled.") -@pytest.mark.asyncio -async def test_variable_server_units(variable_server_path): - msg1 = '{"cmd":"var_add","var_name":"dyn.cannon.pos[0]"}' - msg2 = '{ "cmd" : "units", "var_name" : "dyn.cannon.pos[0]" }' - async with websockets.connect(variable_server_path, ssl=ssl_context) as websocket: - await websocket.send(msg1) - await websocket.send(msg2) - response = await websocket.recv() - assert response == '{ "msg_type": "units", "var_name": "dyn.cannon.pos[0]", "data": "m"}' + @pytest.mark.asyncio + async def test_variable_server_vars(self, variable_server_path): + msg1 = '{"cmd":"var_add","var_name":"dyn.cannon.pos[0]"}' + msg2 = '{ "cmd" : "var_send" }' + async with websockets.connect(variable_server_path, ssl=TestWebserverWs.ssl_context) as websocket: + await websocket.send(msg1) + await websocket.send(msg2) + message = await websocket.recv() + vars = json.loads(message) + assert vars["msg_type"] == "values" + assert "time" in vars + assert len(vars["values"]) == 1 + + @pytest.mark.asyncio + async def test_variable_server_sie(self, variable_server_path): + async with websockets.connect(variable_server_path, ssl=TestWebserverWs.ssl_context) as websocket: + await websocket.send('{ "cmd" : "sie" }') + response = await websocket.recv() + assert response == '{ "msg_type": "sie", "data": ' + + @pytest.mark.asyncio + async def test_variable_server_units(self, variable_server_path): + msg1 = '{"cmd":"var_add","var_name":"dyn.cannon.pos[0]"}' + msg2 = '{ "cmd" : "units", "var_name" : "dyn.cannon.pos[0]" }' + async with websockets.connect(variable_server_path, ssl=TestWebserverWs.ssl_context) as websocket: + await websocket.send(msg1) + await websocket.send(msg2) + response = await websocket.recv() + assert response == '{ "msg_type": "units", "var_name": "dyn.cannon.pos[0]", "data": "m"}' + + @pytest.mark.asyncio + async def test_variable_server_shell_access(self, variable_server_path): + async with websockets.connect(variable_server_path, ssl=TestWebserverWs.ssl_context) as websocket: + file_to_create = "tmp_a.txt" + await websocket.send('{ "cmd" : "python", "pycode" : "print \'Hello World!---------------------\'" }') + await websocket.send('{ "cmd" : "python", "pycode" : "import os" }') + await websocket.send('{ "cmd" : "python", "pycode" : "os.system(\'touch ' + file_to_create + '\')" }') + path = os.path.join(params.get_path_to_sim(), file_to_create) + if os.path.exists(path): + os.remove(path) + warning = "This test proves that we have shell access through the websocket api." + print(warning) + assert 1 + # raise RuntimeError(warning) + else: + assert 0 if __name__ == "__main__": - asyncio.get_event_loop().run_until_complete(test_variable_server_send()) \ No newline at end of file + pass \ No newline at end of file diff --git a/share/trick/pymods/trick/parameters.py b/share/trick/pymods/trick/utils.py similarity index 50% rename from share/trick/pymods/trick/parameters.py rename to share/trick/pymods/trick/utils.py index 701c1fc3..f91700be 100644 --- a/share/trick/pymods/trick/parameters.py +++ b/share/trick/pymods/trick/utils.py @@ -1,16 +1,41 @@ +from time import sleep +import subprocess +import os + #This file contains variables for the civet_server tests class Params: #Change the following to change the default parameters def __init__(self) -> None: - self.__port = 9000 - self.__var_server_port = 9001 + self.__port = 5000 + self.__var_server_port = 5001 self.__host = "localhost" self.__enable_ssl = False self.__test_time = True # self.__ssl_cert_path = "server.pem" # self.__ssl_cert_path = "/home/cherpin/git/trick_fork/trick_sims/Cannon/SIM_cannon_numeric/server.pem" self.__ssl_cert_path = "/home/cherpin/.ssl/server.pem" + self.__build_sim = False + self.__start_sim = False + self.__trick_home = os.environ["TRICK_HOME"] + self.__path_to_sim = os.path.join(self.get_trick_home(), "trick_sims", "Cannon", "SIM_cannon_numeric") + self.__input_folder = "RUN_test" + self.__test_input_file = f"tmp_input_for_test.py" + def get_trick_home(self): + return self.__trick_home + def get_path_to_sim(self): + return self.__path_to_sim + def get_input_folder(self): + return self.__input_folder + def get_test_input_file(self): + return self.__test_input_file + + def get_start_sim(self): + return self.__start_sim + + def get_build_sim(self): + return self.__build_sim + def get_ssl_cert_path(self): return self.__ssl_cert_path @@ -38,3 +63,13 @@ class Params: def get_ws_url(self, endpoint): return f"ws{ 's' if self.get_ssl_enable() else '' }://{self.get_host()}:{self.get_port()}/{endpoint}" + +params = Params() +def is_web_server_started(): + for _ in range(10): #Wait 10 seconds i.e 50 * .1 seconds + p = subprocess.run(f"echo \"netstat -tulpan | grep {params.get_port()}\" | /bin/bash", capture_output=True, shell=True) + # print(f"Checking for port output: {p.stdout}") + sleep(.1) #We sleep to use less recourses + if p.stdout != b"": + return True + return False \ No newline at end of file diff --git a/trick_sims/Cannon/SIM_cannon_numeric/RUN_test/input.py b/trick_sims/Cannon/SIM_cannon_numeric/RUN_test/input.py index 131b7c4a..5535d59d 100644 --- a/trick_sims/Cannon/SIM_cannon_numeric/RUN_test/input.py +++ b/trick_sims/Cannon/SIM_cannon_numeric/RUN_test/input.py @@ -5,7 +5,7 @@ web.server.path_to_ssl_cert = '/home/cherpin/.ssl/server.pem' web.server.path_to_ssl_cert = "server.pem" web.server.port = 5000 -trick.var_server_set_port(7000) +trick.var_server_set_port(5001) exec(open("Modified_data/realtime.py").read()) exec(open("Modified_data/cannon.dr").read())