Revision&Refactoring

src/exceptions.py

@@ -147,7 +147,7 @@ def __init__(
         timeout: T_OS_TIMEOUT,
         output: typing.Optional[T_OUT_DATA] = None,
         error: typing.Optional[T_ERR_DATA] = None,
-        source: typing.Optional[str] = None
+        source: typing.Optional[str] = None,
     ):
         assert type(cmd) in [str, list]
         assert type(timeout) in [int, float]

src/local_ops.py

@@ -1,5 +1,24 @@
 from __future__ import annotations
 
+from .exceptions import ExecUtilException
+from .exceptions import ExecTimeoutException
+from .exceptions import InvalidOperationException
+from .os_ops import ConnectionParams, OsOperations, get_default_encoding
+from .os_ops import OsProcessController
+from .os_ops import OsCommandResult
+from .os_ops import T_OS_CMD
+from .os_ops import T_OS_SIGNAL
+from .os_ops import T_OS_TIMEOUT
+from .os_ops import T_OS_IO
+from .os_ops import T_OS_IO_ID
+from .os_ops import T_OS_EXEC_INPUT
+from .os_ops import T_OS_EXEC_ENV
+from .raise_error import RaiseError
+from .helpers import Helpers
+
+from shutil import which as find_executable
+from shutil import rmtree
+
 import getpass
 import logging
 import os
@@ -21,25 +40,6 @@
 import pathlib
 import io
 
-from .exceptions import ExecUtilException
-from .exceptions import ExecTimeoutException
-from .exceptions import InvalidOperationException
-from .os_ops import ConnectionParams, OsOperations, get_default_encoding
-from .os_ops import OsProcessController
-from .os_ops import OsCommandResult
-from .os_ops import T_OS_CMD
-from .os_ops import T_OS_SIGNAL
-from .os_ops import T_OS_TIMEOUT
-from .os_ops import T_OS_IO
-from .os_ops import T_OS_IO_ID
-from .os_ops import T_OS_EXEC_INPUT
-from .os_ops import T_OS_EXEC_ENV
-from .raise_error import RaiseError
-from .helpers import Helpers
-
-from shutil import which as find_executable
-from shutil import rmtree
-
 CMD_TIMEOUT_SEC = 60
 
 
@@ -385,7 +385,7 @@ def _run_command__generic(
             cwd=cwd,
             text=get_process and (encoding is not None),
             encoding=encoding if get_process else None,
-            **extParams
+            **extParams,
         )
         assert process is not None
         assert isinstance(process, subprocess.Popen)
@@ -498,7 +498,8 @@ def exec_command(
                 exit_code=run_r[0],
                 msg_arg=run_r[2] or run_r[1],
                 error=run_r[2],
-                out=run_r[1])
+                out=run_r[1],
+            )
 
         if verbose:
             return run_r
@@ -825,8 +826,8 @@ def write(
         truncate: bool = False,
         binary: bool = False,
         read_and_write: bool = False,
-        encoding: typing.Optional[str] = None
-    ):
+        encoding: typing.Optional[str] = None,
+    ) -> None:
         """
         Write data to a file locally
         Args:

src/os_ops.py

@@ -195,7 +195,7 @@ def exec_command(
         timeout=None,
         ignore_errors=False,
         exec_env: typing.Optional[dict] = None,
-        cwd: typing.Optional[str] = None
+        cwd: typing.Optional[str] = None,
     ) -> T_EXEC_COMMAND_RESULT:
         assert type(cmd) is str or type(cmd) is list
         assert type(verbose) is bool
@@ -218,7 +218,7 @@ def popen(
         stdout: typing.Optional[T_OS_IO_ID] = None,
         stderr: typing.Optional[T_OS_IO_ID] = None,
         exec_env: typing.Optional[T_OS_EXEC_ENV] = None,
-        cwd: typing.Optional[str] = None
+        cwd: typing.Optional[str] = None,
     ) -> OsProcessController:
         assert type(cmd) is str or type(cmd) is list
         assert text is None or type(text) is bool
@@ -394,7 +394,7 @@ def write(
         binary: bool = False,
         read_and_write: bool = False,
         encoding: typing.Optional[str] = None
-    ):
+    ) -> None:
         assert type(filename) is str
         assert encoding is None or type(encoding) is str
         assert data is not None
@@ -491,7 +491,7 @@ def get_process_children(self, pid: int) -> typing.List:
         assert type(pid) is int
         RaiseError.MethodIsNotImplemented(__class__, "get_process_children")
 
-    def is_port_free(self, number: int):
+    def is_port_free(self, number: int) -> bool:
         assert type(number) is int
         RaiseError.MethodIsNotImplemented(__class__, "is_port_free")
 

src/raise_error.py

@@ -34,7 +34,13 @@ def PropertyIsNotImplemented(
         raise NotImplementedError(err_msg)
 
     @staticmethod
-    def UtilityExitedWithNonZeroCode(cmd, exit_code, msg_arg, error, out) -> typing.NoReturn:
+    def UtilityExitedWithNonZeroCode(
+        cmd,
+        exit_code,
+        msg_arg,
+        error,
+        out,
+    ) -> typing.NoReturn:
         assert type(exit_code) is int
 
         msg_arg_s = __class__._TranslateDataIntoString(msg_arg)
@@ -50,10 +56,17 @@ def UtilityExitedWithNonZeroCode(cmd, exit_code, msg_arg, error, out) -> typing.
             command=cmd,
             exit_code=exit_code,
             out=out,
-            error=error)
+            error=error,
+        )
 
     @staticmethod
-    def CommandExecutionError(cmd, exit_code, message, error, out) -> typing.NoReturn:
+    def CommandExecutionError(
+        cmd,
+        exit_code,
+        message,
+        error,
+        out,
+    ) -> typing.NoReturn:
         assert type(exit_code) is int
         assert type(message) is str
         assert message != ""
@@ -63,15 +76,18 @@ def CommandExecutionError(cmd, exit_code, message, error, out) -> typing.NoRetur
             command=cmd,
             exit_code=exit_code,
             out=out,
-            error=error)
+            error=error,
+        )
 
     @staticmethod
     def _TranslateDataIntoString(data):
         if data is None:
             return ""
 
         if type(data) is bytes:
-            return __class__._TranslateDataIntoString__FromBinary(data)
+            return __class__._TranslateDataIntoString__FromBinary(
+                data,
+            )
 
         return str(data)
 

src/remote_ops.py

@@ -1,21 +1,5 @@
 from __future__ import annotations
 
-import getpass
-import os
-import posixpath
-import subprocess
-import io
-import logging
-import typing
-import copy
-import re
-import signal as os_signal
-import time
-import datetime
-import shlex
-import threading
-import warnings
-
 from .exceptions import ExecUtilException
 from .exceptions import ExecTimeoutException
 from .exceptions import InvalidOperationException
@@ -33,19 +17,37 @@
 from .helpers import Helpers
 from .static_config import OsOperationStaticConfig
 
+import getpass
+import os
+import posixpath
+import subprocess
+import io
+import logging
+import typing
+import copy
+import re
+import signal as os_signal
+import time
+import datetime
+import shlex
+import threading
+import warnings
+
 
 class PsUtilProcessProxy:
     def __init__(self, ssh, pid):
         assert isinstance(ssh, RemoteOperations)
         assert type(pid) is int
         self.ssh = ssh
         self.pid = pid
+        return
 
     def kill(self):
         assert isinstance(self.ssh, RemoteOperations)
         assert type(self.pid) is int
         command = ["kill", str(self.pid)]
         self.ssh._transport_run(command, encoding=get_default_encoding())
+        return
 
     def cmdline(self):
         assert isinstance(self.ssh, RemoteOperations)
@@ -425,7 +427,7 @@ def exec_command(
         timeout=None,
         ignore_errors=False,
         exec_env: typing.Optional[T_ENVS] = None,
-        cwd: typing.Optional[str] = None
+        cwd: typing.Optional[str] = None,
     ) -> OsOperations.T_EXEC_COMMAND_RESULT:
         """
         Execute a command in the SSH session.
