Fix tests and minors (#271)

src/Kathara/cli/command/LstartCommand.py

@@ -175,7 +175,6 @@ def run(self, current_path: str, argv: List[str]) -> Lab:
         if len(lab.machines) <= 0:
             raise EmptyLabError()
 
-
         lab.global_machine_metadata = OptionParser.parse(args['global_machine_metadata'])
 
         lab_ext_path = os.path.join(lab_path, 'lab.ext')
@@ -220,7 +219,11 @@ def run(self, current_path: str, argv: List[str]) -> Lab:
         Kathara.get_instance().deploy_lab(lab, selected_machines=set(args['machine_name']))
 
         if args['list']:
-            machines_stats = Kathara.get_instance().get_machines_stats(lab_hash=lab.hash)
-            print(next(create_table(machines_stats)))
+            with self.console.status(
+                    f"Loading...",
+                    spinner="dots"
+            ) as _:
+                machines_stats = Kathara.get_instance().get_machines_stats(lab_hash=lab.hash)
+                self.console.print(create_table(machines_stats))
 
         return lab

src/Kathara/foundation/model/FilesystemMixin.py

@@ -1,7 +1,7 @@
 import io
 import os.path
 import re
-from typing import Optional, List, BinaryIO, TextIO, Union, Pattern
+from typing import Optional, List, BinaryIO, TextIO, Union
 
 from fs import open_fs
 from fs.base import FS

src/Kathara/manager/Kathara.py

@@ -340,6 +340,9 @@ def get_machines_api_objects(self, lab_hash: Optional[str] = None, lab_name: Opt
 
         Returns:
             List[Any]: API objects of devices, specific for the current manager.
+
+        Raises:
+            InvocationError: If a running network scenario hash or name is not specified.
         """
         return self.manager.get_machines_api_objects(lab_hash, lab_name, lab, all_users)
 
@@ -381,6 +384,9 @@ def get_links_api_objects(self, lab_hash: Optional[str] = None, lab_name: Option
 
         Returns:
             List[Any]: API objects of collision domains, specific for the current manager.
+
+        Raises:
+            InvocationError: If a running network scenario hash or name is not specified.
         """
         return self.manager.get_links_api_objects(lab_hash, lab_name, lab, all_users)
 
@@ -427,6 +433,10 @@ def get_machines_stats(self, lab_hash: Optional[str] = None, lab_name: Optional[
         Returns:
               Generator[Dict[str, IMachineStats], None, None]: A generator containing dicts that has API Object
               identifier as keys and IMachineStats objects as values.
+
+        Raises:
+            InvocationError: If more than one param among lab_hash, lab_name and lab is specified.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         return self.manager.get_machines_stats(lab_hash, lab_name, lab, machine_name, all_users)
 
@@ -450,7 +460,8 @@ def get_machine_stats(self, machine_name: str, lab_hash: Optional[str] = None, l
             with the device info. Returns None if the device is not found.
 
         Raises:
-            InvocationError: If a running network scenario hash or name is not specified.
+            InvocationError: If more than one param among lab_hash, lab_name and lab is specified.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         return self.manager.get_machine_stats(machine_name, lab_hash, lab_name, lab, all_users)
 
@@ -469,6 +480,7 @@ def get_machine_stats_obj(self, machine: Machine, all_users: bool = False) \
         Raises:
             LabNotFoundError: If the specified device is not associated to any network scenario.
             MachineNotRunningError: If the specified device is not running.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         return self.manager.get_machine_stats_obj(machine, all_users)
 
@@ -489,6 +501,10 @@ def get_links_stats(self, lab_hash: Optional[str] = None, lab_name: Optional[str
         Returns:
              Generator[Dict[str, ILinkStats], None, None]: A generator containing dicts that has API Object
              identifier as keys and ILinksStats objects as values.
+
+        Raises:
+            InvocationError: If a running network scenario hash, name or object is not specified.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         return self.manager.get_links_stats(lab_hash, lab_name, lab, link_name, all_users)
 
@@ -513,6 +529,7 @@ def get_link_stats(self, link_name: str, lab_hash: Optional[str] = None, lab_nam
 
         Raises:
             InvocationError: If a running network scenario hash or name is not specified.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         return self.manager.get_link_stats(link_name, lab_hash, lab_name, lab, all_users)
 
@@ -529,6 +546,7 @@ def get_link_stats_obj(self, link: Link, all_users: bool = False) -> Generator[O
 
         Raises:
             LabNotFoundError: If the specified device is not associated to any network scenario.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         return self.manager.get_link_stats_obj(link, all_users)
 

src/Kathara/manager/docker/DockerLink.py

@@ -246,7 +246,13 @@ def get_links_stats(self, lab_hash: str = None, link_name: str = None, user: str
         Returns:
            Generator[Dict[str, DockerMachineStats], None, None]: A generator containing network names as keys and
            DockerLinkStats as values.
+
+        Raises:
+            PrivilegeError: If user param is None and the user does not have root privileges.
         """
+        if user is None and not utils.is_admin():
+            raise PrivilegeError("You must be root to get networks statistics of all users.")
+
         networks_stats = {}
 
         def load_link_stats(network):
@@ -340,25 +346,29 @@ def _delete_external_interfaces(self, external_links: List[str], network: docker
 
         Returns:
             None
+
+        Raises:
+            OSError: If the link is attached to external interfaces and the host OS is not LINUX.
+            PrivilegeError: If the link is attached to external interfaces and the user does not have root privileges.
         """
+        if not (utils.is_platform(utils.LINUX) or utils.is_platform(utils.LINUX2)):
+            raise OSError("External collision domains available only on Linux systems.")
+
+        if not utils.is_admin():
+            raise PrivilegeError("You must be root in order to use external collision domains.")
+
         for external_link in external_links:
-            if utils.is_platform(utils.LINUX) or utils.is_platform(utils.LINUX2):
-                if utils.is_admin():
-                    def vde_delete():
-                        plugin_pid = self.docker_plugin.plugin_pid()
-                        switch_path = os.path.join(self.docker_plugin.plugin_store_path(),
-                                                   self._get_bridge_name(network))
-                        Networking.remove_interface_ns(external_link, switch_path, plugin_pid)
-
-                    self.docker_plugin.exec_by_version(vde_delete, lambda: None)
-
-                    if re.search(r"^\w+\.\d+$", external_link):
-                        # Only remove VLAN interfaces, physical ones cannot be removed.
-                        Networking.remove_interface(external_link)
-                else:
-                    raise PrivilegeError("You must be root in order to delete an External Interface.")
-            else:
-                raise OSError("External Interfaces are only available on UNIX systems.")
+            def vde_delete():
+                plugin_pid = self.docker_plugin.plugin_pid()
+                switch_path = os.path.join(self.docker_plugin.plugin_store_path(),
+                                           self._get_bridge_name(network))
+                Networking.remove_interface_ns(external_link, switch_path, plugin_pid)
+
+            self.docker_plugin.exec_by_version(vde_delete, lambda: None)
+
+            if re.search(r"^\w+\.\d+$", external_link):
+                # Only remove VLAN interfaces, physical ones cannot be removed.
+                Networking.remove_interface(external_link)
 
     @staticmethod
     def _get_bridge_name(network: docker.models.networks.Network) -> str:

src/Kathara/manager/docker/DockerMachine.py

@@ -883,7 +883,7 @@ def get_machines_stats(self, lab_hash: str = None, machine_name: str = None, use
             PrivilegeError: If user param is None and the user does not have root privileges.
         """
         if user is None and not utils.is_admin():
-            raise PrivilegeError("You must be root to get devices stats of other users.")
+            raise PrivilegeError("You must be root to get devices statistics of all users.")
 
         machines_stats = {}
 

src/Kathara/manager/docker/DockerManager.py

@@ -93,7 +93,6 @@ def deploy_machine(self, machine: Machine) -> None:
         if not machine.lab:
             raise LabNotFoundError("Device `%s` is not associated to a network scenario." % machine.name)
 
-        logging.debug(f"Checking `{machine.name}` integrity...")
         machine.check()
 
         self.docker_link.deploy_links(machine.lab, selected_links={x.link.name for x in machine.interfaces.values()})
@@ -134,7 +133,6 @@ def deploy_lab(self, lab: Lab, selected_machines: Set[str] = None) -> None:
             PrivilegeError: If the user start the network scenario in privileged mode without having root privileges.
             MachineNotFoundError: If the specified devices are not in the network scenario.
         """
-        logging.debug("Checking network scenario integrity...")
         lab.check_integrity()
 
         if selected_machines and not lab.has_machines(selected_machines):
@@ -520,6 +518,9 @@ def get_machines_api_objects(self, lab_hash: Optional[str] = None, lab_name: Opt
 
         Returns:
             List[docker.models.containers.Container]: Docker API objects of devices.
+
+        Raises:
+            InvocationError: If a running network scenario hash or name is not specified.
         """
         check_single_not_none_var(lab_hash=lab_hash, lab_name=lab_name, lab=lab)
         if lab:
@@ -584,6 +585,9 @@ def get_links_api_objects(self, lab_hash: Optional[str] = None, lab_name: Option
 
         Returns:
             List[docker.models.networks.Network]: Docker API objects of networks.
+
+        Raises:
+            InvocationError: If a running network scenario hash or name is not specified.
         """
         check_single_not_none_var(lab_hash=lab_hash, lab_name=lab_name, lab=lab)
         if lab:
@@ -826,6 +830,7 @@ def get_machine_stats_obj(self, machine: Machine, all_users: bool = False) \
         Raises:
             LabNotFoundError: If the specified device is not associated to any network scenario.
             MachineNotRunningError: If the specified device is not running.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         if not machine.lab:
             raise LabNotFoundError("Device `%s` is not associated to a network scenario." % machine.name)
@@ -851,6 +856,10 @@ def get_links_stats(self, lab_hash: Optional[str] = None, lab_name: Optional[str
         Returns:
              Generator[Dict[str, DockerLinkStats], None, None]: A generator containing dicts that has API Object
                 identifier as keys and DockerLinksStats objects as values.
+
+        Raises:
+            InvocationError: If a running network scenario hash, name or object is not specified.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         check_single_not_none_var(lab_hash=lab_hash, lab_name=lab_name, lab=lab)
         if lab:
@@ -883,6 +892,7 @@ def get_link_stats(self, link_name: str, lab_hash: Optional[str] = None, lab_nam
 
         Raises:
             InvocationError: If a running network scenario hash or name is not specified.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         check_required_single_not_none_var(lab_hash=lab_hash, lab_name=lab_name, lab=lab)
         if lab:
@@ -912,6 +922,7 @@ def get_link_stats_obj(self, link: Link, all_users: bool = False) \
 
         Raises:
             LabNotFoundError: If the specified device is not associated to any network scenario.
+            PrivilegeError: If all_users is True and the user does not have root privileges.
         """
         if not link.lab:
             raise LabNotFoundError(f"Link `{link.name}` is not associated to a network scenario.")

src/Kathara/manager/kubernetes/KubernetesMachine.py

@@ -151,9 +151,7 @@ def deploy_machines(self, lab: Lab, selected_machines: Set[str] = None) -> None:
         machines = {k: v for (k, v) in lab.machines.items() if k in selected_machines}.items() if selected_machines \
             else lab.machines.items()
 
-        privileged = lab.general_options['privileged_machines'] if 'privileged_machines' in lab.general_options \
-            else False
-        if privileged:
+        if lab.general_options['privileged_machines']:
             logging.warning('Privileged option is not supported on Megalos. It will be ignored.')
 
         # Do not open terminals on Megalos

src/Kathara/manager/kubernetes/KubernetesManager.py

@@ -50,7 +50,6 @@ def deploy_machine(self, machine: Machine) -> None:
         if not machine.lab:
             raise LabNotFoundError("Machine `%s` is not associated to a network scenario." % machine.name)
 
-        logging.debug(f"Checking `{machine.name}` integrity...")
         machine.check()
 
         machine.lab.hash = machine.lab.hash.lower()
@@ -95,7 +94,6 @@ def deploy_lab(self, lab: Lab, selected_machines: Set[str] = None) -> None:
             LabAlreadyExistsError: If a network scenario is deployed while it is terminating its execution.
             ApiError: If the Kubernetes APIs throw an exception.
         """
-        logging.debug("Checking network scenario integrity...")
         lab.check_integrity()
 
         if selected_machines and not lab.has_machines(selected_machines):

src/Kathara/model/Lab.py

@@ -1,4 +1,5 @@
 import collections
+import logging
 from itertools import chain
 from typing import Dict, Set, Any, List, Union, Optional, Tuple
 
@@ -188,6 +189,8 @@ def check_integrity(self) -> None:
         Raises:
             NonSequentialMachineInterfaceError: If there is a missing interface number in any device of the lab.
         """
+        logging.debug("Checking network scenario integrity...")
+
         for machine in self.machines.values():
             machine.check()
 

src/Kathara/model/Machine.py

@@ -276,6 +276,8 @@ def check(self) -> None:
         Raises:
             NonSequentialMachineInterfaceError: If there is a missing interface number.
         """
+        logging.debug(f"Checking `{self.name}` integrity...")
+
         sorted_interfaces = sorted(self.interfaces.items(), key=lambda kv: kv[0])
 
         logging.debug("`%s` interfaces are %s." % (self.name, sorted_interfaces))