fix: TestgresLogger did not use os_ops

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

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

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):

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