[#266] OsOperations::build_path is added (#273)

This thing must be used instead os.path.join(...).

LocalOperations uses os.path.join
RemoteOperations uses posixpath.join

We updated:
 - utils.get_bin_path2
 - utils.get_pg_config2
 - utils.get_pg_version2 (note: new restrictions/asserts are added)
 - RemoteOperations::find_executable
 - RemoteOperations::copytree
 - PostgresNode::logs_dir
 - PostgresNode::data_dir
 - PostgresNode::utils_log_file
 - PostgresNode::pg_log_file
 - PostgresNode::_create_recovery_conf
 - PostgresNode::_collect_special_files
 - PostgresNode::default_conf
 - PostgresNode::append_conf
 - PostgresNode::set_auto_conf
 - PostgresNode::upgrade_from
 - PostgresNode::_get_bin_path
 - NodeApp::__init__
 - NodeApp::make_simple
 - cache.cached_initdb
 - cache.call_initdb
 - NodeBackup.log_file
 - NodeBackup.__init__
 - NodeBackup._prepare_dir

Author: dmitry-lipetsk

testgres/backup.py

@@ -1,7 +1,5 @@
 # coding: utf-8
 
-import os
-
 from six import raise_from
 
 from .enums import XLogMethod
@@ -29,7 +27,9 @@ class NodeBackup(object):
     """
     @property
     def log_file(self):
-        return os.path.join(self.base_dir, BACKUP_LOG_FILE)
+        assert self.os_ops is not None
+        assert isinstance(self.os_ops, OsOperations)
+        return self.os_ops.build_path(self.base_dir, BACKUP_LOG_FILE)
 
     def __init__(self,
                  node,
@@ -75,7 +75,7 @@ def __init__(self,
         # private
         self._available = True
 
-        data_dir = os.path.join(self.base_dir, DATA_DIR)
+        data_dir = self.os_ops.build_path(self.base_dir, DATA_DIR)
 
         _params = [
             get_bin_path2(self.os_ops, "pg_basebackup"),
@@ -112,10 +112,13 @@ def _prepare_dir(self, destroy):
         available = not destroy
 
         if available:
+            assert self.os_ops is not None
+            assert isinstance(self.os_ops, OsOperations)
+
             dest_base_dir = self.os_ops.mkdtemp(prefix=TMP_NODE)
 
-            data1 = os.path.join(self.base_dir, DATA_DIR)
-            data2 = os.path.join(dest_base_dir, DATA_DIR)
+            data1 = self.os_ops.build_path(self.base_dir, DATA_DIR)
+            data2 = self.os_ops.build_path(dest_base_dir, DATA_DIR)
 
             try:
                 # Copy backup to new data dir

testgres/cache.py

@@ -1,7 +1,5 @@
 # coding: utf-8
 
-import os
-
 from six import raise_from
 
 from .config import testgres_config
@@ -39,7 +37,7 @@ def make_utility_path(name):
         assert type(name) == str  # noqa: E721
 
         if bin_path:
-            return os.path.join(bin_path, name)
+            return os_ops.build_path(bin_path, name)
 
         return get_bin_path2(os_ops, name)
 
@@ -72,7 +70,7 @@ def call_initdb(initdb_dir, log=logfile):
                 # XXX: write new unique system id to control file
                 # Some users might rely upon unique system ids, but
                 # our initdb caching mechanism breaks this contract.
-                pg_control = os.path.join(data_dir, XLOG_CONTROL_FILE)
+                pg_control = os_ops.build_path(data_dir, XLOG_CONTROL_FILE)
                 system_id = generate_system_id()
                 cur_pg_control = os_ops.read(pg_control, binary=True)
                 new_pg_control = system_id + cur_pg_control[len(system_id):]

testgres/node.py

@@ -573,7 +573,11 @@ def bin_dir(self):
 
     @property
     def logs_dir(self):
-        path = os.path.join(self.base_dir, LOGS_DIR)
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
+        path = self._os_ops.build_path(self.base_dir, LOGS_DIR)
+        assert type(path) == str  # noqa: E721
 
         # NOTE: it's safe to create a new dir
         if not self.os_ops.path_exists(path):
@@ -583,16 +587,31 @@ def logs_dir(self):
 
     @property
     def data_dir(self):
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
         # NOTE: we can't run initdb without user's args
-        return os.path.join(self.base_dir, DATA_DIR)
+        path = self._os_ops.build_path(self.base_dir, DATA_DIR)
+        assert type(path) == str  # noqa: E721
+        return path
 
     @property
     def utils_log_file(self):
-        return os.path.join(self.logs_dir, UTILS_LOG_FILE)
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
+        path = self._os_ops.build_path(self.logs_dir, UTILS_LOG_FILE)
+        assert type(path) == str  # noqa: E721
+        return path
 
     @property
     def pg_log_file(self):
-        return os.path.join(self.logs_dir, PG_LOG_FILE)
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
+        path = self._os_ops.build_path(self.logs_dir, PG_LOG_FILE)
+        assert type(path) == str  # noqa: E721
+        return path
 
     @property
     def version(self):
@@ -719,7 +738,11 @@ def _create_recovery_conf(self, username, slot=None):
         ).format(options_string(**conninfo))  # yapf: disable
         # Since 12 recovery.conf had disappeared
         if self.version >= PgVer('12'):
-            signal_name = os.path.join(self.data_dir, "standby.signal")
+            assert self._os_ops is not None
+            assert isinstance(self._os_ops, OsOperations)
+
+            signal_name = self._os_ops.build_path(self.data_dir, "standby.signal")
+            assert type(signal_name) == str  # noqa: E721
             self.os_ops.touch(signal_name)
         else:
             line += "standby_mode=on\n"
@@ -768,11 +791,14 @@ def _collect_special_files(self):
         result = []
 
         # list of important files + last N lines
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
         files = [
-            (os.path.join(self.data_dir, PG_CONF_FILE), 0),
-            (os.path.join(self.data_dir, PG_AUTO_CONF_FILE), 0),
-            (os.path.join(self.data_dir, RECOVERY_CONF_FILE), 0),
-            (os.path.join(self.data_dir, HBA_CONF_FILE), 0),
+            (self._os_ops.build_path(self.data_dir, PG_CONF_FILE), 0),
+            (self._os_ops.build_path(self.data_dir, PG_AUTO_CONF_FILE), 0),
+            (self._os_ops.build_path(self.data_dir, RECOVERY_CONF_FILE), 0),
+            (self._os_ops.build_path(self.data_dir, HBA_CONF_FILE), 0),
             (self.pg_log_file, testgres_config.error_log_lines)
         ]  # yapf: disable
 
@@ -840,8 +866,11 @@ def default_conf(self,
             This instance of :class:`.PostgresNode`.
         """
 
-        postgres_conf = os.path.join(self.data_dir, PG_CONF_FILE)
-        hba_conf = os.path.join(self.data_dir, HBA_CONF_FILE)
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
+        postgres_conf = self._os_ops.build_path(self.data_dir, PG_CONF_FILE)
+        hba_conf = self._os_ops.build_path(self.data_dir, HBA_CONF_FILE)
 
         # filter lines in hba file
         # get rid of comments and blank lines
@@ -956,7 +985,7 @@ def append_conf(self, line='', filename=PG_CONF_FILE, **kwargs):
                 # format a new config line
                 lines.append('{} = {}'.format(option, value))
 
-        config_name = os.path.join(self.data_dir, filename)
+        config_name = self._os_ops.build_path(self.data_dir, filename)
         conf_text = ''
         for line in lines:
             conf_text += text_type(line) + '\n'
@@ -2040,8 +2069,11 @@ def set_auto_conf(self, options, config='postgresql.auto.conf', rm_options={}):
             rm_options (set, optional): A set containing the names of the options to remove.
                                          Defaults to an empty set.
         """
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
         # parse postgresql.auto.conf
-        path = os.path.join(self.data_dir, config)
+        path = self.os_ops.build_path(self.data_dir, config)
 
         lines = self.os_ops.readlines(path)
         current_options = {}
@@ -2127,8 +2159,11 @@ def upgrade_from(self, old_node, options=None, expect_error=False):
         return self.os_ops.exec_command(upgrade_command, expect_error=expect_error)
 
     def _get_bin_path(self, filename):
+        assert self._os_ops is not None
+        assert isinstance(self._os_ops, OsOperations)
+
         if self.bin_dir:
-            bin_path = os.path.join(self.bin_dir, filename)
+            bin_path = self._os_ops.build_path(self.bin_dir, filename)
         else:
             bin_path = get_bin_path2(self.os_ops, filename)
         return bin_path
@@ -2333,7 +2368,7 @@ def __init__(self, test_path=None, nodes_to_cleanup=None, os_ops=None):
             if os.path.isabs(test_path):
                 self.test_path = test_path
             else:
-                self.test_path = os.path.join(os_ops.cwd(), test_path)
+                self.test_path = os_ops.build_path(os_ops.cwd(), test_path)
         else:
             self.test_path = os_ops.cwd()
         self.nodes_to_cleanup = nodes_to_cleanup if nodes_to_cleanup else []
@@ -2344,7 +2379,7 @@ def make_empty(
             base_dir=None,
             port=None,
             bin_dir=None):
-        real_base_dir = os.path.join(self.test_path, base_dir)
+        real_base_dir = self.os_ops.build_path(self.test_path, base_dir)
         self.os_ops.rmdirs(real_base_dir, ignore_errors=True)
         self.os_ops.makedirs(real_base_dir)
 
@@ -2373,7 +2408,7 @@ def make_simple(
             initdb_params=initdb_params, allow_streaming=set_replication)
 
         # set major version
-        pg_version_file = self.os_ops.read(os.path.join(node.data_dir, 'PG_VERSION'))
+        pg_version_file = self.os_ops.read(self.os_ops.build_path(node.data_dir, 'PG_VERSION'))
         node.major_version_str = str(pg_version_file.rstrip())
         node.major_version = float(node.major_version_str)
 

testgres/operations/local_ops.py

@@ -219,6 +219,13 @@ def exec_command(self, cmd, wait_exit=False, verbose=False, expect_error=False,
 
         return output
 
+    def build_path(self, a: str, *parts: str) -> str:
+        assert a is not None
+        assert parts is not None
+        assert type(a) == str  # noqa: E721
+        assert type(parts) == tuple  # noqa: E721
+        return os.path.join(a, *parts)
+
     # Environment setup
     def environ(self, var_name):
         return os.environ.get(var_name)

testgres/operations/os_ops.py

@@ -25,6 +25,13 @@ def __init__(self, username=None):
     def exec_command(self, cmd, **kwargs):
         raise NotImplementedError()
 
+    def build_path(self, a: str, *parts: str) -> str:
+        assert a is not None
+        assert parts is not None
+        assert type(a) == str  # noqa: E721
+        assert type(parts) == tuple  # noqa: E721
+        raise NotImplementedError()
+
     # Environment setup
     def environ(self, var_name):
         raise NotImplementedError()

testgres/operations/remote_ops.py

@@ -1,5 +1,6 @@
 import getpass
 import os
+import posixpath
 import platform
 import subprocess
 import tempfile
@@ -138,6 +139,13 @@ def exec_command(self, cmd, wait_exit=False, verbose=False, expect_error=False,
 
         return output
 
+    def build_path(self, a: str, *parts: str) -> str:
+        assert a is not None
+        assert parts is not None
+        assert type(a) == str  # noqa: E721
+        assert type(parts) == tuple  # noqa: E721
+        return __class__._build_path(a, *parts)
+
     # Environment setup
     def environ(self, var_name: str) -> str:
         """
@@ -159,7 +167,7 @@ def find_executable(self, executable):
 
         search_paths = search_paths.split(self.pathsep)
         for path in search_paths:
-            remote_file = os.path.join(path, executable)
+            remote_file = __class__._build_path(path, executable)
             if self.isfile(remote_file):
                 return remote_file
 
@@ -383,7 +391,7 @@ def mkstemp(self, prefix=None):
 
     def copytree(self, src, dst):
         if not os.path.isabs(dst):
-            dst = os.path.join('~', dst)
+            dst = __class__._build_path('~', dst)
         if self.isdir(dst):
             raise FileExistsError("Directory {} already exists.".format(dst))
         return self.exec_command("cp -r {} {}".format(src, dst))
@@ -772,6 +780,14 @@ def _quote_envvar(value: str) -> str:
         result += "\""
         return result
 
+    @staticmethod
+    def _build_path(a: str, *parts: str) -> str:
+        assert a is not None
+        assert parts is not None
+        assert type(a) == str  # noqa: E721
+        assert type(parts) == tuple  # noqa: E721
+        return posixpath.join(a, *parts)
+
 
 def normalize_error(error):
     if isinstance(error, bytes):

testgres/utils.py

@@ -141,17 +141,17 @@ def get_bin_path2(os_ops: OsOperations, filename):
 
     if pg_config:
         bindir = get_pg_config(pg_config, os_ops)["BINDIR"]
-        return os.path.join(bindir, filename)
+        return os_ops.build_path(bindir, filename)
 
     # try PG_BIN
     pg_bin = os_ops.environ("PG_BIN")
     if pg_bin:
-        return os.path.join(pg_bin, filename)
+        return os_ops.build_path(pg_bin, filename)
 
     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)
