Fix Windows Pipe undesired exception (fix #191)

src/Kathara/cli/command/ExecCommand.py

@@ -9,6 +9,8 @@
 from ...parser.netkit.LabParser import LabParser
 from ...strings import strings, wiki_description
 
+pywintypes = utils.import_pywintypes()
+
 
 class ExecCommand(Command):
     def __init__(self) -> None:
@@ -90,5 +92,9 @@ def run(self, current_path: str, argv: List[str]) -> None:
                     sys.stdout.write(stdout)
                 if stderr and not args['no_stderr']:
                     sys.stderr.write(stderr)
+        except pywintypes.error as e:
+            (code, reason, _) = e.args
+            if code == 109 and reason == 'ReadFile':
+                pass
         except StopIteration:
             pass

src/Kathara/foundation/test/Test.py

@@ -6,6 +6,9 @@
 
 from ...manager.Kathara import Kathara
 from ...model.Lab import Lab
+from ...utils import import_pywintypes
+
+pywintypes = import_pywintypes()
 
 
 class Test(ABC):
@@ -46,5 +49,9 @@ def _get_machine_command_output(lab_hash, machine_name, command):
                     result['stdout'] += stdout.decode('utf-8')
                 if stderr:
                     result['stderr'] += stderr.decode('utf-8')
+        except pywintypes.error as e:
+            (code, reason, _) = e.args
+            if code == 109 and reason == 'ReadFile':
+                pass
         except StopIteration:
             return result['stdout'], result['stderr']

src/Kathara/manager/docker/DockerMachine.py

@@ -18,6 +18,8 @@
 from ...model.Machine import Machine
 from ...setting.Setting import Setting
 
+pywintypes = utils.import_pywintypes()
+
 RP_FILTER_NAMESPACE = "net.ipv4.conf.%s.rp_filter"
 
 # Known commands that each container should execute
@@ -515,6 +517,10 @@ def connect(self, lab_hash: str, machine_name: str, user: str = None, shell: str
                 while True:
                     (stdout, _) = next(exec_output)
                     startup_output += stdout.decode('utf-8') if stdout else ""
+            except pywintypes.error as e:
+                (code, reason, _) = e.args
+                if code == 109 and reason == 'ReadFile':
+                    pass
             except StopIteration:
                 pass
 

src/Kathara/manager/docker/DockerManager.py

@@ -25,23 +25,11 @@
 from ...utils import pack_files_for_tar
 from ...exceptions import MachineNotFoundError
 
-
-def pywin_import_stub():
-    """Stub module of pywintypes for Unix systems (so it won't raise any `module not found` exception)."""
-    import types
-    pywintypes = types.ModuleType("pywintypes")
-    pywintypes.error = RequestsConnectionError
-    return pywintypes
-
-
-def pywin_import_win():
-    import pywintypes
-    return pywintypes
+pywintypes = utils.import_pywintypes()
 
 
 def check_docker_status(method):
     """Decorator function to check if Docker daemon is running properly."""
-    pywintypes = utils.exec_by_platform(pywin_import_stub, pywin_import_win, pywin_import_stub)
 
     @privileged
     def check_docker(*args, **kw):

src/Kathara/utils.py

@@ -16,6 +16,7 @@
 from platform import node, machine
 from sys import platform as _platform
 from typing import Any, Optional, Match, Generator, List, Callable, Union, Dict, Iterable
+from types import ModuleType
 
 from binaryornot.check import is_binary
 from slug import slug
@@ -122,6 +123,24 @@ def exec_by_platform(fun_linux: Callable, fun_windows: Callable, fun_mac: Callab
         return fun_mac()
 
 
+def pywintypes_import_stub() -> ModuleType:
+    """Stub module of pywintypes for Unix systems (so it won't raise any `module not found` exception)."""
+    import types
+    from requests.exceptions import ConnectionError as RequestsConnectionError
+    pywintypes = types.ModuleType("pywintypes")
+    pywintypes.error = RequestsConnectionError
+    return pywintypes
+
+
+def pywintypes_import_win() -> ModuleType:
+    import pywintypes
+    return pywintypes
+
+
+def import_pywintypes() -> ModuleType:
+    return exec_by_platform(pywintypes_import_stub, pywintypes_import_win, pywintypes_import_stub)
+
+
 # Architecture Test
 def get_architecture() -> str:
     architecture = machine().lower()