execute_utility2, get_bin_path2, get_pg_config2 are added

testgres/backup.py

@@ -15,9 +15,11 @@
 
 from .exceptions import BackupException
 
+from .operations.os_ops import OsOperations
+
 from .utils import \
-    get_bin_path, \
-    execute_utility, \
+    get_bin_path2, \
+    execute_utility2, \
     clean_on_error
 
 
@@ -44,6 +46,9 @@ def __init__(self,
             username: database user name.
             xlog_method: none | fetch | stream (see docs)
         """
+        assert node.os_ops is not None
+        assert isinstance(node.os_ops, OsOperations)
+
         if not options:
             options = []
         self.os_ops = node.os_ops
@@ -73,15 +78,15 @@ def __init__(self,
         data_dir = os.path.join(self.base_dir, DATA_DIR)
 
         _params = [
-            get_bin_path("pg_basebackup"),
+            get_bin_path2(self.os_ops, "pg_basebackup"),
             "-p", str(node.port),
             "-h", node.host,
             "-U", username,
             "-D", data_dir,
             "-X", xlog_method.value
         ]  # yapf: disable
         _params += options
-        execute_utility(_params, self.log_file)
+        execute_utility2(self.os_ops, _params, self.log_file)
 
     def __enter__(self):
         return self

testgres/cache.py

@@ -15,8 +15,8 @@
     ExecUtilException
 
 from .utils import \
-    get_bin_path, \
-    execute_utility
+    get_bin_path2, \
+    execute_utility2
 
 from .operations.local_ops import LocalOperations
 from .operations.os_ops import OsOperations
@@ -27,11 +27,23 @@ def cached_initdb(data_dir, logfile=None, params=None, os_ops: OsOperations = Lo
     Perform initdb or use cached node files.
     """
 
+    assert os_ops is not None
+    assert isinstance(os_ops, OsOperations)
+
+    def make_utility_path(name):
+        assert name is not None
+        assert type(name) == str  # noqa: E721
+
+        if bin_path:
+            return os.path.join(bin_path, name)
+
+        return get_bin_path2(os_ops, name)
+
     def call_initdb(initdb_dir, log=logfile):
         try:
-            initdb_path = os.path.join(bin_path, 'initdb') if bin_path else get_bin_path("initdb")
+            initdb_path = make_utility_path("initdb")
             _params = [initdb_path, "-D", initdb_dir, "-N"]
-            execute_utility(_params + (params or []), log)
+            execute_utility2(os_ops, _params + (params or []), log)
         except ExecUtilException as e:
             raise_from(InitNodeException("Failed to run initdb"), e)
 
@@ -63,8 +75,8 @@ def call_initdb(initdb_dir, log=logfile):
                 os_ops.write(pg_control, new_pg_control, truncate=True, binary=True, read_and_write=True)
 
                 # XXX: build new WAL segment with our system id
-                _params = [get_bin_path("pg_resetwal"), "-D", data_dir, "-f"]
-                execute_utility(_params, logfile)
+                _params = [make_utility_path("pg_resetwal"), "-D", data_dir, "-f"]
+                execute_utility2(os_ops, _params, logfile)
 
         except ExecUtilException as e:
             msg = "Failed to reset WAL for system id"

testgres/node.py

@@ -89,9 +89,9 @@
 from .utils import \
     PgVer, \
     eprint, \
-    get_bin_path, \
+    get_bin_path2, \
     get_pg_version, \
-    execute_utility, \
+    execute_utility2, \
     options_string, \
     clean_on_error
 
@@ -301,7 +301,7 @@ def base_dir(self):
     @property
     def bin_dir(self):
         if not self._bin_dir:
-            self._bin_dir = os.path.dirname(get_bin_path("pg_config"))
+            self._bin_dir = os.path.dirname(get_bin_path2(self.os_ops, "pg_config"))
         return self._bin_dir
 
     @property
@@ -684,7 +684,7 @@ def status(self):
                 "-D", self.data_dir,
                 "status"
             ]  # yapf: disable
