[#323] TestTestgresCommon::test_restart + PostgresNodeLogReader are updated

pyproject.toml

@@ -65,7 +65,7 @@ dependencies = [
     "six>=1.9.0",
     "psutil",
     "packaging",
-    "testgres.os_ops>=3.1.0,<4.0.0",
+    "testgres.os_ops>=3.2.0,<4.0.0",
 ]
 
 [project.urls]

src/node.py

@@ -2430,9 +2430,14 @@ class LogInfo:
         position: int
         tail: bytes
 
-        def __init__(self, position: int):
+        def __init__(self, position: int, tail: bytes = b''):
+            assert type(position) is int
+            assert type(tail) is bytes
+            assert position >= 0
+
             self.position = position
-            self.tail = b''
+            self.tail = tail
+            return
 
     # --------------------------------------------------------------------
     class LogDataBlock:
@@ -2487,7 +2492,7 @@ def __init__(self, node: PostgresNode, from_beginnig: bool):
         if from_beginnig:
             self._logs = dict()
         else:
-            self._logs = self._collect_logs()
+            self._logs = self._collect_logs(find_line_start=True)
 
         assert type(self._logs) is dict
         return
@@ -2496,7 +2501,7 @@ def read(self) -> typing.List[LogDataBlock]:
         assert self._node is not None
         assert isinstance(self._node, PostgresNode)
 
-        cur_logs: typing.Dict[str, __class__.LogInfo] = self._collect_logs()
+        cur_logs = self._collect_logs(find_line_start=False)
         assert cur_logs is not None
         assert type(cur_logs) is dict
 
@@ -2570,7 +2575,8 @@ def read(self) -> typing.List[LogDataBlock]:
 
         return result
 
-    def _collect_logs(self) -> typing.Dict[str, LogInfo]:
+    def _collect_logs(self, find_line_start: bool) -> typing.Dict[str, LogInfo]:
+        assert type(find_line_start) is bool
         assert self._node is not None
         assert isinstance(self._node, PostgresNode)
 
@@ -2587,14 +2593,92 @@ def _collect_logs(self) -> typing.Dict[str, LogInfo]:
             if not self._node.os_ops.path_exists(f):
                 continue
 
-            file_size = self._node.os_ops.get_file_size(f)
-            assert type(file_size) is int
-            assert file_size >= 0
-
-            result[f] = __class__.LogInfo(file_size)
+            result[f] = self._create_log_info(
+                self._node.os_ops,
+                f,
+                find_line_start,
+            )
+            continue
 
         return result
 
+    @staticmethod
+    def _create_log_info(
+        os_ops: OsOperations,
+        filename: str,
+        find_line_start: bool,
+    ) -> LogInfo:
+        assert type(filename) is str
+        assert type(find_line_start) is bool
+        assert len(filename) > 0
+        assert os_ops is not None
+        assert isinstance(os_ops, OsOperations)
+
+        file_size = os_ops.get_file_size(filename)
+        assert type(file_size) is int
+        assert file_size >= 0
+
+        if not find_line_start:
+            return __class__.LogInfo(
+                position=file_size,
+                tail=b'',
+            )
+
+        read_position = file_size
+        tail_blocks: typing.List[bytes] = []
+
+        C_BACK_READ_BLOCK_SIZE = 4096
+
+        while read_position > 0:
+            if read_position < C_BACK_READ_BLOCK_SIZE:
+                read_offset = 0
+            else:
+                read_offset = read_position - C_BACK_READ_BLOCK_SIZE
+
+            assert read_offset >= 0
+            assert read_offset < file_size
+            assert read_offset < read_position
+
+            block_sz = read_position - read_offset
+
+            assert block_sz > 0
+
+            # read from read_offset to file end
+            block = os_ops.read_binary(filename, read_offset, block_sz)
+
+            assert type(block) is bytes
+
+            if len(block) != block_sz:
+                err_msg = "[BUG CHECK] Readed block has bad size ({}). Expected size is ({}). File name {}.".format(
+                    len(block),
+                    block_sz,
+                    filename,
+                )
+                raise RuntimeError(err_msg)
+
+            assert len(block) == block_sz
+
+            x = block.rfind(b"\n", 0, block_sz)
+
+            if x == -1:
+                tail_blocks.append(block)
+                read_position = read_offset
+                continue
+
+            if x == block_sz - 1:
+                break
+
+            block = block[x + 1:]
+            tail_blocks.append(block)
+            break
+
+        tail = b''.join(reversed(tail_blocks))
+
+        return __class__.LogInfo(
+            position=file_size,
+            tail=tail,
+        )
+
 
 class PostgresNodeUtils:
     @staticmethod

tests/requirements.txt

@@ -4,5 +4,5 @@ pytest-env
 pytest-xdist
 psycopg2
 six
-testgres.os_ops>=3.1.0,<4.0.0
+testgres.os_ops>=3.2.0,<4.0.0
 testgres.postgres_configuration>=0.2.2,<1.0.0

tests/test_testgres_common.py

@@ -456,12 +456,70 @@ def test_restart(self, node_svc: PostgresNodeService):
         assert isinstance(node_svc, PostgresNodeService)
 
         with __class__.helper__get_node(node_svc) as node:
-            node.init().start()
+            node.init()
+
+            nRestartAttempt = 0
+
+            while True:
+                nRestartAttempt += 1
+
+                logging.info("Attempt #{}".format(nRestartAttempt))
+
+                node.start()
+
+                # restart, ok
+                res = node.execute('select 1')
+                assert (res == [(1,)])
+
+                node_log_reader = PostgresNodeLogReader(
+                    node,
+                    from_beginnig=False,
+                )
+
+                try:
+                    node.restart()
+                except StartNodeException as e:
+                    logging.info("Exception ({}): {}".format(
+                        type(e).__name__,
+                        e,
+                    ))
+
+                    if nRestartAttempt == 5:
+                        raise
+
+                    if not PostgresNodeUtils.detect_port_conflict(node_log_reader):
+                        raise
+
+                    logging.info("Node port {} conflicted with another PostgreSQL instance.".format(
+                        node.port
+                    ))
+
+                    logging.info("Wait for node stop")
+
+                    nStopAttemtp = 0
+
+                    while True:
+                        if nStopAttemtp == 5:
+                            raise RuntimeError("Node is not stopped!")
+
+                        nStopAttemtp += 1
+
+                        time.sleep(1)
+
+                        node_status = node.status()
+
+                        logging.info("Node status is {}".format(node_status))
+
+                        if node_status == NodeStatus.Stopped:
+                            break
+                        continue
+
+                    # node is stopped. try again
+                    continue
+
+                assert node.status() == NodeStatus.Running
+                break
 
-            # restart, ok
-            res = node.execute('select 1')
-            assert (res == [(1,)])
-            node.restart()
             res = node.execute('select 2')
             assert (res == [(2,)])
 

tests/units/node/PostgresNodeLogReader/test_setM001__helper_create_log_info.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from ....helpers.global_data import OsOpsDescrs
+from ....helpers.global_data import OsOpsDescr
+from ....helpers.global_data import OsOperations
+
+from src.node import PostgresNodeLogReader
+
+import pytest
+import typing
+import random
+
+
+class TestSetM001__helper_create_log_info:
+    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
+
+    def test_001__common(self, os_ops_descr: OsOpsDescr):
+        assert type(os_ops_descr) is OsOpsDescr
+        os_ops = os_ops_descr.os_ops
+        assert isinstance(os_ops, OsOperations)
+
+        filename = os_ops.mkstemp("data_for_create_log_info")
+
+        # Scenario 0: The log file ends with a normal line feed
+        C_DATA0 = b""
+        os_ops.write(filename, C_DATA0, binary=True, truncate=True)
+
+        log_info1 = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        assert log_info1.tail == b""
+        assert log_info1.position == len(C_DATA0)
+
+        # Scenario 1: The log file ends with a normal line feed
+        C_DATA1 = b"Line 1\nLine 2\n"
+        os_ops.write(filename, C_DATA1, binary=True, truncate=True)
+
+        log_info1 = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        # Since the file ends with \n, the tail must be empty!
+        assert log_info1.tail == b""
+        assert log_info1.position == len(C_DATA1)
+
+        # Scenario 2: The log file contains an unterminated line (our UTF-8 trap)
+        C_DATA2 = b"Line 1\nLine 2\nIncomplete UTF8 \xd0"
+        os_ops.write(filename, C_DATA2, binary=True, truncate=True)
+
+        log_info2 = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        # The tail must contain exactly the piece after the last \n
+        assert log_info2.tail == b"Incomplete UTF8 \xd0"
+        assert log_info2.position == len(C_DATA2)
+
+        # Scenario 3: The file has no line breaks at all (one long line)
+        C_DATA3 = b"Just one long line without newlines"
+        os_ops.write(filename, C_DATA3, binary=True, truncate=True)
+
+        log_info3 = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        # Should take the entire file in tail, and set the position to the file size
+        assert log_info3.tail == C_DATA3
+        assert log_info3.position == len(C_DATA3)
+
+        # 4. Large data (two segments)
+        allowed_bytes = bytes([b for b in range(256) if b != 10])
+        C_DATA4 = bytes(random.choices(allowed_bytes, k=5000))
+        os_ops.write(filename, C_DATA4, binary=True, truncate=True)
+
+        log_info = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        assert log_info.tail == C_DATA4
+        assert log_info.position == len(C_DATA4)
+
+        # 5. Large data (many segments)
+        allowed_bytes = bytes([b for b in range(256) if b != 10])
+        C_DATA5 = bytes(random.choices(allowed_bytes, k=999983))
+        os_ops.write(filename, C_DATA5, binary=True, truncate=True)
+
+        log_info = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        assert log_info.tail == C_DATA5
+        assert log_info.position == len(C_DATA5)
+
+        # 6. Large data (first_line + many segments)
+        allowed_bytes = bytes([b for b in range(256) if b != 10])
+        os_ops.write(filename, b'abcd\n', binary=True, truncate=True)
+        os_ops.write(filename, C_DATA5, binary=True, truncate=False)
+
+        log_info = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        assert log_info.tail == C_DATA5
+        assert log_info.position == 5 + len(C_DATA5)
+
+        # Cleanup
+        os_ops.remove_file(filename)
+        return