+        return os_ops.build_path(bindir, filename)
 
     return filename
 
@@ -213,7 +213,7 @@ def cache_pg_config_data(cmd):
     # try PG_BIN
     pg_bin = os.environ.get("PG_BIN")
     if pg_bin:
-        cmd = os.path.join(pg_bin, "pg_config")
+        cmd = os_ops.build_path(pg_bin, "pg_config")
         return cache_pg_config_data(cmd)
 
     # try plain name
@@ -227,8 +227,17 @@ def get_pg_version2(os_ops: OsOperations, bin_dir=None):
     assert os_ops is not None
     assert isinstance(os_ops, OsOperations)
 
+    C_POSTGRES_BINARY = "postgres"
+
     # Get raw version (e.g., postgres (PostgreSQL) 9.5.7)
-    postgres_path = os.path.join(bin_dir, 'postgres') if bin_dir else get_bin_path2(os_ops, 'postgres')
+    if bin_dir is None:
+        postgres_path = get_bin_path2(os_ops, C_POSTGRES_BINARY)
+    else:
+        # [2025-06-25] OK ?
+        assert type(bin_dir) == str  # noqa: E721
+        assert bin_dir != ""
+        postgres_path = os_ops.build_path(bin_dir, 'postgres')
+
     cmd = [postgres_path, '--version']
     raw_ver = os_ops.exec_command(cmd, encoding='utf-8')