refactoring: internal_utils.read_line_to_pos__bin is added (#420)

- PostgresNodeLogReader::_create_log_info uses it
- impl.FileLineReader now supports a start from any position of file

Author: dmitry-lipetsk

src/impl/file_line_reader.py

@@ -1,5 +1,7 @@
 # coding: utf-8
 
+from . import internal_utils
+
 import typing
 
 from testgres.operations.os_ops import OsOperations
@@ -19,15 +21,20 @@ def __init__(
         os_ops: OsOperations,
         file_name: str,
         file_encoding: str = "utf-8",
+        file_pos: int = 0
     ):
         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._file_pos = file_pos
         self._buffer_pos = 0
-        self._buffer = b''
+        self._buffer = internal_utils.read_line_to_pos__bin(
+            os_ops,
+            file_name,
+            file_pos,
+        )
         return
 
     # interface ----------------------------------------------------------

src/impl/internal_utils.py

@@ -1,4 +1,7 @@
+from testgres.operations.os_ops import OsOperations
+
 import logging
+import typing
 
 
 def send_log(level: int, msg: str) -> None:
@@ -18,3 +21,72 @@ def send_log_debug(msg: str) -> None:
     assert type(msg) is str
 
     return send_log(logging.DEBUG, msg)
+
+
+def read_line_to_pos__bin(
+    os_ops: OsOperations,
+    filename: str,
+    position: int,
+) -> bytes:
+    assert type(filename) is str
+    assert type(position) is int
+    assert len(filename) > 0
+    assert position >= 0
+    assert os_ops is not None
+    assert isinstance(os_ops, OsOperations)
+
+    if position == 0:
+        return b''
+
+    assert position > 0
+
+    read_position = position
+    result_blocks: typing.List[bytes] = []
+
+    C_BACK_READ_BLOCK_SIZE = 4096
+
+    while read_position > 0:
+        if read_position < C_BACK_READ_BLOCK_SIZE:
+            block_sz = read_position
+        else:
+            block_sz = C_BACK_READ_BLOCK_SIZE
+
+        assert block_sz > 0
+        assert block_sz <= C_BACK_READ_BLOCK_SIZE
+
+        read_position -= block_sz
+
+        assert read_position < position
+        assert read_position >= 0
+
+        block = os_ops.read_binary(filename, read_position, 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:
+            result_blocks.append(block)
+            continue
+
+        if x == block_sz - 1:
+            break
+
+        block = block[x + 1:]
+        result_blocks.append(block)
+        break
+
+    result = b''.join(reversed(result_blocks))
+    assert type(result) is bytes
+    assert len(result) <= (position - read_position)
+    return result

src/node.py

@@ -2507,7 +2507,7 @@ def read(self) -> typing.List[LogDataBlock]:
 
         assert type(self._logs) is dict
 
-        result = list()
+        result: typing.List[__class__.LogDataBlock] = []
 
         for file_name, cur_log_info in cur_logs.items():
             assert type(file_name) is str
@@ -2624,55 +2624,11 @@ def _create_log_info(
                 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))
+        tail = internal_utils.read_line_to_pos__bin(
+            os_ops,
+            filename,
+            file_size,
+        )
 
         return __class__.LogInfo(
             position=file_size,

tests/units/impl/test_file_line_reader.py

@@ -38,6 +38,7 @@ class tagStep:
         write_data: bytes
         read_lines: typing.List[typing.Optional[str]]
 
+    # --------------------------------------------------------------------
     sm_Steps001: typing.List[tagStep] = [
         tagStep(
             b"",
@@ -91,14 +92,80 @@ class tagStep:
 
     ]
 
+    # -------------------------------------------------------------------
+    def test_001__from_beginnig(
+        self,
+        os_ops_descr: OsOpsDescr,
+    ):
+        assert type(os_ops_descr) is OsOpsDescr
+        assert isinstance(os_ops_descr.os_ops, OsOperations)
+
+        __class__.helper__player(
+            os_ops_descr,
+            __class__.sm_Steps001,
+            0,
+        )
+        return
+
     # --------------------------------------------------------------------
-    def test_001__common(
+    sm_Steps002: typing.List[tagStep] = [
+        tagStep(
+            b"abc\ndefg\n",
+            ["abc\n", "defg\n", None]
+        ),
+    ]
+
+    # -------------------------------------------------------------------
+    def test_002__from_1(
         self,
         os_ops_descr: OsOpsDescr,
     ):
         assert type(os_ops_descr) is OsOpsDescr
         assert isinstance(os_ops_descr.os_ops, OsOperations)
 
+        __class__.helper__player(
+            os_ops_descr,
+            __class__.sm_Steps002,
+            1,
+        )
+        return
+
+    # --------------------------------------------------------------------
+    sm_Steps003: typing.List[tagStep] = [
+        tagStep(
+            b"abc\ndefg\n",
+            ["defg\n", None]
+        ),
+    ]
+
+    # -------------------------------------------------------------------
+    def test_003__from_second_line(
+        self,
+        os_ops_descr: OsOpsDescr,
+    ):
+        assert type(os_ops_descr) is OsOpsDescr
+        assert isinstance(os_ops_descr.os_ops, OsOperations)
+
+        __class__.helper__player(
+            os_ops_descr,
+            __class__.sm_Steps003,
+            4,
+        )
+        return
+
+    # --------------------------------------------------------------------
+    @staticmethod
+    def helper__player(
+        os_ops_descr: OsOpsDescr,
+        steps: typing.List[tagStep],
+        initial_pos: int,
+    ):
+        assert type(os_ops_descr) is OsOpsDescr
+        assert isinstance(os_ops_descr.os_ops, OsOperations)
+        assert type(steps) is list
+        assert type(initial_pos) is int
+        assert initial_pos >= 0
+
         os_ops = os_ops_descr.os_ops
         assert isinstance(os_ops_descr.os_ops, OsOperations)
 
@@ -110,23 +177,27 @@ def test_001__common(
         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",
-        )
+        file_line_reader: typing.Optional[FileLineReader] = None
 
         # -----------------------
         nStep = 0
 
-        for step in __class__.sm_Steps001:
+        for step in steps:
             nStep += 1
 
             logging.info("-------------------- step: {}".format(nStep))
 
             logging.info("write: {}".format(step.write_data))
             os_ops.write(filename, step.write_data, binary=True)
 
+            if file_line_reader is None:
+                file_line_reader = FileLineReader(
+                    os_ops,
+                    filename,
+                    file_encoding="utf-8",
+                    file_pos=initial_pos,
+                )
+
             nRead = 0
             for expected_line in step.read_lines:
                 nRead += 1

tests/units/node/PostgresNodeLogReader/test_setM001__helper_create_log_info.py

@@ -100,6 +100,13 @@ def test_001__common(self, os_ops_descr: OsOpsDescr):
         assert log_info.tail == C_DATA5
         assert log_info.position == 5 + len(C_DATA5)
 
+        # Scenario 7: The log file has one line with the normal end
+        os_ops.write(filename, b"\n", binary=True, truncate=True)
+
+        log_info = PostgresNodeLogReader._create_log_info(os_ops, filename, find_line_start=True)
+        assert log_info.tail == b""
+        assert log_info.position == 1
+
         # Cleanup
         os_ops.remove_file(filename)
         return