Port scan test module

cmd/install

@@ -4,6 +4,8 @@ python3 -m venv venv
 
 source venv/bin/activate
 
+pip3 install --upgrade requests
+
 pip3 install -r framework/requirements.txt
 
 pip3 install -r net_orc/python/requirements.txt

framework/device.py

@@ -1,12 +1,14 @@
-"""Track device object information."""
-from dataclasses import dataclass
-from network_device import NetworkDevice
-
-
-@dataclass
-class Device(NetworkDevice):
-  """Represents a physical device and it's configuration."""
-
-  make: str = None
-  model: str = None
-  test_modules: str = None
+"""Track device object information."""
+
+from network_device import NetworkDevice
+from dataclasses import dataclass
+
+
+@dataclass
+class Device(NetworkDevice):
+    """Represents a physical device and it's configuration."""
+
+    make: str = None
+    model: str = None
+    mac_addr: str
+    test_modules: str = None

framework/testrun.py

@@ -46,142 +46,142 @@
 
 
 class TestRun:  # pylint: disable=too-few-public-methods
-        """Test Run controller.
+    """Test Run controller.
 
-        Creates an instance of the network orchestrator, test
-        orchestrator and user interface.
-        """
+    Creates an instance of the network orchestrator, test
+    orchestrator and user interface.
+    """
 
-        def __init__(self, config_file=CONFIG_FILE, validate=True, net_only=False, single_intf=False):
-            self._devices = []
-            self._net_only = net_only
-            self._single_intf = single_intf
+    def __init__(self, config_file=CONFIG_FILE, validate=True, net_only=False, single_intf=False):
+        self._devices = []
+        self._net_only = net_only
+        self._single_intf = single_intf
 
-            # Catch any exit signals
-            self._register_exits()
+        # Catch any exit signals
+        self._register_exits()
 
-            # Expand the config file to absolute pathing
-            config_file_abs = self._get_config_abs(config_file=config_file)
+        # Expand the config file to absolute pathing
+        config_file_abs = self._get_config_abs(config_file=config_file)
 
-            self._net_orc = net_orc.NetworkOrchestrator(
-                config_file=config_file_abs, 
-                validate=validate, 
-                async_monitor=not self._net_only,
-                single_intf = self._single_intf)
-            self._test_orc = test_orc.TestOrchestrator()
+        self._net_orc = net_orc.NetworkOrchestrator(
+            config_file=config_file_abs, 
+            validate=validate, 
+            async_monitor=not self._net_only,
+            single_intf = self._single_intf)
+        self._test_orc = test_orc.TestOrchestrator(self._net_orc)
 
-        def start(self):
+    def start(self):
 
-            self._load_all_devices()
+        self._load_all_devices()
 
-            if self._net_only:
-                LOGGER.info(
-                    "Network only option configured, no tests will be run")
-                self._start_network()
-            else:
-                self._start_network()
-                self._test_orc.start()
 
