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
5 changes: 4 additions & 1 deletion src/firebird/driver/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -5767,7 +5767,10 @@ def readline(self) -> str | Sentinel | None:
With the default `.query_timeout` this method blocks until output is available.
"""
while True:
if self.__line_buffer and (self.__line_buffer[0].endswith('\n') or self._eof):
# Only the last buffered piece can be an incomplete line; earlier ones ended on
# another `splitlines()` boundary (e.g. a bare '\r' inside SQL text).
if self.__line_buffer and (len(self.__line_buffer) > 1
or self.__line_buffer[0].endswith('\n') or self._eof):
return self.__line_buffer.pop(0)
if self._eof:
return None
Expand Down
15 changes: 15 additions & 0 deletions tests/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,21 @@ def test_text_output_timeout_preserves_partial_line(mode):
finally:
server.close()

@pytest.mark.parametrize('timeout', [-1, 1])
def test_text_output_non_lf_line_boundary_keeps_following_lines(timeout):
# str.splitlines() also breaks lines on CR, FF and other non-LF separators, which can
# occur in SQL text in trace output. Lines buffered after such a piece must survive.
responses = [output_response(SrvInfoCode.TO_EOF, 'select 1 /* a\rb */ from x\nnext\n', 2),
output_response(SrvInfoCode.TO_EOF, 'last\n', 2),
output_response(SrvInfoCode.TO_EOF)]
server = driver.Server(OutputService(responses), b'', 'localhost', 'utf8', 'strict')
server.mode = SrvInfoCode.TO_EOF
server.query_timeout = timeout
try:
assert ''.join(server.readlines()) == 'select 1 /* a\rb */ from x\nnext\nlast\n'
finally:
server.close()

def test_get_limbo_transaction_ids(server_connection, db_file):
pytest.skip('Not implemented yet')
ids = server_connection.database.get_limbo_transaction_ids(database=str(db_file))
Expand Down