PBCKP-152 use black for formatting

Author: v.shepard

testgres/cache.py

@@ -4,8 +4,8 @@
 
 from six import raise_from
 
-from .op_ops.local_ops import LocalOperations
-from .op_ops.os_ops import OsOperations
+from .os_ops.local_ops import LocalOperations
+from .os_ops.os_ops import OsOperations
 from .config import testgres_config
 
 from .consts import XLOG_CONTROL_FILE
@@ -25,6 +25,7 @@ def cached_initdb(data_dir, logfile=None, params=None, os_ops: OsOperations = Lo
     """
     Perform initdb or use cached node files.
     """
+    testgres_config.os_ops = os_ops
 
     def call_initdb(initdb_dir, log=None):
         try:
@@ -60,7 +61,7 @@ def call_initdb(initdb_dir, log=None):
 
                 # XXX: build new WAL segment with our system id
                 _params = [get_bin_path("pg_resetwal"), "-D", data_dir, "-f"]
-                execute_utility(_params, logfile, os_ops)
+                execute_utility(_params, logfile, os_ops=os_ops)
 
         except ExecUtilException as e:
             msg = "Failed to reset WAL for system id"

testgres/config.py

@@ -5,9 +5,8 @@
 import tempfile
 
 from contextlib import contextmanager
-from shutil import rmtree
 
-from .op_ops.local_ops import LocalOperations
+from .os_ops.local_ops import LocalOperations
 from .consts import TMP_CACHE
 
 
@@ -127,7 +126,7 @@ def copy(self):
 cached_initdb_dirs = set()
 
 # default config object
-testgres_config = GlobalConfig()
+testgres_config = GlobalConfig(os_ops=LocalOperations())
 
 # NOTE: for compatibility
 TestgresConfig = testgres_config
@@ -138,6 +137,7 @@ def copy(self):
 
 @atexit.register
 def _rm_cached_initdb_dirs(os_ops=LocalOperations()):
+    testgres_config.os_ops = os_ops
     for d in cached_initdb_dirs:
         os_ops.rmdirs(d, ignore_errors=True)
 

testgres/defaults.py

@@ -1,9 +1,8 @@
 import datetime
-import os
 import struct
 import uuid
 
-from .op_ops.local_ops import LocalOperations
+from .os_ops.local_ops import LocalOperations
 
 
 def default_dbname():

testgres/node.py

@@ -1,20 +1,16 @@
 # coding: utf-8
 
-import io
 import os
 import random
-import shutil
 import signal
 import threading
 from queue import Queue
 
 import psutil
-import subprocess
 import time
 
-from op_ops.local_ops import LocalOperations
-from op_ops.os_ops import OsOperations
-from op_ops.remote_ops import RemoteOperations
+from .os_ops.local_ops import LocalOperations
+from .os_ops.remote_ops import RemoteOperations
 
 try:
     from collections.abc import Iterable
@@ -32,7 +28,6 @@
 
 from shutil import rmtree
 from six import raise_from, iteritems, text_type
-from tempfile import mkstemp, mkdtemp
 
 from .enums import \
     NodeStatus, \
@@ -96,7 +91,6 @@
     eprint, \
     get_bin_path, \
     get_pg_version, \
-    file_tail, \
     reserve_port, \
     release_port, \
     execute_utility, \
@@ -163,6 +157,7 @@ def __init__(self, name=None, port=None, base_dir=None,
         else:
             self.os_ops = RemoteOperations(host, hostname, ssh_key)
 
+        testgres_config.os_ops = self.os_ops
         # defaults for __exit__()
         self.cleanup_on_good_exit = testgres_config.node_cleanup_on_good_exit
         self.cleanup_on_bad_exit = testgres_config.node_cleanup_on_bad_exit
@@ -628,7 +623,7 @@ def status(self):
                 "-D", self.data_dir,
                 "status"
             ]  # yapf: disable
-            execute_utility(_params, self.utils_log_file)
+            execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
             return NodeStatus.Running
 
         except ExecUtilException as e:
@@ -650,7 +645,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_utility(_params, self.utils_log_file, os_ops=self.os_ops)
 
         out_dict = {}
 
@@ -713,7 +708,7 @@ def start(self, params=[], wait=True):
         ] + params  # yapf: disable
 
         try:
