Fix Server text output splitting multi-byte characters across service responses

src/firebird/driver/core.py

@@ -43,6 +43,7 @@
 from __future__ import annotations
 
 import atexit
+import codecs
 import contextlib
 import datetime
 import decimal
@@ -5622,6 +5623,7 @@ def __init__(self, svc: iService, spb: bytes, host: str, encoding: str,
         "Response buffer used to communicate with service"
         self._eof: bool = False
         self.__line_buffer: list[str] = []
+        self.__decoder: codecs.IncrementalDecoder | None = None
         self.encoding: str = encoding
         "Encoding used for text data exchange with server"
         self.encoding_errors: str = encoding_errors
@@ -5655,6 +5657,7 @@ def _engine_version(self) -> float:
     def _reset_output(self) -> None:
         self._eof = False
         self.__line_buffer.clear()
+        self.__decoder = None
     def _make_request(self, timeout: int) -> bytes:
         if timeout == -1:
             return None
@@ -5686,7 +5689,14 @@ def _query_output(self, timeout: int) -> str | Sentinel:
             return TIMEOUT
         if tag != self.mode:  # pragma: no cover
             raise InterfaceError(f"Service responded with error code: {tag}")
-        return self.response.read_sized_string(encoding=self.encoding, errors=self.encoding_errors)
+        if self.__decoder is None:
+            self.__decoder = codecs.getincrementaldecoder(self.encoding)(self.encoding_errors)
+        # The service fills responses by size, so one can end inside a multi-byte character.
+        # The incremental decoder keeps those bytes until the next response completes it.
+        return self.__decoder.decode(self.response.read_bytes())
+    def _flush_output(self) -> str:
+        """Decode bytes left from an unfinished character at the end of output or of a line."""
+        return self.__decoder.decode(b'', final=True)
     def _read_output(self, *, init: str='') -> bool:
         """Fetch output into the line buffer; return True if the query timed out."""
         data = self._query_output(self.query_timeout)
@@ -5697,6 +5707,8 @@ def _read_output(self, *, init: str='') -> bool:
         timed_out = tag == isc_info_svc_timeout
         if self.mode is SrvInfoCode.TO_EOF:
             self._eof = tag == isc_info_end
+            if self._eof:
+                data += self._flush_output()
         else: # LINE mode
             while tag == isc_info_truncated:
                 part = self._query_output(self.query_timeout)
@@ -5709,6 +5721,7 @@ def _read_output(self, *, init: str='') -> bool:
             if not timed_out and tag != isc_info_end:  # pragma: no cover
                 raise InterfaceError("Malformed result buffer (missing isc_info_end item)")
             if not timed_out:
+                data += self._flush_output()
                 self._eof = not data
         init += data
         if data and self.mode is SrvInfoCode.LINE and not timed_out:

tests/test_server.py

@@ -192,7 +192,7 @@ def detach(self):
 
 
 def output_response(mode, data='', tag=1):
-    encoded = data.encode('utf8')
+    encoded = data if isinstance(data, bytes) else data.encode('utf8')
     return bytes([mode]) + len(encoded).to_bytes(2, 'little') + encoded + bytes([tag])
 
 
@@ -269,6 +269,42 @@ def test_text_output_non_lf_line_boundary_keeps_following_lines(timeout):
     finally:
         server.close()
 
+@pytest.mark.parametrize('mode', [SrvInfoCode.LINE, SrvInfoCode.TO_EOF])
+@pytest.mark.parametrize('timeout', [-1, 1])
+@pytest.mark.parametrize('errors', ['strict', 'replace'])
+def test_text_output_multibyte_character_split_across_responses(mode, timeout, errors):
+    # The service fills each response by size, so a response can end inside a multi-byte
+    # character. The character must survive, also when a timeout falls between the halves.
+    line = 'a\xe7\xe3o'
+    encoded = line.encode('utf8')
+    head, tail = encoded[:2], encoded[2:]  # split inside the 2-byte '\xe7'
+    if mode is SrvInfoCode.TO_EOF:
+        tail += b'\n'
+    responses = [output_response(mode, head, 2)]
+    if timeout > 0:
+        responses.append(bytes([SrvInfoCode.TIMEOUT]))
+    responses.extend([output_response(mode, tail), output_response(mode)])
+    server = driver.Server(OutputService(responses), b'', 'localhost', 'utf8', errors)
+    server.mode = mode
+    server.query_timeout = timeout
+    try:
+        assert server.readlines() == [line + '\n']
+    finally:
+        server.close()
+
+@pytest.mark.parametrize('mode', [SrvInfoCode.LINE, SrvInfoCode.TO_EOF])
+def test_text_output_incomplete_character_at_end_uses_error_handler(mode):
+    # A character still incomplete when the output (or a LINE-mode line) ends is not
+    # silently dropped; the error handler decides, as for any other invalid input.
+    responses = [output_response(mode, b'ab\xc3'), output_response(mode)]
+    server = driver.Server(OutputService(responses), b'', 'localhost', 'utf8', 'replace')
+    server.mode = mode
+    try:
+        expected = 'ab' + chr(0xFFFD)
+        assert server.readlines() == [expected + '\n' if mode is SrvInfoCode.LINE else expected]
+    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))