From a82f5e2588e4a4a9fbadd8fde084b02981767f76 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Fri, 18 Sep 2026 23:44:32 +0300 Subject: [PATCH 1/3] Revision&Refactoring --- src/exceptions.py | 4 +- src/node.py | 421 ++++++++++++++++++++++------------ src/node_app.py | 46 ++-- src/pubsub.py | 18 +- src/raise_error.py | 4 +- src/utils.py | 24 +- tests/test_os_ops_common.py | 81 +++++-- tests/test_os_ops_local.py | 15 +- tests/test_testgres_common.py | 2 +- 9 files changed, 396 insertions(+), 219 deletions(-) diff --git a/src/exceptions.py b/src/exceptions.py index 42153566..13de787a 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -113,7 +113,7 @@ class QueryTimeoutException(QueryException): def __init__( self, message: typing.Optional[str] = None, - query: typing.Optional[str] = None + query: typing.Optional[str] = None, ): assert message is None or type(message) is str assert query is None or type(query) is str @@ -170,7 +170,7 @@ class StartNodeException(TestgresException): def __init__( self, message: typing.Optional[str] = None, - files: typing.Optional[typing.Iterable] = None + files: typing.Optional[typing.Iterable] = None, ): assert message is None or type(message) is str assert files is None or isinstance(files, typing.Iterable) diff --git a/src/node.py b/src/node.py index c8a849c2..d75c09cf 100644 --- a/src/node.py +++ b/src/node.py @@ -88,13 +88,6 @@ from . import utils -from .utils import \ - PgVer, \ - eprint, \ - get_pg_version2, \ - options_string, \ - clean_on_error - from .raise_error import RaiseError from .backup import NodeBackup @@ -220,7 +213,13 @@ def __init__( assert type(self._bin_dir) is str - self._pg_version = PgVer(get_pg_version2(self._os_ops, self._bin_dir)) + raw_version = utils.get_pg_version2( + self._os_ops, + self._bin_dir, + ) + self._pg_version = utils.PgVer( + raw_version, + ) self._base_dir = base_dir self._prefix = prefix self._logger = None @@ -655,7 +654,7 @@ def _try_shutdown_internal(self, max_attempts, with_force): except ExecUtilException: continue # one more time except Exception: - eprint('cannot stop node {}'.format(self.name)) + utils.eprint('cannot stop node {}'.format(self.name)) break return # OK @@ -691,7 +690,10 @@ def _try_shutdown_internal(self, max_attempts, with_force): ps_command) try: - eprint('Force stopping node {0} with PID {1}'.format(self.name, node_pid)) + utils.eprint('Force stopping node {0} with PID {1}'.format( + self.name, + node_pid, + )) self._os_ops.kill(node_pid, signal.SIGKILL) except Exception: # The node has already stopped @@ -706,11 +708,15 @@ def _try_shutdown_internal(self, max_attempts, with_force): assert type(ps_output) is str if ps_output == "": - eprint('Node {0} has been stopped successfully.'.format(self.name)) + utils.eprint('Node {0} has been stopped successfully.'.format( + self.name, + )) return if ps_output == str(node_pid): - eprint('Failed to stop node {0}.'.format(self.name)) + utils.eprint('Failed to stop node {0}.'.format( + self.name, + )) return __class__._throw_bugcheck__unexpected_result_of_ps( @@ -755,11 +761,11 @@ def _create_recovery_conf(self, username, slot=None): except ValueError: conninfo["host"] = master.host - line = ( - "primary_conninfo='{}'\n" - ).format(options_string(**conninfo)) # yapf: disable + line = "primary_conninfo='{}'\n".format( + utils.options_string(**conninfo), + ) # Since 12 recovery.conf had disappeared - if self.version >= PgVer('12'): + if self.version >= utils.PgVer('12'): assert self._os_ops is not None assert isinstance(self._os_ops, OsOperations) @@ -793,7 +799,7 @@ def _create_recovery_conf(self, username, slot=None): line += "primary_slot_name={}\n".format(slot) - if self.version >= PgVer('12'): + if self.version >= utils.PgVer('12'): self.append_conf(line=line) else: self.append_conf(filename=RECOVERY_CONF_FILE, line=line) @@ -898,45 +904,62 @@ def default_conf(self, # hba file is updated self._default_conf__hba() - postgres_conf = self._os_ops.build_path(self.data_dir, PG_CONF_FILE) + postgres_conf = self._os_ops.build_path( + self.data_dir, + PG_CONF_FILE, + ) # overwrite config file self._os_ops.write(postgres_conf, '', truncate=True) - self.append_conf(fsync=fsync, - max_worker_processes=MAX_WORKER_PROCESSES, - log_statement=log_statement, - listen_addresses=self._host, - port=self.port) # yapf:disable + self.append_conf( + fsync=fsync, + max_worker_processes=MAX_WORKER_PROCESSES, + log_statement=log_statement, + listen_addresses=self._host, + port=self.port, + ) # common replication settings if allow_streaming or allow_logical: - self.append_conf(max_replication_slots=MAX_REPLICATION_SLOTS, - max_wal_senders=MAX_WAL_SENDERS) # yapf: disable + self.append_conf( + max_replication_slots=MAX_REPLICATION_SLOTS, + max_wal_senders=MAX_WAL_SENDERS, + ) # binary replication if allow_streaming: # select a proper wal_level for PostgreSQL - wal_level = 'replica' if self._pg_version >= PgVer('9.6') else 'hot_standby' + if self._pg_version >= utils.PgVer('9.6'): + wal_level = 'replica' + else: + wal_level = 'hot_standby' - if self._pg_version < PgVer('13'): - self.append_conf(hot_standby=True, - wal_keep_segments=WAL_KEEP_SEGMENTS, - wal_level=wal_level) # yapf: disable + if self._pg_version < utils.PgVer('13'): + self.append_conf( + hot_standby=True, + wal_keep_segments=WAL_KEEP_SEGMENTS, + wal_level=wal_level, + ) else: - self.append_conf(hot_standby=True, - wal_keep_size=WAL_KEEP_SIZE, - wal_level=wal_level) # yapf: disable + self.append_conf( + hot_standby=True, + wal_keep_size=WAL_KEEP_SIZE, + wal_level=wal_level, + ) # logical replication if allow_logical: - if self._pg_version < PgVer('10'): - raise InitNodeException("Logical replication is only " - "available on PostgreSQL 10 and newer") + if self._pg_version < utils.PgVer('10'): + raise InitNodeException( + "Logical replication is only " + "available on PostgreSQL 10 and newer", + ) self.append_conf( max_logical_replication_workers=MAX_LOGICAL_REPLICATION_WORKERS, - wal_level='logical') + wal_level='logical', + ) # disable UNIX sockets if asked to if not unix_sockets: @@ -1089,7 +1112,7 @@ def _get_node_state(self) -> utils.PostgresNodeState: self._os_ops, self.bin_dir, self.data_dir, - self.utils_log_file + self.utils_log_file, ) def get_control_data(self): @@ -1099,7 +1122,7 @@ def get_control_data(self): # this one is tricky (blame PG 9.4) _params = [self._get_bin_path("pg_controldata")] - _params += ["-D"] if self._pg_version >= PgVer('9.5') else [] + _params += ["-D"] if self._pg_version >= utils.PgVer('9.5') else [] _params += [self.data_dir] exec_r = utils.execute_utility3( @@ -1154,7 +1177,7 @@ def slow_start( InternalError, QueryException, ProgrammingError, - OperationalError + OperationalError, } self.poll_query_until( @@ -1303,7 +1326,10 @@ def LOCAL__raise_cannot_start_node__std(from_exception): assert nAttempt > 0 assert nAttempt <= __class__._C_MAX_START_ATEMPTS if nAttempt == __class__._C_MAX_START_ATEMPTS: - self._raise_cannot_start_node(e, "Cannot start node after multiple attempts.") + self._raise_cannot_start_node( + e, + "Cannot start node after multiple attempts.", + ) is_it_port_conflict = PostgresNodeUtils.detect_port_conflict(log_reader) @@ -1311,7 +1337,10 @@ def LOCAL__raise_cannot_start_node__std(from_exception): LOCAL__raise_cannot_start_node__std(e) logging.warning( - "Detected a conflict with using the port {0}. Trying another port after a {1}-second sleep...".format(self._port, timeout) + "Detected a conflict with using the port {}. Trying another port after a {}-second sleep...".format( + self._port, + timeout, + ), ) time.sleep(timeout) timeout = min(2 * timeout, 5) @@ -1333,14 +1362,18 @@ def LOCAL__raise_cannot_start_node__std(from_exception): def _raise_cannot_start_node( self, from_exception: typing.Optional[Exception], - msg: str + msg: str, ): assert from_exception is None or isinstance(from_exception, Exception) assert type(msg) is str files = self._collect_special_files() raise_from(StartNodeException(msg, files), from_exception) - def stop(self, params=[], wait=True): + def stop( + self, + params=[], + wait=True, + ): """ Stops the PostgreSQL node using pg_ctl if the node has been started. @@ -1355,7 +1388,7 @@ def stop(self, params=[], wait=True): self._get_bin_path("pg_ctl"), "-D", self.data_dir, "-w" if wait else '-W', # --wait or --no-wait - "stop" + "stop", ] + params # yapf: disable try: @@ -1420,7 +1453,7 @@ def restart(self, params=[]): "-D", self.data_dir, "-l", self.pg_log_file, "-w", # wait - "restart" + "restart", ] + params # yapf: disable try: @@ -1457,7 +1490,7 @@ def reload(self, params=[]): _params = [ self._get_bin_path("pg_ctl"), "-D", self.data_dir, - "reload" + "reload", ] + params # yapf: disable utils.execute_utility3( @@ -1494,14 +1527,16 @@ def promote(self, dbname=None, username=None): # for versions below 10 `promote` is asynchronous so we need to wait # until it actually becomes writable - if self._pg_version < PgVer('10'): + if self._pg_version < utils.PgVer('10'): check_query = "SELECT pg_is_in_recovery()" - self.poll_query_until(query=check_query, - expected=False, - dbname=dbname, - username=username, - max_attempts=0) # infinite + self.poll_query_until( + query=check_query, + expected=False, + dbname=dbname, + username=username, + max_attempts=0, + ) # infinite # node becomes master itself self._master = None @@ -1522,7 +1557,7 @@ def pg_ctl(self, params): _params = [ self._get_bin_path("pg_ctl"), "-D", self.data_dir, - "-w" # wait + "-w", # wait ] + params # yapf: disable return utils.execute_utility3( @@ -1544,7 +1579,12 @@ def free_port(self): """ return self._free_port() - def cleanup(self, max_attempts=3, full=False, release_resources=False): + def cleanup( + self, + max_attempts=3, + full=False, + release_resources=False, + ): """ Stop node if needed and remove its data/logs directory. NOTE: take a look at TestgresConfig.node_cleanup_full. @@ -1573,15 +1613,17 @@ def cleanup(self, max_attempts=3, full=False, release_resources=False): return self @method_decorator(positional_args_hack(['dbname', 'query'])) - def psql(self, - query=None, - filename=None, - dbname=None, - username=None, - input=None, - host: typing.Optional[str] = None, - port: typing.Optional[int] = None, - **variables): + def psql( + self, + query=None, + filename=None, + dbname=None, + username=None, + input=None, + host: typing.Optional[str] = None, + port: typing.Optional[int] = None, + **variables, + ): """ Execute a query using psql. @@ -1617,22 +1659,22 @@ def psql(self, input=input, host=host, port=port, - **variables + **variables, ) assert type(r) is OsCommandResult return r.returncode, r.stdout, r.stderr def _psql( - self, - ignore_errors, - query=None, - filename=None, - dbname=None, - username=None, - input=None, - host: typing.Optional[str] = None, - port: typing.Optional[int] = None, - **variables + self, + ignore_errors, + query=None, + filename=None, + dbname=None, + username=None, + input=None, + host: typing.Optional[str] = None, + port: typing.Optional[int] = None, + **variables, ) -> OsCommandResult: assert host is None or type(host) is str assert port is None or type(port) is int @@ -1694,7 +1736,12 @@ def _psql( return r @method_decorator(positional_args_hack(['dbname', 'query'])) - def safe_psql(self, query=None, expect_error=False, **kwargs): + def safe_psql( + self, + query=None, + expect_error=False, + **kwargs, + ): """ Execute a query using psql. @@ -1736,12 +1783,14 @@ def safe_psql(self, query=None, expect_error=False, **kwargs): return exec_r.stdout - def dump(self, - filename=None, - dbname=None, - username=None, - format=DumpFormat.Plain, - options=None): + def dump( + self, + filename=None, + dbname=None, + username=None, + format=DumpFormat.Plain, + options=None, + ): """ Dump database into a file using pg_dump. NOTE: the file is not removed automatically. @@ -1797,7 +1846,12 @@ def tmpfile(): return filename - def restore(self, filename, dbname=None, username=None): + def restore( + self, + filename, + dbname=None, + username=None, + ): """ Restore database from pg_dump's file. @@ -1817,7 +1871,7 @@ def restore(self, filename, dbname=None, username=None): "-h", self._host, "-U", username, "-d", dbname, - filename + filename, ] # yapf: disable # try pg_restore if dump is binary format, and psql if not @@ -1873,10 +1927,12 @@ def poll_query_until( attempts = 0 while max_attempts == 0 or attempts < max_attempts: try: - res = self.execute(dbname=dbname, - query=query, - username=username, - commit=commit) + res = self.execute( + dbname=dbname, + query=query, + username=username, + commit=commit, + ) if expected is None and res is None: return # done @@ -1904,12 +1960,14 @@ def poll_query_until( raise QueryTimeoutException('Query timeout', query) @method_decorator(positional_args_hack(['dbname', 'query'])) - def execute(self, - query, - dbname=None, - username=None, - password=None, - commit=True): + def execute( + self, + query, + dbname=None, + username=None, + password=None, + commit=True, + ): """ Execute a query and return all rows as list. @@ -1924,16 +1982,21 @@ def execute(self, A list of tuples representing rows. """ - with self.connect(dbname=dbname, - username=username, - password=password, - autocommit=commit) as node_con: # yapf: disable + node_con = self.connect( + dbname=dbname, + username=username, + password=password, + autocommit=commit, + ) + with node_con: # yapf: disable res = node_con.execute(query) - return res - def backup(self, **kwargs): + def backup( + self, + **kwargs, + ) -> NodeBackup: """ Perform pg_basebackup. @@ -1946,9 +2009,17 @@ def backup(self, **kwargs): A smart object of type NodeBackup. """ - return NodeBackup(node=self, **kwargs) + return NodeBackup( + node=self, + **kwargs, + ) - def replicate(self, name=None, slot=None, **kwargs): + def replicate( + self, + name=None, + slot=None, + **kwargs, + ): """ Create a binary replica of this node. @@ -1961,8 +2032,12 @@ def replicate(self, name=None, slot=None, **kwargs): """ # transform backup into a replica - with clean_on_error(self.backup(**kwargs)) as backup: - return backup.spawn_replica(name=name, destroy=True, slot=slot) + with utils.clean_on_error(self.backup(**kwargs)) as backup: + return backup.spawn_replica( + name=name, + destroy=True, + slot=slot, + ) def set_synchronous_standbys(self, standbys): """ @@ -1990,7 +2065,7 @@ def set_synchronous_standbys(self, standbys): master.restart() """ - if self._pg_version >= PgVer('9.6'): + if self._pg_version >= utils.PgVer('9.6'): if isinstance(standbys, Iterable): standbys = First(1, standbys) else: @@ -1998,8 +2073,10 @@ def set_synchronous_standbys(self, standbys): standbys = u", ".join(u"\"{}\"".format(r.name) for r in standbys) else: - raise TestgresException("Feature isn't supported in " - "Postgres 9.5 and below") + raise TestgresException( + "Feature isn't supported in " + "Postgres 9.5 and below", + ) self.append_conf("synchronous_standby_names = '{}'".format(standbys)) @@ -2011,7 +2088,7 @@ def catchup(self, dbname=None, username=None): if not self.master: raise TestgresException("Node doesn't have a master") - if self._pg_version >= PgVer('10'): + if self._pg_version >= utils.PgVer('10'): poll_lsn = "select pg_catalog.pg_current_wal_lsn()::text" wait_lsn = "select pg_catalog.pg_last_wal_replay_lsn() >= '{}'::pg_lsn" else: @@ -2020,15 +2097,19 @@ def catchup(self, dbname=None, username=None): try: # fetch latest LSN - lsn = self.master.execute(query=poll_lsn, - dbname=dbname, - username=username)[0][0] # yapf: disable + lsn = self.master.execute( + query=poll_lsn, + dbname=dbname, + username=username, + )[0][0] # wait until this LSN reaches replica - self.poll_query_until(query=wait_lsn.format(lsn), - dbname=dbname, - username=username, - max_attempts=0) # infinite + self.poll_query_until( + query=wait_lsn.format(lsn), + dbname=dbname, + username=username, + max_attempts=0, + ) except Exception as e: raise_from(CatchUpException("Failed to catch up."), e) @@ -2044,12 +2125,14 @@ def publish(self, name, **kwargs): """ return Publication(name=name, node=self, **kwargs) - def subscribe(self, - publication, - name, - dbname=None, - username=None, - **params): + def subscribe( + self, + publication, + name, + dbname=None, + username=None, + **params, + ): """ Create subscription for logical replication @@ -2063,8 +2146,14 @@ def subscribe(self, for details) """ # yapf: disable - return Subscription(name=name, node=self, publication=publication, - dbname=dbname, username=username, **params) + return Subscription( + name=name, + node=self, + publication=publication, + dbname=dbname, + username=username, + **params, + ) # yapf: enable def pgbench( @@ -2113,12 +2202,14 @@ def pgbench( assert isinstance(proc, OsProcessController) return proc - def pgbench_with_wait(self, - dbname=None, - username=None, - stdout=None, - stderr=None, - options=None): + def pgbench_with_wait( + self, + dbname=None, + username=None, + stdout=None, + stderr=None, + options=None, + ): """ Do pgbench command and wait. @@ -2136,7 +2227,10 @@ def pgbench_with_wait(self, pgbench.wait() return - def pgbench_init(self, **kwargs): + def pgbench_init( + self, + **kwargs, + ): """ Small wrapper for pgbench_run(). Sets initialize=True. @@ -2145,11 +2239,20 @@ def pgbench_init(self, **kwargs): This instance of :class:`.PostgresNode`. """ - self.pgbench_run(initialize=True, **kwargs) + self.pgbench_run( + initialize=True, + **kwargs, + ) return self - def pgbench_run(self, dbname=None, username=None, options=[], **kwargs): + def pgbench_run( + self, + dbname=None, + username=None, + options=[], + **kwargs, + ): """ Run pgbench with some options. This event is logged (see self.utils_log_file). @@ -2199,11 +2302,13 @@ def pgbench_run(self, dbname=None, username=None, options=[], **kwargs): self.utils_log_file, ).stdout - def connect(self, - dbname=None, - username=None, - password=None, - autocommit=False): + def connect( + self, + dbname=None, + username=None, + password=None, + autocommit=False, + ): """ Connect to a database. @@ -2219,16 +2324,18 @@ def connect(self, An instance of :class:`.NodeConnection`. """ - return NodeConnection(node=self, - dbname=dbname, - username=username, - password=password, - autocommit=autocommit) # yapf: disable + return NodeConnection( + node=self, + dbname=dbname, + username=username, + password=password, + autocommit=autocommit, + ) def table_checksum( self, table: str, - dbname: str = "postgres" + dbname: str = "postgres", ) -> int: assert type(table) is str assert type(dbname) is str @@ -2250,13 +2357,13 @@ def table_checksum( 'pgbench_branches', 'pgbench_tellers', 'pgbench_accounts', - 'pgbench_history' + 'pgbench_history', ] def pgbench_table_checksums( self, dbname: str = "postgres", - pgbench_tables: typing.Iterable[str] = sm_pgbench_tables + pgbench_tables: typing.Iterable[str] = sm_pgbench_tables, ) -> typing.Set[typing.Tuple[str, int]]: assert type(dbname) is str @@ -2267,7 +2374,12 @@ def pgbench_table_checksums( assert type(r2) is set return r2 - def set_auto_conf(self, options, config='postgresql.auto.conf', rm_options={}): + def set_auto_conf( + self, + options, + config='postgresql.auto.conf', + rm_options={}, + ): """ Update or remove configuration options in the specified configuration file, updates the options specified in the options dictionary, removes any options @@ -2337,7 +2449,12 @@ def set_auto_conf(self, options, config='postgresql.auto.conf', rm_options={}): self._os_ops.write(path, auto_conf, truncate=True) - def upgrade_from(self, old_node, options=None, expect_error=False): + def upgrade_from( + self, + old_node, + options=None, + expect_error=False, + ): """ Upgrade this node from an old node using pg_upgrade. @@ -2366,7 +2483,7 @@ def upgrade_from(self, old_node, options=None, expect_error=False): "--old-datadir", old_node.data_dir, "--new-datadir", self.data_dir, "--old-port", str(old_node.port), - "--new-port", str(self.port) + "--new-port", str(self.port), ] upgrade_command += options @@ -2539,7 +2656,7 @@ def __init__( self, file_name: str, position: int, - data: str + data: str, ): assert type(file_name) is str assert type(position) is int @@ -2728,7 +2845,9 @@ def _create_log_info( class PostgresNodeUtils: @staticmethod - def detect_port_conflict(log_reader: PostgresNodeLogReader) -> bool: + def detect_port_conflict( + log_reader: PostgresNodeLogReader, + ) -> bool: assert type(log_reader) is PostgresNodeLogReader blocks = log_reader.read() @@ -2740,4 +2859,6 @@ def detect_port_conflict(log_reader: PostgresNodeLogReader) -> bool: if 'Is another postmaster already running on port' in block.data: return True + continue + return False diff --git a/src/node_app.py b/src/node_app.py index 6053bf55..a6dfd602 100644 --- a/src/node_app.py +++ b/src/node_app.py @@ -67,10 +67,10 @@ def nodes_to_cleanup(self) -> typing.List[PostgresNode]: return self._nodes_to_cleanup def make_empty( - self, - base_dir: str, - port: typing.Optional[int] = None, - bin_dir: typing.Optional[str] = None + self, + base_dir: str, + port: typing.Optional[int] = None, + bin_dir: typing.Optional[str] = None ) -> PostgresNode: assert type(base_dir) is str assert port is None or type(port) is int @@ -112,15 +112,15 @@ def make_empty( return node def make_simple( - self, - base_dir: str, - port: typing.Optional[int] = None, - set_replication: bool = False, - ptrack_enable: bool = False, - initdb_params: typing.Optional[T_LIST_STR] = None, - pg_options: typing.Optional[T_DICT_STR_STR] = None, - checksum: bool = True, - bin_dir: typing.Optional[str] = None + self, + base_dir: str, + port: typing.Optional[int] = None, + set_replication: bool = False, + ptrack_enable: bool = False, + initdb_params: typing.Optional[T_LIST_STR] = None, + pg_options: typing.Optional[T_DICT_STR_STR] = None, + checksum: bool = True, + bin_dir: typing.Optional[str] = None, ) -> PostgresNode: assert type(base_dir) is str assert port is None or type(port) is int @@ -134,7 +134,7 @@ def make_simple( node = self.make_empty( base_dir, port, - bin_dir=bin_dir + bin_dir=bin_dir, ) final_initdb_params = initdb_params @@ -143,7 +143,7 @@ def make_simple( final_initdb_params = __class__._paramlist_append_if_not_exist( initdb_params, final_initdb_params, - '--data-checksums' + '--data-checksums', ) assert final_initdb_params is not None assert '--data-checksums' in final_initdb_params @@ -209,14 +209,18 @@ def make_simple( # https://github.com/postgrespro/testgres/issues/54 # for PG >= 13 remove 'wal_keep_segments' parameter if node.major_version >= 13: - node.set_auto_conf({}, 'postgresql.conf', ['wal_keep_segments']) + node.set_auto_conf( + {}, + 'postgresql.conf', + ['wal_keep_segments'], + ) return node @staticmethod def _paramlist_has_param( params: typing.Optional[T_LIST_STR], - param: str + param: str, ) -> bool: assert type(param) is str @@ -297,13 +301,17 @@ def _gettempdir(self) -> str: # Paranoid checks # if type(v) is str: - __class__._raise_bugcheck("os_ops.get_tempdir returned a value with type {0}.".format(type(v).__name__)) + __class__._raise_bugcheck("os_ops.get_tempdir returned a value with type {}.".format( + type(v).__name__, + )) if v == "": __class__._raise_bugcheck("os_ops.get_tempdir returned an empty string.") if not self._os_ops.path_exists(v): - __class__._raise_bugcheck("os_ops.get_tempdir returned a not exist path [{0}].".format(v)) + __class__._raise_bugcheck("os_ops.get_tempdir returned a not exist path [{}].".format( + v, + )) # OK return v diff --git a/src/pubsub.py b/src/pubsub.py index cbc55c9b..c8a446ef 100644 --- a/src/pubsub.py +++ b/src/pubsub.py @@ -135,13 +135,15 @@ def add_tables(self, tables, dbname=None, username=None): class Subscription(object): - def __init__(self, - node, - publication, - name=None, - dbname=None, - username=None, - **params): + def __init__( + self, + node, + publication, + name=None, + dbname=None, + username=None, + **params, + ): """ Constructor. Use :meth:`.PostgresNode.subscribe()` instead of direct constructing subscription objects. @@ -174,7 +176,7 @@ def __init__(self, "dbname": self.pub.dbname, "user": self.pub.username, "host": self.pub.node.host, - "port": self.pub.node.port + "port": self.pub.node.port, } query = ( diff --git a/src/raise_error.py b/src/raise_error.py index e30c9315..aa6a910b 100644 --- a/src/raise_error.py +++ b/src/raise_error.py @@ -108,10 +108,10 @@ def _map_node_status_to_reason( if node_status == NodeStatus.Running: return "Node is running (pid: {})".format( - node_pid + node_pid, ) # assert False return "Node has unknown status {}".format( - node_status + node_status, ) diff --git a/src/utils.py b/src/utils.py index 5fa75e1f..9c7d91c2 100644 --- a/src/utils.py +++ b/src/utils.py @@ -312,7 +312,11 @@ def cache_pg_config_data(cmd): pg_config_data = cache_pg_config_data("pg_config") except Exception: raise InvalidOperationException( - "Failed to determine how to start pg_config. Either specify the path to pg_config in PG_CONFIG or specify the path to the Postgres directory containing pg_config in PG_BIN, or put pg_config into the system PATH.") + "Failed to determine how to start pg_config. " + "Either specify the path to pg_config in PG_CONFIG or " + "specify the path to the Postgres directory containing " + "pg_config in PG_BIN, or put pg_config into the system PATH.", + ) return pg_config_data @@ -422,7 +426,7 @@ class PostgresNodeState: def __init__( self, node_status: NodeStatus, - pid: typing.Optional[int] + pid: typing.Optional[int], ): assert type(node_status) is NodeStatus assert pid is None or type(pid) is int @@ -485,7 +489,7 @@ def get(self) -> T_PLATFORM_UTILS: if attempt > 1: internal_utils.send_log_debug("Sleep {} second(s) before an attempt #{}".format( sleep_time, - attempt + attempt, )) time.sleep(sleep_time) sleep_time = sleep_time * C_SLEEP_TIME_MULT @@ -527,7 +531,7 @@ def get(self) -> T_PLATFORM_UTILS: if i == -1: RaiseError.pg_ctl_returns_an_unexpected_string( out, - _params + _params, ) assert i > 0 @@ -542,7 +546,7 @@ def get(self) -> T_PLATFORM_UTILS: if i == len(out): RaiseError.pg_ctl_returns_an_unexpected_string( out, - _params + _params, ) ch = out[i] @@ -556,14 +560,14 @@ def get(self) -> T_PLATFORM_UTILS: RaiseError.pg_ctl_returns_an_unexpected_string( out, - _params + _params, ) assert False if i == start_pid_s: RaiseError.pg_ctl_returns_an_unexpected_string( out, - _params + _params, ) # TODO: Let's verify a length of pid string. @@ -573,7 +577,7 @@ def get(self) -> T_PLATFORM_UTILS: if pid == 0: RaiseError.pg_ctl_returns_a_zero_pid( out, - _params + _params, ) assert pid != 0 @@ -631,14 +635,14 @@ def get(self) -> T_PLATFORM_UTILS: # Postmaster is alive. Let's wait a few seconds and check its status again. internal_utils.send_log_debug( "Postmaster is found and has PID {}.".format( - find_postmaster_r.pid + find_postmaster_r.pid, )) if attempt < C_MAX_ATTEMPTS: continue errMsg = "Getting of a node status [data_dir is {0}] failed.".format( - data_dir + data_dir, ) raise ExecUtilException( diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 68a13862..67f94146 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -1351,8 +1351,9 @@ def test_read_binary__spec__negative_size( filename = os_ops.mkstemp(name_with_surprize.value) with pytest.raises( - ValueError, - match=re.escape("Negative 'size' is not supported.")): + ValueError, + match=re.escape("Negative 'size' is not supported."), + ): os_ops.read_binary(filename, 0, size=-1) os_ops.remove_file(filename) @@ -1998,17 +1999,23 @@ def test_mkdir__mt(self, data001: tagData_OS_OPS__NUMS): LocalCheck.check_path_exists(os_ops, lock_dir) assert os_ops.path_exists(lock_dir) is True - def MAKE_PATH(os_ops: OsOperations, lock_dir: str, num: int) -> str: + def MAKE_PATH( + os_ops: OsOperations, + lock_dir: str, + num: int, + ) -> str: assert isinstance(os_ops, OsOperations) assert type(lock_dir) is str assert type(num) is int return os_ops.build_path(lock_dir, str(num) + ".lock") - def LOCAL_WORKER(os_ops: OsOperations, - workerID: int, - lock_dir: str, - cNumbers: int, - reservedNumbers: typing.Set[int]) -> None: + def LOCAL_WORKER( + os_ops: OsOperations, + workerID: int, + lock_dir: str, + cNumbers: int, + reservedNumbers: typing.Set[int], + ) -> None: assert isinstance(os_ops, OsOperations) assert type(workerID) is int assert type(lock_dir) is str @@ -2064,7 +2071,7 @@ def LOG_INFO(template: str, *args) -> None: threadPool = ThreadPoolExecutor( max_workers=N_WORKERS, - thread_name_prefix="ex_creator" + thread_name_prefix="ex_creator", ) class tadWorkerData: @@ -2089,7 +2096,7 @@ class tadWorkerData: n, lock_dir, N_NUMBERS, - workerDatas[n].reservedNumbers + workerDatas[n].reservedNumbers, ) assert workerDatas[n].future is not None @@ -2152,7 +2159,7 @@ class tadWorkerData: nErrors += 1 logging.error("Number {} was already reserved by worker #{}".format( n, - reservedNumbers[n] + reservedNumbers[n], )) else: reservedNumbers[n] = i @@ -2198,7 +2205,7 @@ class tadWorkerData: logging.error("Cannot delete directory [{}]. Error ({}): {}".format( file_path, type(e).__name__, - str(e) + str(e), )) continue @@ -2215,7 +2222,7 @@ class tadWorkerData: logging.error("Cannot delete directory [{}]. Error ({}): {}".format( lock_dir, type(e).__name__, - str(e) + str(e), )) logging.info("Test is finished! Total error count is {}.".format(nErrors)) @@ -2278,7 +2285,7 @@ def test_kill( "python3", "-u", "-c", - "import os, time; print(os.getpid());time.sleep(300);print('EXIT')" + "import os, time; print(os.getpid());time.sleep(300);print('EXIT')", ] logging.info("Local test process is creating ...") @@ -2532,7 +2539,10 @@ def test_is_abs_path__yes( assert actual_value is True return - def test_is_abs_path__no(self, os_ops_descr: OsOpsDescr): + def test_is_abs_path__no( + self, + os_ops_descr: OsOpsDescr, + ): assert type(os_ops_descr) is OsOpsDescr assert isinstance(os_ops_descr.os_ops, OsOperations) @@ -3085,7 +3095,10 @@ class tagReadLinesData_TXT: for x in sm_ReadLinesData_TXT ] ) - def readlines_data_txt(self, request: pytest.FixtureRequest) -> tagReadLinesData_TXT: + def readlines_data_txt( + self, + request: pytest.FixtureRequest, + ) -> tagReadLinesData_TXT: assert isinstance(request, pytest.FixtureRequest) return request.param @@ -3143,7 +3156,7 @@ def test_readlines__BIN( def test_prove_environment_isolation( self, - os_ops_descr: OsOpsDescr + os_ops_descr: OsOpsDescr, ): # # Author: Marg G. (mark@google.com) @@ -4540,7 +4553,7 @@ def test_popen_returncode_active( controller = os_ops.popen( fx_data_wait_timeout.cmd, - shell=type(fx_data_wait_timeout.cmd) is str + shell=type(fx_data_wait_timeout.cmd) is str, ) assert isinstance(controller, OsProcessController) @@ -4855,7 +4868,7 @@ def test_popen_garbage_collection( local_p = controller._local_process else: raise RuntimeError("Unknown controller type: {}.".format( - type(controller).__name__ + type(controller).__name__, )) assert local_p is not None @@ -5150,6 +5163,8 @@ def test_popen_cwd( s = controller.stderr.read() assert s == "" + return + def test_popen_communicate_timeout( self, os_ops_descr: OsOpsDescr, @@ -6313,11 +6328,20 @@ def test_run_check_exception2__list(self, os_ops_descr: OsOpsDescr): RunConditions.skip_if_windows() os_ops = os_ops_descr.os_ops - cmd = ["sh", "-c", "echo normal_out && echo error_err >&2 && exit 1"] + cmd = [ + "sh", + "-c", + "echo normal_out && echo error_err >&2 && exit 1", + ] # 1. Check default behavior (check=True) with pytest.raises(expected_exception=ExecUtilException) as x: - os_ops.run(cmd, text=True, encoding="utf-8", check=True) + os_ops.run( + cmd, + text=True, + encoding="utf-8", + check=True, + ) assert x.type is ExecUtilException assert type(x.value.out) is str @@ -6343,7 +6367,12 @@ def test_run_check_exception2__list(self, os_ops_descr: OsOpsDescr): ) # 2. Test the negative scenario with validation disabled (check=False) - result = os_ops.run(cmd, text=True, encoding="utf-8", check=False) + result = os_ops.run( + cmd, + text=True, + encoding="utf-8", + check=False, + ) assert isinstance(result, OsCommandResult) assert result.returncode == 1 @@ -6364,7 +6393,13 @@ def test_run_check_exception3__str(self, os_ops_descr: OsOpsDescr): # 1. Check default behavior (check=True) with pytest.raises(expected_exception=ExecUtilException) as x: - os_ops.run(cmd, text=True, encoding="utf-8", shell=True, check=True) + os_ops.run( + cmd, + text=True, + encoding="utf-8", + shell=True, + check=True, + ) assert x.type is ExecUtilException assert type(x.value.out) is str diff --git a/tests/test_os_ops_local.py b/tests/test_os_ops_local.py index 1909824a..4fafe56e 100644 --- a/tests/test_os_ops_local.py +++ b/tests/test_os_ops_local.py @@ -28,7 +28,10 @@ def test_read__unknown_file( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - with pytest.raises(FileNotFoundError, match=re.escape("[Errno 2] No such file or directory: '/dummy'")): + with pytest.raises( + FileNotFoundError, + match=re.escape("[Errno 2] No such file or directory: '/dummy'"), + ): os_ops.read("/dummy") return @@ -46,8 +49,9 @@ def test_read_binary__spec__unk_file( assert isinstance(os_ops, OsOperations) with pytest.raises( - FileNotFoundError, - match=re.escape("[Errno 2] No such file or directory: '/dummy'")): + FileNotFoundError, + match=re.escape("[Errno 2] No such file or directory: '/dummy'"), + ): os_ops.read_binary("/dummy", 0) return @@ -64,7 +68,10 @@ def test_get_file_size__unk_file( os_ops = os_ops_descr.os_ops assert isinstance(os_ops, OsOperations) - with pytest.raises(FileNotFoundError, match=re.escape("[Errno 2] No such file or directory: '/dummy'")): + with pytest.raises( + FileNotFoundError, + match=re.escape("[Errno 2] No such file or directory: '/dummy'"), + ): os_ops.get_file_size("/dummy") return diff --git a/tests/test_testgres_common.py b/tests/test_testgres_common.py index 2f055048..6edc5f18 100644 --- a/tests/test_testgres_common.py +++ b/tests/test_testgres_common.py @@ -9,12 +9,12 @@ from .helpers.pg_cfg_os_ops import PgCfgOsOps from src import __version__ as testgres_version -from src.node import PgVer from src.node import PostgresNode from src.node import NodeConnection from src.node import PostgresNodeLogReader from src.node import PostgresNodeUtils from src.node import ProcessProxy +from src.utils import PgVer from src.utils import get_pg_version2 from src.utils import file_tail from src.utils import get_bin_path2 From 3f27d89e1fd537f23565b649008be34f0a2c7a56 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Sat, 19 Sep 2026 15:40:35 +0300 Subject: [PATCH 2/3] node.py: imports are reordered --- src/node.py | 51 ++++++++++++++++++++++++++------------------------- 1 file changed, 26 insertions(+), 25 deletions(-) diff --git a/src/node.py b/src/node.py index d75c09cf..ded039ec 100644 --- a/src/node.py +++ b/src/node.py @@ -1,29 +1,6 @@ # coding: utf-8 from __future__ import annotations -import logging -import signal -import subprocess - -import time -import typing - -try: - from collections.abc import Iterable -except ImportError: - from collections import Iterable - -# we support both pg8000 and psycopg2 -try: - import psycopg2 as pglib -except ImportError: - try: - import pg8000 as pglib - except ImportError: - raise ImportError("You must have psycopg2 or pg8000 modules installed") - -from six import raise_from, iteritems, text_type - from .enums import \ NodeStatus, \ ProcessType, \ @@ -86,17 +63,41 @@ from .standby import First -from . import utils - from .raise_error import RaiseError from .backup import NodeBackup +from . import utils + 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 +import logging +import signal +import subprocess + +import time +import typing + +try: + from collections.abc import Iterable +except ImportError: + from collections import Iterable + +# we support both pg8000 and psycopg2 +try: + import psycopg2 as pglib +except ImportError: + try: + import pg8000 as pglib + except ImportError: + raise ImportError("You must have psycopg2 or pg8000 modules installed") + +from six import raise_from, iteritems, text_type + + InternalError = pglib.InternalError ProgrammingError = pglib.ProgrammingError OperationalError = pglib.OperationalError From 31d206e938fb9d2543e37b95503765aaa08c56d2 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Sat, 19 Sep 2026 15:52:05 +0300 Subject: [PATCH 3/3] consts.BINARY_NAME__POSTGRES is added --- src/consts.py | 3 +++ src/utils.py | 14 ++++++++++---- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/src/consts.py b/src/consts.py index d3589205..9baf3ebc 100644 --- a/src/consts.py +++ b/src/consts.py @@ -1,5 +1,8 @@ # coding: utf-8 +# binary names +BINARY_NAME__POSTGRES = "postgres" + # names for dirs in base_dir DATA_DIR = "data" LOGS_DIR = "logs" diff --git a/src/utils.py b/src/utils.py index 9c7d91c2..77bf1bd3 100644 --- a/src/utils.py +++ b/src/utils.py @@ -19,6 +19,8 @@ from testgres.operations.local_ops import LocalOperations from testgres.operations.helpers import Helpers as OsHelpers +from . import consts + from .impl.port_manager__generic2 import PortManager__Generic2 from .impl.platforms import internal_platform_utils_factory @@ -327,16 +329,20 @@ 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) if bin_dir is None: - postgres_path = get_bin_path2(os_ops, C_POSTGRES_BINARY) + postgres_path = get_bin_path2( + os_ops, + consts.BINARY_NAME__POSTGRES, + ) else: # [2025-06-25] OK ? assert type(bin_dir) is str assert bin_dir != "" - postgres_path = os_ops.build_path(bin_dir, 'postgres') + postgres_path = os_ops.build_path( + bin_dir, + consts.BINARY_NAME__POSTGRES, + ) cmd = [postgres_path, '--version'] raw_ver = os_ops.run(cmd, encoding='utf-8').stdout