diff --git a/src/impl/platforms/linux/internal_platform_utils.py b/src/impl/platforms/linux/internal_platform_utils.py index 95d9d15a..1afcc0fd 100644 --- a/src/impl/platforms/linux/internal_platform_utils.py +++ b/src/impl/platforms/linux/internal_platform_utils.py @@ -5,7 +5,9 @@ from ....raise_error import RaiseError from testgres.operations.os_ops import OsOperations +from testgres.operations.os_ops import OsCommandResult from testgres.operations.exceptions import ExecUtilException +from testgres.operations.types import T_OS_EXEC_ENV import re import shlex @@ -17,7 +19,7 @@ class InternalPlatformUtils(base.InternalPlatformUtils): C_MAX_FIND_POSTMASTER_ATTEMPTS = 5 C_BASH_EXE = "/bin/bash" - sm_exec_env = { + sm_exec_env: T_OS_EXEC_ENV = { "LANG": "en_US.UTF-8", "LC_ALL": "en_US.UTF-8", } @@ -113,37 +115,32 @@ def _FindPostmaster( "ps -ewwo \"pid=,ppid=,args=\" | grep -E " + shlex.quote(regexp), ] - exec_r = os_ops.exec_command( + exec_r = os_ops.run( cmd=cmd, - ignore_errors=True, - verbose=True, + check=False, exec_env=__class__.sm_exec_env, ) - assert type(exec_r) is tuple - assert len(exec_r) == 3 + assert type(exec_r) is OsCommandResult + assert type(exec_r.returncode) is int + assert type(exec_r.stdout) is bytes + assert type(exec_r.stderr) is bytes - exit_status, output_b, error_b = exec_r - - assert type(exit_status) is int - assert type(output_b) is bytes - assert type(error_b) is bytes - - if exit_status == 1: + if exec_r.returncode == 1: return None - output = output_b.decode("utf-8") - error = error_b.decode("utf-8") + output = exec_r.stdout.decode("utf-8") + error = exec_r.stderr.decode("utf-8") assert type(output) is str assert type(error) is str - if exit_status != 0: - errMsg = f"test command returned an unexpected exit code: {exit_status}" + if exec_r.returncode != 0: + errMsg = f"test command returned an unexpected exit code: {exec_r.returncode}" raise ExecUtilException( message=errMsg, command=cmd, - exit_code=exit_status, + exit_code=exec_r.returncode, out=output, error=error, ) diff --git a/src/node.py b/src/node.py index c18cbae6..c8a849c2 100644 --- a/src/node.py +++ b/src/node.py @@ -100,6 +100,8 @@ from .backup import NodeBackup from testgres.operations.os_ops import OsOperations +from testgres.operations.os_ops import OsCommandResult +from testgres.operations.os_ops import OsProcessController from testgres.operations.local_ops import LocalOperations InternalError = pglib.InternalError @@ -673,7 +675,11 @@ def _try_shutdown_internal(self, max_attempts, with_force): ps_command = ['ps', '-o', 'pid=', '-p', str(node_pid)] - ps_output = self._os_ops.exec_command(cmd=ps_command, shell=True, ignore_errors=True).decode('utf-8') + ps_output = self._os_ops.run( + cmd=ps_command, + shell=True, + check=False, + ).stdout.decode('utf-8') assert type(ps_output) is str if ps_output == "": @@ -692,7 +698,11 @@ def _try_shutdown_internal(self, max_attempts, with_force): pass # Check that node stopped - print only column pid without headers - ps_output = self._os_ops.exec_command(cmd=ps_command, shell=True, ignore_errors=True).decode('utf-8') + ps_output = self._os_ops.run( + cmd=ps_command, + shell=True, + check=False, + ).stdout.decode('utf-8') assert type(ps_output) is str if ps_output == "": @@ -1598,7 +1608,7 @@ def psql(self, assert port is None or type(port) is int assert type(variables) is dict - return self._psql( + r = self._psql( ignore_errors=True, query=query, filename=filename, @@ -1609,6 +1619,8 @@ def psql(self, port=port, **variables ) + assert type(r) is OsCommandResult + return r.returncode, r.stdout, r.stderr def _psql( self, @@ -1620,7 +1632,8 @@ def _psql( input=None, host: typing.Optional[str] = None, port: typing.Optional[int] = None, - **variables): + **variables + ) -> OsCommandResult: assert host is None or type(host) is str assert port is None or type(port) is int assert type(variables) is dict @@ -1670,13 +1683,15 @@ def _psql( else: raise QueryException('Query or filename must be provided') - return self._os_ops.exec_command( + r = self._os_ops.run( psql_params, - verbose=True, input=input, stderr=subprocess.PIPE, stdout=subprocess.PIPE, - ignore_errors=ignore_errors) + check=not ignore_errors, + ) + assert type(r) is OsCommandResult + return r @method_decorator(positional_args_hack(['dbname', 'query'])) def safe_psql(self, query=None, expect_error=False, **kwargs): @@ -1704,7 +1719,7 @@ def safe_psql(self, query=None, expect_error=False, **kwargs): # force this setting kwargs['ON_ERROR_STOP'] = 1 try: - ret, out, err = self._psql(ignore_errors=False, query=query, **kwargs) + exec_r = self._psql(ignore_errors=False, query=query, **kwargs) except ExecUtilException as e: if not expect_error: raise QueryException(e.message, query) @@ -1719,7 +1734,7 @@ def safe_psql(self, query=None, expect_error=False, **kwargs): if expect_error: raise InvalidOperationException("Exception was expected, but query finished successfully: `{}`.".format(query)) - return out + return exec_r.stdout def dump(self, filename=None, @@ -2052,12 +2067,14 @@ def subscribe(self, dbname=dbname, username=username, **params) # yapf: enable - def pgbench(self, - dbname=None, - username=None, - stdout=None, - stderr=None, - options=None): + def pgbench( + self, + dbname=None, + username=None, + stdout=None, + stderr=None, + options=None, + ) -> OsProcessController: """ Spawn a pgbench process. @@ -2069,7 +2086,7 @@ def pgbench(self, options: additional options for pgbench (list). Returns: - Process created by subprocess.Popen. + OsProcessController. """ if options is None: options = [] @@ -2086,10 +2103,14 @@ def pgbench(self, # should be the last one _params.append(dbname) - proc = self._os_ops.exec_command(_params, stdout=stdout, stderr=stderr, get_process=True) + proc = self._os_ops.popen( + _params, + stdout=stdout, + stderr=stderr, + ) # [2026-06-21] It is so - assert isinstance(proc, subprocess.Popen) + assert isinstance(proc, OsProcessController) return proc def pgbench_with_wait(self, @@ -2349,7 +2370,22 @@ def upgrade_from(self, old_node, options=None, expect_error=False): ] upgrade_command += options - return self._os_ops.exec_command(upgrade_command, expect_error=expect_error) + r: typing.Optional[typing.Any] = None + try: + r = self._os_ops.run(upgrade_command).stdout + except BaseException as e: + if not expect_error: + raise + + logging.info("Exception ({}): {}".format( + type(e).__name__, + e, + )) + + if expect_error: + raise RuntimeError("Operation executed without any errors.") + + return r def _release_resources(self): self._free_port() diff --git a/src/utils.py b/src/utils.py index 622ec9e4..5fa75e1f 100644 --- a/src/utils.py +++ b/src/utils.py @@ -264,7 +264,7 @@ def get_pg_config2(os_ops: OsOperations, pg_config_path): def cache_pg_config_data(cmd): # execute pg_config and get the output - out = os_ops.exec_command(cmd, encoding='utf-8') + out = os_ops.run(cmd, encoding='utf-8').stdout assert type(out) is str data = {} @@ -335,7 +335,8 @@ def get_pg_version2(os_ops: OsOperations, bin_dir=None): postgres_path = os_ops.build_path(bin_dir, 'postgres') cmd = [postgres_path, '--version'] - raw_ver = os_ops.exec_command(cmd, encoding='utf-8') + raw_ver = os_ops.run(cmd, encoding='utf-8').stdout + assert type(raw_ver) is str return parse_pg_version(raw_ver)