From c5c793076948159f27ade81629ee9897ba9aba8e Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 8 Jul 2026 19:04:45 +0300 Subject: [PATCH] Method declarations are updated (typing) OsOperations: - find_executable - is_executable - get_user - makedir - rmdir - listdir - path_exists - pathsep - mkdtemp - mkstemp - read - readlines - read_binary - kill - get_pid - get_process_children LocalOperaions: - find_executable - is_executable - makedir - rmdir - listdir - path_exists - pathsep - mkdtemp - mkstemp - read - _read__text_with_encoding - _read__binary - readlines - read_binary - kill - get_pid - get_process_children RemoteOperations: - find_executable - is_executable - makedir - rmdir - listdir - path_exists - pathsep - mkdtemp - mkstemp - read - _read__text_with_encoding - _read__binary - read_binary - kill - get_pid - get_process_children --- src/local_ops.py | 46 ++++++++++++++++++++++++++++------------------ src/os_ops.py | 45 +++++++++++++++++++++++++++++---------------- src/remote_ops.py | 44 ++++++++++++++++++++++++++------------------ 3 files changed, 83 insertions(+), 52 deletions(-) diff --git a/src/local_ops.py b/src/local_ops.py index 241d5fa..217d3c1 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -390,11 +390,16 @@ def environ(self, var_name: str) -> typing.Optional[str]: def cwd(self): return os.getcwd() - def find_executable(self, executable): + def find_executable(self, executable: str) -> typing.Optional[str]: + assert type(executable) is str + assert executable != "" return find_executable(executable) - def is_executable(self, file): + def is_executable(self, file: str) -> bool: # Check if the file is executable + assert type(file) is str + assert file != "" + assert stat.S_IXUSR != 0 return (os.stat(file).st_mode & stat.S_IXUSR) == stat.S_IXUSR @@ -431,7 +436,7 @@ def makedirs( os.makedirs(path, exist_ok=True) return - def makedir(self, path: str): + def makedir(self, path: str) -> None: assert type(path) is str os.mkdir(path) return @@ -485,19 +490,21 @@ def rmdirs( # OK! return True - def rmdir(self, path: str): + def rmdir(self, path: str) -> None: assert type(path) is str os.rmdir(path) return - def listdir(self, path): + def listdir(self, path: str) -> typing.List[str]: + assert type(path) is str return os.listdir(path) - def path_exists(self, path): + def path_exists(self, path: str) -> bool: + assert type(path) is str return os.path.exists(path) @property - def pathsep(self): + def pathsep(self) -> str: os_name = self.get_name() if os_name == "posix": pathsep = ":" @@ -507,10 +514,13 @@ def pathsep(self): raise Exception("Unsupported operating system: {}".format(os_name)) return pathsep - def mkdtemp(self, prefix=None): - return tempfile.mkdtemp(prefix='{}'.format(prefix)) + def mkdtemp(self, prefix: typing.Optional[str] = None) -> str: + assert prefix is None or type(prefix) is str + return tempfile.mkdtemp(prefix=prefix) + + def mkstemp(self, prefix: typing.Optional[str] = None) -> str: + assert prefix is None or type(prefix) is str - def mkstemp(self, prefix=None): fd, filename = tempfile.mkstemp(prefix=prefix) os.close(fd) # Close the file descriptor immediately after creating the file return filename @@ -606,7 +616,7 @@ def read( filename: str, encoding: typing.Optional[str] = None, binary: bool = False, - ): + ) -> OsOperations.T_READ_RESULT: assert type(filename) is str assert encoding is None or type(encoding) is str assert type(binary) is bool @@ -623,7 +633,7 @@ def read( return self._read__text_with_encoding(filename, encoding or get_default_encoding()) - def _read__text_with_encoding(self, filename, encoding): + def _read__text_with_encoding(self, filename: str, encoding: str) -> str: assert type(filename) is str assert type(encoding) is str with open(filename, mode='r', encoding=encoding) as file: # open in a text mode @@ -631,7 +641,7 @@ def _read__text_with_encoding(self, filename, encoding): assert type(content) is str return content - def _read__binary(self, filename): + def _read__binary(self, filename: str) -> bytes: assert type(filename) is str with open(filename, 'rb') as file: # open in a binary mode content = file.read() @@ -644,7 +654,7 @@ def readlines( num_lines: int = 0, binary: bool = False, encoding: typing.Optional[str] = None, - ) -> typing.Union[typing.List[str], typing.List[bytes]]: + ) -> OsOperations.T_READLINES_RESULT: """ Read lines from a local file. If num_lines is greater than 0, only the last num_lines lines will be read. @@ -692,7 +702,7 @@ def readlines( ) # Adjust buffer size return - def read_binary(self, filename, offset): + def read_binary(self, filename: str, offset: int) -> bytes: assert type(filename) is str assert type(offset) is int @@ -728,18 +738,18 @@ def remove_file(self, filename: str) -> None: return # Processes control - def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): + def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]) -> None: # Kill the process assert type(pid) is int assert type(signal) is int or type(signal) is os_signal.Signals os.kill(pid, signal) return - def get_pid(self): + def get_pid(self) -> int: # Get current process id return os.getpid() - def get_process_children(self, pid: int): + def get_process_children(self, pid: int) -> typing.List: assert type(pid) is int return psutil.Process(pid).children() diff --git a/src/os_ops.py b/src/os_ops.py index 41a3fab..3ac6c4f 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -128,11 +128,16 @@ def environ(self, var_name: str) -> typing.Optional[str]: def cwd(self): raise NotImplementedError() - def find_executable(self, executable): + def find_executable(self, executable: str) -> typing.Optional[str]: + assert type(executable) is str + assert executable != "" raise NotImplementedError() - def is_executable(self, file): + def is_executable(self, file: str) -> bool: # Check if the file is executable + assert type(file) is str + assert file != "" + raise NotImplementedError() def set_env( @@ -145,7 +150,7 @@ def set_env( assert var_name != "" raise NotImplementedError() - def get_user(self): + def get_user(self) -> typing.Optional[str]: return self.username def get_name(self): @@ -161,7 +166,7 @@ def makedirs( assert type(remove_existing) is bool raise NotImplementedError() - def makedir(self, path: str): + def makedir(self, path: str) -> None: assert type(path) is str raise NotImplementedError() @@ -182,24 +187,28 @@ def rmdirs( assert delay >= 0 raise NotImplementedError() - def rmdir(self, path: str): + def rmdir(self, path: str) -> None: assert type(path) is str raise NotImplementedError() - def listdir(self, path): + def listdir(self, path: str) -> typing.List[str]: + assert type(path) is str raise NotImplementedError() - def path_exists(self, path): + def path_exists(self, path: str) -> bool: + assert type(path) is str raise NotImplementedError() @property - def pathsep(self): + def pathsep(self) -> str: raise NotImplementedError() - def mkdtemp(self, prefix=None): + def mkdtemp(self, prefix: typing.Optional[str] = None) -> str: + assert prefix is None or type(prefix) is str raise NotImplementedError() - def mkstemp(self, prefix=None): + def mkstemp(self, prefix: typing.Optional[str] = None) -> str: + assert prefix is None or type(prefix) is str raise NotImplementedError() def copytree(self, src: str, dst: str) -> str: @@ -235,24 +244,28 @@ def touch(self, filename: str) -> None: assert filename != "" raise NotImplementedError() + T_READ_RESULT = typing.Union[str, bytes] + def read( self, filename: str, encoding: typing.Optional[str] = None, binary: bool = False, - ): + ) -> T_READ_RESULT: assert type(filename) is str assert encoding is None or type(encoding) is str assert type(binary) is bool raise NotImplementedError() + T_READLINES_RESULT = typing.Union[typing.List[str], typing.List[bytes]] + def readlines( self, filename: str, num_lines: int = 0, binary: bool = False, encoding: typing.Optional[str] = None, - ) -> typing.Union[typing.List[str], typing.List[bytes]]: + ) -> T_READLINES_RESULT: """ Read lines from a local file. If num_lines is greater than 0, only the last num_lines lines will be read. @@ -264,7 +277,7 @@ def readlines( assert num_lines >= 0 raise NotImplementedError() - def read_binary(self, filename, offset): + def read_binary(self, filename: str, offset: int) -> bytes: assert type(filename) is str assert type(offset) is int assert offset >= 0 @@ -291,17 +304,17 @@ def remove_file(self, filename: str) -> None: raise NotImplementedError() # Processes control - def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): + def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]) -> None: # Kill the process assert type(pid) is int assert type(signal) is int or type(signal) is os_signal.Signals raise NotImplementedError() - def get_pid(self): + def get_pid(self) -> int: # Get current process id raise NotImplementedError() - def get_process_children(self, pid: int): + def get_process_children(self, pid: int) -> typing.List: assert type(pid) is int raise NotImplementedError() diff --git a/src/remote_ops.py b/src/remote_ops.py index 10d55c0..bf8b006 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -340,7 +340,10 @@ def cwd(self): assert type(stdout) is str return stdout.rstrip() - def find_executable(self, executable): + def find_executable(self, executable: str) -> typing.Optional[str]: + assert type(executable) is str + assert executable != "" + search_paths = self.environ("PATH") if not search_paths: return None @@ -353,7 +356,7 @@ def find_executable(self, executable): return None - def is_executable(self, file): + def is_executable(self, file: str) -> bool: # Check if the file is executable assert type(file) is str assert file != "" @@ -462,7 +465,7 @@ def makedirs( ) return - def makedir(self, path: str): + def makedir(self, path: str) -> None: assert type(path) is str cmd = "mkdir " + __class__._quote_path(path) self.exec_command(cmd, encoding=get_default_encoding()) @@ -539,13 +542,13 @@ def rmdirs( # OK! return True - def rmdir(self, path: str): + def rmdir(self, path: str) -> None: assert type(path) is str cmd = "rmdir " + __class__._quote_path(path) self.exec_command(cmd, encoding=get_default_encoding()) return - def listdir(self, path): + def listdir(self, path: str) -> typing.List[str]: """ List all files and directories in a directory. Args: @@ -559,7 +562,7 @@ def listdir(self, path): assert type(result) is list return result - def path_exists(self, path): + def path_exists(self, path: str) -> bool: assert type(path) is str command = "test -e " + __class__._quote_path(path) @@ -599,7 +602,7 @@ def path_exists(self, path): ) @property - def pathsep(self): + def pathsep(self) -> str: os_name = self.get_name() if os_name == "posix": pathsep = ":" @@ -609,12 +612,14 @@ def pathsep(self): raise Exception("Unsupported operating system: {}".format(os_name)) return pathsep - def mkdtemp(self, prefix=None): + def mkdtemp(self, prefix: typing.Optional[str] = None) -> str: """ Creates a temporary directory in the remote server. Args: - prefix (str): The prefix of the temporary directory name. """ + assert prefix is None or type(prefix) is str + if prefix: command_p = [ "mktemp", @@ -652,12 +657,14 @@ def mkdtemp(self, prefix=None): temp_dir = exec_output.strip() return temp_dir - def mkstemp(self, prefix=None): + def mkstemp(self, prefix: typing.Optional[str] = None) -> str: """ Creates a temporary file in the remote server. Args: - prefix (str): The prefix of the temporary directory name. """ + assert prefix is None or type(prefix) is str + if prefix: command_p = [ "mktemp", @@ -819,7 +826,7 @@ def read( filename: str, encoding: typing.Optional[str] = None, binary: bool = False, - ): + ) -> OsOperations.T_READ_RESULT: assert type(filename) is str assert encoding is None or type(encoding) is str assert type(binary) is bool @@ -836,7 +843,7 @@ def read( return self._read__text_with_encoding(filename, encoding or get_default_encoding()) - def _read__text_with_encoding(self, filename, encoding): + def _read__text_with_encoding(self, filename: str, encoding: str) -> str: assert type(filename) is str assert type(encoding) is str content = self._read__binary(filename) @@ -847,7 +854,7 @@ def _read__text_with_encoding(self, filename, encoding): assert type(content_s) is str return content_s - def _read__binary(self, filename): + def _read__binary(self, filename: str) -> bytes: assert type(filename) is str cmd = "cat " + __class__._quote_path(filename) content = self.exec_command(cmd) @@ -860,7 +867,7 @@ def readlines( num_lines: int = 0, binary: bool = False, encoding: typing.Optional[str] = None, - ) -> typing.Union[typing.List[str], typing.List[bytes]]: + ) -> OsOperations.T_READLINES_RESULT: assert type(num_lines) is int assert type(filename) is str assert type(binary) is bool @@ -904,7 +911,7 @@ def readlines( assert type(lines) is list return lines - def read_binary(self, filename, offset): + def read_binary(self, filename: str, offset: int) -> bytes: assert type(filename) is str assert type(offset) is int @@ -966,21 +973,22 @@ def remove_file(self, filename: str) -> None: return # Processes control - def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): + def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]) -> None: # Kill the process assert type(pid) is int assert type(signal) is int or type(signal) is os_signal.Signals assert int(signal) == signal cmd = "kill -{} {}".format(int(signal), pid) - return self.exec_command(cmd, encoding=get_default_encoding()) + self.exec_command(cmd, encoding=get_default_encoding()) + return - def get_pid(self): + def get_pid(self) -> int: # Get current process id x = self.exec_command("echo $$", encoding=get_default_encoding()) assert type(x) is str return int(x) - def get_process_children(self, pid: int): + def get_process_children(self, pid: int) -> typing.List: assert type(pid) is int exec_r = self.exec_command(