-            execute_utility(_params, self.utils_log_file)
+            execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
         except ExecUtilException as e:
             msg = 'Cannot start node'
             files = self._collect_special_files()
@@ -744,7 +739,7 @@ def stop(self, params=[], wait=True):
             "stop"
         ] + params  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
 
         self._maybe_stop_logger()
         self.is_started = False
@@ -786,7 +781,7 @@ def restart(self, params=[]):
         ] + params  # yapf: disable
 
         try:
-            execute_utility(_params, self.utils_log_file)
+            execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
         except ExecUtilException as e:
             msg = 'Cannot restart node'
             files = self._collect_special_files()
@@ -813,7 +808,7 @@ def reload(self, params=[]):
             "reload"
         ] + params  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
 
         return self
 
@@ -835,7 +830,7 @@ def promote(self, dbname=None, username=None):
             "promote"
         ]  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
 
         # for versions below 10 `promote` is asynchronous so we need to wait
         # until it actually becomes writable
@@ -870,7 +865,7 @@ def pg_ctl(self, params):
             "-w"  # wait
         ] + params  # yapf: disable
 
-        return execute_utility(_params, self.utils_log_file)
+        return execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
 
     def free_port(self):
         """
@@ -1035,10 +1030,9 @@ def dump(self,
         # Generate tmpfile or tmpdir
         def tmpfile():
             if format == DumpFormat.Directory:
-                fname = mkdtemp(prefix=TMP_DUMP)
+                fname = self.os_ops.mkdtemp(prefix=TMP_DUMP)
             else:
-                fd, fname = mkstemp(prefix=TMP_DUMP)
-                os.close(fd)
+                fname = self.os_ops.mkstemp(prefix=TMP_DUMP)
             return fname
 
         # Set default arguments
@@ -1056,7 +1050,7 @@ def tmpfile():
             "-F", format.value
         ]  # yapf: disable
 
-        execute_utility(_params, self.utils_log_file)
+        execute_utility(_params, self.utils_log_file, os_ops=self.os_ops)
 
         return filename
 
@@ -1085,7 +1079,7 @@ def restore(self, filename, dbname=None, username=None):
 
         # try pg_restore if dump is binary formate, and psql if not
         try:
-            execute_utility(_params, self.utils_log_name)
+            execute_utility(_params, self.utils_log_name, os_ops=self.os_ops)
         except ExecUtilException:
             self.psql(filename=filename, dbname=dbname, username=username)
 
@@ -1417,7 +1411,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_utility(_params, self.utils_log_file, os_ops=self.os_ops)
 
     def connect(self,
                 dbname=None,

testgres/op_ops/local_ops.py renamed to testgres/os_ops/local_ops.py

@@ -14,30 +14,35 @@
 
 
 class LocalOperations(OsOperations):
-
     def __init__(self, username=None):
         super().__init__()
         self.username = username or self.get_user()
 
     # Command execution
     def exec_command(self, cmd, wait_exit=False, verbose=False, expect_error=False):
         if isinstance(cmd, list):
-            cmd = ' '.join(cmd)
+            cmd = " ".join(cmd)
         log.debug(f"os_ops.exec_command: `{cmd}`; remote={self.remote}")
         # Source global profile file + execute command
         try:
-            process = subprocess.run(cmd, shell=True, text=True,
-                                     stdout=subprocess.PIPE,
-                                     stderr=subprocess.PIPE,
-                                     timeout=CMD_TIMEOUT_SEC)
+            process = subprocess.run(
+                cmd,
+                shell=True,
+                text=True,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                timeout=CMD_TIMEOUT_SEC,
+            )
             exit_status = process.returncode
             result = process.stdout
             error = process.stderr
 
             if expect_error:
                 raise Exception(result, error)
-            if exit_status != 0 or 'error' in error.lower():
-                log.error(f"Problem in executing command: `{cmd}`\nerror: {error}\nexit_code: {exit_status}")
+            if exit_status != 0 or "error" in error.lower():
+                log.error(
+                    f"Problem in executing command: `{cmd}`\nerror: {error}\nexit_code: {exit_status}"
+                )
                 exit(1)
 
             if verbose:
@@ -55,7 +60,7 @@ def environ(self, var_name):
         return self.exec_command(cmd).strip()
 
     def find_executable(self, executable):
-        search_paths = self.environ('PATH')
+        search_paths = self.environ("PATH")
         if not search_paths:
             return None
 
@@ -74,12 +79,12 @@ def is_executable(self, file):
     def add_to_path(self, new_path):
         pathsep = self.pathsep
         # Check if the directory is already in PATH
-        path = self.environ('PATH')
+        path = self.environ("PATH")
         if new_path not in path.split(pathsep):
             if self.remote:
                 self.exec_command(f"export PATH={new_path}{pathsep}{path}")
             else:
-                os.environ['PATH'] = f"{new_path}{pathsep}{path}"
+                os.environ["PATH"] = f"{new_path}{pathsep}{path}"
         return pathsep
 
     def set_env(self, var_name, var_val):
@@ -112,17 +117,22 @@ def path_exists(self, path):
     @property
     def pathsep(self):
         os_name = self.get_name()
-        if os_name == 'posix':
-            pathsep = ':'
-        elif os_name == 'nt':
-            pathsep = ';'
+        if os_name == "posix":
+            pathsep = ":"
+        elif os_name == "nt":
+            pathsep = ";"
         else:
             raise Exception(f"Unsupported operating system: {os_name}")
         return pathsep
 
     def mkdtemp(self, prefix=None):
         return tempfile.mkdtemp(prefix=prefix)
 
+    def mkstemp(self, prefix=None):
+        fd, filename = tempfile.mkstemp(prefix=prefix)
+        os.close(fd)  # Close the file descriptor immediately after creating the file
+        return filename
+
     def copytree(self, src, dst):
         return shutil.copytree(src, dst)
 
@@ -140,11 +150,11 @@ def write(self, filename, data, truncate=False, binary=False, read_and_write=Fal
             read_and_write: If True, the file will be opened with read and write permissions ('r+' option);
                             if False (default), only write permission will be used ('w', 'a', 'wb', or 'ab' option)
         """