-                self._net_orc.listener.register_callback(
+        if self._net_only:
+            LOGGER.info("Network only option configured, no tests will be run")
+            self._start_network()
+        else:
+            self._start_network()
+            self._test_orc.start()
+
+            self._net_orc.listener.register_callback(
                     self._device_stable,
                     [NetworkEvent.DEVICE_STABLE]
                 )
 
-                LOGGER.info("Waiting for devices on the network...")
+            LOGGER.info("Waiting for devices on the network...")
 
-                # Check timeout and whether testing is currently in progress before stopping
-                time.sleep(RUNTIME)
+            # Check timeout and whether testing is currently in progress before stopping
+            time.sleep(RUNTIME)
 
-            self.stop()
+        self.stop()
 
-        def stop(self, kill=False):
-            self._stop_tests()
-            self._stop_network(kill=kill)
+    def stop(self, kill=False):
+        self._stop_tests()
+        self._stop_network(kill=kill)
 
-        def _register_exits(self):
-            signal.signal(signal.SIGINT, self._exit_handler)
-            signal.signal(signal.SIGTERM, self._exit_handler)
-            signal.signal(signal.SIGABRT, self._exit_handler)
-            signal.signal(signal.SIGQUIT, self._exit_handler)
+    def _register_exits(self):
+        signal.signal(signal.SIGINT, self._exit_handler)
+        signal.signal(signal.SIGTERM, self._exit_handler)
+        signal.signal(signal.SIGABRT, self._exit_handler)
+        signal.signal(signal.SIGQUIT, self._exit_handler)
 
-        def _exit_handler(self, signum, arg):  # pylint: disable=unused-argument
-            LOGGER.debug("Exit signal received: " + str(signum))
-            if signum in (2, signal.SIGTERM):
-                LOGGER.info("Exit signal received.")
-                self.stop(kill=True)
-                sys.exit(1)
+    def _exit_handler(self, signum, arg):  # pylint: disable=unused-argument
+        LOGGER.debug("Exit signal received: " + str(signum))
+        if signum in (2, signal.SIGTERM):
+            LOGGER.info("Exit signal received.")
+            self.stop(kill=True)
+            sys.exit(1)
 
-        def _get_config_abs(self, config_file=None):
-            if config_file is None:
-                # If not defined, use relative pathing to local file
-                config_file = os.path.join(parent_dir, CONFIG_FILE)
+    def _get_config_abs(self, config_file=None):
+        if config_file is None:
+            # If not defined, use relative pathing to local file
+            config_file = os.path.join(parent_dir, CONFIG_FILE)
 
-            # Expand the config file to absolute pathing
-            return os.path.abspath(config_file)
+        # Expand the config file to absolute pathing
+        return os.path.abspath(config_file)
 
-        def _start_network(self):
-            # Load in local device configs to the network orchestrator
-            self._net_orc._devices = self._devices
+    def _start_network(self):
+        # Load in local device configs to the network orchestrator
+        self._net_orc._devices = self._devices
 
-            # Start the network orchestrator
-            self._net_orc.start()
+        # Start the network orchestrator
+        self._net_orc.start()
 
-        def _run_tests(self, device):
-            """Iterate through and start all test modules."""
+    def _run_tests(self, device):
+        """Iterate through and start all test modules."""
 
-            # TODO: Make this configurable
-            time.sleep(60) # Let device bootup
+        # To Do: Make this configurable
+        time.sleep(60)  # Let device bootup
 
-            self._test_orc.run_test_modules(device)
+        self._test_orc._run_test_modules(device)
 
-        def _stop_network(self, kill=False):
-            self._net_orc.stop(kill=kill)
+    def _stop_network(self, kill=False):
+        self._net_orc.stop(kill=kill)
 
-        def _stop_tests(self):
-            self._test_orc.stop()
+    def _stop_tests(self):
+        self._test_orc.stop()
 
-        def _load_all_devices(self):
-            self._load_devices(device_dir=LOCAL_DEVICES_DIR)
-            LOGGER.info('Loaded ' + str(len(self._devices)) + ' devices')
+    def _load_all_devices(self):
+        self._load_devices(device_dir=LOCAL_DEVICES_DIR)
+        self._load_devices(device_dir=RESOURCE_DEVICES_DIR)
 
-        def _load_devices(self, device_dir):
-            LOGGER.debug('Loading devices from ' + device_dir)
+    def _load_devices(self, device_dir):
+        LOGGER.debug('Loading devices from ' + device_dir)
 
-            os.makedirs(device_dir, exist_ok=True)
+        os.makedirs(device_dir, exist_ok=True)
 
-            for device_folder in os.listdir(device_dir):
-                with open(os.path.join(device_dir, device_folder, DEVICE_CONFIG),
-                          encoding='utf-8') as device_config_file:
-                    device_config_json = json.load(device_config_file)
+        for device_folder in os.listdir(device_dir):
+            with open(os.path.join(device_dir, device_folder, DEVICE_CONFIG),
+                      encoding='utf-8') as device_config_file:
+                device_config_json = json.load(device_config_file)
 
-                    device_make = device_config_json.get(DEVICE_MAKE)
-                    device_model = device_config_json.get(DEVICE_MODEL)
-                    mac_addr = device_config_json.get(DEVICE_MAC_ADDR)
-                    test_modules = device_config_json.get(DEVICE_TEST_MODULES)
+                device_make = device_config_json.get(DEVICE_MAKE)
+                device_model = device_config_json.get(DEVICE_MODEL)
+                mac_addr = device_config_json.get(DEVICE_MAC_ADDR)
+                test_modules = device_config_json.get(DEVICE_TEST_MODULES)
 
-                    device = Device(make=device_make, model=device_model,
-                                    mac_addr=mac_addr, test_modules=json.dumps(test_modules))
-                    self._devices.append(device)
-
-        def get_device(self, mac_addr):
-            """Returns a loaded device object from the device mac address."""
-            for device in self._devices:
-                if device.mac_addr == mac_addr:
-                    return device
-            return None
-
-        def _device_discovered(self, mac_addr):
-            device = self.get_device(mac_addr)
-            if device is not None:
-                LOGGER.info(
-                    f'Discovered {device.make} {device.model} on the network')
-            else:
-                device = Device(mac_addr=mac_addr)
+                device = Device(make=device_make, model=device_model,
+                                mac_addr=mac_addr, test_modules=json.dumps(test_modules))
                 self._devices.append(device)
-                LOGGER.info(
-                    f'A new device has been discovered with mac address {mac_addr}')
 
-        def _device_stable(self, mac_addr):
-            device = self.get_device(mac_addr)
-            LOGGER.info(f'Device with mac address {mac_addr} is ready for testing.')
-            self._test_orc.run_test_modules(device)
+    def get_device(self, mac_addr):
+        """Returns a loaded device object from the device mac address."""
+        for device in self._devices:
+            if device.mac_addr == mac_addr:
+                return device
+        return None
+
+    def _device_discovered(self, mac_addr):
+        device = self.get_device(mac_addr)
+        if device is not None:
+            LOGGER.info(
+                f'Discovered {device.make} {device.model} on the network')
+        else:
+            device = Device(mac_addr=mac_addr)
+            self._devices.append(device)
+            LOGGER.info(
+                f'A new device has been discovered with mac address {mac_addr}')
+
+    def _device_stable(self, mac_addr):
+        device = self.get_device(mac_addr)
+        LOGGER.info(f'Device with mac address {mac_addr} is ready for testing.')
+        self._test_orc.run_test_modules(device)