Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions src/impl/file_line_reader.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
# coding: utf-8

from . import internal_utils

import typing

from testgres.operations.os_ops import OsOperations
Expand All @@ -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 ----------------------------------------------------------
Expand Down
72 changes: 72 additions & 0 deletions src/impl/internal_utils.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
from testgres.operations.os_ops import OsOperations

import logging
import typing


def send_log(level: int, msg: str) -> None:
Expand All @@ -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
56 changes: 6 additions & 50 deletions src/node.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down
85 changes: 78 additions & 7 deletions tests/units/impl/test_file_line_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ class tagStep:
write_data: bytes
read_lines: typing.List[typing.Optional[str]]

# --------------------------------------------------------------------
sm_Steps001: typing.List[tagStep] = [
tagStep(
b"",
Expand Down Expand Up @@ -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)

Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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