ExternalVendorCode/trick
1
0
Fork 0
mirror of https://github.com/nasa/trick.git synced 2025-01-18 18:56:31 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Exit code fixed

Browse Source
This commit is contained in:
Caleb Herpin 2021-07-27 16:50:30 -05:00
parent aab2ed927b
commit d52350fedf
4 changed files with 112 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
share/trick/pymods/trick/conftest.py
View File

@ -6,32 +6,36 @@ 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
from utils import is_web_server_started, params, pause
# store history of failures per test class name and per index in parametrize (if parametrize used)
web_server_status = {}
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:
print(f"Building and starting sim for class {cls_name}")
build_sim()
status = is_web_server_started()
web_server_status[cls_name] = status
print(f"Web server status for {cls_name} = {status}")
if not web_server_status[cls_name]:
pytest.fail("web server is not started.")
pause("start of test")
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:
print(f"Building and starting sim for class {cls_name}")
pause("before build sim")
build_sim()
pause("here 2")
status = is_web_server_started()
web_server_status[cls_name] = status
print(f"Web server status for {cls_name} = {status}")
pause("here 1")
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
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()}'
@ -46,20 +50,28 @@ trick.itimer_enable()
trick.exec_set_enable_freeze(True)
trick.exec_set_freeze_command(True)""")
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)
if params.get_build_sim():
#TODO: Need make file to only rebuild only when necessary, otherwise, test need to rebuild and this is time consuming.
print("#"*10)
print("Auto rebuilding sim. Auto rebuild will build the SIM everytime the test is run, which can take some time.")
print("To turn auto rebuild off, in utils.py, self.__build_sim = False. Note: it's important that SIM rebuild is current.")
print("#"*10)
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():
pathToSim=params.get_path_to_sim()
if not os.path.exists(os.path.join(pathToSim, "S_main_Linux_9.3_x86_64.exe")):
raise RuntimeError(f"Sim executable does not exist in {pathToSim}. Buid this sim before running this test.")
cmd = f'echo "cd {pathToSim} && ./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()))
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()))

5
share/trick/pymods/trick/run_tests.py
View File

@ -1,5 +1,8 @@
#!/usr/bin/env python3
import pytest
import sys
if __name__ == "__main__":
pytest.main(["tests/civet_server"])
args = list(sys.argv[1:]) + ["tests/civet_server"]
print(args)
sys.exit(pytest.main(args))

1
share/trick/pymods/trick/tests/civet_server/test_ws.py
View File

@ -13,7 +13,6 @@ import ssl
sys.path.append("../..")
from utils import params, is_web_server_started
@pytest.mark.webserver
class TestWebserverWs:
if params.get_ssl_enable():

125
share/trick/pymods/trick/utils.py
View File

@ -2,77 +2,82 @@ from time import sleep
import subprocess
import os
def pause(my_str):
print("Type exit to continue:" + my_str)
os.system("/bin/bash")
# input()
#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.__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 = True
self.__start_sim = True
self.__trick_home = os.environ.get("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
#Change the following to change the default parameters
def __init__(self) -> None:
self.__port = 9000
self.__var_server_port = 9001
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 = True
self.__start_sim = True
self.__trick_home = os.environ.get("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_start_sim(self):
return self.__start_sim
def get_build_sim(self):
return self.__build_sim
def get_build_sim(self):
return self.__build_sim
def get_ssl_cert_path(self):
return self.__ssl_cert_path
def get_ssl_cert_path(self):
return self.__ssl_cert_path
def get_port(self):
return self.__port
def get_port(self):
return self.__port
def get_host(self):
if self.get_ssl_enable():
return self.__host + ".ssl"
else:
return self.__host
def get_host(self):
if self.get_ssl_enable():
return self.__host + ".ssl"
else:
return self.__host
def get_ssl_enable(self):
return self.__enable_ssl
def get_ssl_enable(self):
return self.__enable_ssl
def get_var_server_port(self):
return self.__var_server_port
def get_test_time(self):
return self.__test_time
def get_var_server_port(self):
return self.__var_server_port
def get_test_time(self):
return self.__test_time
def get_url(self, endpoint):
server_port = self.get_port()
server_host = self.get_host()
ssl_enable = self.get_ssl_enable()
base_url = f"http{ 's' if ssl_enable else '' }://{server_host}:{server_port}"
return f"{base_url}/{endpoint}"
def get_url(self, endpoint):
server_port = self.get_port()
server_host = self.get_host()
ssl_enable = self.get_ssl_enable()
base_url = f"http{ 's' if ssl_enable else '' }://{server_host}:{server_port}"
return f"{base_url}/{endpoint}"
def get_ws_url(self, endpoint):
return f"ws{ 's' if self.get_ssl_enable() else '' }://{self.get_host()}:{self.get_port()}/{endpoint}"
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(20): #Wait 2 seconds i.e 20 * .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 "LISTEN" in p.stdout.decode():
return True
return False
for _ in range(20): #Wait 2 seconds i.e 20 * .1 seconds, must wait for service to get to listening state.
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 "LISTEN" in p.stdout.decode():
return True
return False