From 9ac0450ac71739f2568a783d5eb7a202c371b0be Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 13:56:13 +0300 Subject: [PATCH 1/9] [E721] ... is int --- src/exceptions.py | 4 ++-- src/local_ops.py | 12 ++++++------ src/os_ops.py | 6 +++--- src/raise_error.py | 4 ++-- src/remote_ops.py | 22 +++++++++++----------- tests/test_os_ops_common.py | 22 +++++++++++----------- tests/test_os_ops_remote.py | 2 +- 7 files changed, 36 insertions(+), 36 deletions(-) diff --git a/src/exceptions.py b/src/exceptions.py index b5c7dc8..fdfc7be 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -28,7 +28,7 @@ def __init__( ): 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 exit_code is None or type(exit_code) is int assert out is None or type(out) in [str, bytes] # noqa: E721 assert error is None or type(error) in [str, bytes] # noqa: E721 @@ -76,7 +76,7 @@ def command(self) -> typing.Optional[T_CMD]: @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 diff --git a/src/local_ops.py b/src/local_ops.py index 52b77d9..9deefa8 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -333,7 +333,7 @@ def rmdirs(self, path, ignore_errors=True, attempts=3, delay=1): """ assert type(path) == str # noqa: E721 assert type(ignore_errors) == bool # noqa: E721 - assert type(attempts) == int # noqa: E721 + assert type(attempts) is int assert type(delay) == int or type(delay) == float # noqa: E721 assert attempts > 0 assert delay >= 0 @@ -503,7 +503,7 @@ 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(num_lines) is int assert type(filename) == str # noqa: E721 assert type(binary) == bool # noqa: E721 assert encoding is None or type(encoding) == str # noqa: E721 @@ -547,7 +547,7 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): def read_binary(self, filename, offset): assert type(filename) == str # noqa: E721 - assert type(offset) == int # noqa: E721 + assert type(offset) is int if offset < 0: raise ValueError("Negative 'offset' is not supported.") @@ -575,7 +575,7 @@ 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(pid) is int assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 os.kill(pid, signal) @@ -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? diff --git a/src/os_ops.py b/src/os_ops.py index 3c9cae4..3ea30ed 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -113,7 +113,7 @@ def readlines(self, filename): def read_binary(self, filename, offset): assert type(filename) == str # noqa: E721 - assert type(offset) == int # noqa: E721 + assert type(offset) is int assert offset >= 0 raise NotImplementedError() @@ -133,7 +133,7 @@ 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(pid) is int assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 raise NotImplementedError() @@ -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..0ef6549 100644 --- a/src/raise_error.py +++ b/src/raise_error.py @@ -5,7 +5,7 @@ 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 @@ -24,7 +24,7 @@ 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(exit_code) is int assert type(message) == str # noqa: E721 assert message != "" diff --git a/src/remote_ops.py b/src/remote_ops.py index 1e791d9..ecbe9c9 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -24,19 +24,19 @@ 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 @@ -387,7 +387,7 @@ 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_exitcode) is int assert type(exec_output) == str # noqa: E721 assert type(exec_error) == str # noqa: E721 @@ -415,7 +415,7 @@ 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_exitcode) is int assert type(exec_output) == str # noqa: E721 assert type(exec_error) == str # noqa: E721 @@ -537,7 +537,7 @@ def _read__binary(self, filename): return content def readlines(self, filename, num_lines=0, binary=False, encoding=None): - assert type(num_lines) == int # noqa: E721 + assert type(num_lines) is int assert type(filename) == str # noqa: E721 assert type(binary) == bool # noqa: E721 assert encoding is None or type(encoding) == str # noqa: E721 @@ -572,7 +572,7 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): def read_binary(self, filename, offset): assert type(filename) == str # noqa: E721 - assert type(offset) == int # noqa: E721 + assert type(offset) is int if offset < 0: raise ValueError("Negative 'offset' is not supported.") @@ -663,7 +663,7 @@ 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(pid) is int assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 assert int(signal) == signal cmd = "kill -{} {}".format(int(signal), pid) @@ -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,7 +743,7 @@ def get_tempdir(self) -> str: ignore_errors=True ) - assert type(exec_exitcode) == int # noqa: E721 + assert type(exec_exitcode) is int assert type(exec_output) == str # noqa: E721 assert type(exec_error) == str # noqa: E721 diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 8e9a0ee..5426b66 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -91,7 +91,7 @@ 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 @@ -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): @@ -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 @@ -968,7 +968,7 @@ def test_mkdir__mt(self, data001: 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) @@ -983,7 +983,7 @@ def test_mkdir__mt(self, data001: tagData_OS_OPS__NUMS): def MAKE_PATH(lock_dir: str, num: int) -> str: assert type(lock_dir) == str # noqa: E721 - assert type(num) == int # noqa: E721 + assert type(num) is int return os.path.join(lock_dir, str(num) + ".lock") def LOCAL_WORKER(os_ops: OsOperations, @@ -992,9 +992,9 @@ 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(workerID) is int assert type(lock_dir) == str # noqa: E721 - assert type(cNumbers) == int # noqa: E721 + assert type(cNumbers) is int assert type(reservedNumbers) == set # noqa: E721 assert cNumbers > 0 assert len(reservedNumbers) == 0 @@ -1263,7 +1263,7 @@ def test_kill( assert proc is not None assert type(proc) == subprocess.Popen # noqa: E721 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 ...") @@ -1334,7 +1334,7 @@ def test_kill__unk_pid( assert proc is not None assert type(proc) == subprocess.Popen # noqa: E721 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 ...") @@ -1345,7 +1345,7 @@ def test_kill__unk_pid( assert type(perr) == str # noqa: E721 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 ...") diff --git a/tests/test_os_ops_remote.py b/tests/test_os_ops_remote.py index 0cdcb78..3bca117 100755 --- a/tests/test_os_ops_remote.py +++ b/tests/test_os_ops_remote.py @@ -38,7 +38,7 @@ def test_rmdirs__try_to_delete_file(self, os_ops: OsOperations): assert x.value.message.startswith(x.value.description) assert type(x.value.error) == str # noqa: E721 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): From 5637e5791c61e51270a38442656dfba848adcaa2 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 13:57:24 +0300 Subject: [PATCH 2/9] [E721] ... is str --- src/exceptions.py | 6 +-- src/helpers.py | 4 +- src/local_ops.py | 54 ++++++++++----------- src/os_ops.py | 10 ++-- src/raise_error.py | 4 +- src/remote_ops.py | 94 ++++++++++++++++++------------------ tests/helpers/global_data.py | 2 +- tests/test_os_ops_common.py | 40 +++++++-------- tests/test_os_ops_local.py | 4 +- tests/test_os_ops_remote.py | 4 +- 10 files changed, 111 insertions(+), 111 deletions(-) diff --git a/src/exceptions.py b/src/exceptions.py index fdfc7be..1faa900 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -26,7 +26,7 @@ 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 message is None or type(message) is str assert command is None or type(command) in [str, list] # noqa: E721 assert exit_code is None or type(exit_code) is int assert out is None or type(out) in [str, bytes] # noqa: E721 @@ -61,12 +61,12 @@ 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 diff --git a/src/helpers.py b/src/helpers.py index ebbf0f7..2e5c725 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 @@ -47,7 +47,7 @@ def PrepareProcessInput(input, encoding): 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! diff --git a/src/local_ops.py b/src/local_ops.py index 9deefa8..918785d 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -103,7 +103,7 @@ def _run_command__nt( 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 cwd is None or type(cwd) is str # TODO: why don't we use the data from input? @@ -119,13 +119,13 @@ def _run_command__nt( for v in exec_env.items(): assert type(v) == tuple # noqa: E721 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 @@ -158,7 +158,7 @@ def _run_command__generic( 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 cwd is None or type(cwd) is str input_prepared = None if not get_process: @@ -178,13 +178,13 @@ def _run_command__generic( for v in exec_env.items(): assert type(v) == tuple # noqa: E721 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 @@ -223,7 +223,7 @@ 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 cwd is None or type(cwd) is str if os.name == 'nt' and stdout is None: # Windows method = __class__._run_command__nt @@ -245,7 +245,7 @@ def exec_command( 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 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,7 +282,7 @@ 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(a) is str assert type(parts) == tuple # noqa: E721 return os.path.join(a, *parts) @@ -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,7 +331,7 @@ 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(path) is str assert type(ignore_errors) == bool # noqa: E721 assert type(attempts) is int assert type(delay) == int or type(delay) == float # noqa: E721 @@ -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: @@ -441,7 +441,7 @@ def _prepare_line_to_write(data, binary): assert type(data) == bytes # noqa: E721 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,8 +467,8 @@ 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(filename) is str + assert encoding is None or type(encoding) is str assert type(binary) == bool # noqa: E721 if binary: @@ -484,15 +484,15 @@ 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 @@ -504,9 +504,9 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): If num_lines is greater than 0, only the last num_lines lines will be read. """ assert type(num_lines) is int - assert type(filename) == str # noqa: E721 + assert type(filename) is str assert type(binary) == bool # noqa: E721 - assert encoding is None or type(encoding) == str # noqa: E721 + 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,7 +546,7 @@ 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(filename) is str assert type(offset) is int if offset < 0: @@ -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): @@ -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 3ea30ed..326f4a5 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -36,7 +36,7 @@ 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(a) is str assert type(parts) == tuple # noqa: E721 raise NotImplementedError() @@ -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,7 +112,7 @@ def readlines(self, filename): raise NotImplementedError() def read_binary(self, filename, offset): - assert type(filename) == str # noqa: E721 + assert type(filename) is str assert type(offset) is int assert offset >= 0 raise NotImplementedError() @@ -127,7 +127,7 @@ 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 diff --git a/src/raise_error.py b/src/raise_error.py index 0ef6549..d6f28db 100644 --- a/src/raise_error.py +++ b/src/raise_error.py @@ -8,7 +8,7 @@ def UtilityExitedWithNonZeroCode(cmd, exit_code, msg_arg, error, out): 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 == "": @@ -25,7 +25,7 @@ def UtilityExitedWithNonZeroCode(cmd, exit_code, msg_arg, error, out): @staticmethod def CommandExecutionError(cmd, exit_code, message, error, out): assert type(exit_code) is int - assert type(message) == str # noqa: E721 + assert type(message) is str assert message != "" raise ExecUtilException( diff --git a/src/remote_ops.py b/src/remote_ops.py index ecbe9c9..78f01e7 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -39,7 +39,7 @@ def cmdline(self): 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() @@ -115,7 +115,7 @@ def exec_command( 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 cwd is None or type(cwd) is str input_prepared = None if not get_process: @@ -126,7 +126,7 @@ def exec_command( 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] @@ -183,7 +183,7 @@ 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(a) is str assert type(parts) == tuple # noqa: E721 return __class__._build_path(a, *parts) @@ -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,7 +286,7 @@ 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(path) is str assert type(ignore_errors) == bool # noqa: E721 # ENOENT = 2 - No such file or 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,7 +332,7 @@ 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 return result @@ -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 @@ -388,8 +388,8 @@ 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) is int - assert type(exec_output) == str # noqa: E721 - assert type(exec_error) == str # noqa: E721 + assert type(exec_output) is str + assert type(exec_error) is str if exec_exitcode != 0: RaiseError.CommandExecutionError( @@ -416,8 +416,8 @@ 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) is int - assert type(exec_output) == str # noqa: E721 - assert type(exec_error) == str # noqa: E721 + assert type(exec_output) is str + assert type(exec_error) is str if exec_exitcode != 0: RaiseError.CommandExecutionError( @@ -477,7 +477,7 @@ def _prepare_line_to_write(data, binary, encoding): assert type(data) == bytes # noqa: E721 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,8 +502,8 @@ 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(filename) is str + assert encoding is None or type(encoding) is str assert type(binary) == bool # noqa: E721 if binary: @@ -519,18 +519,18 @@ 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 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 @@ -538,9 +538,9 @@ def _read__binary(self, filename): def readlines(self, filename, num_lines=0, binary=False, encoding=None): assert type(num_lines) is int - assert type(filename) == str # noqa: E721 + assert type(filename) is str assert type(binary) == bool # noqa: E721 - assert encoding is None or type(encoding) == str # noqa: E721 + assert encoding is None or type(encoding) is str if num_lines > 0: cmd = ["tail", "-n", str(num_lines), filename] @@ -552,9 +552,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 result = self.exec_command(cmd, encoding=encoding) @@ -564,14 +564,14 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): assert type(result) == bytes # noqa: E721 lines = result.splitlines() else: - assert type(result) == str # noqa: E721 + assert type(result) is str lines = result.splitlines() assert type(lines) == list # noqa: E721 return lines def read_binary(self, filename, offset): - assert type(filename) == str # noqa: E721 + assert type(filename) is str assert type(offset) is int if offset < 0: @@ -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( @@ -744,8 +744,8 @@ def get_tempdir(self) -> str: ) assert type(exec_exitcode) is int - assert type(exec_output) == str # noqa: E721 - assert type(exec_error) == str # noqa: E721 + 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 @@ -787,7 +787,7 @@ def _build_cmdline(cmd, exec_env: typing.Dict = None) -> str: 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 @@ -799,7 +799,7 @@ def _ensure_cmdline(cmd) -> typing.List[str]: 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 @@ -818,7 +818,7 @@ def _create_exec_env_list(exec_env: typing.Dict) -> typing.List[str]: for envvar in exec_env.items(): assert type(envvar) == tuple # noqa: E721 assert len(envvar) == 2 - assert type(envvar[0]) == str # noqa: E721 + assert type(envvar[0]) is str env[envvar[0]] = envvar[1] # ---------------------------------- FINAL BUILD @@ -826,14 +826,14 @@ def _create_exec_env_list(exec_env: typing.Dict) -> typing.List[str]: for envvar in env.items(): assert type(envvar) == tuple # noqa: E721 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,7 +869,7 @@ 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(a) is str assert type(parts) == tuple # noqa: E721 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 5426b66..8357146 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): @@ -94,7 +94,7 @@ def test_exec_command_failure(self, os_ops: OsOperations): assert type(e.exit_code) is int assert e.exit_code == 127 - assert type(e.message) == str # noqa: E721 + assert type(e.message) is str assert type(e.error) == bytes # noqa: E721 assert e.message.startswith("Utility exited with non-zero code (127). Error:") @@ -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)) @@ -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): @@ -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: @@ -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,7 +927,7 @@ 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());"] @@ -935,7 +935,7 @@ def test_get_tmpdir__compare_with_py_info(self, os_ops: OsOperations): expected_dir_b = os_ops.exec_command(cmd) assert type(expected_dir_b) == bytes # noqa: E721 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 @@ -982,7 +982,7 @@ 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(lock_dir) is str assert type(num) is int return os.path.join(lock_dir, str(num) + ".lock") @@ -993,7 +993,7 @@ def LOCAL_WORKER(os_ops: OsOperations, reservedNumbers: typing.Set[int]) -> None: assert isinstance(os_ops, OsOperations) assert type(workerID) is int - assert type(lock_dir) == str # noqa: E721 + assert type(lock_dir) is str assert type(cNumbers) is int assert type(reservedNumbers) == set # noqa: E721 assert cNumbers > 0 @@ -1002,11 +1002,11 @@ def LOCAL_WORKER(os_ops: OsOperations, assert os.path.exists(lock_dir) def LOG_INFO(template: str, *args) -> None: - assert type(template) == str # noqa: E721 + assert type(template) is str assert type(args) == tuple # noqa: E721 msg = template.format(*args) - assert type(msg) == str # noqa: E721 + assert type(msg) is str logging.info("[Worker #{}] {}".format(workerID, msg)) return @@ -1341,8 +1341,8 @@ def test_kill__unk_pid( 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) is int 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 3bca117..19a1a89 100755 --- a/tests/test_os_ops_remote.py +++ b/tests/test_os_ops_remote.py @@ -25,7 +25,7 @@ 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: @@ -36,7 +36,7 @@ def test_rmdirs__try_to_delete_file(self, os_ops: OsOperations): assert type(x.value.description) == str # noqa: E721 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) is int assert x.value.exit_code == 20 From 70ded1f94685895523ff0bb030e7feb889295691 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 14:02:52 +0300 Subject: [PATCH 3/9] [E721] ... is bool --- src/local_ops.py | 10 +++++----- src/remote_ops.py | 10 +++++----- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/src/local_ops.py b/src/local_ops.py index 918785d..dd6f603 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -242,8 +242,8 @@ 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 type(expect_error) is bool + assert type(ignore_errors) is bool assert exec_env is None or type(exec_env) == dict # noqa: E721 assert cwd is None or type(cwd) is str @@ -332,7 +332,7 @@ def rmdirs(self, path, ignore_errors=True, attempts=3, delay=1): :param delay: Delay between attempts in seconds. """ assert type(path) is str - assert type(ignore_errors) == bool # noqa: E721 + assert type(ignore_errors) is bool assert type(attempts) is int assert type(delay) == int or type(delay) == float # noqa: E721 assert attempts > 0 @@ -469,7 +469,7 @@ def touch(self, filename): def read(self, filename, encoding=None, binary=False): assert type(filename) is str assert encoding is None or type(encoding) is str - assert type(binary) == bool # noqa: E721 + assert type(binary) is bool if binary: if encoding is not None: @@ -505,7 +505,7 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): """ assert type(num_lines) is int assert type(filename) is str - assert type(binary) == bool # noqa: E721 + assert type(binary) is bool assert encoding is None or type(encoding) is str assert num_lines >= 0 diff --git a/src/remote_ops.py b/src/remote_ops.py index 78f01e7..7d6f92a 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -112,8 +112,8 @@ 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 type(expect_error) is bool + assert type(ignore_errors) is bool assert exec_env is None or type(exec_env) == dict # noqa: E721 assert cwd is None or type(cwd) is str @@ -287,7 +287,7 @@ def rmdirs(self, path, ignore_errors=True): - ignore_errors (bool): If True, do not raise error if directory does not exist. """ assert type(path) is str - assert type(ignore_errors) == bool # noqa: E721 + assert type(ignore_errors) is bool # ENOENT = 2 - No such file or directory # ENOTDIR = 20 - Not a directory @@ -504,7 +504,7 @@ def touch(self, filename): def read(self, filename, binary=False, encoding=None): assert type(filename) is str assert encoding is None or type(encoding) is str - assert type(binary) == bool # noqa: E721 + assert type(binary) is bool if binary: if encoding is not None: @@ -539,7 +539,7 @@ def _read__binary(self, filename): def readlines(self, filename, num_lines=0, binary=False, encoding=None): assert type(num_lines) is int assert type(filename) is str - assert type(binary) == bool # noqa: E721 + assert type(binary) is bool assert encoding is None or type(encoding) is str if num_lines > 0: From acc98c371cfa6951579256880dcf9ce8a64eaa71 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 14:04:36 +0300 Subject: [PATCH 4/9] [E721] ... is dict --- src/local_ops.py | 12 ++++++------ src/remote_ops.py | 2 +- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/local_ops.py b/src/local_ops.py index dd6f603..bc6eaf4 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -102,7 +102,7 @@ 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 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,7 +115,7 @@ 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 len(v) == 2 @@ -157,7 +157,7 @@ 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 exec_env is None or type(exec_env) is dict assert cwd is None or type(cwd) is str input_prepared = None @@ -174,7 +174,7 @@ 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 len(v) == 2 @@ -222,7 +222,7 @@ 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 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 @@ -244,7 +244,7 @@ def exec_command( """ assert type(expect_error) is bool assert type(ignore_errors) is bool - assert exec_env is None or type(exec_env) == dict # noqa: E721 + 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( diff --git a/src/remote_ops.py b/src/remote_ops.py index 7d6f92a..e2cf15a 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -114,7 +114,7 @@ def exec_command( """ assert type(expect_error) is bool assert type(ignore_errors) is bool - assert exec_env is None or type(exec_env) == dict # 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 From 0e9026a3cc41cac2ddd2d24bb0412a334f674d4c Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 14:05:10 +0300 Subject: [PATCH 5/9] [E721] ... is list --- src/remote_ops.py | 6 +++--- tests/test_os_ops_common.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/remote_ops.py b/src/remote_ops.py index e2cf15a..3db52e6 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -334,7 +334,7 @@ def listdir(self, path): output = self.exec_command(cmd=command, encoding=get_default_encoding()) 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): @@ -567,7 +567,7 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): 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): @@ -782,7 +782,7 @@ 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)) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 8357146..cd92576 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -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 From 98e2337c91fe4ee4c7c5a8c10d0fc2d9c073ceaa Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 14:06:40 +0300 Subject: [PATCH 6/9] [E721] ... is bytes --- src/helpers.py | 2 +- src/local_ops.py | 10 ++++----- src/raise_error.py | 2 +- src/remote_ops.py | 14 ++++++------- tests/test_os_ops_common.py | 42 ++++++++++++++++++------------------- 5 files changed, 35 insertions(+), 35 deletions(-) diff --git a/src/helpers.py b/src/helpers.py index 2e5c725..a502a06 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -51,5 +51,5 @@ def PrepareProcessInput(input, encoding): 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 bc6eaf4..2317afd 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -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) @@ -438,7 +438,7 @@ 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) is str @@ -495,7 +495,7 @@ def _read__binary(self, filename): 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): @@ -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): diff --git a/src/raise_error.py b/src/raise_error.py index d6f28db..55bfbc3 100644 --- a/src/raise_error.py +++ b/src/raise_error.py @@ -47,7 +47,7 @@ def _TranslateDataIntoString(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 3db52e6..cc25392 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -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) @@ -474,7 +474,7 @@ 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) is str @@ -522,7 +522,7 @@ def _read__text_with_encoding(self, filename, encoding): 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() @@ -533,7 +533,7 @@ def _read__binary(self, filename): 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): @@ -561,7 +561,7 @@ def readlines(self, filename, num_lines=0, binary=False, encoding=None): 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) is str @@ -579,7 +579,7 @@ def read_binary(self, filename, offset): 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): diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index cd92576..f7dc0d7 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -95,7 +95,7 @@ def test_exec_command_failure(self, os_ops: OsOperations): assert e.exit_code == 127 assert type(e.message) is str - assert type(e.error) == bytes # noqa: E721 + 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): @@ -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): @@ -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): @@ -933,7 +933,7 @@ def test_get_tmpdir__compare_with_py_info(self, os_ops: OsOperations): 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) is str assert actual_dir + "\n" == expected_dir From 337cb585401a17bfe7ba1aaad59b8c455282cc36 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 14:07:22 +0300 Subject: [PATCH 7/9] [E721] ... is tuple --- src/local_ops.py | 6 +++--- src/os_ops.py | 2 +- src/remote_ops.py | 8 ++++---- tests/test_os_ops_common.py | 10 +++++----- 4 files changed, 13 insertions(+), 13 deletions(-) diff --git a/src/local_ops.py b/src/local_ops.py index 2317afd..4d9e8dd 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -117,7 +117,7 @@ def _run_command__nt( env = os.environ.copy() 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]) is str assert v[0] != "" @@ -176,7 +176,7 @@ def _run_command__generic( env = os.environ.copy() 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]) is str assert v[0] != "" @@ -283,7 +283,7 @@ def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None assert type(a) is str - assert type(parts) == tuple # noqa: E721 + assert type(parts) is tuple return os.path.join(a, *parts) # Environment setup diff --git a/src/os_ops.py b/src/os_ops.py index 326f4a5..2d12fcd 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -37,7 +37,7 @@ def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None assert type(a) is str - assert type(parts) == tuple # noqa: E721 + assert type(parts) is tuple raise NotImplementedError() # Environment setup diff --git a/src/remote_ops.py b/src/remote_ops.py index cc25392..0f92c68 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -184,7 +184,7 @@ def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None assert type(a) is str - assert type(parts) == tuple # noqa: E721 + assert type(parts) is tuple return __class__._build_path(a, *parts) # Environment setup @@ -816,7 +816,7 @@ 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]) is str env[envvar[0]] = envvar[1] @@ -824,7 +824,7 @@ def _create_exec_env_list(exec_env: typing.Dict) -> typing.List[str]: # ---------------------------------- 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]) is str @@ -870,7 +870,7 @@ def _build_path(a: str, *parts: str) -> str: assert a is not None assert parts is not None assert type(a) is str - assert type(parts) == tuple # noqa: E721 + assert type(parts) is tuple return posixpath.join(a, *parts) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index f7dc0d7..862da4f 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -1003,7 +1003,7 @@ def LOCAL_WORKER(os_ops: OsOperations, def LOG_INFO(template: str, *args) -> None: assert type(template) is str - assert type(args) == tuple # noqa: E721 + assert type(args) is tuple msg = template.format(*args) assert type(msg) is str @@ -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, @@ -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, From d1d4918001ed1850c1cc5e84e9737f4410332b24 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 14:08:30 +0300 Subject: [PATCH 8/9] [E721] ... is set --- tests/test_os_ops_common.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 862da4f..2d30780 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -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,7 +843,7 @@ 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 @@ -995,7 +995,7 @@ def LOCAL_WORKER(os_ops: OsOperations, assert type(workerID) is int assert type(lock_dir) is str assert type(cNumbers) is int - assert type(reservedNumbers) == set # noqa: E721 + assert type(reservedNumbers) is set assert cNumbers > 0 assert len(reservedNumbers) == 0 @@ -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: From 3b9dc6de31dc1a64afd749e20c1e6bb368180e86 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Wed, 25 Feb 2026 14:14:38 +0300 Subject: [PATCH 9/9] [E721] mix --- src/exceptions.py | 12 ++++++------ src/helpers.py | 2 +- src/local_ops.py | 10 +++++----- src/os_ops.py | 2 +- src/raise_error.py | 2 +- src/remote_ops.py | 8 ++++---- tests/test_os_ops_common.py | 20 ++++++++++---------- tests/test_os_ops_remote.py | 4 ++-- 8 files changed, 30 insertions(+), 30 deletions(-) diff --git a/src/exceptions.py b/src/exceptions.py index 1faa900..9598323 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -27,10 +27,10 @@ def __init__( error: typing.Optional[T_ERR_DATA] = None, ): assert message is None or type(message) is str - assert command is None or type(command) in [str, list] # noqa: E721 + 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] # noqa: E721 - assert error is None or type(error) in [str, bytes] # noqa: E721 + assert out is None or type(out) in [str, bytes] + assert error is None or type(error) in [str, bytes] super().__init__(message) @@ -71,7 +71,7 @@ def description(self) -> typing.Optional[str]: @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 @@ -81,12 +81,12 @@ def exit_code(self) -> typing.Optional[int]: @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 a502a06..bd357f4 100644 --- a/src/helpers.py +++ b/src/helpers.py @@ -43,7 +43,7 @@ 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()) diff --git a/src/local_ops.py b/src/local_ops.py index 4d9e8dd..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: @@ -164,7 +164,7 @@ def _run_command__generic( 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() @@ -334,7 +334,7 @@ def rmdirs(self, path, ignore_errors=True, attempts=3, delay=1): assert type(path) is str assert type(ignore_errors) is bool assert type(attempts) is int - assert type(delay) == int or type(delay) == float # noqa: E721 + assert type(delay) is int or type(delay) is float assert attempts > 0 assert delay >= 0 @@ -576,7 +576,7 @@ def remove_file(self, filename): def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): # Kill the process assert type(pid) is int - assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 + assert type(signal) is int or type(signal) is os_signal.Signals os.kill(pid, signal) def get_pid(self): diff --git a/src/os_ops.py b/src/os_ops.py index 2d12fcd..01ee347 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -134,7 +134,7 @@ def remove_file(self, filename): def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): # Kill the process assert type(pid) is int - assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 + assert type(signal) is int or type(signal) is os_signal.Signals raise NotImplementedError() def get_pid(self): diff --git a/src/raise_error.py b/src/raise_error.py index 55bfbc3..fe7ee9e 100644 --- a/src/raise_error.py +++ b/src/raise_error.py @@ -40,7 +40,7 @@ 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) diff --git a/src/remote_ops.py b/src/remote_ops.py index 0f92c68..727833d 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -121,7 +121,7 @@ def exec_command( 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 = [] @@ -664,7 +664,7 @@ def remove_file(self, filename): def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]): # Kill the process assert type(pid) is int - assert type(signal) == int or type(signal) == os_signal.Signals # noqa: E721 E501 + 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()) @@ -792,9 +792,9 @@ def _build_cmdline(cmd, exec_env: typing.Dict = None) -> str: @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__)) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 2d30780..fb6c3a6 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -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): """ @@ -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+" @@ -847,7 +847,7 @@ def test_is_port_free__false(self, os_ops: OsOperations): 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: @@ -964,7 +964,7 @@ 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 @@ -1261,7 +1261,7 @@ 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) is int logging.info("Test process pid is {}".format(proc_pid)) @@ -1332,7 +1332,7 @@ 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) is int logging.info("Test process pid is {}".format(proc_pid)) @@ -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_remote.py b/tests/test_os_ops_remote.py index 19a1a89..7dcd6e1 100755 --- a/tests/test_os_ops_remote.py +++ b/tests/test_os_ops_remote.py @@ -32,8 +32,8 @@ def test_rmdirs__try_to_delete_file(self, os_ops: OsOperations): 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) is str