From 46c140e819222f1e03872e20e5570d095e3841af Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Thu, 24 Sep 2026 09:38:39 +0300 Subject: [PATCH] utils: public API surface is restored --- src/impl/internal_node_utils.py | 221 +++++++++++++ src/impl/internal_utils.py | 58 ++++ .../internal_platform_utils_provider.py | 32 ++ src/node_state.py | 20 ++ src/utils.py | 304 +----------------- 5 files changed, 341 insertions(+), 294 deletions(-) create mode 100644 src/impl/internal_node_utils.py create mode 100644 src/impl/platforms/internal_platform_utils_provider.py create mode 100644 src/node_state.py diff --git a/src/impl/internal_node_utils.py b/src/impl/internal_node_utils.py new file mode 100644 index 00000000..d0cc42ef --- /dev/null +++ b/src/impl/internal_node_utils.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +from . import internal_utils +from . platforms.internal_platform_utils_provider import InternalPlaformUtilsProvider + +from .. import consts +from ..node_state import PostgresNodeState +from ..enums import NodeStatus +from ..raise_error import RaiseError +from ..exceptions import ExecUtilException + +from testgres.operations.os_ops import OsOperations + +import typing +import time + + +class InternalNodeUtils: + T_PLATFORM_UTILS = InternalPlaformUtilsProvider.T_PLATFORM_UTILS + + @staticmethod + def get_pg_node_state( + os_ops: OsOperations, + bin_dir: str, + data_dir: str, + utils_log_file: typing.Optional[str], + ) -> PostgresNodeState: + assert isinstance(os_ops, OsOperations) + assert type(bin_dir) is str + assert type(data_dir) is str + assert utils_log_file is None or type(utils_log_file) is str + + C_MAX_ATTEMPTS = 3 + C_SLEEP_TIME1 = 1 + C_SLEEP_TIME_MULT = 2 + + pg_ctl_params = [ + os_ops.build_path(bin_dir, consts.BINARY_NAME__PG_CTL), + "-D", + data_dir, + "status", + ] + + attempt = 0 + sleep_time = C_SLEEP_TIME1 + + platform_utils_provider = InternalPlaformUtilsProvider( + os_ops, + ) + + while True: + assert type(attempt) is int + assert attempt >= 0 + assert attempt < C_MAX_ATTEMPTS + + attempt += 1 + + if attempt > 1: + internal_utils.send_log_debug("Sleep {} second(s) before an attempt #{}".format( + sleep_time, + attempt, + )) + time.sleep(sleep_time) + sleep_time = sleep_time * C_SLEEP_TIME_MULT + + exec_r = internal_utils.execute_utility3( + os_ops, + pg_ctl_params, + utils_log_file, + check=False, + ) + + status_code = exec_r.returncode + out = exec_r.stdout + error = exec_r.stderr + + assert type(status_code) is int + assert type(out) is str + assert type(error) is str + + # ----------------- + if status_code == consts.PG_CTL__STATUS__NODE_IS_STOPPED: + return PostgresNodeState(NodeStatus.Stopped, None) + + # ----------------- + if status_code == consts.PG_CTL__STATUS__BAD_DATADIR: + return PostgresNodeState(NodeStatus.Uninitialized, None) + + # ----------------- + if status_code == consts.PG_CTL__STATUS__OK: + pid = __class__._parse_pid( + out, + pg_ctl_params, + ) + assert type(pid) is int + assert pid != 0 + + # ----------------- detect zombie + if platform_utils_provider.get().ProcessIsZombie_soft_check(os_ops, pid) is True: + internal_utils.send_log_debug("Postmaster process {} is a zombie.".format( + pid, + )) + return PostgresNodeState(NodeStatus.Zombie, pid) + + # ----------------- + return PostgresNodeState(NodeStatus.Running, pid) + + assert status_code != consts.PG_CTL__STATUS__OK + + errMsg = "Getting of a node status [data_dir is {0}] failed.".format( + data_dir, + ) + + e1 = ExecUtilException( + message=errMsg, + command=pg_ctl_params, + exit_code=status_code, + out=out, + error=error, + ) + + if status_code == consts.PG_CTL__STATUS__FAILED: + internal_utils.send_log_debug( + "pg_ctl fails with an error: {}".format( + exec_r.stderr, + )) + + try: + find_postmaster_r = platform_utils_provider.get().FindPostmaster( + os_ops, + bin_dir, + data_dir, + ) + except Exception as e2: + raise e2 from e1 + + assert type(find_postmaster_r) is __class__.T_PLATFORM_UTILS.FindPostmasterResult + + if find_postmaster_r.code == __class__.T_PLATFORM_UTILS.FindPostmasterResultCode.ok: + # 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, + )) + + if attempt < C_MAX_ATTEMPTS: + continue + + raise e1 + + @staticmethod + def _parse_pid( + out: str, + pg_ctl_params, + ) -> int: + assert type(out) is str + + if out == "": + RaiseError.pg_ctl_returns_an_empty_string( + pg_ctl_params, + ) + + C_PID_PREFIX = "(PID: " + + i = out.find(C_PID_PREFIX) + + if i == -1: + RaiseError.pg_ctl_returns_an_unexpected_string( + out, + pg_ctl_params, + ) + + assert i > 0 + assert i < len(out) + assert len(C_PID_PREFIX) <= len(out) + assert i <= len(out) - len(C_PID_PREFIX) + + i += len(C_PID_PREFIX) + start_pid_s = i + + while True: + if i == len(out): + RaiseError.pg_ctl_returns_an_unexpected_string( + out, + pg_ctl_params, + ) + + ch = out[i] + + if ch == ")": + break + + if ch.isdigit(): + i += 1 + continue + + RaiseError.pg_ctl_returns_an_unexpected_string( + out, + pg_ctl_params, + ) + assert False + + if i == start_pid_s: + RaiseError.pg_ctl_returns_an_unexpected_string( + out, + pg_ctl_params, + ) + + # TODO: Let's verify a length of pid string. + + pid = int(out[start_pid_s:i]) + + if pid == 0: + RaiseError.pg_ctl_returns_a_zero_pid( + out, + pg_ctl_params, + ) + + assert pid != 0 + + return pid diff --git a/src/impl/internal_utils.py b/src/impl/internal_utils.py index 05d5a465..56843223 100644 --- a/src/impl/internal_utils.py +++ b/src/impl/internal_utils.py @@ -1,4 +1,10 @@ from testgres.operations.os_ops import OsOperations +from testgres.operations.os_ops import OsCommandResult +from testgres.operations.types import T_OS_CMD +from testgres.operations.types import T_OS_EXEC_ENV +from testgres.operations.helpers import Helpers as OsHelpers + +from ..exceptions import ExecUtilException import logging import typing @@ -90,3 +96,55 @@ def read_line_to_pos__bin( assert type(result) is bytes assert len(result) <= (position - read_position) return result + + +def execute_utility3( + os_ops: OsOperations, + args: T_OS_CMD, + logfile: typing.Optional[str] = None, + check: bool = True, + exec_env: typing.Optional[T_OS_EXEC_ENV] = None, +) -> OsCommandResult: + assert os_ops is not None + assert isinstance(os_ops, OsOperations) + assert type(check) is bool + assert exec_env is None or type(exec_env) is dict + + exec_r = os_ops.run( + args, + check=check, + encoding=OsHelpers.GetDefaultEncoding(), + exec_env=exec_env, + ) + + assert type(exec_r) is OsCommandResult + + # write new log entry if possible + if logfile: + try: + log_lines = [ + os_ops.join_command_arguments(args), + ] + + if exec_r.stdout is None: + log_lines.append("# #NONE#") + else: + # comment-out lines + assert type(exec_r.stdout) is str + log_lines += ['# ' + line for line in exec_r.stdout.splitlines()] + + log_lines.append("") + + os_ops.write( + filename=logfile, + data="\n".join(log_lines), + truncate=False, + ) + except IOError: + raise ExecUtilException( + "Problem with writing to logfile `{}` during run command `{}`".format( + logfile, + args, + )) + + return exec_r diff --git a/src/impl/platforms/internal_platform_utils_provider.py b/src/impl/platforms/internal_platform_utils_provider.py new file mode 100644 index 00000000..411c7ed9 --- /dev/null +++ b/src/impl/platforms/internal_platform_utils_provider.py @@ -0,0 +1,32 @@ +from .internal_platform_utils_factory import create_internal_platform_utils +from .internal_platform_utils_factory import InternalPlatformUtils + +from testgres.operations.os_ops import OsOperations + +import typing + + +class InternalPlaformUtilsProvider: + T_PLATFORM_UTILS = InternalPlatformUtils + + _os_ops: OsOperations + _platform_utils: typing.Optional[T_PLATFORM_UTILS] = None + + def __init__( + self, + os_ops: OsOperations, + ): + assert isinstance(os_ops, OsOperations) + self._os_ops = os_ops + self._platform_utils = None + return + + def get(self) -> T_PLATFORM_UTILS: + if self._platform_utils is None: + self._platform_utils = create_internal_platform_utils( + self._os_ops, + ) + assert isinstance(self._platform_utils, __class__.T_PLATFORM_UTILS) + + assert isinstance(self._platform_utils, __class__.T_PLATFORM_UTILS) + return self._platform_utils diff --git a/src/node_state.py b/src/node_state.py new file mode 100644 index 00000000..25d7277e --- /dev/null +++ b/src/node_state.py @@ -0,0 +1,20 @@ +from .enums import NodeStatus + +import typing + + +class PostgresNodeState: + node_status: NodeStatus + pid: typing.Optional[int] + + def __init__( + self, + node_status: NodeStatus, + pid: typing.Optional[int], + ): + assert type(node_status) is NodeStatus + assert pid is None or type(pid) is int + + self.node_status = node_status + self.pid = pid + return diff --git a/src/utils.py b/src/utils.py index 546af917..718f8353 100644 --- a/src/utils.py +++ b/src/utils.py @@ -4,10 +4,9 @@ from __future__ import division from __future__ import print_function -from .exceptions import ExecUtilException, InvalidOperationException +from .exceptions import InvalidOperationException from .config import testgres_config as tconf -from .raise_error import RaiseError -from .enums import NodeStatus +from .node_state import PostgresNodeState from testgres.operations.types import T_OS_CMD from testgres.operations.types import T_OS_EXEC_ENV @@ -15,18 +14,15 @@ from testgres.operations.os_ops import OsCommandResult from testgres.operations.remote_ops import RemoteOperations 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 from .impl import internal_utils +from .impl.internal_node_utils import InternalNodeUtils +from .impl.port_manager__generic2 import PortManager__Generic2 import os import sys -import time import re import typing @@ -138,45 +134,14 @@ def execute_utility3( assert type(check) is bool assert exec_env is None or type(exec_env) is dict - exec_r = os_ops.run( + return internal_utils.execute_utility3( + os_ops, args, - check=check, - encoding=OsHelpers.GetDefaultEncoding(), - exec_env=exec_env, + logfile, + check, + exec_env, ) - assert type(exec_r) is OsCommandResult - - # write new log entry if possible - if logfile: - try: - log_lines = [ - os_ops.join_command_arguments(args), - ] - - if exec_r.stdout is None: - log_lines.append("# #NONE#") - else: - # comment-out lines - assert type(exec_r.stdout) is str - log_lines += ['# ' + line for line in exec_r.stdout.splitlines()] - - log_lines.append("") - - os_ops.write( - filename=logfile, - data="\n".join(log_lines), - truncate=False, - ) - except IOError: - raise ExecUtilException( - "Problem with writing to logfile `{}` during run command `{}`".format( - logfile, - args, - )) - - return exec_r - def get_bin_path(filename): """ @@ -422,23 +387,6 @@ def clean_on_error(node): raise -class PostgresNodeState: - node_status: NodeStatus - pid: typing.Optional[int] - - def __init__( - self, - node_status: NodeStatus, - pid: typing.Optional[int], - ): - assert type(node_status) is NodeStatus - assert pid is None or type(pid) is int - - self.node_status = node_status - self.pid = pid - return - - def get_pg_node_state( os_ops: OsOperations, bin_dir: str, @@ -450,241 +398,9 @@ def get_pg_node_state( assert type(data_dir) is str assert utils_log_file is None or type(utils_log_file) is str - return PostgresNodeStateUtils.get_pg_node_state( + return InternalNodeUtils.get_pg_node_state( os_ops, bin_dir, data_dir, utils_log_file, ) - - -class InternalPlaformUtilsProvider: - T_PLATFORM_UTILS = internal_platform_utils_factory.InternalPlatformUtils - - _os_ops: OsOperations - _platform_utils: typing.Optional[T_PLATFORM_UTILS] = None - - def __init__( - self, - os_ops: OsOperations, - ): - assert isinstance(os_ops, OsOperations) - self._os_ops = os_ops - self._platform_utils = None - return - - def get(self) -> T_PLATFORM_UTILS: - if self._platform_utils is None: - self._platform_utils = internal_platform_utils_factory.create_internal_platform_utils( - self._os_ops, - ) - assert isinstance(self._platform_utils, __class__.T_PLATFORM_UTILS) - - assert isinstance(self._platform_utils, __class__.T_PLATFORM_UTILS) - return self._platform_utils - - -class PostgresNodeStateUtils: - T_PLATFORM_UTILS = InternalPlaformUtilsProvider.T_PLATFORM_UTILS - - @staticmethod - def get_pg_node_state( - os_ops: OsOperations, - bin_dir: str, - data_dir: str, - utils_log_file: typing.Optional[str], - ) -> PostgresNodeState: - assert isinstance(os_ops, OsOperations) - assert type(bin_dir) is str - assert type(data_dir) is str - assert utils_log_file is None or type(utils_log_file) is str - - C_MAX_ATTEMPTS = 3 - C_SLEEP_TIME1 = 1 - C_SLEEP_TIME_MULT = 2 - - pg_ctl_params = [ - os_ops.build_path(bin_dir, consts.BINARY_NAME__PG_CTL), - "-D", - data_dir, - "status", - ] - - attempt = 0 - sleep_time = C_SLEEP_TIME1 - - platform_utils_provider = InternalPlaformUtilsProvider( - os_ops, - ) - - while True: - assert type(attempt) is int - assert attempt >= 0 - assert attempt < C_MAX_ATTEMPTS - - attempt += 1 - - if attempt > 1: - internal_utils.send_log_debug("Sleep {} second(s) before an attempt #{}".format( - sleep_time, - attempt, - )) - time.sleep(sleep_time) - sleep_time = sleep_time * C_SLEEP_TIME_MULT - - exec_r = execute_utility3( - os_ops, - pg_ctl_params, - utils_log_file, - check=False, - ) - - status_code = exec_r.returncode - out = exec_r.stdout - error = exec_r.stderr - - assert type(status_code) is int - assert type(out) is str - assert type(error) is str - - # ----------------- - if status_code == consts.PG_CTL__STATUS__NODE_IS_STOPPED: - return PostgresNodeState(NodeStatus.Stopped, None) - - # ----------------- - if status_code == consts.PG_CTL__STATUS__BAD_DATADIR: - return PostgresNodeState(NodeStatus.Uninitialized, None) - - # ----------------- - if status_code == consts.PG_CTL__STATUS__OK: - pid = __class__._parse_pid( - out, - pg_ctl_params, - ) - assert type(pid) is int - assert pid != 0 - - # ----------------- detect zombie - if platform_utils_provider.get().ProcessIsZombie_soft_check(os_ops, pid) is True: - internal_utils.send_log_debug("Postmaster process {} is a zombie.".format( - pid, - )) - return PostgresNodeState(NodeStatus.Zombie, pid) - - # ----------------- - return PostgresNodeState(NodeStatus.Running, pid) - - assert status_code != consts.PG_CTL__STATUS__OK - - errMsg = "Getting of a node status [data_dir is {0}] failed.".format( - data_dir, - ) - - e1 = ExecUtilException( - message=errMsg, - command=pg_ctl_params, - exit_code=status_code, - out=out, - error=error, - ) - - if status_code == consts.PG_CTL__STATUS__FAILED: - internal_utils.send_log_debug( - "pg_ctl fails with an error: {}".format( - exec_r.stderr, - )) - - try: - find_postmaster_r = platform_utils_provider.get().FindPostmaster( - os_ops, - bin_dir, - data_dir, - ) - except Exception as e2: - raise e2 from e1 - - assert type(find_postmaster_r) is __class__.T_PLATFORM_UTILS.FindPostmasterResult - - if find_postmaster_r.code == __class__.T_PLATFORM_UTILS.FindPostmasterResultCode.ok: - # 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, - )) - - if attempt < C_MAX_ATTEMPTS: - continue - - raise e1 - - @staticmethod - def _parse_pid( - out: str, - pg_ctl_params, - ) -> int: - assert type(out) is str - - if out == "": - RaiseError.pg_ctl_returns_an_empty_string( - pg_ctl_params, - ) - - C_PID_PREFIX = "(PID: " - - i = out.find(C_PID_PREFIX) - - if i == -1: - RaiseError.pg_ctl_returns_an_unexpected_string( - out, - pg_ctl_params, - ) - - assert i > 0 - assert i < len(out) - assert len(C_PID_PREFIX) <= len(out) - assert i <= len(out) - len(C_PID_PREFIX) - - i += len(C_PID_PREFIX) - start_pid_s = i - - while True: - if i == len(out): - RaiseError.pg_ctl_returns_an_unexpected_string( - out, - pg_ctl_params, - ) - - ch = out[i] - - if ch == ")": - break - - if ch.isdigit(): - i += 1 - continue - - RaiseError.pg_ctl_returns_an_unexpected_string( - out, - pg_ctl_params, - ) - assert False - - if i == start_pid_s: - RaiseError.pg_ctl_returns_an_unexpected_string( - out, - pg_ctl_params, - ) - - # TODO: Let's verify a length of pid string. - - pid = int(out[start_pid_s:i]) - - if pid == 0: - RaiseError.pg_ctl_returns_a_zero_pid( - out, - pg_ctl_params, - ) - - assert pid != 0 - - return pid