diff --git a/src/exceptions.py b/src/exceptions.py index 2dc1b9c..1f305b5 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -1,19 +1,24 @@ # coding: utf-8 +from .types import T_OS_CMD +from .types import T_OS_TIMEOUT + from testgres.common.exceptions import TestgresException from testgres.common.exceptions import InvalidOperationException + import six import typing -T_CMD = typing.Union[str, list] +# [2026-09-15] deprecated +T_CMD = T_OS_CMD T_OUT_DATA = typing.Union[str, bytes] T_ERR_DATA = typing.Union[str, bytes] class ExecUtilException(TestgresException): _description: typing.Optional[str] - _command: typing.Optional[T_CMD] + _command: typing.Optional[T_OS_CMD] _exit_code: typing.Optional[int] _out: typing.Optional[T_OUT_DATA] _error: typing.Optional[T_ERR_DATA] @@ -21,7 +26,7 @@ class ExecUtilException(TestgresException): def __init__( self, message: typing.Optional[str] = None, - command: typing.Optional[T_CMD] = None, + command: typing.Optional[T_OS_CMD] = None, exit_code: typing.Optional[int] = None, out: typing.Optional[T_OUT_DATA] = None, error: typing.Optional[T_ERR_DATA] = None, @@ -71,7 +76,7 @@ def description(self) -> typing.Optional[str]: return self._description @property - def command(self) -> typing.Optional[T_CMD]: + def command(self) -> typing.Optional[T_OS_CMD]: assert self._command is None or type(self._command) in [str, list] return self._command @@ -120,15 +125,106 @@ def __repr__(self) -> str: @staticmethod def convert_and_join(msg_list): # Convert each byte element in the list to str - str_list = [six.text_type(item, 'utf-8') if isinstance(item, bytes) else six.text_type(item) for item in - msg_list] + str_list = [ + six.text_type(item, 'utf-8') if isinstance(item, bytes) else six.text_type(item) + for item in msg_list + ] # Join the list into a single string with the specified delimiter return six.text_type('\n').join(str_list) +class ExecTimeoutException(TestgresException): + _cmd: T_OS_CMD + _timeout: T_OS_TIMEOUT + _output: typing.Optional[T_OUT_DATA] + _error: typing.Optional[T_ERR_DATA] + _source: typing.Optional[str] + + def __init__( + self, + cmd: T_OS_CMD, + timeout: T_OS_TIMEOUT, + output: typing.Optional[T_OUT_DATA] = None, + error: typing.Optional[T_ERR_DATA] = None, + source: typing.Optional[str] = None + ): + assert type(cmd) in [str, list] + assert type(timeout) in [int, float] + assert output is None or type(output) in [str, bytes] + assert error is None or type(error) in [str, bytes] + assert source is None or type(source) is str + + super().__init__() + + self._cmd = cmd + self._timeout = timeout + self._output = output + self._error = error + self._source = source + return + + @property + def message(self) -> str: + # Construct a clear log message, similar to the standard subprocess module + r = "Command '{}' timed out after {} second(s).".format( + self._cmd, + self._timeout, + ) + assert type(r) is str + return r + + @property + def source(self) -> typing.Optional[str]: + return self._source + + @property + def cmd(self) -> T_OS_CMD: + return self._cmd + + @property + def timeout(self) -> T_OS_TIMEOUT: + return self._timeout + + @property + def output(self) -> typing.Optional[T_OUT_DATA]: + return self._output + + @property + def error(self) -> typing.Optional[T_ERR_DATA]: + return self._error + + def __repr__(self) -> str: + args = [] + + if self._cmd is not None: + args.append(("cmd", self._cmd)) + + if self._timeout is not None: + args.append(("timeout", self._timeout)) + + if self._output is not None: + args.append(("output", self._output)) + + if self._error is not None: + args.append(("error", self._error)) + + if self._source is not None: + args.append(("source", self._source)) + + result = "{}(".format(type(self).__name__) + sep = "" + for a in args: + result += sep + a[0] + "=" + repr(a[1]) + sep = ", " + continue + result += ")" + return result + + __all__ = [ "TestgresException", "InvalidOperationException", "ExecUtilException", + "ExecTimeoutException", ] diff --git a/src/local_ops.py b/src/local_ops.py index 25e805c..fc846fc 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -19,10 +19,19 @@ import signal as os_signal import datetime import pathlib +import io from .exceptions import ExecUtilException +from .exceptions import ExecTimeoutException from .exceptions import InvalidOperationException from .os_ops import ConnectionParams, OsOperations, get_default_encoding +from .os_ops import OsProcessController +from .os_ops import OsCommandResult +from .os_ops import T_OS_CMD +from .os_ops import T_OS_SIGNAL +from .os_ops import T_OS_TIMEOUT +from .os_ops import T_OS_IO +from .os_ops import T_OS_IO_ID from .raise_error import RaiseError from .helpers import Helpers @@ -32,6 +41,121 @@ CMD_TIMEOUT_SEC = 60 +class LocalProcessController(OsProcessController): + _cmd: T_OS_CMD + _local_process: typing.Optional[subprocess.Popen] + + def __init__( + self, + cmd: T_OS_CMD, + ): + assert type(cmd) is str or type(cmd) is list + + self._cmd = copy.copy(cmd) + assert type(self._cmd) is type(cmd) + + self._local_process = None + return + + def __enter__(self) -> OsProcessController: + assert type(self._local_process) is subprocess.Popen + self._local_process.__enter__() + return self + + def __exit__(self, exc_type, value, traceback) -> typing.Optional[bool]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.__exit__(exc_type, value, traceback) + + @property + def args(self) -> T_OS_CMD: + assert type(self._cmd) is str or type(self._cmd) is list + return self._cmd + + @property + def pid(self) -> int: + assert type(self._local_process) is subprocess.Popen + return self._local_process.pid + + @property + def stdin(self) -> typing.Optional[T_OS_IO]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.stdin + + @property + def stdout(self) -> typing.Optional[T_OS_IO]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.stdout + + @property + def stderr(self) -> typing.Optional[T_OS_IO]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.stderr + + @property + def returncode(self) -> typing.Optional[int]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.poll() + + def communicate( + self, + input=None, + timeout: typing.Optional[T_OS_TIMEOUT] = None, + ) -> OsProcessController.T_COMMUNICATE_RESULT: + assert timeout is None or type(timeout) in [int, float] + assert type(self._local_process) is subprocess.Popen + + try: + return self._local_process.communicate( + input=input, + timeout=timeout, + ) + except subprocess.TimeoutExpired as e: + # Transforming a "foreign" exception into one native to the Testgres architecture + raise ExecTimeoutException( + cmd=self._cmd, + timeout=e.timeout, + output=e.output, + error=e.stderr, + source="LocalProcessController::communicate", + ) from e + + def send_signal(self, sig: T_OS_SIGNAL) -> None: + assert type(sig) in [int, os_signal.Signals] + assert type(self._local_process) is subprocess.Popen + self._local_process.send_signal(sig) + return + + def kill(self) -> None: + assert type(self._local_process) is subprocess.Popen + self.send_signal(os_signal.SIGKILL) + return + + def terminate(self) -> None: + assert type(self._local_process) is subprocess.Popen + self.send_signal(os_signal.SIGTERM) + return + + def poll(self) -> typing.Optional[int]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.poll() + + def wait(self, timeout: typing.Optional[T_OS_TIMEOUT] = None) -> int: + assert timeout is None or type(timeout) in [int, float] + assert type(self._local_process) is subprocess.Popen + + try: + return self._local_process.wait(timeout) + except subprocess.TimeoutExpired as e: + # Transforming a "foreign" exception into one native to the Testgres architecture + raise ExecTimeoutException( + cmd=e.cmd, + timeout=e.timeout, + output=e.output, + error=e.stderr, + source="LocalProcessController::wait", + ) from e + + class LocalOperations(OsOperations): sm_dummy_conn_params = ConnectionParams() sm_single_instance: typing.Optional[OsOperations] = None @@ -366,6 +490,146 @@ def exec_command( return run_r[1] + def popen( + self, + cmd: OsOperations.T_CMD, + text: typing.Optional[bool] = None, + encoding: typing.Optional[str] = None, + shell: bool = False, + stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + exec_env: typing.Optional[dict] = None, + cwd: typing.Optional[str] = None + ) -> OsProcessController: + assert text is None or type(text) is bool + assert encoding is None or type(encoding) is str + assert type(shell) is bool + assert stdin is None or type(stdin) is int or isinstance(stdin, io.IOBase) + assert stdout is None or type(stdout) is int or isinstance(stdout, io.IOBase) + assert stderr is None or type(stderr) is int or isinstance(stderr, io.IOBase) + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str + + extParams: typing.Dict[str, typing.Any] = dict() + + if exec_env is None: + pass + elif len(exec_env) == 0: + pass + else: + env = os.environ.copy() + assert type(env) is dict + for v in exec_env.items(): + assert type(v) is tuple + assert len(v) == 2 + assert type(v[0]) is str + assert v[0] != "" + + if v[1] is None: + env.pop(v[0], None) + else: + assert type(v[1]) is str + env[v[0]] = v[1] + + extParams["env"] = env + + if encoding is not None and text is None: + text = True + + result = LocalProcessController(cmd) + + result._local_process = subprocess.Popen( + cmd, + shell=shell, + stdin=stdin, + stdout=stdout, + stderr=stderr, + text=text, + encoding=encoding, + cwd=cwd, + **extParams, + ) + assert type(result._local_process) is subprocess.Popen + return result + + def run( + self, + cmd: T_OS_CMD, + text: typing.Optional[bool] = None, + encoding: typing.Optional[str] = None, + shell: bool = False, + input: typing.Optional[OsOperations.T_INPUT] = None, + stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + cwd: typing.Optional[str] = None, + timeout: typing.Optional[T_OS_TIMEOUT] = None, + check: bool = True, + ) -> OsCommandResult: + assert type(cmd) in [str, list] + assert text is None or type(text) is bool + assert encoding is None or type(encoding) is str + assert type(shell) is bool + assert input is None or type(input) in [str, bytes] or isinstance(input, io.IOBase) + assert stdin is None or type(stdin) is int or isinstance(stdin, io.IOBase) + assert stdout is None or type(stdout) is int or isinstance(stdout, io.IOBase) + assert stderr is None or type(stderr) is int or isinstance(stderr, io.IOBase) + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str + assert timeout is None or type(timeout) in [int, float] + assert type(check) is bool + + controller = self.popen( + cmd=cmd, + text=text, + encoding=encoding, + shell=shell, + stdin=stdin, + stdout=stdout, + stderr=stderr, + exec_env=exec_env, + cwd=cwd, + ) + assert isinstance(controller, OsProcessController) + + with controller: + try: + communicate_r = controller.communicate( + input=input, + timeout=timeout, + ) + except BaseException: + controller.kill() + raise + + assert type(communicate_r) is tuple + assert len(communicate_r) == 2 + + rc = controller.returncode + assert type(rc) is int + + result = OsCommandResult( + cmd=cmd, + returncode=rc, + stdout=communicate_r[0], + stderr=communicate_r[1], + ) + + if result.returncode == 0: + pass + elif check: + RaiseError.UtilityExitedWithNonZeroCode( + cmd=cmd, + exit_code=result.returncode, + msg_arg=result.stderr, + error=result.stderr, + out=result.stdout, + ) + + return result + def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None @@ -389,7 +653,7 @@ def environ(self, var_name: str) -> typing.Optional[str]: assert var_name != "" return os.environ.get(var_name) - def cwd(self): + def cwd(self) -> str: return os.getcwd() def find_executable(self, executable: str) -> typing.Optional[str]: diff --git a/src/os_ops.py b/src/os_ops.py index 9ef487b..a5c5c9d 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -1,5 +1,10 @@ from __future__ import annotations +from .types import T_OS_CMD +from .types import T_OS_SIGNAL +from .types import T_OS_TIMEOUT +from .types import T_OS_IO +from .types import T_OS_IO_ID from .raise_error import RaiseError import locale @@ -37,6 +42,94 @@ def get_default_encoding(): return locale.getencoding() or 'UTF-8' +class OsProcessController: + def __enter__(self) -> OsProcessController: + RaiseError.PropertyIsNotImplemented(__class__, "__enter__") + + def __exit__(self, exc_type, value, traceback) -> typing.Optional[bool]: + RaiseError.PropertyIsNotImplemented(__class__, "__exit__") + + @property + def args(self) -> T_OS_CMD: + RaiseError.PropertyIsNotImplemented(__class__, "get_args") + + @property + def pid(self) -> int: + RaiseError.PropertyIsNotImplemented(__class__, "get_pid") + + @property + def stdin(self) -> typing.Optional[T_OS_IO]: + RaiseError.PropertyIsNotImplemented(__class__, "get_stdin") + + @property + def stdout(self) -> typing.Optional[T_OS_IO]: + RaiseError.PropertyIsNotImplemented(__class__, "get_stdout") + + @property + def stderr(self) -> typing.Optional[T_OS_IO]: + RaiseError.PropertyIsNotImplemented(__class__, "get_stderr") + + @property + def returncode(self) -> typing.Optional[int]: + RaiseError.PropertyIsNotImplemented(__class__, "get_returncode") + + T_COMMUNICATE_RESULT = typing.Union[ + typing.Tuple[bytes, bytes], + typing.Tuple[str, str], + ] + + def communicate( + self, + input=None, + timeout: typing.Optional[T_OS_TIMEOUT] = None + ) -> T_COMMUNICATE_RESULT: + assert timeout is not None or type(timeout) in [int, float] + RaiseError.MethodIsNotImplemented(__class__, "communicate") + + def send_signal(self, sig: T_OS_SIGNAL) -> None: + assert type(sig) in [int, os_signal.Signals] + RaiseError.MethodIsNotImplemented(__class__, "send_signal") + + def kill(self) -> None: + RaiseError.MethodIsNotImplemented(__class__, "kill") + + def terminate(self) -> None: + RaiseError.MethodIsNotImplemented(__class__, "terminate") + + def poll(self) -> typing.Optional[int]: + RaiseError.MethodIsNotImplemented(__class__, "poll") + + def wait(self, timeout: typing.Optional[T_OS_TIMEOUT] = None) -> int: + assert timeout is None or type(timeout) in [int, float] + RaiseError.MethodIsNotImplemented(__class__, "wait") + + +class OsCommandResult: + T_IO_RESULT = typing.Union[str, bytes] + + cmd: T_OS_CMD + returncode: int + stdout: typing.Optional[T_IO_RESULT] + stderr: typing.Optional[T_IO_RESULT] + + def __init__( + self, + cmd: T_OS_CMD, + returncode: int, + stdout: typing.Optional[T_IO_RESULT], + stderr: typing.Optional[T_IO_RESULT], + ): + assert type(cmd) in [str, list] + assert type(returncode) is int + assert stdout is None or type(stdout) in [str, bytes] + assert stderr is None or type(stderr) in [str, bytes] + self.cmd = cmd + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + return + + class OsOperations: def __init__(self): pass @@ -70,7 +163,7 @@ def create_clone(self) -> OsOperations: RaiseError.MethodIsNotImplemented(__class__, "create_clone") # Command execution - T_CMD = typing.Union[str, typing.List[str]] + T_CMD = T_OS_CMD T_EXEC_COMMAND_RESULT = typing.Union[ subprocess.Popen, str, @@ -81,7 +174,7 @@ def create_clone(self) -> OsOperations: def exec_command( self, - cmd: T_CMD, + cmd: T_OS_CMD, wait_exit=False, verbose=False, expect_error=False, @@ -109,6 +202,62 @@ def exec_command( assert cwd is None or type(cwd) is str RaiseError.MethodIsNotImplemented(__class__, "exec_command") + T_EXEC_ENV = typing.Dict[str, typing.Optional[str]] + + def popen( + self, + cmd: T_OS_CMD, + text: typing.Optional[bool] = None, + encoding: typing.Optional[str] = None, + shell: bool = False, + stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + exec_env: typing.Optional[T_EXEC_ENV] = None, + cwd: typing.Optional[str] = None + ) -> OsProcessController: + assert type(cmd) is str or type(cmd) is list + assert text is None or type(text) is bool + assert encoding is None or type(encoding) is str + assert type(shell) is bool + assert stdin is None or type(stdin) is int or isinstance(stdin, typing.IO) + assert stdout is None or type(stdout) is int or isinstance(stdout, typing.IO) + assert stderr is None or type(stderr) is int or isinstance(stderr, typing.IO) + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str + RaiseError.MethodIsNotImplemented(__class__, "popen") + + T_INPUT = typing.Union[str, bytes, typing.IO[typing.Any]] + + def run( + self, + cmd: T_OS_CMD, + text: typing.Optional[bool] = None, + encoding: typing.Optional[str] = None, + shell: bool = False, + input: typing.Optional[T_INPUT] = None, + stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + exec_env: typing.Optional[T_EXEC_ENV] = None, + cwd: typing.Optional[str] = None, + timeout: typing.Optional[T_OS_TIMEOUT] = None, + check: bool = True, + ) -> OsCommandResult: + assert type(cmd) in [str, list] + assert text is None or type(text) is bool + assert encoding is None or type(encoding) is str + assert type(shell) is bool + assert input is None or type(input) in [str, bytes] or isinstance(input, typing.IO) + assert stdin is None or type(stdin) is int or isinstance(stdin, typing.IO) + assert stdout is None or type(stdout) is int or isinstance(stdout, typing.IO) + assert stderr is None or type(stderr) is int or isinstance(stderr, typing.IO) + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str + assert timeout is None or type(timeout) in [int, float] + assert type(check) is bool + RaiseError.MethodIsNotImplemented(__class__, "run") + def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None @@ -326,7 +475,7 @@ def remove_file(self, filename: str) -> None: RaiseError.MethodIsNotImplemented(__class__, "remove_file") # Processes control - def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]) -> None: + def kill(self, pid: int, signal: T_OS_SIGNAL) -> None: # Kill the process assert type(pid) is int assert type(signal) is int or type(signal) is os_signal.Signals diff --git a/src/remote_ops.py b/src/remote_ops.py index 420f271..bf82fd9 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -14,12 +14,22 @@ import datetime import shlex import threading +import warnings from .exceptions import ExecUtilException +from .exceptions import ExecTimeoutException from .exceptions import InvalidOperationException from .os_ops import OsOperations, ConnectionParams, get_default_encoding +from .os_ops import OsProcessController +from .os_ops import OsCommandResult +from .os_ops import T_OS_CMD +from .os_ops import T_OS_SIGNAL +from .os_ops import T_OS_TIMEOUT +from .os_ops import T_OS_IO +from .os_ops import T_OS_IO_ID from .raise_error import RaiseError from .helpers import Helpers +from .static_config import OsOperationStaticConfig class PsUtilProcessProxy: @@ -46,7 +56,236 @@ def cmdline(self): return cmdline.split() +class RemoteProcessController(OsProcessController): + _C_MAX_RESP_RC_FILE_SIZE = 32 + + _remote_ops: RemoteOperations + _remote_cmd: T_OS_CMD + _remote_rc_file: typing.Optional[str] + _remote_pid: typing.Optional[int] + _remote_rc: typing.Optional[int] + _local_process: typing.Optional[subprocess.Popen] + + def __init__( + self, + remote_ops: RemoteOperations, + remote_cmd: T_OS_CMD, + ): + assert isinstance(remote_ops, RemoteOperations) + assert type(remote_cmd) is str or type(remote_cmd) is list + + self._remote_ops = remote_ops + self._remote_cmd = copy.copy(remote_cmd) + self._remote_rc_file = None + self._remote_pid = None + self._remote_rc = None + + # IT IS LAST STATEMENT ! + self._local_process = None + return + + def __enter__(self) -> OsProcessController: + assert type(self._local_process) is subprocess.Popen + self._local_process.__enter__() + return self + + def __exit__(self, exc_type, value, traceback) -> typing.Optional[bool]: + assert type(self._local_process) is subprocess.Popen + assert type(self._remote_rc_file) is str + assert self._remote_cmd is not None + + self.wait() + + try: + self._local_process.kill() + except BaseException as e: + msg_lines = [] + + msg_lines.append("RemoteProcessController::__exit__ catches an exception ({}): {}".format( + type(e).__name__, + e, + )) + msg_lines.append("Remote command is {}".format( + self._remote_cmd, + )) + logging.debug("\n".join(msg_lines)) + + self._remote_ops.remove_file(self._remote_rc_file) + + return self._local_process.__exit__(exc_type, value, traceback) + + def __del__(self, _warn=warnings.warn): + assert isinstance(_warn, typing.Callable) + + # 1. If the process hasn't even managed to initialize, we do nothing. + if not getattr(self, "_local_process", None): + return + + if self._local_process is None: + return + + # 2. If the process is still active (we did not wait for it to complete) + if self._remote_rc is None: + # Issue a system warning, just like the standard subprocess module does. + if type(self._remote_pid) is int: + _warn( + f"Remote process {self._remote_pid} is still running inside RemoteProcessController", + ResourceWarning, + source=self + ) + + # Issue a system warning, just like the standard subprocess module does. + try: + if self._local_process.stdin: + self._local_process.stdin.close() + if self._local_process.stdout: + self._local_process.stdout.close() + if self._local_process.stderr: + self._local_process.stderr.close() + except Exception: + pass + + # 4. Terminate the local SSH transport. + # We do NOT invoke a remote remove_file or kill over the network here, + # as the destructor must execute immediately. + # However, killing the local SSH client will close the socket, + # and the remote shell will eventually close on its own (due to HUP or wait completion). + try: + self._local_process.kill() + # Implementing a fast, non-blocking wait for a local process + self._local_process.wait(timeout=0.1) + except Exception: + pass + return + + @property + def pid(self) -> int: + assert type(self._remote_pid) is int + return self._remote_pid + + @property + def args(self) -> T_OS_CMD: + assert type(self._remote_cmd) is str or type(self._remote_cmd) is list + return self._remote_cmd + + @property + def stdin(self) -> typing.Optional[T_OS_IO]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.stdin + + @property + def stdout(self) -> typing.Optional[T_OS_IO]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.stdout + + @property + def stderr(self) -> typing.Optional[T_OS_IO]: + assert type(self._local_process) is subprocess.Popen + return self._local_process.stderr + + @property + def returncode(self) -> typing.Optional[int]: + return self._poll() + + def communicate( + self, + input=None, + timeout: typing.Optional[T_OS_TIMEOUT] = None, + ) -> OsProcessController.T_COMMUNICATE_RESULT: + assert timeout is None or type(timeout) in [int, float] + assert type(self._local_process) is subprocess.Popen + + try: + return self._local_process.communicate( + input=input, + timeout=timeout, + ) + except subprocess.TimeoutExpired as e: + # Transforming a "foreign" exception into one native to the Testgres architecture + raise ExecTimeoutException( + cmd=self._remote_cmd, + timeout=e.timeout, + output=e.output, + error=e.stderr, + source="RemoteProcessController::communicate", + ) from e + + def send_signal(self, sig: T_OS_SIGNAL) -> None: + assert type(sig) in [int, os_signal.Signals] + assert type(self._local_process) is subprocess.Popen + assert type(self._remote_pid) is int + + self._remote_ops.kill(self._remote_pid, sig) + return + + def kill(self) -> None: + assert type(self._local_process) is subprocess.Popen + self.send_signal(os_signal.SIGKILL) + return + + def terminate(self) -> None: + assert type(self._local_process) is subprocess.Popen + + if self._remote_rc is not None: + return + + self.send_signal(os_signal.SIGTERM) + return + + def poll(self) -> typing.Optional[int]: + assert type(self._local_process) is subprocess.Popen + return self._poll() + + def wait(self, timeout: typing.Optional[T_OS_TIMEOUT] = None) -> int: + assert timeout is None or type(timeout) in [int, float] + + start_time = time.monotonic() + nPass = 0 + while True: + nPass += 1 + + r = self._poll() + + if r is not None: + assert type(r) is int + return r + + if timeout is not None and (time.monotonic() - start_time) >= timeout: + raise ExecTimeoutException( + self._remote_cmd, + timeout=timeout, + source="RemoteProcessController::wait", + ) + + time.sleep(0.05) + continue + + def _poll(self) -> typing.Optional[int]: + assert type(self._remote_rc_file) is str + + if self._remote_rc is not None: + return self._remote_rc + + # Read the file containing the return code + rc_bytes = self._remote_ops.read_binary( + self._remote_rc_file, + offset=0, + size=__class__._C_MAX_RESP_RC_FILE_SIZE, + ) + + if len(rc_bytes) == __class__._C_MAX_RESP_RC_FILE_SIZE: + raise RuntimeError("Responce rc-file [{}] is too long.".format( + self._remote_rc_file, + )) + + self._remote_rc = self._remote_ops._parse_resp_data(rc_bytes) + + assert self._remote_rc is None or type(self._remote_rc) is int + return self._remote_rc + + class RemoteOperations(OsOperations): + _C_MAX_RESP_PID_FILE_SIZE = 32 _C_EOL = "\n" T_ENVS = typing.Dict[str, typing.Optional[str]] @@ -286,6 +525,305 @@ def exec_command( return run_r[1] + def popen( + self, + cmd: OsOperations.T_CMD, + text: typing.Optional[bool] = None, + encoding: typing.Optional[str] = None, + shell: bool = False, + stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + exec_env: typing.Optional[dict] = None, + cwd: typing.Optional[str] = None + ) -> OsProcessController: + assert type(cmd) is str or type(cmd) is list + assert text is None or type(text) is bool + assert encoding is None or type(encoding) is str + assert type(shell) is bool + assert stdin is None or type(stdin) is int or isinstance(stdin, io.IOBase) + assert stdout is None or type(stdout) is int or isinstance(stdout, io.IOBase) + assert stderr is None or type(stderr) is int or isinstance(stderr, io.IOBase) + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str + + result = RemoteProcessController( + self, + cmd, + ) + + # 1. Create a temporary file on the remote machine + pid_file = self.mkstemp(prefix="testgres_pid_") + assert type(pid_file) is str and pid_file != "" + + try: + result._remote_rc_file = self.mkstemp(prefix="testgres_rc_") + assert type(result._remote_rc_file) is str and result._remote_rc_file != "" + + cmds = [] + cmds.append("trap '' HUP") + + if cwd is not None: + cmds.append(__class__._build_cmdline(["cd", cwd])) + + assert self._remote_env_guard is not None + assert type(self._remote_env) is dict + + exec_env2: typing.Optional[__class__.T_ENVS] = None + with self._remote_env_guard: + if len(self._remote_env) > 0: + exec_env2 = self._remote_env.copy() + + if exec_env2 is None: + exec_env2 = exec_env + elif exec_env is not None: + exec_env2.update(exec_env) + + # Construct the final command, recording the PID and replacing the process via exec + cmd2 = "exec " + __class__._ensure_cmdline(cmd) + + target_cmdline = __class__._build_cmdline(cmd2, exec_env2) + + # Escape the file path for bash + q_pid_file = __class__._quote_path(pid_file) + + q_rc_file = __class__._quote_path(result._remote_rc_file) + + handshake_timeout = OsOperationStaticConfig.remote_ops__popen__handshake_timeout + assert type(handshake_timeout) is float + assert handshake_timeout > 0 + + # Поскольку sleep у нас 0.01с, количество итераций — это таймаут * 100 + handshake_max_iterations = int(handshake_timeout * 100) + assert handshake_max_iterations > 0 + + # A robust Bash handshake script: + # 1. Write the current shell's PID to a file. + # 2. Loop while checking only for the file's existence. + # 3. If the counter 'i' exceeds 500, it's a timeout; exit with code 1. + # 4. Otherwise, sleep for 0.01s and increment the counter. + # 5. Once the Python process successfully deletes the file, execute the target command. + ping_pong_script_s = ( + f"printf \"%s!\" \"$$\" > {q_pid_file} && " + f"i=0 && " + f"while [ -f {q_pid_file} ]; do " + f"if [ $i -ge {handshake_max_iterations} ]; then " + f"printf \"testgres error: popen handshake timeout expired\\n\" >&2; " + f"exit 1; " + f"fi; " + f"sleep 0.01; i=$((i+1)); " + f"done && " + f"{target_cmdline}" + ) + + ping_pong_script2 = [ + "sh", + "-c", + ping_pong_script_s, + ] + + ping_pong_script2_s = self._join_command_arguments( + ping_pong_script2 + ) + + # Run script within isolated env to get a true return codes of kill/terminate + final_script_s = ( + f"({ping_pong_script2_s}); " + f"printf \"%s!\" \"$?\" > {q_rc_file};" + ) + + cmds.append("(" + final_script_s + ")") + + cmdline = " && ".join(cmds) + + assert type(self._ssh_cmd) is list + assert len(self._ssh_cmd) > 0 + ssh_cmd = self._ssh_cmd + [cmdline] + + if encoding is not None and text is None: + text = True + + # 2. Run a local SSH client in the background + result._local_process = subprocess.Popen( + ssh_cmd, + stdin=stdin, + stdout=stdout, + stderr=stderr, + text=text, + encoding=encoding, + shell=False, + ) + assert result._local_process is not None + assert type(result._local_process) is subprocess.Popen + + # 3. Wait for the PID file to appear and be populated on the remote side. + + # A short wait loop (up to 5 seconds; 0.05s is usually sufficient) + start_time = time.monotonic() + nPass = 0 + while True: + if result._remote_pid is not None: + break + + if time.monotonic() - start_time < handshake_timeout: + pass + elif nPass < 10: + pass + else: + # If the PID could not be read, something went seriously wrong + # (e.g., the SSH connection dropped). + raise RuntimeError("Failed to retrieve remote process PID via temporary file.") + + nPass += 1 + + if nPass > 1: + time.sleep(0.05) + + # Reading the file contents + pid_bytes = self.read_binary( + pid_file, + offset=0, + size=__class__._C_MAX_RESP_PID_FILE_SIZE, + ) + assert type(pid_bytes) is bytes + + if len(pid_bytes) == __class__._C_MAX_RESP_PID_FILE_SIZE: + raise RuntimeError("Responce pid-file [{}] is too long.".format( + pid_file, + )) + + result._remote_pid = __class__._parse_resp_data( + pid_bytes, + ) + continue + + except BaseException: + p = result._local_process + result._local_process = None + if p is not None: + with p: + p.kill() + raise + finally: + # 4. Delete the temporary file; we no longer need it. + try: + self.remove_file(pid_file) + except BaseException: + p = result._local_process + result._local_process = None + if p is not None: + with p: + p.kill() + raise + + assert type(result._local_process) is subprocess.Popen + assert type(result._remote_pid) is int + + # 5. Putting back our brand-new control controller + return result + + def run( + self, + cmd: T_OS_CMD, + text: typing.Optional[bool] = None, + encoding: typing.Optional[str] = None, + shell: bool = False, + input: typing.Optional[OsOperations.T_INPUT] = None, + stdin: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stdout: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + stderr: typing.Optional[T_OS_IO_ID] = subprocess.PIPE, + exec_env: typing.Optional[OsOperations.T_EXEC_ENV] = None, + cwd: typing.Optional[str] = None, + timeout: typing.Optional[T_OS_TIMEOUT] = None, + check: bool = True, + ) -> OsCommandResult: + assert type(cmd) in [str, list] + assert text is None or type(text) is bool + assert encoding is None or type(encoding) is str + assert type(shell) is bool + assert input is None or type(input) in [str, bytes] or isinstance(input, io.IOBase) + assert stdin is None or type(stdin) is int or isinstance(stdin, io.IOBase) + assert stdout is None or type(stdout) is int or isinstance(stdout, io.IOBase) + assert stderr is None or type(stderr) is int or isinstance(stderr, io.IOBase) + assert exec_env is None or type(exec_env) is dict + assert cwd is None or type(cwd) is str + assert timeout is None or type(timeout) in [int, float] + assert type(check) is bool + + controller = self.popen( + cmd=cmd, + text=text, + encoding=encoding, + shell=shell, + stdin=stdin, + stdout=stdout, + stderr=stderr, + exec_env=exec_env, + cwd=cwd, + ) + assert isinstance(controller, OsProcessController) + + with controller: + try: + communicate_r = controller.communicate( + input=input, + timeout=timeout, + ) + except BaseException: + controller.kill() + raise + + assert type(communicate_r) is tuple + assert len(communicate_r) == 2 + + rc = controller.returncode + assert type(rc) is int + + result = OsCommandResult( + cmd=cmd, + returncode=rc, + stdout=communicate_r[0], + stderr=communicate_r[1], + ) + + if result.returncode == 0: + pass + elif check: + RaiseError.UtilityExitedWithNonZeroCode( + cmd=cmd, + exit_code=result.returncode, + msg_arg=result.stderr, + error=result.stderr, + out=result.stdout, + ) + + return result + + @staticmethod + def _parse_resp_data(data: bytes) -> typing.Optional[int]: + assert type(data) is bytes + + i = 0 + c = len(data) + + while True: + if i == c: + return None + + b = data[i] + assert type(b) is int + + if (b >= ord('0') and b <= ord('9')): + i += 1 + continue + + if b == ord('!') and i > 0 and (i + 1) == c: + return int(data[:i]) + + raise RuntimeError("[BUG CHECK] Bad data in pid responsed file: {!r}.".format( + data, + )) + def build_path(self, a: str, *parts: str) -> str: assert a is not None assert parts is not None @@ -1004,7 +1542,7 @@ def remove_file(self, filename: str) -> None: return # Processes control - def kill(self, pid: int, signal: typing.Union[int, os_signal.Signals]) -> None: + def kill(self, pid: int, signal: T_OS_SIGNAL) -> None: # Kill the process assert type(pid) is int assert type(signal) is int or type(signal) is os_signal.Signals diff --git a/src/static_config.py b/src/static_config.py new file mode 100644 index 0000000..8c2c357 --- /dev/null +++ b/src/static_config.py @@ -0,0 +1,90 @@ +# ////////////////////////////////////////////////////////////////////////////// +from __future__ import annotations + +import os +import typing +import logging + + +# ////////////////////////////////////////////////////////////////////////////// + +def _debug_print(tmpl: str, *args) -> None: + assert type(tmpl) is str + s = tmpl.format(*args) + logging.debug("[testgres.os_ops] DEBUG: {}".format(s)) + return + + +# ------------------------------------------------------------------------ +def _setup_new_cfg_opt_value( + env_param_name: str, + value: typing.Any, +) -> typing.Any: + assert type(env_param_name) is str + _debug_print( + "New value for cfg option {} is used: {!r}", + env_param_name, + value, + ) + return value + + +# ------------------------------------------------------------------------ +def _get_opt_float( + default_value: float, + env_param_name: str, + min_value: float, + max_value: float, +) -> float: + assert type(default_value) is float + assert type(env_param_name) is str + assert type(min_value) is float + assert type(max_value) is float + + env_val = os.environ.get(env_param_name) + + if env_val is None: + return default_value + + try: + val = float(env_val) + except ValueError: + _debug_print( + "Cfg property {} has wrong value {!r}. Default value is used {}.", + env_param_name, + env_val, + default_value, + ) + return default_value + + assert type(val) is float + + if val < min_value or max_value < val: + _debug_print( + "Cfg property {} has out of range value {}. Valid range is [{}..{}]. Default value {} is used.", + env_param_name, + val, + min_value, + max_value, + default_value, + ) + return default_value + + return _setup_new_cfg_opt_value( + env_param_name, + val, + ) + + +# ////////////////////////////////////////////////////////////////////////////// +# OsOperationStaticConfig + +class OsOperationStaticConfig: + remote_ops__popen__handshake_timeout = _get_opt_float( + 10.0, + "TESTGRES_OS_OPS_CFG__REMOTE_OPS__POPEN__HANDSHAKE_TIMEOUT", + 5.0, + 4 * 3600.0, + ) + +# ////////////////////////////////////////////////////////////////////////////// diff --git a/src/types.py b/src/types.py new file mode 100644 index 0000000..d614ece --- /dev/null +++ b/src/types.py @@ -0,0 +1,10 @@ +import typing +import signal as os_signal + + +T_OS_CMD = typing.Union[str, typing.List[str]] + +T_OS_SIGNAL = typing.Union[int, os_signal.Signals] +T_OS_TIMEOUT = typing.Union[int, float] +T_OS_IO = typing.IO[typing.Any] +T_OS_IO_ID = typing.Union[int, T_OS_IO] diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 6a32b66..b163b23 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -8,6 +8,10 @@ from tests.helpers.local_check import LocalCheck from tests.helpers.local_check import OsOpsHelpers +from src.os_ops import OsProcessController +from src.os_ops import OsCommandResult +from src.exceptions import ExecTimeoutException + import os import sys @@ -25,6 +29,9 @@ import datetime import threading import queue +import gc +import warnings +import tempfile from src.exceptions import InvalidOperationException from src.exceptions import ExecUtilException @@ -1843,12 +1850,11 @@ def test_is_port_free__false( py_server_code = py_server_code_templ.format(port) # Start a background process on the target machine - p = os_ops.exec_command( + p = os_ops.popen( ["python3", "-u", "-c", py_server_code], - get_process=True, encoding="utf-8", ) - assert isinstance(p, subprocess.Popen) + assert isinstance(p, OsProcessController) assert p.stdout is not None try: @@ -2275,14 +2281,15 @@ def test_kill( ] logging.info("Local test process is creating ...") - proc = os_ops.exec_command( + proc = os_ops.popen( cmd, encoding="utf-8", - get_process=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, ) assert proc is not None - assert type(proc) is subprocess.Popen + assert isinstance(proc, OsProcessController) assert proc.stdout is not None line = proc.stdout.readline() assert line is not None @@ -2375,14 +2382,13 @@ def test_kill__unk_pid( ] logging.info("Local test process is creating ...") - proc = os_ops.exec_command( + proc = os_ops.popen( cmd, encoding="utf-8", - get_process=True, ) assert proc is not None - assert type(proc) is subprocess.Popen + assert isinstance(proc, OsProcessController) proc_pid = proc.pid assert type(proc_pid) is int @@ -2790,13 +2796,12 @@ def test_get_process_children__no_children( sh_cmd = ["sh", "-c", "python3 -u -c 'import time;import os; print(os.getpid()); time.sleep(60)'"] - p1 = os_ops.exec_command( + p1 = os_ops.popen( sh_cmd, - get_process=True, encoding="utf-8", ) - assert isinstance(p1, subprocess.Popen) + assert isinstance(p1, OsProcessController) assert p1.stdout is not None line = p1.stdout.readline() @@ -2834,13 +2839,12 @@ def test_get_process_children__with_child( ) sh_cmd = ["python3", "-u", "-c", script] - p1 = os_ops.exec_command( + p1 = os_ops.popen( sh_cmd, - get_process=True, encoding="utf-8", ) - assert isinstance(p1, subprocess.Popen) + assert isinstance(p1, OsProcessController) assert p1.stdout is not None line = p1.stdout.readline() @@ -2896,13 +2900,12 @@ def test_get_process_children__with_three_children( ) sh_cmd = ["python3", "-u", "-c", script] - p = os_ops.exec_command( + p = os_ops.popen( sh_cmd, - get_process=True, encoding="utf-8", ) - assert isinstance(p, subprocess.Popen) + assert isinstance(p, OsProcessController) assert p.stdout is not None line = p.stdout.readline() @@ -3944,6 +3947,2252 @@ def thread_worker( logging.info("Total number of errors: {}".format(total_error_count)) return + @dataclasses.dataclass + class tagPOpenTestData: + param_text: typing.Optional[bool] + param_encoding: typing.Optional[str] + expected_result: typing.Union[str, bytes] + + def gen_sign(self) -> str: + return "text={!r}; encoding={!r}".format( + self.param_text, + self.param_encoding, + ) + + sm_POpenTestDatas: typing.List[tagPOpenTestData] = [ + tagPOpenTestData( + param_text=None, + param_encoding=None, + expected_result=b"hello\n", + ), + tagPOpenTestData( + param_text=False, + param_encoding=None, + expected_result=b"hello\n", + ), + tagPOpenTestData( + param_text=True, + param_encoding=None, + expected_result="hello\n", + ), + tagPOpenTestData( + param_text=None, + param_encoding="utf-8", + expected_result="hello\n", + ), + tagPOpenTestData( + param_text=True, + param_encoding="utf-8", + expected_result="hello\n", + ), + ] + + @pytest.fixture( + params=[ + pytest.param( + x, + id=x.gen_sign(), + ) + for x in sm_POpenTestDatas + ] + ) + def popen_data(self, request: pytest.FixtureRequest) -> tagPOpenTestData: + assert isinstance(request, pytest.FixtureRequest) + return request.param + + def test_popen_controller_stdout( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello"] + + controller = os_ops.popen( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + ) + assert isinstance(controller, OsProcessController) + assert controller.args == cmd + assert controller.args is not cmd + + with controller: + returncode = controller.wait() + assert returncode == 0 + assert controller.stdout is not None + v = controller.stdout.read() + assert type(v) is type(popen_data.expected_result) + assert len(v) > 0 + logging.info("stdout: {!r}".format(v)) + assert v == popen_data.expected_result + + assert controller.stderr is not None + x = controller.stderr.read() + assert len(x) == 0 + pass + + return + + def test_popen_controller_stderr( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello >&2"] + + controller = os_ops.popen( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + ) + assert isinstance(controller, OsProcessController) + assert controller.args == cmd + assert controller.args is not cmd + + with controller: + returncode = controller.wait() + assert returncode == 0 + assert controller.stderr is not None + v = controller.stderr.read() + assert type(v) is type(popen_data.expected_result) + assert len(v) > 0 + logging.info("stderr: {!r}".format(v)) + assert v == popen_data.expected_result + + assert controller.stdout is not None + x = controller.stdout.read() + assert len(x) == 0 + pass + + return + + def test_popen_controller_stdin( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "cat"] + + controller = os_ops.popen( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + ) + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.stdin is not None + controller.stdin.write(popen_data.expected_result) + controller.stdin.close() + + returncode = controller.wait() + assert returncode == 0 + assert controller.stdout is not None + v = controller.stdout.read() + assert type(v) is type(popen_data.expected_result) + assert len(v) > 0 + logging.info("stdout: {!r}".format(v)) + assert v == popen_data.expected_result + + assert controller.stderr is not None + x = controller.stderr.read() + assert len(x) == 0 + pass + + return + + def test_popen_stdin_stream( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "cat"] + + file_mode = "w+t" if popen_data.param_text or popen_data.param_encoding else "w+b" + + def LOCAL_f(data): + f = tempfile.TemporaryFile( + mode=file_mode, + encoding=popen_data.param_encoding, + ) + f.write(data) + f.seek(0) + return f + + # Yes, it it is not good. I know. + tmp_stdin = LOCAL_f(popen_data.expected_result) + + controller = os_ops.popen( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + stdin=tmp_stdin, + ) + + with tmp_stdin, controller: + assert isinstance(controller, OsProcessController) + + returncode = controller.wait() + assert returncode == 0 + assert controller.stdout is not None + v = controller.stdout.read() + assert type(v) is type(popen_data.expected_result) + assert len(v) > 0 + logging.info("stdout: {!r}".format(v)) + assert v == popen_data.expected_result + + assert controller.stderr is not None + x = controller.stderr.read() + assert len(x) == 0 + pass + + return + + @dataclasses.dataclass + class tagPOpenTestData2: + param_text: typing.Optional[bool] + param_encoding: typing.Optional[str] + expected_result1: typing.Union[str, bytes] + expected_result2: typing.Union[str, bytes] + + def gen_sign(self) -> str: + return "text={!r}; encoding={!r}".format( + self.param_text, + self.param_encoding, + ) + + sm_POpenTestDatas2: typing.List[tagPOpenTestData2] = [ + tagPOpenTestData2( + param_text=None, + param_encoding=None, + expected_result1=b"hello1\n", + expected_result2=b"hello2\n", + ), + tagPOpenTestData2( + param_text=False, + param_encoding=None, + expected_result1=b"hello1\n", + expected_result2=b"hello2\n", + ), + tagPOpenTestData2( + param_text=True, + param_encoding=None, + expected_result1="hello1\n", + expected_result2="hello2\n", + ), + tagPOpenTestData2( + param_text=None, + param_encoding="utf-8", + expected_result1="hello1\n", + expected_result2="hello2\n", + ), + tagPOpenTestData2( + param_text=True, + param_encoding="utf-8", + expected_result1="hello1\n", + expected_result2="hello2\n", + ), + ] + + @pytest.fixture( + params=[ + pytest.param( + x, + id=x.gen_sign(), + ) + for x in sm_POpenTestDatas2 + ] + ) + def popen_data2(self, request: pytest.FixtureRequest) -> tagPOpenTestData2: + assert isinstance(request, pytest.FixtureRequest) + assert type(request.param).__name__ == "tagPOpenTestData2" + return request.param + + def test_popen_controller_stderr_and_stdout( + self, + os_ops_descr: OsOpsDescr, + popen_data2: tagPOpenTestData2, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello1 && echo hello2 >&2"] + + controller = os_ops.popen( + cmd, + text=popen_data2.param_text, + encoding=popen_data2.param_encoding, + ) + assert isinstance(controller, OsProcessController) + + with controller: + returncode = controller.wait() + assert returncode == 0 + + assert controller.stdout is not None + v1 = controller.stdout.read() + assert type(v1) is type(popen_data2.expected_result1) + assert len(v1) > 0 + logging.info("stdout: {!r}".format(v1)) + assert v1 == popen_data2.expected_result1 + + assert controller.stderr is not None + v2 = controller.stderr.read() + assert type(v2) is type(popen_data2.expected_result2) + assert len(v2) > 0 + logging.info("stderr: {!r}".format(v2)) + assert v2 == popen_data2.expected_result2 + pass + + return + + def test_popen_stderr_and_stdout__streams( + self, + os_ops_descr: OsOpsDescr, + popen_data2: tagPOpenTestData2, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello1 && echo hello2 >&2"] + + # Configure the file opening mode (text or binary) based on the data matrix + file_mode = "w+t" if popen_data2.param_text or popen_data2.param_encoding else "w+b" + + def LOCAL_f(): + return tempfile.TemporaryFile( + mode=file_mode, + encoding=popen_data2.param_encoding, + ) + + # Yes, it it is not good. I know. + tmp_stderr = LOCAL_f() + tmp_stdout = LOCAL_f() + controller = os_ops.popen( + cmd, + text=popen_data2.param_text, + encoding=popen_data2.param_encoding, + stdout=tmp_stdout, + stderr=tmp_stderr, + ) + + with tmp_stderr, tmp_stdout, controller: + assert isinstance(controller, OsProcessController) + assert controller.wait() == 0 + assert controller.returncode == 0 + assert controller.stdout is None + assert controller.stderr is None + + tmp_stdout.seek(0) + v1 = tmp_stdout.read() + assert type(v1) is type(popen_data2.expected_result1) + assert len(v1) > 0 + logging.info("stdout: {!r}".format(v1)) + assert v1 == popen_data2.expected_result1 + + tmp_stderr.seek(0) + v2 = tmp_stderr.read() + assert type(v2) is type(popen_data2.expected_result2) + assert len(v2) > 0 + logging.info("stderr: {!r}".format(v2)) + assert v2 == popen_data2.expected_result2 + + return + + def test_popen_pid( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "printf \"%s!\" \"$$\""] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + ) + assert isinstance(controller, OsProcessController) + assert controller.stdout is not None + + with controller: + framework_pid = controller.pid + + v = "" + + nPass = 0 + + while True: + if nPass > 1: + time.sleep(0.1) + + if nPass == 100: + raise RuntimeError("Test hanged.") + + v += controller.stdout.read() + + actual_pid = __class__.helper__parse_pid_resp_str(v) + + if actual_pid is not None: + break + + continue + + assert framework_pid == actual_pid + pass + return + + @staticmethod + def helper__parse_pid_resp_str(data: str) -> typing.Optional[int]: + assert type(data) is str + + i = 0 + c = len(data) + + while True: + if i == c: + return None + + ch = data[i] + assert type(ch) is str + + if (ch.isdigit()): + i += 1 + continue + + if ch == '!' and i > 0 and (i + 1) == c: + return int(data[:i]) + + raise RuntimeError("Bad data in pid responsed data: {!r}.".format( + data, + )) + + def test_popen_returncode_stopped( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + for rc in range(256): + logging.info("test result code {}".format(rc)) + + cmd = ["sh", "-c", "exit {}".format(rc)] + + controller = os_ops.popen(cmd) + assert isinstance(controller, OsProcessController) + + with controller: + controller.wait() + returncode = controller.returncode + assert returncode == rc + assert controller.stderr is not None + v = controller.stderr.read() + assert len(v) == 0 + + assert controller.stdout is not None + v = controller.stdout.read() + assert len(v) == 0 + pass + continue + + return + + def test_popen_returncode_active( + self, + os_ops_descr: OsOpsDescr, + fx_data_wait_timeout: tagPOpenWaitTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + assert type(fx_data_wait_timeout) is __class__.tagPOpenWaitTestData + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + logging.info("cmd={}".format( + fx_data_wait_timeout.cmd, + )) + + controller = os_ops.popen( + fx_data_wait_timeout.cmd, + shell=type(fx_data_wait_timeout.cmd) is str + ) + + assert isinstance(controller, OsProcessController) + assert controller.args == fx_data_wait_timeout.cmd + assert type(controller.args) is type(fx_data_wait_timeout.cmd) + # it must be a copy + if type(controller.args) is str: + pass + else: + assert controller.args is not fx_data_wait_timeout.cmd + pass + + with controller: + try: + logging.info("controller.pid={}".format( + controller.pid, + )) + + assert controller.returncode is None + pass + finally: + logging.info("kill") + controller.kill() + pass + logging.info("EXIT1") + exit1_ts = time.monotonic() + + logging.info("EXIT2") + exit2_ts = time.monotonic() + + assert exit1_ts <= exit2_ts + + duration = exit2_ts - exit1_ts + + if 15 < duration: + raise RuntimeError("Test stops too long - {} second(s).".format( + duration, + )) + + return + + def test_popen_wait_returncode( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + for rc in range(256): + logging.info("test result code {}".format(rc)) + + cmd = ["sh", "-c", "exit {}".format(rc)] + + controller = os_ops.popen(cmd) + assert isinstance(controller, OsProcessController) + + with controller: + returncode = controller.wait() + assert returncode == rc + assert controller.stderr is not None + v = controller.stderr.read() + assert len(v) == 0 + + assert controller.stdout is not None + v = controller.stdout.read() + assert len(v) == 0 + pass + continue + + return + + def test_popen_terminate(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + cmd1 = ["sleep", "100"] + controller = os_ops.popen(cmd1) + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.pid > 0 + + # Check that the process is alive (returncode is still None) + assert controller.returncode is None + + # Send polite terminate + controller.terminate() + + # Waiting for completion. In Unix, a process killed by SIGTERM returns exit code -15 (or 143), + # depending on how `wait` reports the result locally versus remotely. + # The standard `subprocess` module returns the negative signal number (-15). + rc = controller.wait(timeout=5.0) + assert rc is not None + logging.info(f"Process terminated with exit code: {rc}") + + if type(os_ops).__name__ == "LocalOperations": + assert rc == -15 + elif type(os_ops).__name__ == "RemoteOperations": + assert rc == 143 + else: + raise RuntimeError("Unknown os_ops type: {}".format( + type(os_ops).__name__, + )) + pass + pass + return + + def test_popen_kill(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + # 2. Теперь тестируем жесткий kill() + cmd2 = ["sleep", "100"] + controller = os_ops.popen(cmd2) + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.pid > 0 + assert controller.returncode is None + + # Forcefully kill the process + controller.kill() + + # Waiting for completion. A process killed via SIGKILL returns code -9 (or 137). + rc = controller.wait(timeout=5.0) + assert rc is not None + logging.info(f"Process killed with exit code: {rc}") + + if type(os_ops).__name__ == "LocalOperations": + assert rc == -9 + elif type(os_ops).__name__ == "RemoteOperations": + assert rc == 137 + else: + raise RuntimeError("Unknown os_ops type: {}".format( + type(os_ops).__name__, + )) + pass + return + + @dataclasses.dataclass + class tagPOpenWaitTestData: + cmd: OsOperations.T_CMD + + def gen_sign(self) -> str: + return type(self.cmd).__name__ + ":" + repr(self.cmd) + + sm_POpenWaitTestDatas: typing.List[tagPOpenWaitTestData] = [ + tagPOpenWaitTestData( + cmd="sleep 100", + ), + tagPOpenWaitTestData( + cmd="sh -c \"sleep 100\"", + ), + tagPOpenWaitTestData( + cmd=["sleep", "100"], + ), + tagPOpenWaitTestData( + cmd=["sh", "-c", "sleep 100"], + ), + tagPOpenWaitTestData( + cmd=["bash", "-c", "sleep 100"], + ), + ] + + @pytest.fixture( + params=[ + pytest.param( + x, + id=x.gen_sign(), + ) + for x in sm_POpenWaitTestDatas + ] + ) + def fx_data_wait_timeout(self, request: pytest.FixtureRequest) -> tagPOpenWaitTestData: + assert isinstance(request, pytest.FixtureRequest) + assert type(request.param).__name__ == "tagPOpenWaitTestData" + return request.param + + def test_popen_wait_timeout( + self, + os_ops_descr: OsOpsDescr, + fx_data_wait_timeout: tagPOpenWaitTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + assert type(fx_data_wait_timeout) is __class__.tagPOpenWaitTestData + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + logging.info("cmd={}".format( + fx_data_wait_timeout.cmd, + )) + + controller = os_ops.popen( + fx_data_wait_timeout.cmd, + shell=type(fx_data_wait_timeout.cmd) is str + ) + + assert isinstance(controller, OsProcessController) + assert controller.args == fx_data_wait_timeout.cmd + assert type(controller.args) is type(fx_data_wait_timeout.cmd) + # it must be a copy + if type(controller.args) is str: + pass + else: + assert controller.args is not fx_data_wait_timeout.cmd + pass + + with controller: + try: + logging.info("controller.pid={}".format( + controller.pid, + )) + + with pytest.raises(expected_exception=ExecTimeoutException) as x: + controller.wait(1) + + assert type(x.value) is ExecTimeoutException + assert type(x.value.cmd) is type(fx_data_wait_timeout.cmd) + assert x.value.cmd == fx_data_wait_timeout.cmd + assert x.value.timeout == 1 + assert x.value.output is None + assert x.value.error is None + assert x.value.source == type(controller).__name__ + "::wait" + pass + finally: + logging.info("kill") + controller.kill() + pass + logging.info("EXIT1") + exit1_ts = time.monotonic() + + logging.info("EXIT2") + exit2_ts = time.monotonic() + + assert exit1_ts <= exit2_ts + + duration = exit2_ts - exit1_ts + + if 15 < duration: + raise RuntimeError("Test stops too long - {} second(s).".format( + duration, + )) + + return + + def test_popen_del( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "printf \"%s!\" \"$$\""] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + ) + + del controller + return + + def test_popen_garbage_collection( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + if type(os_ops).__name__ == "LocalOperations": + pytest.skip("It is not required") + + # Перехватываем системные предупреждения (ResourceWarning) + with warnings.catch_warnings(record=True) as caught_warnings: + # Включаем отображение ResourceWarning (по умолчанию в Python они могут быть скрыты) + warnings.simplefilter("always", ResourceWarning) + + # 1. Запускаем бесконечный процесс и умышленно НЕ используем контекстный менеджер + # Переменная 'controller' держит единственную ссылку на объект + cmd = ["sleep", "100"] + controller = os_ops.popen(cmd) + assert isinstance(controller, OsProcessController) + + # Запоминаем локальный процесс транспорта, чтобы проверить его смерть в конце + if type(controller).__name__ == "LocalProcessController": + local_p = controller._local_process + elif type(controller).__name__ == "RemoteProcessController": + local_p = controller._local_process + else: + raise RuntimeError("Unknown controller type: {}.".format( + type(controller).__name__ + )) + + assert local_p is not None + + # 2. Уничтожаем ЕДИНСТВЕННУЮ ссылку на контроллер (имитируем неаккуратность разработчика) + del controller + + # 3. Принудительно запускаем сборщик мусора Python, чтобы он очистил память + # и вызвал наш __del__ прямо здесь + gc.collect() + + # 4. Проверяем, что деструктор честно предупредил нас об утечке + assert len(caught_warnings) >= 1, "__del__ did not trigger any ResourceWarning!" + + # Ищем наше кастомное предупреждение в списке пойманных + has_our_warning = any( + "is still running inside" in str(w.message) or "is still running" in str(w.message) + for w in caught_warnings + ) + assert has_our_warning is True, "Our specific process leak warning was not found" + + # 5. Проверяем, что деструктор отработал как санитар: + # Локальный процесс SSH-клиента или sleep должен быть принудительно убит, + # чтобы дескрипторы не утекли в систему. + # Ожидаем завершения с коротким таймаутом (деструктор должен был сделать kill) + rc = local_p.wait(timeout=1.0) + assert rc is not None + logging.info(f"Leaked transport process reaped by __del__ with exit code: {rc}") + return + + def test_popen_set_env( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + envs: OsOperations.T_EXEC_ENV = { + "AAA": "abcdefg", + } + + cmd = ["sh", "-c", "printf \"%s!\" \"$AAA\""] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + exec_env=envs, + ) + + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.wait() == 0 + + assert controller.stdout is not None + s = controller.stdout.read() + + assert s == "abcdefg!" + + return + + def test_popen_set_env_via_os_ops( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + os_ops.set_env("AAA", "12345") + + cmd = ["sh", "-c", "printf \"%s!\" \"$AAA\""] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + ) + + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.wait() == 0 + assert controller.stdout is not None + s = controller.stdout.read() + assert s == "12345!" + + return + + def test_popen_unset_env( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = [printenv, "PATH"] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + ) + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.wait() == 0 + assert controller.stdout is not None + s = controller.stdout.read() + assert s != "" + + envs: OsOperations.T_EXEC_ENV = { + "PATH": None, + } + + controller = os_ops.popen( + cmd, + encoding="utf-8", + exec_env=envs, + ) + assert isinstance(controller, OsProcessController) + + with controller: + r = controller.wait() + assert controller.stdout is not None + s = controller.stdout.read() + assert s == "" + assert controller.stderr is not None + s = controller.stderr.read() + assert s == "" + assert r == 1 + + return + + def test_popen_unset_env_of_os_ops( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + os_ops.set_env("AAA", "abcdef") + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = [printenv, "AAA"] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + ) + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.wait() == 0 + assert controller.stdout is not None + s = controller.stdout.read() + assert s == "abcdef\n" + assert controller.stderr is not None + s = controller.stderr.read() + assert s == "" + + envs: OsOperations.T_EXEC_ENV = { + "AAA": None, + } + + controller = os_ops.popen( + cmd, + encoding="utf-8", + exec_env=envs, + ) + assert isinstance(controller, OsProcessController) + + with controller: + r = controller.wait() + assert controller.stdout is not None + s = controller.stdout.read() + assert s == "" + assert controller.stderr is not None + s = controller.stderr.read() + assert s == "" + assert r == 1 + + return + + def test_popen_replace_env_of_os_ops( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + os_ops.set_env("AAA", "abcdef") + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = [printenv, "AAA"] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + ) + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.wait() == 0 + assert controller.stdout is not None + s = controller.stdout.read() + assert s == "abcdef\n" + assert controller.stderr is not None + s = controller.stderr.read() + assert s == "" + + envs: OsOperations.T_EXEC_ENV = { + "AAA": "xyz", + } + + controller = os_ops.popen( + cmd, + encoding="utf-8", + exec_env=envs, + ) + assert isinstance(controller, OsProcessController) + + with controller: + r = controller.wait() + assert controller.stdout is not None + s = controller.stdout.read() + assert s == "xyz\n" + assert controller.stderr is not None + s = controller.stderr.read() + assert s == "" + assert r == 0 + + return + + def test_popen_cwd( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = ["pwd"] + + controller = os_ops.popen( + cmd, + encoding="utf-8", + cwd="/etc", + ) + assert isinstance(controller, OsProcessController) + + with controller: + assert controller.wait() == 0 + assert controller.stdout is not None + s = controller.stdout.read() + assert s == "/etc\n" + assert controller.stderr is not None + s = controller.stderr.read() + assert s == "" + + def test_popen_communicate_timeout( + self, + os_ops_descr: OsOpsDescr, + fx_data_wait_timeout: tagPOpenWaitTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + assert type(fx_data_wait_timeout) is __class__.tagPOpenWaitTestData + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + logging.info("cmd={}".format( + fx_data_wait_timeout.cmd, + )) + + controller = os_ops.popen( + fx_data_wait_timeout.cmd, + shell=type(fx_data_wait_timeout.cmd) is str + ) + + assert isinstance(controller, OsProcessController) + assert controller.args == fx_data_wait_timeout.cmd + assert type(controller.args) is type(fx_data_wait_timeout.cmd) + # it must be a copy + if type(controller.args) is str: + pass + else: + assert controller.args is not fx_data_wait_timeout.cmd + pass + + with controller: + try: + logging.info("controller.pid={}".format( + controller.pid, + )) + + with pytest.raises(expected_exception=ExecTimeoutException) as x: + controller.communicate(timeout=1) + + assert type(x.value) is ExecTimeoutException + assert type(x.value.cmd) is type(fx_data_wait_timeout.cmd) + assert x.value.cmd == fx_data_wait_timeout.cmd + assert x.value.timeout == 1 + assert x.value.output is None + assert x.value.error is None + assert x.value.source == type(controller).__name__ + "::communicate" + pass + finally: + logging.info("kill") + controller.kill() + pass + logging.info("EXIT1") + exit1_ts = time.monotonic() + + logging.info("EXIT2") + exit2_ts = time.monotonic() + + assert exit1_ts <= exit2_ts + + duration = exit2_ts - exit1_ts + + if 15 < duration: + raise RuntimeError("Test stops too long - {} second(s).".format( + duration, + )) + + return + + def test_popen_communicate( + self, + os_ops_descr: OsOpsDescr, + popen_data2: tagPOpenTestData2, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello1 && echo hello2 >&2"] + + controller = os_ops.popen( + cmd, + text=popen_data2.param_text, + encoding=popen_data2.param_encoding, + ) + assert isinstance(controller, OsProcessController) + + with controller: + r = controller.communicate() + assert type(r) is tuple + assert len(r) == 2 + + v1 = r[0] + assert v1 is not None + assert type(v1) is type(popen_data2.expected_result1) + assert len(v1) > 0 + logging.info("stdout: {!r}".format(v1)) + assert v1 == popen_data2.expected_result1 + + v2 = r[1] + assert v2 is not None + assert type(v2) is type(popen_data2.expected_result2) + assert len(v2) > 0 + logging.info("stderr: {!r}".format(v2)) + assert v2 == popen_data2.expected_result2 + pass + + return + + def test_popen_communicate_with_input( + self, + os_ops_descr: OsOpsDescr, + popen_data2: tagPOpenTestData2, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "cat && echo hello2 >&2"] + + controller = os_ops.popen( + cmd, + text=popen_data2.param_text, + encoding=popen_data2.param_encoding, + ) + assert isinstance(controller, OsProcessController) + + with controller: + r = controller.communicate(input=popen_data2.expected_result1) + assert type(r) is tuple + assert len(r) == 2 + + v1 = r[0] + assert v1 is not None + assert type(v1) is type(popen_data2.expected_result1) + assert len(v1) > 0 + logging.info("stdout: {!r}".format(v1)) + assert v1 == popen_data2.expected_result1 + + v2 = r[1] + assert v2 is not None + assert type(v2) is type(popen_data2.expected_result2) + assert len(v2) > 0 + logging.info("stderr: {!r}".format(v2)) + assert v2 == popen_data2.expected_result2 + pass + + return + + def test_popen_pool_stopped( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + for rc in range(256): + logging.info("test result code {}".format(rc)) + + cmd = ["sh", "-c", "exit {}".format(rc)] + + controller = os_ops.popen(cmd) + assert isinstance(controller, OsProcessController) + + with controller: + controller.wait() + returncode = controller.poll() + assert returncode is not None + assert type(returncode) is int + assert returncode == rc + assert controller.stderr is not None + v = controller.stderr.read() + assert len(v) == 0 + + assert controller.stdout is not None + v = controller.stdout.read() + assert len(v) == 0 + pass + continue + + return + + def test_popen_poll_active( + self, + os_ops_descr: OsOpsDescr, + fx_data_wait_timeout: tagPOpenWaitTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + assert type(fx_data_wait_timeout) is __class__.tagPOpenWaitTestData + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + logging.info("cmd={}".format( + fx_data_wait_timeout.cmd, + )) + + controller = os_ops.popen( + fx_data_wait_timeout.cmd, + shell=type(fx_data_wait_timeout.cmd) is str + ) + + assert isinstance(controller, OsProcessController) + assert controller.args == fx_data_wait_timeout.cmd + assert type(controller.args) is type(fx_data_wait_timeout.cmd) + # it must be a copy + if type(controller.args) is str: + pass + else: + assert controller.args is not fx_data_wait_timeout.cmd + pass + + with controller: + try: + logging.info("controller.pid={}".format( + controller.pid, + )) + + assert controller.poll() is None + pass + finally: + logging.info("kill") + controller.kill() + pass + logging.info("EXIT1") + exit1_ts = time.monotonic() + + logging.info("EXIT2") + exit2_ts = time.monotonic() + + assert exit1_ts <= exit2_ts + + duration = exit2_ts - exit1_ts + + if 15 < duration: + raise RuntimeError("Test stops too long - {} second(s).".format( + duration, + )) + + return + + def test_run_success(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "python3 --version"] + + exec_r = os_ops.run(cmd) + assert type(exec_r) is OsCommandResult + assert type(exec_r.stdout) is bytes + assert type(exec_r.stderr) is bytes + assert b'Python 3.' in exec_r.stdout + assert exec_r.stderr == b'' + return + + def test_run_failure(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + RunConditions.skip_if_windows() + + cmd = ["sh", "-c", "nonexistent_command"] + + while True: + try: + os_ops.run(cmd) + except ExecUtilException as e: + assert type(e.exit_code) is int + assert e.exit_code == 127 + + 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 + assert "not found" in e.message + assert b"nonexistent_command" in e.error + assert b"not found" in e.error + break + raise Exception("We wait an exception!") + return + + def test_run_stdout( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello"] + + exec_r = os_ops.run( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + v = exec_r.stdout + assert type(v) is type(popen_data.expected_result) + assert len(v) > 0 + logging.info("stdout: {!r}".format(v)) + assert v == popen_data.expected_result + + assert exec_r.stderr is not None + x = exec_r.stderr + assert len(x) == 0 + + return + + def test_run_stdout_stream( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + # Author: Mark G + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + cmd = ["sh", "-c", "echo hello"] + + # Configure the file opening mode (text or binary) based on the data matrix + file_mode = "w+t" if popen_data.param_text or popen_data.param_encoding else "w+b" + + # Create a temporary file where the utility will write its stdout + with tempfile.TemporaryFile(mode=file_mode, encoding=popen_data.param_encoding) as tmp_stdout: + + exec_r = os_ops.run( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + # Forwarding our stream for writing! + stdout=tmp_stdout, + ) + assert isinstance(exec_r, OsCommandResult) + assert exec_r.returncode == 0 + assert exec_r.stdout is None + assert exec_r.stderr is not None + assert len(exec_r.stderr) == 0 + + # By design, since stdout is redirected to a file, + # communicate() returns an empty placeholder (None, "" or b"") + # Let's verify that the result structure is empty here: + assert not exec_r.stdout, f"Expected empty result.stdout, but got: {exec_r.stdout!r}" + + # Now we verify that the data has been physically written to our stream: + tmp_stdout.seek(0) + file_content = tmp_stdout.read() + + assert type(file_content) is type(popen_data.expected_result) + assert file_content == popen_data.expected_result + logging.info("stdout stream content: {!r}".format(file_content)) + + return + + def test_run_stderr( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello >&2"] + + exec_r = os_ops.run( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stderr is not None + v = exec_r.stderr + assert type(v) is type(popen_data.expected_result) + assert len(v) > 0 + logging.info("stderr: {!r}".format(v)) + assert v == popen_data.expected_result + + assert exec_r.stdout is not None + x = exec_r.stdout + assert len(x) == 0 + + return + + def test_run_stderr_stream( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + # Author: Mark G + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + cmd = ["sh", "-c", "echo hello >&2"] + + # Configure the file opening mode (text or binary) based on the data matrix + file_mode = "w+t" if popen_data.param_text or popen_data.param_encoding else "w+b" + + # Create a temporary file where the utility will write its stdout + with tempfile.TemporaryFile(mode=file_mode, encoding=popen_data.param_encoding) as tmp_stderr: + + exec_r = os_ops.run( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + # Forwarding our stream for writing! + stderr=tmp_stderr, + ) + assert isinstance(exec_r, OsCommandResult) + assert exec_r.returncode == 0 + assert exec_r.stderr is None + assert exec_r.stdout is not None + assert len(exec_r.stdout) == 0 + + assert not exec_r.stdout, f"Expected empty result.stdout, but got: {exec_r.stdout!r}" + + tmp_stderr.seek(0) + file_content = tmp_stderr.read() + + assert type(file_content) is type(popen_data.expected_result) + assert file_content == popen_data.expected_result + logging.info("stdout stream content: {!r}".format(file_content)) + + return + + def test_run_stderr_and_stdout( + self, + os_ops_descr: OsOpsDescr, + popen_data2: tagPOpenTestData2, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello1 && echo hello2 >&2"] + + exec_r = os_ops.run( + cmd, + text=popen_data2.param_text, + encoding=popen_data2.param_encoding, + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + + assert exec_r.stdout is not None + v1 = exec_r.stdout + assert type(v1) is type(popen_data2.expected_result1) + assert len(v1) > 0 + logging.info("stdout: {!r}".format(v1)) + assert v1 == popen_data2.expected_result1 + + assert exec_r.stderr is not None + v2 = exec_r.stderr + assert type(v2) is type(popen_data2.expected_result2) + assert len(v2) > 0 + logging.info("stderr: {!r}".format(v2)) + assert v2 == popen_data2.expected_result2 + + return + + def test_run_stderr_and_stdout__streams( + self, + os_ops_descr: OsOpsDescr, + popen_data2: tagPOpenTestData2, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo hello1 && echo hello2 >&2"] + + # Configure the file opening mode (text or binary) based on the data matrix + file_mode = "w+t" if popen_data2.param_text or popen_data2.param_encoding else "w+b" + + def LOCAL_f(): + return tempfile.TemporaryFile( + mode=file_mode, + encoding=popen_data2.param_encoding, + ) + + with LOCAL_f() as tmp_stderr, LOCAL_f() as tmp_stdout: + exec_r = os_ops.run( + cmd, + text=popen_data2.param_text, + encoding=popen_data2.param_encoding, + stdout=tmp_stdout, + stderr=tmp_stderr, + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is None + assert exec_r.stderr is None + + tmp_stdout.seek(0) + v1 = tmp_stdout.read() + assert type(v1) is type(popen_data2.expected_result1) + assert len(v1) > 0 + logging.info("stdout: {!r}".format(v1)) + assert v1 == popen_data2.expected_result1 + + tmp_stderr.seek(0) + v2 = tmp_stderr.read() + assert type(v2) is type(popen_data2.expected_result2) + assert len(v2) > 0 + logging.info("stderr: {!r}".format(v2)) + assert v2 == popen_data2.expected_result2 + + return + + def test_run_input( + self, + os_ops_descr: OsOpsDescr, + popen_data: tagPOpenTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "cat"] + + exec_r = os_ops.run( + cmd, + text=popen_data.param_text, + encoding=popen_data.param_encoding, + input=popen_data.expected_result, + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + v = exec_r.stdout + assert type(v) is type(popen_data.expected_result) + assert len(v) > 0 + logging.info("stdout: {!r}".format(v)) + assert v == popen_data.expected_result + + assert exec_r.stderr is not None + x = exec_r.stderr + assert len(x) == 0 + pass + + return + + def test_run_cwd( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = ["pwd"] + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + cwd="/etc", + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + assert exec_r.stdout == "/etc\n" + assert exec_r.stderr is not None + assert exec_r.stderr == "" + return + + def test_run_set_env( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + envs: OsOperations.T_EXEC_ENV = { + "AAA": "abcdefg", + } + + cmd = ["sh", "-c", "printf \"%s!\" \"$AAA\""] + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + exec_env=envs, + ) + + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + assert exec_r.stdout == "abcdefg!" + return + + def test_run_set_env_via_os_ops( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + os_ops.set_env("AAA", "12345") + + cmd = ["sh", "-c", "printf \"%s!\" \"$AAA\""] + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + ) + + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + assert exec_r.stdout == "12345!" + return + + def test_run_unset_env( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = [printenv, "PATH"] + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + assert exec_r.stdout != "" + + envs: OsOperations.T_EXEC_ENV = { + "PATH": None, + } + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + exec_env=envs, + check=False, + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 1 + assert exec_r.stdout == "" + assert exec_r.stderr == "" + return + + def test_run_unset_env_of_os_ops( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + os_ops.set_env("AAA", "abcdef") + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = [printenv, "AAA"] + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + s = exec_r.stdout + assert s == "abcdef\n" + assert exec_r.stderr is not None + s = exec_r.stderr + assert s == "" + + envs: OsOperations.T_EXEC_ENV = { + "AAA": None, + } + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + exec_env=envs, + check=False, + ) + assert type(exec_r) is OsCommandResult + + r = exec_r.returncode + assert exec_r.stdout is not None + s = exec_r.stdout + assert s == "" + assert exec_r.stderr is not None + s = exec_r.stderr + assert s == "" + assert r == 1 + + return + + def test_run_replace_env_of_os_ops( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + + os_ops.set_env("AAA", "abcdef") + + printenv = os_ops.find_executable("printenv") + assert type(printenv) is str + assert printenv != "" + + cmd = [printenv, "AAA"] + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + ) + assert type(exec_r) is OsCommandResult + + assert exec_r.returncode == 0 + assert exec_r.stdout is not None + s = exec_r.stdout + assert s == "abcdef\n" + assert exec_r.stderr is not None + s = exec_r.stderr + assert s == "" + + envs: OsOperations.T_EXEC_ENV = { + "AAA": "xyz", + } + + exec_r = os_ops.run( + cmd, + encoding="utf-8", + exec_env=envs, + ) + assert type(exec_r) is OsCommandResult + + r = exec_r.returncode + assert exec_r.stdout is not None + s = exec_r.stdout + assert s == "xyz\n" + assert exec_r.stderr is not None + s = exec_r.stderr + assert s == "" + assert r == 0 + + return + + def test_run_timeout( + self, + os_ops_descr: OsOpsDescr, + fx_data_wait_timeout: tagPOpenWaitTestData, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + assert type(fx_data_wait_timeout) is __class__.tagPOpenWaitTestData + + RunConditions.skip_if_windows() + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + logging.info("cmd={}".format( + fx_data_wait_timeout.cmd, + )) + + start_ts = time.monotonic() + + with pytest.raises(expected_exception=ExecTimeoutException) as x: + os_ops.run( + fx_data_wait_timeout.cmd, + shell=type(fx_data_wait_timeout.cmd) is str, + timeout=1, + ) + + stop_ts = time.monotonic() + + assert type(x.value) is ExecTimeoutException + assert type(x.value.cmd) is type(fx_data_wait_timeout.cmd) + assert x.value.cmd == fx_data_wait_timeout.cmd + assert x.value.timeout == 1 + assert x.value.output is None + assert x.value.error is None + + if type(os_ops).__name__ == "LocalOperations": + assert x.value.source == "LocalProcessController::communicate" + elif type(os_ops).__name__ == "RemoteOperations": + assert x.value.source == "RemoteProcessController::communicate" + else: + raise RuntimeError("Unknown os_ops type {}".format( + type(os_ops).__name__, + )) + + assert start_ts <= stop_ts + duration = stop_ts - start_ts + assert duration <= 10 # OK? + + return + + def test_run_large_input_output(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + # Author: Mark G + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + # Generate a 256 KB test string (deliberately larger than the 64 KB pipe buffer) + # If we used a simple wait() instead of communicate(), this volume of data would cause a deadlock. + chunk = "A" * 1024 + "\n" + large_text = chunk * 256 # 256 Килобайт текста + + # Run `cat` in text mode. It should ingest the entire input and spit it back out. + cmd = ["cat"] + + start_ts = time.monotonic() + + result = os_ops.run( + cmd=cmd, + text=True, + encoding="utf-8", + input=large_text, + # With a margin for network transmission in RemoteOperations + timeout=30.0, + ) + + duration = time.monotonic() - start_ts + logging.info(f"Processed 256KB via run() in {duration:.4f} seconds.") + + # Strict result checks + assert isinstance(result, OsCommandResult) + assert result.returncode == 0 + assert type(result.stdout) is str + assert len(result.stdout) == len(large_text) + assert result.stdout == large_text + # cat should not write anything to stderr + assert result.stderr == "" + return + + def test_run_check_exception(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + # Author: Mark G + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + # A command guaranteed to fail (exit code 1) + # and write messages to both streams + cmd = ["sh", "-c", "echo normal_out && echo error_err >&2 && exit 1"] + + # 1. Check default behavior (check=True) + # Expect our strict framework exception + with pytest.raises(expected_exception=Exception) as x: + os_ops.run(cmd, text=True, encoding="utf-8", check=True) + + # Verify that the correct exception was raised + assert x.type is ExecUtilException + + # Verify that the exception contains the correct output streams + # Check the field names (error, out, exit_code) against your RaiseError + assert getattr(x.value, "exit_code", 1) == 1 + assert getattr(x.value, "out", "").strip() == "normal_out" + assert getattr(x.value, "error", "").strip() == "error_err" + + # 2. Test the negative scenario with validation disabled (check=False) + # The method must not fail; instead, it should return a valid result object. + result = os_ops.run(cmd, text=True, encoding="utf-8", check=False) + + assert isinstance(result, OsCommandResult) + assert result.returncode == 1 + assert type(result.stdout) is str + assert type(result.stderr) is str + assert result.stdout.strip() == "normal_out" + assert result.stderr.strip() == "error_err" + return + + def test_run_stdin_stream(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + # Author: Mark G + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + expected_text = "hello from file stream stdin\n" + + # 1. Create a local temporary file with data + # Use tempfile, which is guaranteed to clean itself up + with tempfile.TemporaryFile(mode="w+t", encoding="utf-8") as tmp_file: + tmp_file.write(expected_text) + # Reset the pointer to the beginning so the process can read the data + tmp_file.seek(0) + + # Launch the cat utility. + # Instead of an input parameter, pass an active file descriptor to stdin. + cmd = ["cat"] + + result = os_ops.run( + cmd=cmd, + text=True, + encoding="utf-8", + stdin=tmp_file, # Forwarding the stream! + input=None, # Explicitly test the scenario without input + check=True, + ) + + # 2. Strict checks + assert isinstance(result, OsCommandResult) + assert result.returncode == 0 + assert type(result.stdout) is str + assert result.stdout == expected_text + assert result.stderr == "" + return + + def test_run_check_exception2__list(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + cmd = ["sh", "-c", "echo normal_out && echo error_err >&2 && exit 1"] + + # 1. Check default behavior (check=True) + with pytest.raises(expected_exception=ExecUtilException) as x: + os_ops.run(cmd, text=True, encoding="utf-8", check=True) + + assert x.type is ExecUtilException + assert type(x.value.out) is str + assert type(x.value.error) is str + assert x.value.exit_code == 1 + assert x.value.out == "normal_out\n" + assert x.value.error == "error_err\n" + assert x.value.command == cmd + assert type(x.value.description) is str + assert x.value.description == ( + """Utility exited with non-zero code (1). Error: `error_err`""" + ) + assert type(x.value.message) is str + assert x.value.message == ( + """Utility exited with non-zero code (1). Error: `error_err`\n""" + """Command: sh -c echo normal_out && echo error_err >&2 && exit 1\n""" + """Exit code: 1\n""" + """---- Error:\n""" + """error_err\n""" + """\n""" + """---- Out:\n""" + """normal_out\n""" + ) + + # 2. Test the negative scenario with validation disabled (check=False) + result = os_ops.run(cmd, text=True, encoding="utf-8", check=False) + + assert isinstance(result, OsCommandResult) + assert result.returncode == 1 + assert type(result.stdout) is str + assert type(result.stderr) is str + assert result.stdout == "normal_out\n" + assert result.stderr == "error_err\n" + return + + def test_run_check_exception3__str(self, os_ops_descr: OsOpsDescr): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + RunConditions.skip_if_windows() + os_ops = os_ops_descr.os_ops + + cmd = "sh -c \"echo normal_out && echo error_err >&2 && exit 1\"" + + # 1. Check default behavior (check=True) + with pytest.raises(expected_exception=ExecUtilException) as x: + os_ops.run(cmd, text=True, encoding="utf-8", shell=True, check=True) + + assert x.type is ExecUtilException + assert type(x.value.out) is str + assert type(x.value.error) is str + assert x.value.exit_code == 1 + assert x.value.out == "normal_out\n" + assert x.value.error == "error_err\n" + assert x.value.command == cmd + assert type(x.value.description) is str + assert x.value.description == ( + """Utility exited with non-zero code (1). Error: `error_err`""" + ) + assert type(x.value.message) is str + assert x.value.message == ( + """Utility exited with non-zero code (1). Error: `error_err`\n""" + """Command: sh -c \"echo normal_out && echo error_err >&2 && exit 1\"\n""" + """Exit code: 1\n""" + """---- Error:\n""" + """error_err\n""" + """\n""" + """---- Out:\n""" + """normal_out\n""" + ) + + # 2. Test the negative scenario with validation disabled (check=False) + result = os_ops.run(cmd, text=True, encoding="utf-8", shell=True, check=False) + + assert isinstance(result, OsCommandResult) + assert result.returncode == 1 + assert type(result.stdout) is str + assert type(result.stderr) is str + assert result.stdout == "normal_out\n" + assert result.stderr == "error_err\n" + return + @staticmethod def helper__get_os_ops( use_clone: bool, diff --git a/tests/test_os_ops_remote.py b/tests/test_os_ops_remote.py index 002afb7..acb081a 100755 --- a/tests/test_os_ops_remote.py +++ b/tests/test_os_ops_remote.py @@ -6,6 +6,7 @@ from tests.helpers.local_check import LocalCheck from src.exceptions import ExecUtilException +from src.remote_ops import RemoteProcessController import pytest @@ -123,3 +124,25 @@ def test_get_file_size__unk_file( assert "No such file or directory" in str(x.value) assert "/dummy" in str(x.value) return + + def test_popen_communicate__rc_file_is_deleted( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + cmd = ["sh", "-c", "echo 123"] + + with os_ops.popen(cmd) as controller: + assert type(controller) is RemoteProcessController + + assert controller._remote_rc_file is not None + assert os_ops.path_exists(controller._remote_rc_file) + rc_file = controller._remote_rc_file + + assert not os_ops.path_exists(rc_file) + return diff --git a/tests/units/exceptions/ExecTimeoutException/__init__.py b/tests/units/exceptions/ExecTimeoutException/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/units/exceptions/ExecTimeoutException/test_set001__constructor.py b/tests/units/exceptions/ExecTimeoutException/test_set001__constructor.py new file mode 100644 index 0000000..cabbecc --- /dev/null +++ b/tests/units/exceptions/ExecTimeoutException/test_set001__constructor.py @@ -0,0 +1,51 @@ +from src.exceptions import ExecTimeoutException + + +class TestSet001_Constructor: + def test_001__minimal(self): + e = ExecTimeoutException("cat", 1) + assert e.source is None + assert e.message == "Command 'cat' timed out after 1 second(s)." + assert e.cmd == "cat" + assert e.timeout == 1 + assert e.output is None + assert e.error is None + assert str(e) == "Command 'cat' timed out after 1 second(s)." + assert repr(e) == "ExecTimeoutException(cmd='cat', timeout=1)" + return + + def test_002__source(self): + e = ExecTimeoutException("cat", 1, source="aa") + assert e.source == "aa" + assert e.message == "Command 'cat' timed out after 1 second(s)." + assert e.cmd == "cat" + assert e.timeout == 1 + assert e.output is None + assert e.error is None + assert str(e) == "Command 'cat' timed out after 1 second(s)." + assert repr(e) == "ExecTimeoutException(cmd='cat', timeout=1, source='aa')" + return + + def test_003__output(self): + e = ExecTimeoutException("cat", 1, output=b"bb") + assert e.source is None + assert e.message == "Command 'cat' timed out after 1 second(s)." + assert e.cmd == "cat" + assert e.timeout == 1 + assert e.output == b"bb" + assert e.error is None + assert str(e) == "Command 'cat' timed out after 1 second(s)." + assert repr(e) == "ExecTimeoutException(cmd='cat', timeout=1, output=b'bb')" + return + + def test_004__error(self): + e = ExecTimeoutException("cat", 1, error=b"ee") + assert e.source is None + assert e.message == "Command 'cat' timed out after 1 second(s)." + assert e.cmd == "cat" + assert e.timeout == 1 + assert e.output is None + assert e.error == b"ee" + assert str(e) == "Command 'cat' timed out after 1 second(s)." + assert repr(e) == "ExecTimeoutException(cmd='cat', timeout=1, error=b'ee')" + return