From b171874bcd8cb3b1f904a84f0c560e3716da5082 Mon Sep 17 00:00:00 2001 From: "d.kovalenko" Date: Fri, 10 Jul 2026 18:49:06 +0300 Subject: [PATCH] fix: TestgresLogger did not use os_ops --- src/impl/file_line_reader.py | 80 ++++++++++++ src/logger.py | 99 +++++++++++--- src/node.py | 6 +- tests/units/impl/test_file_line_reader.py | 152 ++++++++++++++++++++++ 4 files changed, 320 insertions(+), 17 deletions(-) create mode 100755 src/impl/file_line_reader.py create mode 100755 tests/units/impl/test_file_line_reader.py diff --git a/src/impl/file_line_reader.py b/src/impl/file_line_reader.py new file mode 100755 index 00000000..ed6579e8 --- /dev/null +++ b/src/impl/file_line_reader.py @@ -0,0 +1,80 @@ +# coding: utf-8 + +import typing + +from testgres.operations.os_ops import OsOperations + + +class FileLineReader: + _os_ops: OsOperations + _file_name: str + _file_encoding: str + _file_pos: int + _buffer_pos: int + _buffer: bytes + + # -------------------------------------------------------------------- + def __init__( + self, + os_ops: OsOperations, + file_name: str, + file_encoding: str = "utf-8", + ): + assert isinstance(os_ops, OsOperations) + assert type(file_encoding) is str + self._os_ops = os_ops + self._file_name = file_name + self._file_encoding = file_encoding + self._file_pos = 0 + self._buffer_pos = 0 + self._buffer = b'' + return + + # interface ---------------------------------------------------------- + def read_line(self) -> typing.Optional[str]: + assert isinstance(self._os_ops, OsOperations) + assert type(self._buffer_pos) is int + assert type(self._buffer) is bytes + assert self._buffer_pos >= 0 + assert self._buffer_pos <= len(self._buffer) + + scan_pos = self._buffer_pos + + while True: + sz1 = len(self._buffer) + + if scan_pos == sz1: + block = self._os_ops.read_binary( + self._file_name, + self._file_pos, + ) + assert type(block) is bytes + self._buffer += block + self._file_pos += len(block) + + x = self._buffer.find(b'\n', scan_pos) + + sz2 = len(self._buffer) + + if x == -1: + if scan_pos == sz2: + return None + + if self._buffer_pos == 0: + scan_pos = sz2 + else: + assert self._buffer_pos > 0 + self._buffer = self._buffer[self._buffer_pos:] + scan_pos = sz2 - self._buffer_pos + self._buffer_pos = 0 + continue + + assert x >= 0 + assert x < sz2 + + b = self._buffer[self._buffer_pos:(x+1)] + + s = b.decode(self._file_encoding) + + self._buffer_pos = x + 1 + return s diff --git a/src/logger.py b/src/logger.py index b4648f44..51e11b73 100644 --- a/src/logger.py +++ b/src/logger.py @@ -1,49 +1,116 @@ # coding: utf-8 +from .config import testgres_config as tconf +from .exceptions import ExecUtilException +from .impl.file_line_reader import FileLineReader + +from testgres.operations.os_ops import OsOperations + import logging -import select import threading import time +import typing class TestgresLogger(threading.Thread): + _C_SLEEP_MIN = 0.01 + _C_SLEEP_MAX = 60 + """ Helper class to implement reading from log files. """ - def __init__(self, node_name, log_file_name): + def __init__( + self, + node_name: str, + log_file_name: str, + log_file_encoding: str = "utf-8", + os_ops: typing.Optional[OsOperations] = None, + ): + assert type(node_name) is str + assert type(log_file_name) is str + assert type(log_file_encoding) is str + assert os_ops is None or isinstance(os_ops, OsOperations) + threading.Thread.__init__(self) + if os_ops is None: + os_ops = tconf.os_ops + + assert os_ops is None or isinstance(os_ops, OsOperations) + + self._os_ops = os_ops self._node_name = node_name self._log_file_name = log_file_name + self._log_file_encoding = log_file_encoding self._stop_event = threading.Event() self._logger = logging.getLogger(node_name) self._logger.setLevel(logging.INFO) + return def run(self): # open log file for reading - with open(self._log_file_name, 'r') as fd: + file_line_reader = FileLineReader( + self._os_ops, + self._log_file_name, + self._log_file_encoding, + ) + + sleep_time = __class__._C_SLEEP_MIN + + try: # work until we're asked to stop while not self._stop_event.is_set(): - sleep_time = 0.1 - new_lines = False + line = None - # do we have new lines? - if fd in select.select([fd], [], [], 0)[0]: - for line in fd.readlines(): - line = line.strip() - if line: - new_lines = True - extra = {'node': self._node_name} - self._logger.info(line, extra=extra) - - if not new_lines: + while True: + try: + line = file_line_reader.read_line() # raise + except Exception as e: + if __class__._is_file_not_found_exception(e): + if not self._os_ops.path_exists(self._log_file_name): + break + raise + break + + if line is None: time.sleep(sleep_time) + sleep_time = min(__class__._C_SLEEP_MAX, 2 * sleep_time) + continue + + assert type(line) is str + + sleep_time = __class__._C_SLEEP_MIN + + # do we have new lines? + line = line.strip() + extra = {'node': self._node_name} + self._logger.info(line, extra=extra) + continue + except Exception as e: + self._logger.error(e) + raise + finally: # don't forget to clear event - self._stop_event.clear() + # [2026-07-10] legacy cargo cult, thread is single-use only. + # self._stop_event.clear() + pass + return def stop(self, wait=True): self._stop_event.set() if wait: self.join() + return + + @staticmethod + def _is_file_not_found_exception(e: Exception) -> bool: + if isinstance(e, FileNotFoundError): + return True + + if isinstance(e, ExecUtilException): + if e.exit_code == 2: + return True + + return False diff --git a/src/node.py b/src/node.py index 883a60eb..80acfe84 100644 --- a/src/node.py +++ b/src/node.py @@ -729,7 +729,11 @@ def _maybe_start_logger(self): if testgres_config.use_python_logging: # spawn new logger if it doesn't exist or is stopped if not self._logger or not self._logger.is_alive(): - self._logger = TestgresLogger(self.name, self.pg_log_file) + self._logger = TestgresLogger( + self.name, + self.pg_log_file, + os_ops=self._os_ops, + ) self._logger.start() def _maybe_stop_logger(self): diff --git a/tests/units/impl/test_file_line_reader.py b/tests/units/impl/test_file_line_reader.py new file mode 100755 index 00000000..5b8d1126 --- /dev/null +++ b/tests/units/impl/test_file_line_reader.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +from ...helpers.global_data import OsOpsDescrs +from ...helpers.global_data import OsOpsDescr +from ...helpers.global_data import OsOperations + +from src.impl.file_line_reader import FileLineReader + +import pytest +import typing +import dataclasses +import logging + + +class TestFileLineReader: + sm_os_ops_descrs: typing.List[OsOpsDescr] = [ + OsOpsDescrs.sm_local_os_ops_descr, + OsOpsDescrs.sm_remote_os_ops_descr + ] + + @pytest.fixture( + params=[ + pytest.param( + descr, + id=descr.sign, + ) + for descr in sm_os_ops_descrs + ], + ) + def os_ops_descr(self, request: pytest.FixtureRequest) -> OsOpsDescr: + assert isinstance(request, pytest.FixtureRequest) + assert isinstance(request.param, OsOpsDescr) + return request.param + + # -------------------------------------------------------------------- + @dataclasses.dataclass + class tagStep: + write_data: bytes + read_lines: typing.List[typing.Optional[str]] + + sm_Steps001: typing.List[tagStep] = [ + tagStep( + b"", + [None, None, None] + ), + tagStep( + b"a", + [None, None] + ), + tagStep( + b"b", + [None, None] + ), + tagStep( + b"c\n", + ["abc\n", None] + ), + tagStep( + b"d", + [None, None] + ), + tagStep( + b"efg\n1\n\n3", + ["defg\n", "1\n", "\n", None, None] + ), + tagStep( + b" \n 1\n", + ["3 \n", " 1\n", None, None] + ), + # russian text ma: b'\xd0\xbc\xd0\xb0' + tagStep( + b'\xd0', + [None] + ), + tagStep( + b'\xbc', + [None] + ), + tagStep( + b'\xd0', + [None] + ), + tagStep( + b'\xb0', + [None] + ), + tagStep( + b'\n', # FINISH + ["\u043c\u0430\n", None] + ), + + ] + + # -------------------------------------------------------------------- + def test_001__common( + self, + os_ops_descr: OsOpsDescr, + ): + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops_descr.os_ops, OsOperations) + + tmpdir = os_ops.mkdtemp() + filename = os_ops.build_path(tmpdir, "my.log") + assert not os_ops.path_exists(filename) + + os_ops.touch(filename) + assert os_ops.path_exists(filename) + assert os_ops.get_file_size(filename) == 0 + + file_line_reader = FileLineReader( + os_ops, + filename, + file_encoding="utf-8", + ) + + # ----------------------- + nStep = 0 + + for step in __class__.sm_Steps001: + nStep += 1 + + logging.info("-------------------- step: {}".format(nStep)) + + logging.info("write: {}".format(step.write_data)) + os_ops.write(filename, step.write_data, binary=True) + + nRead = 0 + for expected_line in step.read_lines: + nRead += 1 + logging.info("read [{}]. expected line is {!r}".format( + nRead, + expected_line, + )) + actual_line = file_line_reader.read_line() + + if actual_line != expected_line: + err_msg = "Read bad line {!r}. Expected {!r}".format( + actual_line, + expected_line, + ) + raise RuntimeError(err_msg) + continue + continue + + assert file_line_reader.read_line() is None + + os_ops.rmdirs(tmpdir) + assert not os_ops.path_exists(tmpdir) + return