-            status_code, out, error = execute_utility(_params, self.utils_log_file, verbose=True)
+            status_code, out, error = execute_utility2(self.os_ops, _params, self.utils_log_file, verbose=True)
             if error and 'does not exist' in error:
                 return NodeStatus.Uninitialized
             elif 'no server running' in out:
@@ -710,7 +710,7 @@ def get_control_data(self):
         _params += ["-D"] if self._pg_version >= PgVer('9.5') else []
         _params += [self.data_dir]
 
-        data = execute_utility(_params, self.utils_log_file)
+        data = execute_utility2(self.os_ops, _params, self.utils_log_file)
 
         out_dict = {}
 
@@ -793,7 +793,7 @@ def start(self, params=[], wait=True):
 
         def LOCAL__start_node():
             # 'error' will be None on Windows
-            _, _, error = execute_utility(_params, self.utils_log_file, verbose=True)
+            _, _, error = execute_utility2(self.os_ops, _params, self.utils_log_file, verbose=True)
             assert error is None or type(error) == str  # noqa: E721
             if error and 'does not exist' in error:
                 raise Exception(error)
@@ -882,7 +882,7 @@ def stop(self, params=[], wait=True):
             "stop"
         ] + params  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility2(self.os_ops, _params, self.utils_log_file)
 
         self._maybe_stop_logger()
         self.is_started = False
@@ -924,7 +924,7 @@ def restart(self, params=[]):
         ] + params  # yapf: disable
 
         try:
-            error_code, out, error = execute_utility(_params, self.utils_log_file, verbose=True)
+            error_code, out, error = execute_utility2(self.os_ops, _params, self.utils_log_file, verbose=True)
             if error and 'could not start server' in error:
                 raise ExecUtilException
         except ExecUtilException as e:
@@ -953,7 +953,7 @@ def reload(self, params=[]):
             "reload"
         ] + params  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility2(self.os_ops, _params, self.utils_log_file)
 
         return self
 
@@ -975,7 +975,7 @@ def promote(self, dbname=None, username=None):
             "promote"
         ]  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility2(self.os_ops, _params, self.utils_log_file)
 
         # for versions below 10 `promote` is asynchronous so we need to wait
         # until it actually becomes writable
@@ -1010,7 +1010,7 @@ def pg_ctl(self, params):
             "-w"  # wait
         ] + params  # yapf: disable
 
-        return execute_utility(_params, self.utils_log_file)
+        return execute_utility2(self.os_ops, _params, self.utils_log_file)
 
     def free_port(self):
         """
@@ -1230,7 +1230,7 @@ def tmpfile():
             "-F", format.value
         ]  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility2(self.os_ops, _params, self.utils_log_file)
 
         return filename
 
@@ -1259,7 +1259,7 @@ def restore(self, filename, dbname=None, username=None):
 
         # try pg_restore if dump is binary format, and psql if not
         try:
-            execute_utility(_params, self.utils_log_name)
+            execute_utility2(self.os_ops, _params, self.utils_log_name)
         except ExecUtilException:
             self.psql(filename=filename, dbname=dbname, username=username)
 
@@ -1612,7 +1612,7 @@ def pgbench_run(self, dbname=None, username=None, options=[], **kwargs):
         # should be the last one
         _params.append(dbname)
 
-        return execute_utility(_params, self.utils_log_file)
+        return execute_utility2(self.os_ops, _params, self.utils_log_file)
 
     def connect(self,
                 dbname=None,
@@ -1809,7 +1809,7 @@ def _get_bin_path(self, filename):
         if self.bin_dir:
             bin_path = os.path.join(self.bin_dir, filename)
         else:
-            bin_path = get_bin_path(filename)
+            bin_path = get_bin_path2(self.os_ops, filename)
         return bin_path
 
     def _escape_config_value(value):

testgres/utils.py

@@ -16,6 +16,8 @@
 from .helpers.port_manager import PortManager
 from .exceptions import ExecUtilException
 from .config import testgres_config as tconf
+from .operations.os_ops import OsOperations
+from .operations.remote_ops import RemoteOperations
 
 # rows returned by PG_CONFIG
 _pg_config_data = {}
@@ -68,7 +70,14 @@ def execute_utility(args, logfile=None, verbose=False):
     Returns:
         stdout of executed utility.
     """
