diff --git a/src/exceptions.py b/src/exceptions.py index b5c7dc8..9598323 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -26,11 +26,11 @@ def __init__( out: typing.Optional[T_OUT_DATA] = None, error: typing.Optional[T_ERR_DATA] = None, ): - assert message is None or type(message) == str # noqa: E721 - assert command is None or type(command) in [str, list] # noqa: E721 - assert exit_code is None or type(exit_code) == int # noqa: E721 - assert out is None or type(out) in [str, bytes] # noqa: E721 - assert error is None or type(error) in [str, bytes] # noqa: E721 + assert message is None or type(message) is str + assert command is None or type(command) in [str, list] + assert exit_code is None or type(exit_code) is int + assert out is None or type(out) in [str, bytes] + assert error is None or type(error) in [str, bytes] super().__init__(message) @@ -61,32 +61,32 @@ def message(self) -> str: msg.append(u'---- Out:\n{}'.format(self.out)) r = self.convert_and_join(msg) - assert type(r) == str # noqa: E721 + assert type(r) is str return r @property def description(self) -> typing.Optional[str]: - assert self._description is None or type(self._description) == str # noqa: E721 + assert self._description is None or type(self._description) is str return self._description @property def command(self) -> typing.Optional[T_CMD]: - assert self._command is None or type(self._command) in [str, list] # noqa: E721 + assert self._command is None or type(self._command) in [str, list] return self._command @property def exit_code(self) -> typing.Optional[int]: - assert self._exit_code is None or type(self._exit_code) == int # noqa: E721 + assert self._exit_code is None or type(self._exit_code) is int return self._exit_code @property def out(self) -> typing.Optional[T_OUT_DATA]: - assert self._out is None or type(self._out) in [str, bytes] # noqa: E721 + assert self._out is None or type(self._out) in [str, bytes] return self._out @property def error(self) -> typing.Optional[T_ERR_DATA]: - assert self._error is None or type(self._error) in [str, bytes] # noqa: E721 + assert self._error is None or type(self._error) in [str, bytes] return self._error def __repr__(self) -> str: diff --git a/src/helpers.py b/src/helpers.py index ebbf0f7..bd357f4 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -31,7 +31,7 @@ def GetDefaultEncoding(): if r: assert r is not None - assert type(r) == str # noqa: E721 + assert type(r) is str assert r != "" return r @@ -43,13 +43,13 @@ def PrepareProcessInput(input, encoding): if not input: return None - if type(input) == str: # noqa: E721 + if type(input) is str: if encoding is None: return input.encode(__class__.GetDefaultEncoding()) - assert type(encoding) == str # noqa: E721 + assert type(encoding) is str return input.encode(encoding) # It is expected! - assert type(input) == bytes # noqa: E721 + assert type(input) is bytes return input diff --git a/src/local_ops.py b/src/local_ops.py index 52b77d9..0664ab1 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -66,14 +66,14 @@ def get_single_instance() -> OsOperations: assert __class__.sm_single_instance_guard is not None if __class__.sm_single_instance is not None: - assert type(__class__.sm_single_instance) == __class__ # noqa: E721 + assert type(__class__.sm_single_instance) is __class__ return __class__.sm_single_instance with __class__.sm_single_instance_guard: if __class__.sm_single_instance is None: __class__.sm_single_instance = __class__() assert __class__.sm_single_instance is not None - assert type(__class__.sm_single_instance) == __class__ # noqa: E721 + assert type(__class__.sm_single_instance) is __class__ return __class__.sm_single_instance def get_platform(self) -> str: @@ -102,8 +102,8 @@ def _run_command__nt( exec_env: typing.Optional[dict], cwd: typing.Optional[str], ): - assert exec_env is None or type(exec_env) == dict # noqa: E721 - assert cwd is None or type(cwd) == str # noqa: E721 + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str # TODO: why don't we use the data from input? @@ -115,17 +115,17 @@ def _run_command__nt( pass else: env = os.environ.copy() - assert type(env) == dict # noqa: E721 + assert type(env) is dict for v in exec_env.items(): - assert type(v) == tuple # noqa: E721 + assert type(v) is tuple assert len(v) == 2 - assert type(v[0]) == str # noqa: E721 + assert type(v[0]) is str assert v[0] != "" if v[1] is None: env.pop(v[0], None) else: - assert type(v[1]) == str # noqa: E721 + assert type(v[1]) is str env[v[0]] = v[1] extParams["env"] = env @@ -157,14 +157,14 @@ def _run_command__generic( exec_env: typing.Optional[dict], cwd: typing.Optional[str], ): - assert exec_env is None or type(exec_env) == dict # noqa: E721 - assert cwd is None or type(cwd) == str # noqa: E721 + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str input_prepared = None if not get_process: input_prepared = Helpers.PrepareProcessInput(input, encoding) # throw - assert input_prepared is None or (type(input_prepared) == bytes) # noqa: E721 + assert input_prepared is None or type(input_prepared) is bytes extParams: typing.Dict[str, str] = dict() @@ -174,17 +174,17 @@ def _run_command__generic( pass else: env = os.environ.copy() - assert type(env) == dict # noqa: E721 + assert type(env) is dict for v in exec_env.items(): - assert type(v) == tuple # noqa: E721 + assert type(v) is tuple assert len(v) == 2 - assert type(v[0]) == str # noqa: E721 + assert type(v[0]) is str assert v[0] != "" if v[1] is None: env.pop(v[0], None) else: - assert type(v[1]) == str # noqa: E721 + assert type(v[1]) is str env[v[0]] = v[1] extParams["env"] = env @@ -207,8 +207,8 @@ def _run_command__generic( process.kill() raise ExecUtilException("Command timed out after {} seconds.".format(timeout)) - assert type(output) == bytes # noqa: E721 - assert type(error) == bytes # noqa: E721 + assert type(output) is bytes + assert type(error) is bytes if encoding: output = output.decode(encoding) @@ -222,8 +222,8 @@ def _run_command( ): """Execute a command and return the process and its output.""" - assert exec_env is None or type(exec_env) == dict # noqa: E721 - assert cwd is None or type(cwd) == str # noqa: E721 + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str if os.name == 'nt' and stdout is None: # Windows method = __class__._run_command__nt @@ -242,10 +242,10 @@ def exec_command( """ Execute a command in a subprocess and handle the output based on the provided parameters. """ - assert type(expect_error) == bool # noqa: E721 - assert type(ignore_errors) == bool # noqa: E721 - assert exec_env is None or type(exec_env) == dict # noqa: E721 - assert cwd is None or type(cwd) == str # noqa: E721 + assert type(expect_error) is bool + assert type(ignore_errors) is bool + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str process, output, error = self._run_command( cmd, shell, input, stdin, stdout, stderr, get_process, timeout, encoding, @@ -282,8 +282,8 @@ def exec_command( def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None - assert type(a) == str # noqa: E721 - assert type(parts) == tuple # noqa: E721 + assert type(a) is str + assert type(parts) is tuple return os.path.join(a, *parts) # Environment setup @@ -318,7 +318,7 @@ def makedirs(self, path, remove_existing=False): pass def makedir(self, path: str): - assert type(path) == str # noqa: E721 + assert type(path) is str os.mkdir(path) # [2025-02-03] Old name of parameter attempts is "retries". @@ -331,10 +331,10 @@ def rmdirs(self, path, ignore_errors=True, attempts=3, delay=1): :param retries: Number of attempts to remove the directory. :param delay: Delay between attempts in seconds. """ - assert type(path) == str # noqa: E721 - assert type(ignore_errors) == bool # noqa: E721 - assert type(attempts) == int # noqa: E721 - assert type(delay) == int or type(delay) == float # noqa: E721 + assert type(path) is str + assert type(ignore_errors) is bool + assert type(attempts) is int + assert type(delay) is int or type(delay) is float assert attempts > 0 assert delay >= 0 @@ -365,7 +365,7 @@ def rmdirs(self, path, ignore_errors=True, attempts=3, delay=1): return True def rmdir(self, path: str): - assert type(path) == str # noqa: E721 + assert type(path) is str os.rmdir(path) def listdir(self, path): @@ -422,7 +422,7 @@ def write(self, filename, data, truncate=False, binary=False, read_and_write=Fal if binary: mode += "b" - assert type(mode) == str # noqa: E721 + assert type(mode) is str assert mode != "" with open(filename, mode) as file: @@ -438,10 +438,10 @@ def _prepare_line_to_write(data, binary): data = __class__._prepare_data_to_write(data, binary) if binary: - assert type(data) == bytes # noqa: E721 + assert type(data) is bytes return data.rstrip(b'\n') + b'\n' - assert type(data) == str # noqa: E721 + assert type(data) is str return data.rstrip('\n') + '\n' @staticmethod @@ -467,9 +467,9 @@ def touch(self, filename): os.utime(filename, None) def read(self, filename, encoding=None, binary=False): - assert type(filename) == str # noqa: E721 - assert encoding is None or type(encoding) == str # noqa: E721 - assert type(binary) == bool # noqa: E721 + assert type(filename) is str + assert encoding is None or type(encoding) is str + assert type(binary) is bool if binary: if encoding is not None: @@ -484,18 +484,18 @@ def read(self, filename, encoding=None, binary=False): return self._read__text_with_encoding(filename, encoding or get_default_encoding()) def _read__text_with_encoding(self, filename, encoding): - assert type(filename) == str # noqa: E721 - assert type(encoding) == str # noqa: E721 + 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) == str # noqa: E721 + assert type(content) is str return content def _read__binary(self, filename): - assert type(filename) == str # noqa: E721 + assert type(filename) is str with open(filename, 'rb') as file: # open in a binary mode content = file.read() - assert type(content) == bytes # noqa: E721 + assert type(content) is bytes return content def readlines(self, filename, num_lines=0, binary=False, encoding=None): @@ -503,10 +503,10 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): Read lines from a local file. If num_lines is greater than 0, only the last num_lines lines will be read. """ - assert type(num_lines) == int # noqa: E721 - assert type(filename) == str # noqa: E721 - assert type(binary) == bool # noqa: E721 - assert encoding is None or type(encoding) == str # noqa: E721 + assert type(num_lines) is int + assert type(filename) is str + assert type(binary) is bool + assert encoding is None or type(encoding) is str assert num_lines >= 0 if binary: @@ -514,9 +514,9 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): pass elif encoding is None: encoding = get_default_encoding() - assert type(encoding) == str # noqa: E721 + assert type(encoding) is str else: - assert type(encoding) == str # noqa: E721 + assert type(encoding) is str pass mode = 'rb' if binary else 'r' @@ -546,8 +546,8 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): ) # Adjust buffer size def read_binary(self, filename, offset): - assert type(filename) == str # noqa: E721 - assert type(offset) == int # noqa: E721 + assert type(filename) is str + assert type(offset) is int if offset < 0: raise ValueError("Negative 'offset' is not supported.") @@ -555,7 +555,7 @@ def read_binary(self, filename, offset): with open(filename, 'rb') as file: # open in a binary mode file.seek(offset, os.SEEK_SET) r = file.read() - assert type(r) == bytes # noqa: E721 + assert type(r) is bytes return r def isfile(self, remote_file): @@ -566,7 +566,7 @@ def isdir(self, dirname): def get_file_size(self, filename): assert filename is not None - assert type(filename) == str # noqa: E721 + assert type(filename) is str return os.path.getsize(filename) def remove_file(self, filename): @@ -575,8 +575,8 @@ def remove_file(self, filename): # Processes control def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): # Kill the process - assert type(pid) == int # noqa: E721 - assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 + assert type(pid) is int + assert type(signal) is int or type(signal) is os_signal.Signals os.kill(pid, signal) def get_pid(self): @@ -584,11 +584,11 @@ def get_pid(self): return os.getpid() def get_process_children(self, pid): - assert type(pid) == int # noqa: E721 + assert type(pid) is int return psutil.Process(pid).children() def is_port_free(self, number: int) -> bool: - assert type(number) == int # noqa: E721 + assert type(number) is int assert number >= 0 assert number <= 65535 # OK? @@ -602,6 +602,6 @@ def is_port_free(self, number: int) -> bool: def get_tempdir(self) -> str: r = tempfile.gettempdir() assert r is not None - assert type(r) == str # noqa: E721 + assert type(r) is str assert os.path.exists(r) return r diff --git a/src/os_ops.py b/src/os_ops.py index 3c9cae4..01ee347 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -36,8 +36,8 @@ def exec_command(self, cmd, **kwargs): def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None - assert type(a) == str # noqa: E721 - assert type(parts) == tuple # noqa: E721 + assert type(a) is str + assert type(parts) is tuple raise NotImplementedError() # Environment setup @@ -69,14 +69,14 @@ def makedirs(self, path, remove_existing=False): raise NotImplementedError() def makedir(self, path: str): - assert type(path) == str # noqa: E721 + assert type(path) is str raise NotImplementedError() def rmdirs(self, path, ignore_errors=True): raise NotImplementedError() def rmdir(self, path: str): - assert type(path) == str # noqa: E721 + assert type(path) is str raise NotImplementedError() def listdir(self, path): @@ -112,8 +112,8 @@ def readlines(self, filename): raise NotImplementedError() def read_binary(self, filename, offset): - assert type(filename) == str # noqa: E721 - assert type(offset) == int # noqa: E721 + assert type(filename) is str + assert type(offset) is int assert offset >= 0 raise NotImplementedError() @@ -127,14 +127,14 @@ def get_file_size(self, filename): raise NotImplementedError() def remove_file(self, filename): - assert type(filename) == str # noqa: E721 + assert type(filename) is str raise NotImplementedError() # Processes control def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): # Kill the process - assert type(pid) == int # noqa: E721 - assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 + assert type(pid) is int + assert type(signal) is int or type(signal) is os_signal.Signals raise NotImplementedError() def get_pid(self): @@ -145,7 +145,7 @@ def get_process_children(self, pid): raise NotImplementedError() def is_port_free(self, number: int): - assert type(number) == int # noqa: E721 + assert type(number) is int raise NotImplementedError() def get_tempdir(self) -> str: diff --git a/src/raise_error.py b/src/raise_error.py index 345ed40..fe7ee9e 100644 --- a/src/raise_error.py +++ b/src/raise_error.py @@ -5,10 +5,10 @@ class RaiseError: @staticmethod def UtilityExitedWithNonZeroCode(cmd, exit_code, msg_arg, error, out): - assert type(exit_code) == int # noqa: E721 + assert type(exit_code) is int msg_arg_s = __class__._TranslateDataIntoString(msg_arg) - assert type(msg_arg_s) == str # noqa: E721 + assert type(msg_arg_s) is str msg_arg_s = msg_arg_s.strip() if msg_arg_s == "": @@ -24,8 +24,8 @@ def UtilityExitedWithNonZeroCode(cmd, exit_code, msg_arg, error, out): @staticmethod def CommandExecutionError(cmd, exit_code, message, error, out): - assert type(exit_code) == int # noqa: E721 - assert type(message) == str # noqa: E721 + assert type(exit_code) is int + assert type(message) is str assert message != "" raise ExecUtilException( @@ -40,14 +40,14 @@ def _TranslateDataIntoString(data): if data is None: return "" - if type(data) == bytes: # noqa: E721 + if type(data) is bytes: return __class__._TranslateDataIntoString__FromBinary(data) return str(data) @staticmethod def _TranslateDataIntoString__FromBinary(data): - assert type(data) == bytes # noqa: E721 + assert type(data) is bytes try: return data.decode(Helpers.GetDefaultEncoding()) diff --git a/src/remote_ops.py b/src/remote_ops.py index 1e791d9..727833d 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -24,22 +24,22 @@ class PsUtilProcessProxy: def __init__(self, ssh, pid): assert isinstance(ssh, RemoteOperations) - assert type(pid) == int # noqa: E721 + assert type(pid) is int self.ssh = ssh self.pid = pid def kill(self): assert isinstance(self.ssh, RemoteOperations) - assert type(self.pid) == int # noqa: E721 + assert type(self.pid) is int command = ["kill", str(self.pid)] self.ssh.exec_command(command, encoding=get_default_encoding()) def cmdline(self): assert isinstance(self.ssh, RemoteOperations) - assert type(self.pid) == int # noqa: E721 + assert type(self.pid) is int command = ["ps", "-p", str(self.pid), "-o", "cmd", "--no-headers"] output = self.ssh.exec_command(command, encoding=get_default_encoding()) - assert type(output) == str # noqa: E721 + assert type(output) is str cmdline = output.strip() # TODO: This code work wrong if command line contains quoted values. Yes? return cmdline.split() @@ -112,21 +112,21 @@ def exec_command( Args: - cmd (str): The command to be executed. """ - assert type(expect_error) == bool # noqa: E721 - assert type(ignore_errors) == bool # noqa: E721 - assert exec_env is None or type(exec_env) == dict # noqa: E721 - assert cwd is None or type(cwd) == str # noqa: E721 + assert type(expect_error) is bool + assert type(ignore_errors) is bool + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str input_prepared = None if not get_process: input_prepared = Helpers.PrepareProcessInput(input, encoding) # throw - assert input_prepared is None or (type(input_prepared) == bytes) # noqa: E721 + assert input_prepared is None or type(input_prepared) is bytes cmds = [] if cwd is not None: - assert type(cwd) == str # noqa: E721 + assert type(cwd) is str cmds.append(__class__._build_cmdline(["cd", cwd])) cmds.append(__class__._build_cmdline(cmd, exec_env)) @@ -134,7 +134,7 @@ def exec_command( assert len(cmds) >= 1 cmdline = ";".join(cmds) - assert type(cmdline) == str # noqa: E721 + assert type(cmdline) is str assert cmdline != "" ssh_cmd = ['ssh', self.ssh_dest] + self.ssh_args + [cmdline] @@ -150,8 +150,8 @@ def exec_command( process.kill() raise ExecUtilException("Command timed out after {} seconds.".format(timeout)) - assert type(output) == bytes # noqa: E721 - assert type(error) == bytes # noqa: E721 + assert type(output) is bytes + assert type(error) is bytes if encoding: output = output.decode(encoding) @@ -183,8 +183,8 @@ def exec_command( def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None - assert type(a) == str # noqa: E721 - assert type(parts) == tuple # noqa: E721 + assert type(a) is str + assert type(parts) is tuple return __class__._build_path(a, *parts) # Environment setup @@ -220,8 +220,8 @@ def is_executable(self, file): exit_status, output, error = self.exec_command(cmd=command, encoding=get_default_encoding(), ignore_errors=True, verbose=True) - assert type(output) == str # noqa: E721 - assert type(error) == str # noqa: E721 + assert type(output) is str + assert type(error) is str if exit_status == 0: return True @@ -275,7 +275,7 @@ def makedirs(self, path, remove_existing=False): return result def makedir(self, path: str): - assert type(path) == str # noqa: E721 + assert type(path) is str cmd = ["mkdir", path] self.exec_command(cmd) @@ -286,8 +286,8 @@ def rmdirs(self, path, ignore_errors=True): - path (str): The path to the directory to be removed. - ignore_errors (bool): If True, do not raise error if directory does not exist. """ - assert type(path) == str # noqa: E721 - assert type(ignore_errors) == bool # noqa: E721 + assert type(path) is str + assert type(ignore_errors) is bool # ENOENT = 2 - No such file or directory # ENOTDIR = 20 - Not a directory @@ -320,7 +320,7 @@ def rmdirs(self, path, ignore_errors=True): return True def rmdir(self, path: str): - assert type(path) == str # noqa: E721 + assert type(path) is str cmd = ["rmdir", path] self.exec_command(cmd) @@ -332,9 +332,9 @@ def listdir(self, path): """ command = ["ls", path] output = self.exec_command(cmd=command, encoding=get_default_encoding()) - assert type(output) == str # noqa: E721 + assert type(output) is str result = output.splitlines() - assert type(result) == list # noqa: E721 + assert type(result) is list return result def path_exists(self, path): @@ -342,8 +342,8 @@ def path_exists(self, path): exit_status, output, error = self.exec_command(cmd=command, encoding=get_default_encoding(), ignore_errors=True, verbose=True) - assert type(output) == str # noqa: E721 - assert type(error) == str # noqa: E721 + assert type(output) is str + assert type(error) is str if exit_status == 0: return True @@ -387,9 +387,9 @@ def mkdtemp(self, prefix=None): exec_exitcode, exec_output, exec_error = self.exec_command(command, verbose=True, encoding=get_default_encoding(), ignore_errors=True) - assert type(exec_exitcode) == int # noqa: E721 - assert type(exec_output) == str # noqa: E721 - assert type(exec_error) == str # noqa: E721 + assert type(exec_exitcode) is int + assert type(exec_output) is str + assert type(exec_error) is str if exec_exitcode != 0: RaiseError.CommandExecutionError( @@ -415,9 +415,9 @@ def mkstemp(self, prefix=None): exec_exitcode, exec_output, exec_error = self.exec_command(command, verbose=True, encoding=get_default_encoding(), ignore_errors=True) - assert type(exec_exitcode) == int # noqa: E721 - assert type(exec_output) == str # noqa: E721 - assert type(exec_error) == str # noqa: E721 + assert type(exec_exitcode) is int + assert type(exec_output) is str + assert type(exec_error) is str if exec_exitcode != 0: RaiseError.CommandExecutionError( @@ -474,10 +474,10 @@ def _prepare_line_to_write(data, binary, encoding): data = __class__._prepare_data_to_write(data, binary, encoding) if binary: - assert type(data) == bytes # noqa: E721 + assert type(data) is bytes return data.rstrip(b'\n') + b'\n' - assert type(data) == str # noqa: E721 + assert type(data) is str return data.rstrip('\n') + '\n' @staticmethod @@ -502,9 +502,9 @@ def touch(self, filename): self.exec_command("touch {}".format(filename)) def read(self, filename, binary=False, encoding=None): - assert type(filename) == str # noqa: E721 - assert encoding is None or type(encoding) == str # noqa: E721 - assert type(binary) == bool # noqa: E721 + assert type(filename) is str + assert encoding is None or type(encoding) is str + assert type(binary) is bool if binary: if encoding is not None: @@ -519,28 +519,28 @@ def read(self, filename, binary=False, encoding=None): return self._read__text_with_encoding(filename, encoding or get_default_encoding()) def _read__text_with_encoding(self, filename, encoding): - assert type(filename) == str # noqa: E721 - assert type(encoding) == str # noqa: E721 + assert type(filename) is str + assert type(encoding) is str content = self._read__binary(filename) - assert type(content) == bytes # noqa: E721 + assert type(content) is bytes buf0 = io.BytesIO(content) buf1 = io.TextIOWrapper(buf0, encoding=encoding) content_s = buf1.read() - assert type(content_s) == str # noqa: E721 + assert type(content_s) is str return content_s def _read__binary(self, filename): - assert type(filename) == str # noqa: E721 + assert type(filename) is str cmd = ["cat", filename] content = self.exec_command(cmd) - assert type(content) == bytes # noqa: E721 + assert type(content) is bytes return content def readlines(self, filename, num_lines=0, binary=False, encoding=None): - assert type(num_lines) == int # noqa: E721 - assert type(filename) == str # noqa: E721 - assert type(binary) == bool # noqa: E721 - assert encoding is None or type(encoding) == str # noqa: E721 + assert type(num_lines) is int + assert type(filename) is str + assert type(binary) is bool + assert encoding is None or type(encoding) is str if num_lines > 0: cmd = ["tail", "-n", str(num_lines), filename] @@ -552,34 +552,34 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): pass elif encoding is None: encoding = get_default_encoding() - assert type(encoding) == str # noqa: E721 + assert type(encoding) is str else: - assert type(encoding) == str # noqa: E721 + assert type(encoding) is str pass result = self.exec_command(cmd, encoding=encoding) assert result is not None if binary: - assert type(result) == bytes # noqa: E721 + assert type(result) is bytes lines = result.splitlines() else: - assert type(result) == str # noqa: E721 + assert type(result) is str lines = result.splitlines() - assert type(lines) == list # noqa: E721 + assert type(lines) is list return lines def read_binary(self, filename, offset): - assert type(filename) == str # noqa: E721 - assert type(offset) == int # noqa: E721 + assert type(filename) is str + assert type(offset) is int if offset < 0: raise ValueError("Negative 'offset' is not supported.") cmd = ["tail", "-c", "+{}".format(offset + 1), filename] r = self.exec_command(cmd) - assert type(r) == bytes # noqa: E721 + assert type(r) is bytes return r def isfile(self, remote_file): @@ -596,11 +596,11 @@ def get_file_size(self, filename): C_ERR_SRC = "RemoteOpertions::get_file_size" assert filename is not None - assert type(filename) == str # noqa: E721 + assert type(filename) is str cmd = ["du", "-b", filename] s = self.exec_command(cmd, encoding=get_default_encoding()) - assert type(s) == str # noqa: E721 + assert type(s) is str if len(s) == 0: raise Exception( @@ -663,8 +663,8 @@ def remove_file(self, filename): # Processes control def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): # Kill the process - assert type(pid) == int # noqa: E721 - assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 + 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()) @@ -674,7 +674,7 @@ def get_pid(self): return int(self.exec_command("echo $$", encoding=get_default_encoding())) def get_process_children(self, pid): - assert type(pid) == int # noqa: E721 + assert type(pid) is int command = ["ssh"] + self.ssh_args + [self.ssh_dest, "pgrep", "-P", str(pid)] result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) @@ -686,7 +686,7 @@ def get_process_children(self, pid): raise ExecUtilException(f"Error in getting process children. Error: {result.stderr}") def is_port_free(self, number: int) -> bool: - assert type(number) == int # noqa: E721 + assert type(number) is int assert number >= 0 assert number <= 65535 # OK? @@ -743,9 +743,9 @@ def get_tempdir(self) -> str: ignore_errors=True ) - assert type(exec_exitcode) == int # noqa: E721 - assert type(exec_output) == str # noqa: E721 - assert type(exec_error) == str # noqa: E721 + assert type(exec_exitcode) is int + assert type(exec_output) is str + assert type(exec_error) is str if exec_exitcode != 0: RaiseError.CommandExecutionError( @@ -756,14 +756,14 @@ def get_tempdir(self) -> str: out=exec_output) temp_subdir = exec_output.strip() - assert type(temp_subdir) == str # noqa: E721 + assert type(temp_subdir) is str temp_dir = os.path.dirname(temp_subdir) - assert type(temp_dir) == str # noqa: E721 + assert type(temp_dir) is str return temp_dir @staticmethod def _is_port_free__process_0(error: str) -> bool: - assert type(error) == str # noqa: E721 + assert type(error) is str # # Example of error text: # "Connection to localhost (127.0.0.1) 1024 port [tcp/*] succeeded!\n" @@ -774,7 +774,7 @@ def _is_port_free__process_0(error: str) -> bool: @staticmethod def _is_port_free__process_1(error: str) -> bool: - assert type(error) == str # noqa: E721 + assert type(error) is str # May be here is needed to check error message? return True @@ -782,24 +782,24 @@ def _is_port_free__process_1(error: str) -> bool: def _build_cmdline(cmd, exec_env: typing.Dict = None) -> str: cmd_items = __class__._create_exec_env_list(exec_env) - assert type(cmd_items) == list # noqa: E721 + assert type(cmd_items) is list cmd_items.append(__class__._ensure_cmdline(cmd)) cmdline = ';'.join(cmd_items) - assert type(cmdline) == str # noqa: E721 + assert type(cmdline) is str return cmdline @staticmethod def _ensure_cmdline(cmd) -> typing.List[str]: - if type(cmd) == str: # noqa: E721 + if type(cmd) is str: cmd_s = cmd - elif type(cmd) == list: # noqa: E721 + elif type(cmd) is list: cmd_s = subprocess.list2cmdline(cmd) else: raise ValueError("Invalid 'cmd' argument type - {0}".format(type(cmd).__name__)) - assert type(cmd_s) == str # noqa: E721 + assert type(cmd_s) is str return cmd_s @staticmethod @@ -816,24 +816,24 @@ def _create_exec_env_list(exec_env: typing.Dict) -> typing.List[str]: pass else: for envvar in exec_env.items(): - assert type(envvar) == tuple # noqa: E721 + assert type(envvar) is tuple assert len(envvar) == 2 - assert type(envvar[0]) == str # noqa: E721 + assert type(envvar[0]) is str env[envvar[0]] = envvar[1] # ---------------------------------- FINAL BUILD result: typing.List[str] = list() for envvar in env.items(): - assert type(envvar) == tuple # noqa: E721 + assert type(envvar) is tuple assert len(envvar) == 2 - assert type(envvar[0]) == str # noqa: E721 + assert type(envvar[0]) is str if envvar[1] is None: result.append("unset " + envvar[0]) else: - assert type(envvar[1]) == str # noqa: E721 + assert type(envvar[1]) is str qvalue = __class__._quote_envvar(envvar[1]) - assert type(qvalue) == str # noqa: E721 + assert type(qvalue) is str result.append("export " + envvar[0] + "=" + qvalue) continue @@ -843,7 +843,7 @@ def _create_exec_env_list(exec_env: typing.Dict) -> typing.List[str]: @staticmethod def _does_put_envvar_into_exec_cmd(name: str) -> bool: - assert type(name) == str # noqa: E721 + assert type(name) is str name = name.upper() if name.startswith("LC_"): return True @@ -853,7 +853,7 @@ def _does_put_envvar_into_exec_cmd(name: str) -> bool: @staticmethod def _quote_envvar(value: str) -> str: - assert type(value) == str # noqa: E721 + assert type(value) is str result = "\"" for ch in value: if ch == "\"": @@ -869,8 +869,8 @@ def _quote_envvar(value: str) -> str: def _build_path(a: str, *parts: str) -> str: assert a is not None assert parts is not None - assert type(a) == str # noqa: E721 - assert type(parts) == tuple # noqa: E721 + assert type(a) is str + assert type(parts) is tuple return posixpath.join(a, *parts) diff --git a/tests/helpers/global_data.py b/tests/helpers/global_data.py index 3b98ab7..f66772e 100644 --- a/tests/helpers/global_data.py +++ b/tests/helpers/global_data.py @@ -11,7 +11,7 @@ class OsOpsDescr: os_ops: OsOperations def __init__(self, sign: str, os_ops: OsOperations): - assert type(sign) == str # noqa: E721 + assert type(sign) is str assert isinstance(os_ops, OsOperations) self.sign = sign self.os_ops = os_ops diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 8e9a0ee..fb6c3a6 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -46,14 +46,14 @@ def test_get_platform(self, os_ops: OsOperations): assert isinstance(os_ops, OsOperations) p = os_ops.get_platform() assert p is not None - assert type(p) == str # noqa: E721 + assert type(p) is str assert p == sys.platform def test_get_platform__is_known(self, os_ops: OsOperations): assert isinstance(os_ops, OsOperations) p = os_ops.get_platform() assert p is not None - assert type(p) == str # noqa: E721 + assert type(p) is str assert p in {"win32", "linux"} def test_create_clone(self, os_ops: OsOperations): @@ -61,7 +61,7 @@ def test_create_clone(self, os_ops: OsOperations): clone = os_ops.create_clone() assert clone is not None assert clone is not os_ops - assert type(clone) == type(os_ops) # noqa: E721 + assert type(clone) is type(os_ops) def test_exec_command_success(self, os_ops: OsOperations): """ @@ -91,11 +91,11 @@ def test_exec_command_failure(self, os_ops: OsOperations): try: os_ops.exec_command(cmd) except ExecUtilException as e: - assert type(e.exit_code) == int # noqa: E721 + assert type(e.exit_code) is int assert e.exit_code == 127 - assert type(e.message) == str # noqa: E721 - assert type(e.error) == bytes # noqa: E721 + assert type(e.message) is str + assert type(e.error) is bytes assert e.message.startswith("Utility exited with non-zero code (127). Error:") assert "nonexistent_command" in e.message @@ -119,7 +119,7 @@ def test_exec_command_failure__expect_error(self, os_ops: OsOperations): assert exit_status == 127 assert result == b'' - assert type(error) == bytes # noqa: E721 + assert type(error) is bytes assert b"nonexistent_command" in error assert b"not found" in error @@ -136,12 +136,12 @@ def test_exec_command_with_exec_env(self, os_ops: OsOperations): response = os_ops.exec_command(cmd, exec_env=exec_env) assert response is not None - assert type(response) == bytes # noqa: E721 + assert type(response) is bytes assert response == b'Hello!\n' response = os_ops.exec_command(cmd) assert response is not None - assert type(response) == bytes # noqa: E721 + assert type(response) is bytes assert response == b'\n' def test_exec_command_with_exec_env__2(self, os_ops: OsOperations): @@ -156,7 +156,7 @@ def test_exec_command_with_exec_env__2(self, os_ops: OsOperations): logging.info("content is [{}]".format(tmp_file_content)) tmp_file = os_ops.mkstemp() - assert type(tmp_file) == str # noqa: E721 + assert type(tmp_file) is str assert tmp_file != "" logging.info("file is [{}]".format(tmp_file)) @@ -170,12 +170,12 @@ def test_exec_command_with_exec_env__2(self, os_ops: OsOperations): response = os_ops.exec_command(cmd, exec_env=exec_env) assert response is not None - assert type(response) == bytes # noqa: E721 + assert type(response) is bytes assert response == b'Hello!\n' response = os_ops.exec_command(cmd) assert response is not None - assert type(response) == bytes # noqa: E721 + assert type(response) is bytes assert response == b'\n' os_ops.remove_file(tmp_file) @@ -191,12 +191,12 @@ def test_exec_command_with_cwd(self, os_ops: OsOperations): response = os_ops.exec_command(cmd, cwd="/tmp") assert response is not None - assert type(response) == bytes # noqa: E721 + assert type(response) is bytes assert response == b'/tmp\n' response = os_ops.exec_command(cmd) assert response is not None - assert type(response) == bytes # noqa: E721 + assert type(response) is bytes assert response != b'/tmp\n' def test_exec_command__test_unset(self, os_ops: OsOperations): @@ -210,7 +210,7 @@ def test_exec_command__test_unset(self, os_ops: OsOperations): response1 = os_ops.exec_command(cmd) assert response1 is not None - assert type(response1) == bytes # noqa: E721 + assert type(response1) is bytes if response1 == b'\n': logging.warning("Environment variable {} is not defined.".format(C_ENV_NAME)) @@ -219,12 +219,12 @@ def test_exec_command__test_unset(self, os_ops: OsOperations): exec_env = {C_ENV_NAME: None} response2 = os_ops.exec_command(cmd, exec_env=exec_env) assert response2 is not None - assert type(response2) == bytes # noqa: E721 + assert type(response2) is bytes assert response2 == b'\n' response3 = os_ops.exec_command(cmd) assert response3 is not None - assert type(response3) == bytes # noqa: E721 + assert type(response3) is bytes assert response3 == response1 def test_exec_command__test_unset_dummy_var(self, os_ops: OsOperations): @@ -239,7 +239,7 @@ def test_exec_command__test_unset_dummy_var(self, os_ops: OsOperations): exec_env = {C_ENV_NAME: None} response2 = os_ops.exec_command(cmd, exec_env=exec_env) assert response2 is not None - assert type(response2) == bytes # noqa: E721 + assert type(response2) is bytes assert response2 == b'\n' def test_is_executable_true(self, os_ops: OsOperations): @@ -315,7 +315,7 @@ def test_listdir(self, os_ops: OsOperations): assert isinstance(files, list) for f in files: assert f is not None - assert type(f) == str # noqa: E721 + assert type(f) is str def test_path_exists_true__directory(self, os_ops: OsOperations): """ @@ -520,22 +520,22 @@ def test_read__text(self, os_ops: OsOperations): with open(filename, 'r') as file: # open in a text mode response0 = file.read() - assert type(response0) == str # noqa: E721 + assert type(response0) is str response1 = os_ops.read(filename) - assert type(response1) == str # noqa: E721 + assert type(response1) is str assert response1 == response0 response2 = os_ops.read(filename, encoding=None, binary=False) - assert type(response2) == str # noqa: E721 + assert type(response2) is str assert response2 == response0 response3 = os_ops.read(filename, encoding="") - assert type(response3) == str # noqa: E721 + assert type(response3) is str assert response3 == response0 response4 = os_ops.read(filename, encoding="UTF-8") - assert type(response4) == str # noqa: E721 + assert type(response4) is str assert response4 == response0 def test_read__binary(self, os_ops: OsOperations): @@ -547,10 +547,10 @@ def test_read__binary(self, os_ops: OsOperations): with open(filename, 'rb') as file: # open in a binary mode response0 = file.read() - assert type(response0) == bytes # noqa: E721 + assert type(response0) is bytes response1 = os_ops.read(filename, binary=True) - assert type(response1) == bytes # noqa: E721 + assert type(response1) is bytes assert response1 == response0 def test_read__binary_and_encoding(self, os_ops: OsOperations): @@ -577,29 +577,29 @@ def test_read_binary__spec(self, os_ops: OsOperations): with open(filename, 'rb') as file: # open in a binary mode response0 = file.read() - assert type(response0) == bytes # noqa: E721 + assert type(response0) is bytes response1 = os_ops.read_binary(filename, 0) - assert type(response1) == bytes # noqa: E721 + assert type(response1) is bytes assert response1 == response0 response2 = os_ops.read_binary(filename, 1) - assert type(response2) == bytes # noqa: E721 + assert type(response2) is bytes assert len(response2) < len(response1) assert len(response2) + 1 == len(response1) assert response2 == response1[1:] response3 = os_ops.read_binary(filename, len(response1)) - assert type(response3) == bytes # noqa: E721 + assert type(response3) is bytes assert len(response3) == 0 response4 = os_ops.read_binary(filename, len(response2)) - assert type(response4) == bytes # noqa: E721 + assert type(response4) is bytes assert len(response4) == 1 assert response4[0] == response1[len(response1) - 1] response5 = os_ops.read_binary(filename, len(response1) + 1) - assert type(response5) == bytes # noqa: E721 + assert type(response5) is bytes assert len(response5) == 0 def test_read_binary__spec__negative_offset(self, os_ops: OsOperations): @@ -622,10 +622,10 @@ def test_get_file_size(self, os_ops: OsOperations): filename = __file__ # current file sz0 = os.path.getsize(filename) - assert type(sz0) == int # noqa: E721 + assert type(sz0) is int sz1 = os_ops.get_file_size(filename) - assert type(sz1) == int # noqa: E721 + assert type(sz1) is int assert sz1 == sz0 def test_isfile_true(self, os_ops: OsOperations): @@ -713,7 +713,7 @@ def test_cwd(self, os_ops: OsOperations): v = os_ops.cwd() assert v is not None - assert type(v) == str # noqa: E721 + assert type(v) is str assert v != "" class tagWriteData001: @@ -759,11 +759,11 @@ def __init__(self, sign, source, cp_rw, cp_truncate, cp_binary, cp_data, result) ) def write_data001(self, request): assert isinstance(request, pytest.FixtureRequest) - assert type(request.param) == __class__.tagWriteData001 # noqa: E721 + assert type(request.param) is __class__.tagWriteData001 return request.param def test_write(self, write_data001: tagWriteData001, os_ops: OsOperations): - assert type(write_data001) == __class__.tagWriteData001 # noqa: E721 + assert type(write_data001) is __class__.tagWriteData001 assert isinstance(os_ops, OsOperations) mode = "w+b" if write_data001.call_param__binary else "w+" @@ -807,7 +807,7 @@ def test_is_port_free__true(self, os_ops: OsOperations): C_LIMIT = 128 ports = set(range(1024, 65535)) - assert type(ports) == set # noqa: E721 + assert type(ports) is set ok_count = 0 no_count = 0 @@ -843,11 +843,11 @@ def test_is_port_free__false(self, os_ops: OsOperations): C_LIMIT = 10 ports = set(range(1024, 65535)) - assert type(ports) == set # noqa: E721 + assert type(ports) is set def LOCAL_server(s: socket.socket): assert s is not None - assert type(s) == socket.socket # noqa: E721 + assert type(s) is socket.socket try: while True: @@ -873,7 +873,7 @@ def LOCAL_server(s: socket.socket): s.listen(10) - assert type(th) == threading.Thread # noqa: E721 + assert type(th) is threading.Thread th.start() try: @@ -902,7 +902,7 @@ def test_get_tmpdir(self, os_ops: OsOperations): assert isinstance(os_ops, OsOperations) dir = os_ops.get_tempdir() - assert type(dir) == str # noqa: E721 + assert type(dir) is str assert os_ops.path_exists(dir) assert os.path.exists(dir) @@ -927,15 +927,15 @@ def test_get_tmpdir__compare_with_py_info(self, os_ops: OsOperations): actual_dir = os_ops.get_tempdir() assert actual_dir is not None - assert type(actual_dir) == str # noqa: E721 + assert type(actual_dir) is str # -------- cmd = [sys.executable, "-c", "import tempfile;print(tempfile.gettempdir());"] expected_dir_b = os_ops.exec_command(cmd) - assert type(expected_dir_b) == bytes # noqa: E721 + assert type(expected_dir_b) is bytes expected_dir = expected_dir_b.decode() - assert type(expected_dir) == str # noqa: E721 + assert type(expected_dir) is str assert actual_dir + "\n" == expected_dir return @@ -945,7 +945,7 @@ class tagData_OS_OPS__NUMS: def __init__(self, os_ops_descr: OsOpsDescr, nums: int): assert isinstance(os_ops_descr, OsOpsDescr) - assert type(nums) == int # noqa: E721 + assert type(nums) is int self.os_ops_descr = os_ops_descr self.nums = nums @@ -964,11 +964,11 @@ def data001(self, request: pytest.FixtureRequest) -> tagData_OS_OPS__NUMS: return request.param def test_mkdir__mt(self, data001: tagData_OS_OPS__NUMS): - assert type(data001) == __class__.tagData_OS_OPS__NUMS # noqa: E721 + assert type(data001) is __class__.tagData_OS_OPS__NUMS N_WORKERS = 4 N_NUMBERS = data001.nums - assert type(N_NUMBERS) == int # noqa: E721 + assert type(N_NUMBERS) is int os_ops = data001.os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) @@ -982,8 +982,8 @@ def test_mkdir__mt(self, data001: tagData_OS_OPS__NUMS): assert os.path.exists(lock_dir) def MAKE_PATH(lock_dir: str, num: int) -> str: - assert type(lock_dir) == str # noqa: E721 - assert type(num) == int # noqa: E721 + assert type(lock_dir) is str + assert type(num) is int return os.path.join(lock_dir, str(num) + ".lock") def LOCAL_WORKER(os_ops: OsOperations, @@ -992,21 +992,21 @@ def LOCAL_WORKER(os_ops: OsOperations, cNumbers: int, reservedNumbers: typing.Set[int]) -> None: assert isinstance(os_ops, OsOperations) - assert type(workerID) == int # noqa: E721 - assert type(lock_dir) == str # noqa: E721 - assert type(cNumbers) == int # noqa: E721 - assert type(reservedNumbers) == set # noqa: E721 + assert type(workerID) is int + assert type(lock_dir) is str + assert type(cNumbers) is int + assert type(reservedNumbers) is set assert cNumbers > 0 assert len(reservedNumbers) == 0 assert os.path.exists(lock_dir) def LOG_INFO(template: str, *args) -> None: - assert type(template) == str # noqa: E721 - assert type(args) == tuple # noqa: E721 + assert type(template) is str + assert type(args) is tuple msg = template.format(*args) - assert type(msg) == str # noqa: E721 + assert type(msg) is str logging.info("[Worker #{}] {}".format(workerID, msg)) return @@ -1085,7 +1085,7 @@ class tadWorkerData: nWorkers = 0 - assert type(workerDatas) == list # noqa: E721 + assert type(workerDatas) is list for i in range(len(workerDatas)): worker = workerDatas[i].future @@ -1122,7 +1122,7 @@ class tadWorkerData: logging.info("Worker #{} is checked ...".format(i)) workerNumbers = workerDatas[i].reservedNumbers - assert type(workerNumbers) == set # noqa: E721 + assert type(workerNumbers) is set for n in workerNumbers: if n < 0 or n >= N_NUMBERS: @@ -1226,14 +1226,14 @@ class tadWorkerData: ) def kill_signal_id(self, request: pytest.FixtureRequest) -> T_KILL_SIGNAL_DESCR: assert isinstance(request, pytest.FixtureRequest) - assert type(request.param) == tuple # noqa: E721 + assert type(request.param) is tuple return request.param def test_kill_signal( self, kill_signal_id: T_KILL_SIGNAL_DESCR, ): - assert type(kill_signal_id) == tuple # noqa: E721 + assert type(kill_signal_id) is tuple assert "{}".format(kill_signal_id[1]) == kill_signal_id[2] assert "{}".format(int(kill_signal_id[1])) == kill_signal_id[2] @@ -1246,7 +1246,7 @@ def test_kill( Test listdir for listing directory contents. """ assert isinstance(os_ops, OsOperations) - assert type(kill_signal_id) == tuple # noqa: E721 + assert type(kill_signal_id) is tuple cmd = [ sys.executable, @@ -1261,9 +1261,9 @@ def test_kill( ) assert proc is not None - assert type(proc) == subprocess.Popen # noqa: E721 + assert type(proc) is subprocess.Popen proc_pid = proc.pid - assert type(proc_pid) == int # noqa: E721 + assert type(proc_pid) is int logging.info("Test process pid is {}".format(proc_pid)) logging.info("Get this test process ...") @@ -1315,7 +1315,7 @@ def test_kill__unk_pid( Test listdir for listing directory contents. """ assert isinstance(os_ops, OsOperations) - assert type(kill_signal_id) == tuple # noqa: E721 + assert type(kill_signal_id) is tuple cmd = [ sys.executable, @@ -1332,20 +1332,20 @@ def test_kill__unk_pid( ) assert proc is not None - assert type(proc) == subprocess.Popen # noqa: E721 + assert type(proc) is subprocess.Popen proc_pid = proc.pid - assert type(proc_pid) == int # noqa: E721 + assert type(proc_pid) is int logging.info("Test process pid is {}".format(proc_pid)) logging.info("Wait for finish ...") pout, perr = proc.communicate() logging.info("STDOUT: {}".format(pout)) logging.info("STDERR: {}".format(pout)) - assert type(pout) == str # noqa: E721 - assert type(perr) == str # noqa: E721 + assert type(pout) is str + assert type(perr) is str assert pout == "a\n" assert perr == "b\n" - assert type(proc.returncode) == int # noqa: E721 + assert type(proc.returncode) is int assert proc.returncode == 0 logging.info("Try to get this test process ...") @@ -1387,10 +1387,10 @@ def test_kill__unk_pid( logging.info("Our exception has type [{}]".format(type(x.value).__name__)) if type(os_ops).__name__ == "LocalOsOperations": - assert type(x.value) == ProcessLookupError # noqa: E721 + assert type(x.value) is ProcessLookupError assert "No such process" in str(x.value) elif type(os_ops).__name__ == "RemoteOsOperations": - assert type(x.value) == ExecUtilException # noqa: E721 + assert type(x.value) is ExecUtilException assert "No such process" in str(x.value) else: RuntimeError("Unknown os_ops type: {}".format(type(os_ops).__name__)) diff --git a/tests/test_os_ops_local.py b/tests/test_os_ops_local.py index ae45c87..c2f7568 100644 --- a/tests/test_os_ops_local.py +++ b/tests/test_os_ops_local.py @@ -49,11 +49,11 @@ def test_cwd(self, os_ops: OsOperations): v = os_ops.cwd() assert v is not None - assert type(v) == str # noqa: E721 + assert type(v) is str expectedValue = os.getcwd() assert expectedValue is not None - assert type(expectedValue) == str # noqa: E721 + assert type(expectedValue) is str assert expectedValue != "" # research # Comp result diff --git a/tests/test_os_ops_remote.py b/tests/test_os_ops_remote.py index 0cdcb78..7dcd6e1 100755 --- a/tests/test_os_ops_remote.py +++ b/tests/test_os_ops_remote.py @@ -25,20 +25,20 @@ def test_rmdirs__try_to_delete_file(self, os_ops: OsOperations): assert isinstance(os_ops, OsOperations) path = os_ops.mkstemp() - assert type(path) == str # noqa: E721 + assert type(path) is str assert os.path.exists(path) with pytest.raises(ExecUtilException) as x: os_ops.rmdirs(path, ignore_errors=False) assert os.path.exists(path) - assert type(x.value) == ExecUtilException # noqa: E721 - assert type(x.value.description) == str # noqa: E721 + assert type(x.value) is ExecUtilException + assert type(x.value.description) is str assert x.value.description == "Utility exited with non-zero code (20). Error: `cannot remove '" + path + "': it is not a directory`" assert x.value.message.startswith(x.value.description) - assert type(x.value.error) == str # noqa: E721 + assert type(x.value.error) is str assert x.value.error.strip() == "cannot remove '" + path + "': it is not a directory" - assert type(x.value.exit_code) == int # noqa: E721 + assert type(x.value.exit_code) is int assert x.value.exit_code == 20 def test_read__unknown_file(self, os_ops: OsOperations):