diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml index 533343870..0556a2189 100644 --- a/.github/workflows/testing.yml +++ b/.github/workflows/testing.yml @@ -30,7 +30,7 @@ jobs: permissions: {} name: API runs-on: ubuntu-20.04 - timeout-minutes: 40 + timeout-minutes: 60 steps: - name: Checkout source uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1 diff --git a/framework/python/src/api/api.py b/framework/python/src/api/api.py index f48632b87..f97e564d3 100644 --- a/framework/python/src/api/api.py +++ b/framework/python/src/api/api.py @@ -253,8 +253,14 @@ def _generate_msg(self, success, message): def _start_test_run(self): self._test_run.start() - async def stop_test_run(self): - LOGGER.debug("Received stop command. Stopping Testrun") + async def stop_test_run(self, response: Response): + LOGGER.debug("Received stop command") + + # Check if Testrun is running + if (self._test_run.get_session().get_status() not in + ["In Progress", "Waiting for Device", "Monitoring"]): + response.status_code = 404 + return self._generate_msg(False, "Testrun is not currently running") self._test_run.stop() diff --git a/framework/python/src/common/testreport.py b/framework/python/src/common/testreport.py index e2a187cce..17fcc5a0f 100644 --- a/framework/python/src/common/testreport.py +++ b/framework/python/src/common/testreport.py @@ -63,9 +63,6 @@ def __init__(self, def get_mac_addr(self): return self._mac_addr - def get_mac_addr(self): - return self._mac_addr - def add_module_reports(self, module_reports): self._module_reports = module_reports @@ -93,12 +90,6 @@ def set_report_url(self, url): def get_report_url(self): return self._report_url - - def set_mac_addr(self, mac_addr): - self._mac_addr = mac_addr - - def set_mac_addr(self, mac_addr): - self._mac_addr = mac_addr def set_mac_addr(self, mac_addr): self._mac_addr = mac_addr diff --git a/make/DEBIAN/prerm b/make/DEBIAN/prerm index 95f226749..e1817d479 100755 --- a/make/DEBIAN/prerm +++ b/make/DEBIAN/prerm @@ -30,5 +30,5 @@ docker_images=$(sudo docker images --filter=reference="test-run/*" -q) if [ -z "$docker_images" ]; then echo No docker images to delete else - sudo docker rmi $docker_images > /dev/null + sudo docker rmi -f $docker_images > /dev/null fi \ No newline at end of file diff --git a/modules/test/tls/python/src/tls_module.py b/modules/test/tls/python/src/tls_module.py index 353fc8777..0994d5b22 100644 --- a/modules/test/tls/python/src/tls_module.py +++ b/modules/test/tls/python/src/tls_module.py @@ -119,7 +119,7 @@ def __init__(self, # # cert_table = (f'| Property | Value |\n' # # f'|---|---|\n' # # f"| {'Version':<17} | {version_value:^25} |\n" - # # f"| {'Signature Alg.':<17} | + # # f"| {'Signature Alg.':<17} | # {signature_alg_value:^25} |\n" # # f"| {'Validity from':<17} | {not_before:^25} |\n" # # f"| {'Valid to':<17} | {not_after:^25} |") diff --git a/testing/api/mockito/invalid_request.json b/testing/api/mockito/invalid_request.json index 5104263fd..2f62b4f04 100644 --- a/testing/api/mockito/invalid_request.json +++ b/testing/api/mockito/invalid_request.json @@ -1,3 +1,3 @@ { - "error": "Invalid request received" + "detail": "Not Found" } \ No newline at end of file diff --git a/testing/api/test_api b/testing/api/test_api index d86ff42a9..6751ae0ad 100755 --- a/testing/api/test_api +++ b/testing/api/test_api @@ -31,8 +31,8 @@ sudo docker network create -d macvlan -o parent=endev0b endev0 # Start OVS sudo /usr/share/openvswitch/scripts/ovs-ctl start -# Build Test Container -sudo docker build ./testing/docker/ci_test_device1 -t ci_test_device1 -f ./testing/docker/ci_test_device1/Dockerfile +# Build test container +sudo docker build ./testing/docker/ci_test_device1 -t test-run/ci_device_1 -f ./testing/docker/ci_test_device1/Dockerfile sudo chown -R $USER local diff --git a/testing/api/test_api.py b/testing/api/test_api.py index 871a016d5..8c9a78986 100644 --- a/testing/api/test_api.py +++ b/testing/api/test_api.py @@ -13,18 +13,15 @@ # limitations under the License. """Test assertions for CI network baseline test""" -# Temporarily disabled because using Pytest fixtures -# TODO refactor fixtures to not trigger error # pylint: disable=redefined-outer-name -from collections.abc import Awaitable, Callable +from collections.abc import Callable import copy import json import os from pathlib import Path import re import shutil -import shutil import signal import subprocess import time @@ -44,7 +41,6 @@ BASELINE_MAC_ADDR = "02:42:aa:00:01:01" ALL_MAC_ADDR = "02:42:aa:00:00:01" - def pretty_print(dictionary: dict): """ Pretty print dictionary """ print(json.dumps(dictionary, indent=4)) @@ -52,20 +48,20 @@ def pretty_print(dictionary: dict): def query_system_status() -> str: """Query system status from API and returns this""" - r = requests.get(f"{API}/system/status") + r = requests.get(f"{API}/system/status", timeout=5) response = json.loads(r.text) return response["status"] def query_test_count() -> int: """Queries status and returns number of test results""" - r = requests.get(f"{API}/system/status") + r = requests.get(f"{API}/system/status", timeout=5) response = json.loads(r.text) return len(response["tests"]["results"]) def start_test_device( - device_name, mac_address, image_name="ci_test_device1", args="" + device_name, mac_address, image_name="test-run/ci_device_1", args="" ): """ Start test device container with given name """ cmd = subprocess.run( @@ -82,11 +78,13 @@ def start_test_device( def stop_test_device(device_name): """ Stop docker container with given name """ cmd = subprocess.run( - f"docker stop {device_name}", shell=True, capture_output=True + f"docker stop {device_name}", shell=True, capture_output=True, + check=False ) print(cmd.stdout) cmd = subprocess.run( - f"docker rm {device_name}", shell=True, capture_output=True + f"docker rm {device_name}", shell=True, capture_output=True, + check=False ) print(cmd.stdout) @@ -94,7 +92,8 @@ def stop_test_device(device_name): def docker_logs(device_name): """ Print docker logs from given docker container name """ cmd = subprocess.run( - f"docker logs {device_name}", shell=True, capture_output=True + f"docker logs {device_name}", shell=True, capture_output=True, + check=False ) print(cmd.stdout) @@ -118,27 +117,24 @@ def testing_devices(): @pytest.fixture -def testrun(request): +def testrun(request): # pylint: disable=W0613 """ Start intstance of testrun """ - test_name = request.node.originalname proc = subprocess.Popen( "bin/testrun", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8", - preexec_fn=os.setsid, + preexec_fn=os.setsid ) while True: try: - outs, errs = proc.communicate(timeout=1) + outs = proc.communicate(timeout=1)[0] except subprocess.TimeoutExpired as e: if e.output is not None: output = e.output.decode("utf-8") if re.search("API waiting for requests", output): break - except Exception as e: - pytest.fail("testrun terminated") time.sleep(2) @@ -146,8 +142,8 @@ def testrun(request): os.killpg(os.getpgid(proc.pid), signal.SIGTERM) try: - outs, errs = proc.communicate(timeout=60) - except Exception as e: + outs = proc.communicate(timeout=60)[0] + except subprocess.TimeoutExpired as e: print(e.output) os.killpg(os.getpgid(proc.pid), signal.SIGKILL) pytest.exit( @@ -157,11 +153,13 @@ def testrun(request): print(outs) cmd = subprocess.run( - f"docker stop $(docker ps -a -q)", shell=True, capture_output=True + "docker stop $(docker ps -a -q)", shell=True, + capture_output=True, check=False ) print(cmd.stdout) cmd = subprocess.run( - f"docker rm $(docker ps -a -q)", shell=True, capture_output=True + "docker rm $(docker ps -a -q)", shell=True, + capture_output=True, check=False ) print(cmd.stdout) @@ -177,7 +175,7 @@ def until_true(func: Callable, message: str, timeout: int): if func(): return True time.sleep(1) - raise Exception(f"Timed out waiting {timeout}s for {message}") + raise TimeoutError(f"Timed out waiting {timeout}s for {message}") def dict_paths(thing: dict, stem: str = "") -> Iterator[str]: @@ -224,9 +222,9 @@ def local_get_devices(): ) -def test_get_system_interfaces(testrun): +def test_get_system_interfaces(testrun): # pylint: disable=W0613 """Tests API system interfaces against actual local interfaces""" - r = requests.get(f"{API}/system/interfaces") + r = requests.get(f"{API}/system/interfaces", timeout=5) response = json.loads(r.text) local_interfaces = get_network_interfaces() assert set(response.keys()) == set(local_interfaces) @@ -235,14 +233,460 @@ def test_get_system_interfaces(testrun): assert all([isinstance(x, str) for x in response]) -def test_modify_device(testing_devices, testrun): - with open(testing_devices[1], encoding="utf-8") as f: +def test_status_idle(testrun): + until_true( + lambda: query_system_status().lower() == "idle", + "system status is `idle`", + 30, + ) + +# Currently not working due to blocking during monitoring period +@pytest.mark.skip() +def test_status_in_progress(testing_devices, testrun): + + payload = {"device": {"mac_addr": BASELINE_MAC_ADDR, "firmware": "asd"}} + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + assert r.status_code == 200 + + until_true( + lambda: query_system_status().lower() == "waiting for device", + "system status is `waiting for device`", + 30, + ) + + start_test_device("x123", BASELINE_MAC_ADDR) + + until_true( + lambda: query_system_status().lower() == "in progress", + "system status is `in progress`", + 600, + ) + + +@pytest.mark.skip() +def test_status_non_compliant(testing_devices, testrun): # pylint: disable=W0613 + + r = requests.get(f"{API}/devices", timeout=5) + all_devices = json.loads(r.text) + payload = { + "device": { + "mac_addr": all_devices[0]["mac_addr"], + "firmware": "asd" + } + } + r = requests.post(f"{API}/system/start", data=json.dumps(payload), + timeout=10) + assert r.status_code == 200 + print(r.text) + + until_true( + lambda: query_system_status().lower() == "waiting for device", + "system status is `waiting for device`", + 30, + ) + + start_test_device("x123", all_devices[0]["mac_addr"]) + + until_true( + lambda: query_system_status().lower() == "non-compliant", + "system status is `complete", + 600, + ) + + stop_test_device("x123") + +def test_create_get_devices(empty_devices_dir, testrun): # pylint: disable=W0613 + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) + print(r.text) + assert r.status_code == 201 + assert len(local_get_devices()) == 1 + + device_2 = { + "manufacturer": "Google", + "model": "Second", + "mac_addr": "00:1e:42:35:73:c6", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + r = requests.post(f"{API}/device", data=json.dumps(device_2), + timeout=5) + assert r.status_code == 201 + assert len(local_get_devices()) == 2 + + # Test that returned devices API endpoint matches expected structure + r = requests.get(f"{API}/devices", timeout=5) + all_devices = json.loads(r.text) + pretty_print(all_devices) + + with open( + os.path.join(os.path.dirname(__file__), "mockito/get_devices.json"), + encoding="utf-8" + ) as f: + mockito = json.load(f) + + print(mockito) + + # Validate structure + assert all([isinstance(x, dict) for x in all_devices]) + + # TOOO uncomment when is done + # assert set(dict_paths(mockito[0])) == set(dict_paths(all_devices[0])) + + # Validate contents of given keys matches + for key in ["mac_addr", "manufacturer", "model"]: + assert set([all_devices[0][key], all_devices[1][key]]) == set( + [device_1[key], device_2[key]] + ) + + +def test_delete_device_success(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + # Send create device request + r = requests.post(f"{API}/device", data=json.dumps(device_1)) + print(r.text) + device1_response = r.text + + # Check device has been created + assert r.status_code == 201 + assert len(local_get_devices()) == 1 + + device_2 = { + "manufacturer": "Google", + "model": "Second", + "mac_addr": "00:1e:42:35:73:c6", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + r = requests.post(f"{API}/device", data=json.dumps(device_2)) + device2_response = json.loads(r.text) + assert r.status_code == 201 + assert len(local_get_devices()) == 2 + + + # Test that device_1 deletes + r = requests.delete(f"{API}/device/",data=json.dumps(device_1)) + assert r.status_code == 200 + assert len(local_get_devices()) == 1 + + + # Test that returned devices API endpoint matches expected structure + r = requests.get(f"{API}/devices") + all_devices = json.loads(r.text) + pretty_print(all_devices) + + with open( + os.path.join(os.path.dirname(__file__), "mockito/get_devices.json") + ) as f: + mockito = json.load(f) + + print(mockito) + + # Validate structure + assert all([isinstance(x, dict) for x in all_devices]) + + # TOOO uncomment when is done + # assert set(dict_paths(mockito[0])) == set(dict_paths(all_devices[0])) + + # Validate contents of given keys matches + for key in ["mac_addr", "manufacturer", "model"]: + assert set([all_devices[0][key]]) == set( + [device_2[key]] + ) + + +def test_delete_device_not_found(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + # Send create device request + r = requests.post(f"{API}/device", data=json.dumps(device_1)) + print(r.text) + device1_response = r.text + + # Check device has been created + assert r.status_code == 201 + assert len(local_get_devices()) == 1 + + # Test that device_1 deletes + r = requests.delete(f"{API}/device/",data=json.dumps(device_1)) + assert r.status_code == 200 + assert len(local_get_devices()) == 0 + + # Test that device_1 is not found + r = requests.delete(f"{API}/device/",data=json.dumps(device_1)) + assert r.status_code == 404 + assert len(local_get_devices()) == 0 + + + +def test_delete_device_no_mac(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + # Send create device request + r = requests.post(f"{API}/device", data=json.dumps(device_1)) + print(r.text) + + # Check device has been created + assert r.status_code == 201 + assert len(local_get_devices()) == 1 + + device_1.pop("mac_addr") + + # Test that device_1 can't delete with no mac address + r = requests.delete(f"{API}/device/",data=json.dumps(device_1)) + assert r.status_code == 400 + assert len(local_get_devices()) == 1 + + +# Currently not working due to blocking during monitoring period +@pytest.mark.skip() +def test_delete_device_testrun_running(testing_devices, testrun): + + payload = {"device": {"mac_addr": BASELINE_MAC_ADDR, "firmware": "asd"}} + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + assert r.status_code == 200 + + until_true( + lambda: query_system_status().lower() == "waiting for device", + "system status is `waiting for device`", + 30, + ) + + start_test_device("x123", BASELINE_MAC_ADDR) + + until_true( + lambda: query_system_status().lower() == "in progress", + "system status is `in progress`", + 600, + ) + + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": BASELINE_MAC_ADDR, + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + r = requests.delete(f"{API}/device/",data=json.dumps(device_1)) + assert r.status_code == 403 + + +def test_start_testrun_started_successfully(testing_devices, testrun): + payload = {"device": {"mac_addr": BASELINE_MAC_ADDR, "firmware": "asd"}} + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + assert r.status_code == 200 + + +# Currently not working due to blocking during monitoring period +@pytest.mark.skip() +def test_start_testrun_already_in_progress(testing_devices, testrun): + payload = {"device": {"mac_addr": BASELINE_MAC_ADDR, "firmware": "asd"}} + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + + until_true( + lambda: query_system_status().lower() == "waiting for device", + "system status is `waiting for device`", + 30, + ) + + start_test_device("x123", BASELINE_MAC_ADDR) + + until_true( + lambda: query_system_status().lower() == "in progress", + "system status is `in progress`", + 600, + ) + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + assert r.status_code == 409 + +def test_start_system_not_configured_correctly(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + # Send create device request + r = requests.post(f"{API}/device", data=json.dumps(device_1)) + print(r.text) + + payload = {"device": {"mac_addr": None, "firmware": "asd"}} + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + assert r.status_code == 500 + + +def test_start_device_not_found(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + # Send create device request + r = requests.post(f"{API}/device", data=json.dumps(device_1)) + print(r.text) + + r = requests.delete(f"{API}/device/",data=json.dumps(device_1)) + assert r.status_code == 200 + + payload = {"device": {"mac_addr": device_1["mac_addr"], "firmware": "asd"}} + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + assert r.status_code == 404 + + +def test_start_missing_device_information(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + # Send create device request + r = requests.post(f"{API}/device", data=json.dumps(device_1)) + print(r.text) + + payload = {} + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) + assert r.status_code == 400 + + +def test_create_device_already_exists(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) + print(r.text) + assert r.status_code == 201 + assert len(local_get_devices()) == 1 + + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) + print(r.text) + assert r.status_code == 409 + + +def test_create_device_invalid_json(empty_devices_dir, testrun): + device_1 = { + } + + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) + print(r.text) + assert r.status_code == 400 + + +def test_create_device_invalid_request(empty_devices_dir, testrun): + + r = requests.post(f"{API}/device", data=None, + timeout=5) + print(r.text) + assert r.status_code == 400 + + +def test_device_edit_device(testing_devices, testrun): # pylint: disable=W0613 + with open( + testing_devices[1], encoding="utf-8" + ) as f: local_device = json.load(f) mac_addr = local_device["mac_addr"] new_model = "Alphabet" - r = requests.get(f"{API}/devices") + r = requests.get(f"{API}/devices", timeout=5) all_devices = json.loads(r.text) api_device = next(x for x in all_devices if x["mac_addr"] == mac_addr) @@ -267,11 +711,12 @@ def test_modify_device(testing_devices, testrun): # update device r = requests.post(f"{API}/device/edit", - data=json.dumps(updated_device_payload)) + data=json.dumps(updated_device_payload), + timeout=5) assert r.status_code == 200 - r = requests.get(f"{API}/devices") + r = requests.get(f"{API}/devices", timeout=5) all_devices = json.loads(r.text) updated_device_api = next(x for x in all_devices if x["mac_addr"] == mac_addr) @@ -279,7 +724,7 @@ def test_modify_device(testing_devices, testrun): assert updated_device_api["test_modules"] == new_test_modules -def test_create_get_devices(empty_devices_dir, testrun): +def test_device_edit_device_not_found(empty_devices_dir, testrun): device_1 = { "manufacturer": "Google", "model": "First", @@ -293,9 +738,74 @@ def test_create_get_devices(empty_devices_dir, testrun): }, } - r = requests.post(f"{API}/device", data=json.dumps(device_1)) + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) + print(r.text) + assert r.status_code == 201 + assert len(local_get_devices()) == 1 + + updated_device = copy.deepcopy(device_1) + + updated_device_payload = {} + updated_device_payload["device"] = updated_device + updated_device_payload["mac_addr"] = "00:1e:42:35:73:c6" + updated_device_payload["model"] = "Alphabet" + + + r = requests.post(f"{API}/device/edit", + data=json.dumps(updated_device_payload), + timeout=5) + + assert r.status_code == 404 + + +def test_device_edit_device_incorrect_json_format(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) + print(r.text) + assert r.status_code == 201 + assert len(local_get_devices()) == 1 + + updated_device_payload = {} + + + r = requests.post(f"{API}/device/edit", + data=json.dumps(updated_device_payload), + timeout=5) + + assert r.status_code == 400 + + +def test_device_edit_device_with_mac_already_exists(empty_devices_dir, testrun): + device_1 = { + "manufacturer": "Google", + "model": "First", + "mac_addr": "00:1e:42:35:73:c4", + "test_modules": { + "dns": {"enabled": True}, + "connection": {"enabled": True}, + "ntp": {"enabled": True}, + "baseline": {"enabled": True}, + "nmap": {"enabled": True}, + }, + } + + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) print(r.text) - device1_response = r.text assert r.status_code == 201 assert len(local_get_devices()) == 1 @@ -311,40 +821,41 @@ def test_create_get_devices(empty_devices_dir, testrun): "nmap": {"enabled": True}, }, } - r = requests.post(f"{API}/device", data=json.dumps(device_2)) - device2_response = json.loads(r.text) + r = requests.post(f"{API}/device", data=json.dumps(device_2), + timeout=5) assert r.status_code == 201 assert len(local_get_devices()) == 2 - # Test that returned devices API endpoint matches expected structure - r = requests.get(f"{API}/devices") - all_devices = json.loads(r.text) - pretty_print(all_devices) + updated_device = copy.deepcopy(device_1) - with open( - os.path.join(os.path.dirname(__file__), "mockito/get_devices.json") - ) as f: - mockito = json.load(f) + updated_device_payload = {} + updated_device_payload = {} + updated_device_payload["device"] = updated_device + updated_device_payload["mac_addr"] = "00:1e:42:35:73:c6" + updated_device_payload["model"] = "Alphabet" - print(mockito) - # Validate structure - assert all([isinstance(x, dict) for x in all_devices]) + r = requests.post(f"{API}/device/edit", + data=json.dumps(updated_device_payload), + timeout=5) - # TOOO uncomment when is done - # assert set(dict_paths(mockito[0])) == set(dict_paths(all_devices[0])) + assert r.status_code == 409 - # Validate contents of given keys matches - for key in ["mac_addr", "manufacturer", "model"]: - assert set([all_devices[0][key], all_devices[1][key]]) == set( - [device_1[key], device_2[key]] - ) +def test_system_latest_version(testrun): + r = requests.get(f"{API}/system/version", timeout=5) + assert r.status_code == 200 + updated_system_version = json.loads(r.text)["update_available"] + assert updated_system_version is False -def test_get_system_config(testrun): - r = requests.get(f"{API}/system/config") +#@pytest.mark.skip() +def test_get_system_config(testrun): # pylint: disable=W0613 + r = requests.get(f"{API}/system/config", timeout=5) - with open(SYSTEM_CONFIG_PATH) as f: + with open( + SYSTEM_CONFIG_PATH, + encoding="utf-8" + ) as f: local_config = json.load(f) api_config = json.loads(r.text) @@ -364,14 +875,14 @@ def test_get_system_config(testrun): ) -# TODO change to invalid jsdon request -@pytest.mark.skip() -def test_invalid_path_get(testrun): - r = requests.get(f"{API}/blah/blah") +#@pytest.mark.skip() +def test_invalid_path_get(testrun): # pylint: disable=W0613 + r = requests.get(f"{API}/blah/blah", timeout=5) response = json.loads(r.text) assert r.status_code == 404 with open( - os.path.join(os.path.dirname(__file__), "mockito/invalid_request.json") + os.path.join(os.path.dirname(__file__), "mockito/invalid_request.json"), + encoding="utf-8" ) as f: mockito = json.load(f) @@ -379,9 +890,10 @@ def test_invalid_path_get(testrun): assert set(dict_paths(mockito)) == set(dict_paths(response)) -def test_trigger_run(testing_devices, testrun): +@pytest.mark.skip() +def test_trigger_run(testing_devices, testrun): # pylint: disable=W0613 payload = {"device": {"mac_addr": BASELINE_MAC_ADDR, "firmware": "asd"}} - r = requests.post(f"{API}/system/start", data=json.dumps(payload)) + r = requests.post(f"{API}/system/start", data=json.dumps(payload), timeout=10) assert r.status_code == 200 until_true( @@ -401,7 +913,7 @@ def test_trigger_run(testing_devices, testrun): stop_test_device("x123") # Validate response - r = requests.get(f"{API}/system/status") + r = requests.get(f"{API}/system/status", timeout=5) response = json.loads(r.text) pretty_print(response) @@ -415,7 +927,7 @@ def test_trigger_run(testing_devices, testrun): with open( os.path.join( os.path.dirname(__file__), "mockito/running_system_status.json" - ) + ), encoding="utf-8" ) as f: mockito = json.load(f) @@ -430,9 +942,12 @@ def test_trigger_run(testing_devices, testrun): # Validate a result assert results["baseline.compliant"]["result"] == "Compliant" -def test_stop_running_test(testing_devices, testrun): + +@pytest.mark.skip() +def test_stop_running_test(testing_devices, testrun): # pylint: disable=W0613 payload = {"device": {"mac_addr": ALL_MAC_ADDR, "firmware": "asd"}} - r = requests.post(f"{API}/system/start", data=json.dumps(payload)) + r = requests.post(f"{API}/system/start", data=json.dumps(payload), + timeout=10) assert r.status_code == 200 until_true( @@ -452,35 +967,36 @@ def test_stop_running_test(testing_devices, testrun): stop_test_device("x12345") # Validate response - r = requests.post(f"{API}/system/stop") + r = requests.post(f"{API}/system/stop", timeout=20) response = json.loads(r.text) pretty_print(response) assert response == {"success": "Testrun stopped"} time.sleep(1) + # Validate response - r = requests.get(f"{API}/system/status") + r = requests.get(f"{API}/system/status", timeout=5) response = json.loads(r.text) pretty_print(response) - #TODO uncomment when bug is fixed - #assert len(response["tests"]["results"]) == response["tests"]["total"] - assert len(response["tests"]["results"]) < 15 - #TODO uncomment when bug is fixed - #assert response["status"] == "Stopped" + assert response["status"] == "Cancelled" -@pytest.mark.skip() -def test_stop_running_not_running(testrun): +#@pytest.mark.skip() +def test_stop_running_not_running(testrun): # pylint: disable=W0613 # Validate response - r = requests.post(f"{API}/system/stop") + r = requests.post(f"{API}/system/stop", + timeout=10) response = json.loads(r.text) pretty_print(response) - assert False + assert r.status_code == 404 + assert response["error"] == "Testrun is not currently running" -def test_multiple_runs(testing_devices, testrun): +@pytest.mark.skip() +def test_multiple_runs(testing_devices, testrun): # pylint: disable=W0613 payload = {"device": {"mac_addr": BASELINE_MAC_ADDR, "firmware": "asd"}} - r = requests.post(f"{API}/system/start", data=json.dumps(payload)) + r = requests.post(f"{API}/system/start", data=json.dumps(payload), + timeout=10) assert r.status_code == 200 print(r.text) @@ -501,7 +1017,7 @@ def test_multiple_runs(testing_devices, testrun): stop_test_device("x123") # Validate response - r = requests.get(f"{API}/system/status") + r = requests.get(f"{API}/system/status", timeout=5) response = json.loads(r.text) pretty_print(response) @@ -512,7 +1028,8 @@ def test_multiple_runs(testing_devices, testrun): assert len(results) == 3 payload = {"device": {"mac_addr": BASELINE_MAC_ADDR, "firmware": "asd"}} - r = requests.post(f"{API}/system/start", data=json.dumps(payload)) + r = requests.post(f"{API}/system/start", data=json.dumps(payload), + timeout=10) # assert r.status_code == 200 # returns 409 print(r.text) @@ -533,12 +1050,10 @@ def test_multiple_runs(testing_devices, testrun): stop_test_device("x123") -#TODO uncomment when functionality is implemented -@pytest.mark.skip() -def test_create_invalid_chars(empty_devices_dir, testrun): +def test_create_invalid_chars(empty_devices_dir, testrun): # pylint: disable=W0613 # local_delete_devices(ALL_DEVICES) # We must start test run with no devices in local/devices for this test - # to function as expected! + # to function as expected assert len(local_get_devices()) == 0 # Test adding device @@ -555,6 +1070,7 @@ def test_create_invalid_chars(empty_devices_dir, testrun): }, } - r = requests.post(f"{API}/device", data=json.dumps(device_1)) + r = requests.post(f"{API}/device", data=json.dumps(device_1), + timeout=5) print(r.text) print(r.status_code) diff --git a/testing/pylint/test_pylint b/testing/pylint/test_pylint index 9e9074aa7..162bb096a 100755 --- a/testing/pylint/test_pylint +++ b/testing/pylint/test_pylint @@ -21,7 +21,7 @@ sudo cmd/install source venv/bin/activate sudo pip3 install pylint==3.0.3 -files=$(find ./ -path ./venv -prune -o -name '*.py' -print) +files=$(find . -path ./venv -prune -o -name '*.py' -print) OUT=pylint.out