diff --git a/pyproject.toml b/pyproject.toml index cd649fed..1a790e80 100644 --- a/pyproject.toml +++ b/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] diff --git a/src/node.py b/src/node.py index c60c82dc..3359b095 100644 --- a/src/node.py +++ b/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 diff --git a/tests/requirements.txt b/tests/requirements.txt index b035e99a..c4adb2c6 100644 --- a/tests/requirements.txt +++ b/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 diff --git a/tests/test_testgres_common.py b/tests/test_testgres_common.py index ac0dc90b..d71908b0 100644 --- a/tests/test_testgres_common.py +++ b/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,)]) diff --git a/tests/units/node/PostgresNodeLogReader/__init__.py b/tests/units/node/PostgresNodeLogReader/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/units/node/PostgresNodeLogReader/test_setM001__helper_create_log_info.py b/tests/units/node/PostgresNodeLogReader/test_setM001__helper_create_log_info.py new file mode 100644 index 00000000..73267c65 --- /dev/null +++ b/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