Method declarations are updated (typing)

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,15 +633,15 @@ 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
             content = file.read()
             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()
 

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()