From 2f405db907addc20cee8abfcc0a35415d245948b Mon Sep 17 00:00:00 2001 From: "F.D.Castel" Date: Sat, 10 Oct 2026 00:01:35 -0300 Subject: [PATCH] Fix Server.readline() dropping buffered lines after a non-LF line boundary readline() treated any buffered piece not ending in '\n' as an incomplete line and re-read it via _read_output(init=...), which replaces the whole line buffer. str.splitlines() also breaks on CR, FF and other non-LF separators, so a bare CR inside SQL text in trace output discarded every line buffered after it. Only the last buffered piece can be incomplete. Refs #70. --- src/firebird/driver/core.py | 5 ++++- tests/test_server.py | 15 +++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/firebird/driver/core.py b/src/firebird/driver/core.py index 0c54b01..e7c6dec 100644 --- a/src/firebird/driver/core.py +++ b/src/firebird/driver/core.py @@ -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 diff --git a/tests/test_server.py b/tests/test_server.py index 900a336..34a2aac 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -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))