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

Updated tests

Browse Source
This commit is contained in:
Caleb Herpin 2021-07-22 13:51:47 -05:00
parent f9feac77c4
commit 85b1a09632
7 changed files with 234 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
share/trick/pymods/trick/tests/civet_server/conftest.py Normal file
View File

@ -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()))

3
share/trick/pymods/trick/tests/civet_server/pytest.ini Normal file
View File

@ -0,0 +1,3 @@
[pytest]
markers =
webserver: Tests relies on the webserver

87
share/trick/pymods/trick/tests/civet_server/test_http.py
View File

@ -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"

9
share/trick/pymods/trick/tests/civet_server/test_misc.py Normal file
View File

@ -0,0 +1,9 @@
import websockets
import pytest
import sys
# sys.path.append("../..")
# from parameters import Params
# from test_ws import ssl_context

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

@ -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())
pass

39
share/trick/pymods/trick/parameters.py → share/trick/pymods/trick/utils.py
View File

@ -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

2
trick_sims/Cannon/SIM_cannon_numeric/RUN_test/input.py
View File

@ -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())