diff --git a/share/trick/pymods/trick/conftest.py b/share/trick/pymods/trick/conftest.py index d18458fb..532ff238 100644 --- a/share/trick/pymods/trick/conftest.py +++ b/share/trick/pymods/trick/conftest.py @@ -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())) \ No newline at end of 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())) \ No newline at end of file diff --git a/share/trick/pymods/trick/run_tests.py b/share/trick/pymods/trick/run_tests.py index 400abfb8..3fbaf975 100755 --- a/share/trick/pymods/trick/run_tests.py +++ b/share/trick/pymods/trick/run_tests.py @@ -1,5 +1,8 @@ #!/usr/bin/env python3 import pytest +import sys if __name__ == "__main__": - pytest.main(["tests/civet_server"]) \ No newline at end of file + args = list(sys.argv[1:]) + ["tests/civet_server"] + print(args) + sys.exit(pytest.main(args)) \ No newline at end of file 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 56f6bb6a..3c229193 100644 --- a/share/trick/pymods/trick/tests/civet_server/test_ws.py +++ b/share/trick/pymods/trick/tests/civet_server/test_ws.py @@ -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(): diff --git a/share/trick/pymods/trick/utils.py b/share/trick/pymods/trick/utils.py index 9bbf0691..6479a007 100644 --- a/share/trick/pymods/trick/utils.py +++ b/share/trick/pymods/trick/utils.py @@ -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