-        mode = 'wb' if binary else 'w'
+        mode = "wb" if binary else "w"
         if not truncate:
-            mode = 'a' + mode
+            mode = "a" + mode
         if read_and_write:
-            mode = 'r+' + mode
+            mode = "r+" + mode
 
         with open(filename, mode) as file:
             if isinstance(data, list):
@@ -161,11 +171,11 @@ def touch(self, filename):
         This method behaves as the 'touch' command in Unix. It's equivalent to calling 'touch filename' in the shell.
         """
         # cross-python touch(). It is vulnerable to races, but who cares?
-        with open(filename, 'a'):
+        with open(filename, "a"):
             os.utime(filename, None)
 
     def read(self, filename):
-        with open(filename, 'r') as file:
+        with open(filename, "r") as file:
             return file.read()
 
     def readlines(self, filename, num_lines=0, encoding=None):
@@ -176,14 +186,14 @@ def readlines(self, filename, num_lines=0, encoding=None):
         assert num_lines >= 0
 
         if num_lines == 0:
-            with open(filename, 'r', encoding=encoding) as file:
+            with open(filename, "r", encoding=encoding) as file:
                 return file.readlines()
 
         else:
             bufsize = 8192
             buffers = 1
 
-            with open(filename, 'r', encoding=encoding) as file:
+            with open(filename, "r", encoding=encoding) as file:
                 file.seek(0, os.SEEK_END)
                 end_pos = file.tell()
 
@@ -197,23 +207,25 @@ def readlines(self, filename, num_lines=0, encoding=None):
                     if cur_lines >= num_lines or pos == 0:
                         return lines[-num_lines:]
 
-                    buffers = int(buffers * max(2, int(num_lines / max(cur_lines, 1))))  # Adjust buffer size
+                    buffers = int(
+                        buffers * max(2, int(num_lines / max(cur_lines, 1)))
+                    )  # Adjust buffer size
 
     def isfile(self, remote_file):
         return os.path.isfile(remote_file)
 
     # Processes control
     def kill(self, pid, signal):
         # Kill the process
-        cmd = f'kill -{signal} {pid}'
+        cmd = f"kill -{signal} {pid}"
         return self.exec_command(cmd)
 
     def get_pid(self):
         # Get current process id
         return os.getpid()
 
     # Database control
-    def db_connect(self, dbname, user, password=None, host='localhost', port=5432):
+    def db_connect(self, dbname, user, password=None, host="localhost", port=5432):
         conn = pglib.connect(
             host=host,
             port=port,