-    exit_status, out, error = tconf.os_ops.exec_command(args, verbose=True)
+    return execute_utility2(tconf.os_ops, args, logfile, verbose)
+
+
+def execute_utility2(os_ops: OsOperations, args, logfile=None, verbose=False):
+    assert os_ops is not None
+    assert isinstance(os_ops, OsOperations)
+
+    exit_status, out, error = os_ops.exec_command(args, verbose=True)
     # decode result
     out = '' if not out else out
     if isinstance(out, bytes):
@@ -79,11 +88,11 @@ def execute_utility(args, logfile=None, verbose=False):
     # write new log entry if possible
     if logfile:
         try:
-            tconf.os_ops.write(filename=logfile, data=args, truncate=True)
+            os_ops.write(filename=logfile, data=args, truncate=True)
             if out:
                 # comment-out lines
                 lines = [u'\n'] + ['# ' + line for line in out.splitlines()] + [u'\n']
-                tconf.os_ops.write(filename=logfile, data=lines)
+                os_ops.write(filename=logfile, data=lines)
         except IOError:
             raise ExecUtilException(
                 "Problem with writing to logfile `{}` during run command `{}`".format(logfile, args))
@@ -98,25 +107,32 @@ def get_bin_path(filename):
     Return absolute path to an executable using PG_BIN or PG_CONFIG.
     This function does nothing if 'filename' is already absolute.
     """
+    return get_bin_path2(tconf.os_ops, filename)
+
+
+def get_bin_path2(os_ops: OsOperations, filename):
+    assert os_ops is not None
+    assert isinstance(os_ops, OsOperations)
+
     # check if it's already absolute
     if os.path.isabs(filename):
         return filename
-    if tconf.os_ops.remote:
+    if isinstance(os_ops, RemoteOperations):
         pg_config = os.environ.get("PG_CONFIG_REMOTE") or os.environ.get("PG_CONFIG")
     else:
         # try PG_CONFIG - get from local machine
         pg_config = os.environ.get("PG_CONFIG")
 
     if pg_config:
-        bindir = get_pg_config()["BINDIR"]
+        bindir = get_pg_config(pg_config, os_ops)["BINDIR"]
         return os.path.join(bindir, filename)
 
     # try PG_BIN
-    pg_bin = tconf.os_ops.environ("PG_BIN")
+    pg_bin = os_ops.environ("PG_BIN")
     if pg_bin:
         return os.path.join(pg_bin, filename)
 
-    pg_config_path = tconf.os_ops.find_executable('pg_config')
+    pg_config_path = os_ops.find_executable('pg_config')
     if pg_config_path:
         bindir = get_pg_config(pg_config_path)["BINDIR"]
         return os.path.join(bindir, filename)
@@ -129,12 +145,20 @@ def get_pg_config(pg_config_path=None, os_ops=None):
     Return output of pg_config (provided that it is installed).
     NOTE: this function caches the result by default (see GlobalConfig).
     """
-    if os_ops:
-        tconf.os_ops = os_ops
+
+    if os_ops is None:
+        os_ops = tconf.os_ops
+
+    return get_pg_config2(os_ops, pg_config_path)
+
+
+def get_pg_config2(os_ops: OsOperations, pg_config_path):
+    assert os_ops is not None
+    assert isinstance(os_ops, OsOperations)
 
     def cache_pg_config_data(cmd):
         # execute pg_config and get the output
-        out = tconf.os_ops.exec_command(cmd, encoding='utf-8')
+        out = os_ops.exec_command(cmd, encoding='utf-8')
 
         data = {}
         for line in out.splitlines():
@@ -158,7 +182,7 @@ def cache_pg_config_data(cmd):
         return _pg_config_data
 
     # try specified pg_config path or PG_CONFIG
-    if tconf.os_ops.remote:
+    if isinstance(os_ops, RemoteOperations):
         pg_config = pg_config_path or os.environ.get("PG_CONFIG_REMOTE") or os.environ.get("PG_CONFIG")
     else:
         # try PG_CONFIG - get from local machine