Fix K8s exec w/ and w/o stream (#271)

src/Kathara/manager/docker/DockerMachine.py

@@ -133,7 +133,6 @@ def deploy_machines(self, lab: Lab, selected_machines: Set[str] = None) -> None:
 
         shared_mount = lab.general_options['shared_mount'] if 'shared_mount' in lab.general_options \
             else Setting.get_instance().shared_mount
-
         if shared_mount:
             if Setting.get_instance().remote_url is not None:
                 logging.warning("Shared folder cannot be mounted with a remote Docker connection.")
@@ -258,18 +257,17 @@ def create(self, machine: Machine) -> None:
         volumes = {}
 
         lab_options = machine.lab.general_options
-
-        if lab_options['shared_mount'] and machine.lab.shared_path:
+        shared_mount = ['shared_mount'] if 'shared_mount' in lab_options else Setting.get_instance().shared_mount
+        if shared_mount and machine.lab.shared_path:
             volumes[machine.lab.shared_path] = {'bind': '/shared', 'mode': 'rw'}
 
         # Mount the host home only if specified in settings.
-
-        if lab_options['hosthome_mount'] and Setting.get_instance().remote_url is None:
+        hosthome_mount = lab_options['hosthome_mount'] if 'hosthome_mount' in lab_options else \
+            Setting.get_instance().hosthome_mount
+        if hosthome_mount and Setting.get_instance().remote_url is None:
             volumes[utils.get_current_user_home()] = {'bind': '/hosthome', 'mode': 'rw'}
 
-        privileged = lab_options['privileged_machines'] if 'privileged_machines' in lab_options \
-            else False
-
+        privileged = lab_options['privileged_machines']
         if Setting.get_instance().remote_url is not None and privileged:
             privileged = False
             logging.warning("Privileged flag is ignored with a remote Docker connection.")
@@ -678,7 +676,7 @@ def exec(self, lab_hash: str, machine_name: str, command: Union[str, List], user
         if should_wait:
             self._wait_startup_execution(container, n_retries=n_retries, retry_interval=retry_interval)
 
-        command = shlex.split(command) if type(command) == str else command
+        command = shlex.split(command) if type(command) is str else command
         exec_result = self._exec_run(container,
                                      cmd=command,
                                      stdout=True,
@@ -756,7 +754,7 @@ def _exec_run(self, container: docker.models.containers.Container,
             (stdout_out, _) = exec_output if demux else (exec_output, None)
             exec_stdout = ""
             if stdout_out:
-                if type(stdout_out) == bytes:
+                if type(stdout_out) is bytes:
                     char_encoding = chardet.detect(stdout_out)
                     exec_stdout = stdout_out.decode(char_encoding['encoding'])
                 else:

src/Kathara/manager/kubernetes/KubernetesMachine.py

@@ -9,7 +9,7 @@
 import threading
 import uuid
 from multiprocessing.dummy import Pool
-from typing import Optional, Set, List, Union, Generator, Tuple, Dict
+from typing import Optional, Set, List, Union, Generator, Tuple, Dict, Any
 
 import chardet
 from kubernetes import client
@@ -624,7 +624,7 @@ def connect(self, lab_hash: str, machine_name: str, shell: Union[str, List[str]]
             shell_env_value = self.get_env_var_value_from_pod(deployment, "_MEGALOS_SHELL")
             shell = shlex.split(shell_env_value if shell_env_value else Setting.get_instance().device_shell)
         else:
-            shell = shlex.split(shell) if type(shell) == str else shell
+            shell = shlex.split(shell) if type(shell) is str else shell
 
         logging.debug("Connect to device `%s` with shell: %s" % (machine_name, shell))
 
@@ -698,7 +698,7 @@ def exec(self, lab_hash: str, machine_name: str, command: Union[str, List], tty:
         Args:
             lab_hash (str): The hash of the network scenario containing the device.
             machine_name (str): The name of the device.
-            command (str): The command to execute.
+            command (Union[str, List]): The command to execute.
             tty (bool): If True, open a new tty.
             stdin (bool): If True, open the stdin channel.
             stdin_buffer (List[Union[str, bytes]]): List of command to pass to the stdin.
@@ -714,7 +714,7 @@ def exec(self, lab_hash: str, machine_name: str, command: Union[str, List], tty:
             MachineNotRunningError: If the specified device is not running.
             MachineBinaryError: If the command specified is not found on the device.
         """
-        command = shlex.split(command) if command is str else command
+        command = shlex.split(command) if type(command) is str else command
 
         logging.debug("Executing command `%s` to device with name: %s" % (command, machine_name))
 
@@ -735,44 +735,87 @@ def exec(self, lab_hash: str, machine_name: str, command: Union[str, List], tty:
                               tty=tty,
                               _preload_content=False
                               )
-        except ApiException:
-            return None
+        except ApiException as e:
+            raise e
 
         if stdin_buffer is None:
             stdin_buffer = []
 
-        result = {
-            'stdout': '',
-            'stderr': ''
-        }
+        if is_stream:
+            return self._exec_stream(response, stdin, stdin_buffer, stderr)
+        else:
+            return self._exec_all(response, machine_name, command, stdin, stdin_buffer, stderr)
+
+    @staticmethod
+    def _exec_stream(response: Any, stdin: bool = False, stdin_buffer: List[Union[str, bytes]] = None,
+                     has_stderr: bool = False) -> Generator[Tuple[bytes, bytes], None, None]:
+        """Execute the command on the Kubernetes Pod, returning a generator.
+
+        Args:
+            response (Any): The stream response from Kubernetes API.
+            stdin (bool): If True, open the stdin channel.
+            stdin_buffer (List[Union[str, bytes]]): List of command to pass to the stdin.
+            has_stderr (bool): If True, return the stderr.
 
+        Returns:
+            Tuple[bytes, bytes, int]: A tuple containing the stdout, the stderr and the return code of the command.
+        """
         while response.is_open():
             stdout = None
             stderr = None
             if response.peek_stdout():
                 stdout = response.read_stdout()
-                if not is_stream:
-                    result['stdout'] += stdout
-            if stderr and response.peek_stderr():
+            if has_stderr and response.peek_stderr():
                 stderr = response.read_stderr()
-                if not is_stream:
-                    result['stderr'] += stderr
             if stdin and stdin_buffer:
                 param = stdin_buffer.pop(0)
                 response.write_stdin(param)
                 if len(stdin_buffer) <= 0:
                     break
 
-            if is_stream and (stdout or stderr):
-                yield stdout.encode('utf-8') if stdout else None, stderr.encode('utf-8') if stderr else None
+            yield stdout.encode('utf-8') if stdout else None, stderr.encode('utf-8') if stderr else None
+
+        response.close()
+
+    @staticmethod
+    def _exec_all(response: Any, machine_name: str, command: List, stdin: bool = False,
+                  stdin_buffer: List[Union[str, bytes]] = None, has_stderr: bool = False) -> Tuple[bytes, bytes, int]:
+        """Execute the command on the Kubernetes Pod, returning the full output.
+
+        Args:
+            response (Any): The stream response from Kubernetes API.
+            machine_name (str): The name of the device.
+            command (List): The command to execute.
+            stdin (bool): If True, open the stdin channel.
+            stdin_buffer (List[Union[str, bytes]]): List of command to pass to the stdin.
+            has_stderr (bool): If True, return the stderr.
+
+        Returns:
+            Generator[Tuple[bytes, bytes]]: A generator of tuples containing the stdout and stderr in bytes.
+
+        Raises:
+            MachineBinaryError: If the command specified is not found on the device.
+        """
+        result = {'stdout': '', 'stderr': ''}
+
+        while response.is_open():
+            if response.peek_stdout():
+                result['stdout'] += response.read_stdout()
+            if has_stderr and response.peek_stderr():
+                result['stderr'] += response.read_stderr()
+            if stdin and stdin_buffer:
+                param = stdin_buffer.pop(0)
+                response.write_stdin(param)
+                if len(stdin_buffer) <= 0:
+                    break
 
         response.close()
 
         try:
             error_code = response.returncode
         except ValueError as e:
             if OCI_RUNTIME_RE.search(str(e)):
-                raise MachineBinaryError(command, machine_name)
+                raise MachineBinaryError(shlex.join(command), machine_name)
             error_code = 1
 
         return result['stdout'].encode('utf-8'), result['stderr'].encode('utf-8'), error_code

src/Kathara/model/Lab.py

@@ -13,7 +13,6 @@
 from .. import utils
 from ..exceptions import LinkNotFoundError, MachineNotFoundError, MachineAlreadyExistsError, LinkAlreadyExistsError
 from ..foundation.model.FilesystemMixin import FilesystemMixin
-from ..setting.Setting import Setting
 
 LAB_METADATA: List[str] = ["LAB_NAME", "LAB_DESCRIPTION", "LAB_VERSION", "LAB_AUTHOR", "LAB_EMAIL", "LAB_WEB"]
 
@@ -67,8 +66,6 @@ def __init__(self, name: Optional[str], path: Optional[str] = None) -> None:
 
         self.general_options: Dict[str, Any] = {
             'privileged_machines': False,
-            'shared_mount': Setting.get_instance().shared_mount,
-            'hosthome_mount': Setting.get_instance().hosthome_mount
         }
 
         self.global_machine_metadata: Dict[str, Any] = {}

src/Kathara/model/Machine.py

@@ -485,7 +485,7 @@ def is_ipv6_enabled(self) -> bool:
             elif "ipv6" in self.meta:
                 is_v6_enabled = self.meta["ipv6"]
 
-            return is_v6_enabled if type(is_v6_enabled) == bool else strtobool(is_v6_enabled)
+            return is_v6_enabled if type(is_v6_enabled) is bool else strtobool(is_v6_enabled)
         except ValueError:
             raise MachineOptionError("IPv6 value not valid on `%s`." % self.name)
 

tests/model/lab_test.py

@@ -59,8 +59,7 @@ def test_default_scenario_creation(default_scenario: Lab):
     assert default_scenario.web is None
     assert default_scenario.machines == {}
     assert default_scenario.links == {}
-    assert default_scenario.general_options == {'privileged_machines': False, 'shared_mount': True,
-                                                'hosthome_mount': False}
+    assert default_scenario.general_options == {'privileged_machines': False}
     assert not default_scenario.has_dependencies
     assert default_scenario.fs_type() == "memory"
     assert default_scenario.shared_path is None
@@ -81,9 +80,7 @@ def test_directory_scenario_creation_with_shared_files(directory_scenario: Lab,
     assert directory_scenario.web is None
     assert directory_scenario.machines == {}
     assert directory_scenario.links == {}
-    assert directory_scenario.general_options == {'privileged_machines': False,
-                                                  'shared_mount': True,
-                                                  'hosthome_mount': False}
+    assert directory_scenario.general_options == {'privileged_machines': False}
     assert not directory_scenario.has_dependencies
     assert os.path.normpath(directory_scenario.fs_path()) == os.path.normpath(temporary_path)
     assert directory_scenario.shared_path is None