Migration to new os_ops methods: run and popen

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

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

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)