@@ -533,7 +535,8 @@ def exec_command(
                 exit_code=run_r[0],
                 msg_arg=error,
                 error=run_r[2],
-                out=run_r[1])
+                out=run_r[1],
+            )
 
         if verbose:
             return run_r
@@ -550,7 +553,7 @@ def popen(
         stdout: typing.Optional[T_OS_IO_ID] = None,
         stderr: typing.Optional[T_OS_IO_ID] = None,
         exec_env: typing.Optional[T_OS_EXEC_ENV] = None,
-        cwd: typing.Optional[str] = None
+        cwd: typing.Optional[str] = None,
     ) -> OsProcessController:
         assert type(cmd) is str or type(cmd) is list
         assert text is None or type(text) is bool
@@ -639,7 +642,7 @@ def popen(
             ]
 
             ping_pong_script2_s = self._join_command_arguments(
-                ping_pong_script2
+                ping_pong_script2,
             )
 
             # Run script within isolated env to get a true return codes of kill/terminate
@@ -1021,7 +1024,7 @@ def makedirs(
                 "&&",
                 "mkdir",
                 "-p",
-                path_q
+                path_q,
             ]
         else:
             cmd_p = [
@@ -1077,7 +1080,7 @@ def rmdirs(
             "elif", "[", "-e", path_q, "]", ";",
             "then", "{", "echo", "cannot remove " + path_q + ": it is not a directory", ">&2", ";", "exit", "20", ";", "}", ";",
             "else", "{", "echo", "directory " + path_q + " does not exist", ">&2", ";", "exit", "2", ";", "}", ";",
-            "fi"
+            "fi",
         ]
 
         cmd1 = " ".join(cmd1_p)
@@ -1160,7 +1163,8 @@ def path_exists(self, path: str) -> bool:
 
         errMsg = "Test operation returns an unknown result code: {0}. Path is [{1}].".format(
             exec_r.returncode,
-            path)
+            path,
+        )
 
         RaiseError.CommandExecutionError(
             cmd=command,
@@ -1290,8 +1294,8 @@ def write(
         truncate: bool = False,
         binary: bool = False,
         read_and_write: bool = False,
-        encoding: typing.Optional[str] = None
-    ):
+        encoding: typing.Optional[str] = None,
+    ) -> None:
         assert type(filename) is str
         assert encoding is None or type(encoding) is str
         assert data is not None
@@ -1580,7 +1584,7 @@ def get_process_children(self, pid: int) -> typing.List:
 
         if exec_r.returncode == 100:
             err_msg = "Failed to get process children. Reason: No such process with PID {}.".format(
-                pid
+                pid,
             )
 
             raise ExecUtilException(
@@ -1787,7 +1791,7 @@ def _transport_popen(
         stdout: typing.Optional[T_OS_IO_ID] = None,
         stderr: typing.Optional[T_OS_IO_ID] = None,
         exec_env: typing.Optional[T_OS_EXEC_ENV] = None,
-        cwd: typing.Optional[str] = None
+        cwd: typing.Optional[str] = None,
     ) -> subprocess.Popen:
         assert type(cmd) in [str, list]
         assert text is None or type(text) is bool