Fix decoding error (issue 168)

[pevogam]

By decoding incomplete byte-input, we risk losing multi-byte characters, if their byte-representation is not aligned with the end of our buffer.

Fix this by concatenating bytes first, and only decode when we have to.

In case of Tail that is a little complicated, because we need to return text in-between read() calls. Outsource to a helper function with lots of documentation to clarify and reduce complexity.

Clarify error policy for decode: why not switch from "ignore" to "replace" to notify callers of decoding problems instead of hiding them?

Original author: Christian Herdtweck christian.herdtweck@intra2net.com

[pevogam]

@christian-intra2net Thanks for contributing! Before any review though - have you run the unit tests? I believe they also need adaptation.

[christian-intra2net]

Oops, sorry, I was not aware of unit tests. I think I found the problem, fixing it now...

[pevogam]

Great to see the CI passing, to clarify for everyone else - this PR fixes issue #168.

[pevogam]

This is an initial review from me, overall this pull request fixes a really non-trivial issue!

[pevogam]

I assume from your comment in #168 (comment) this could be instead be turned into replace and thus provide the clarity you mentioned there. So let's see if we collect some feedback there on the original choices first and until then a "replace" setting here would rather be a requested change for additional improvement there here.

[ldoktor]

I'd suggest another commit to do mass ignore->replace which could be reverted in case someone depended on the ignore.

[pevogam]

Definitely better not to decode raw data until the very end, I think this change improves the clarity and related better to the choice of naming.

[pevogam]

Maybe we could add a few unit tests for what behavior should be contracted with this function?

[pevogam]

Maybe there is also a better location for it e.g. in utils folder or something like that since the current module is purely structuring the classes in order of composition.

[ldoktor]

yep, utils.astring would be IMO the best location

[ldoktor]

This seems quite complex, how about:

def partial_decode(input_bytes, encoding):
    """
    Helper for decoding as many bytes as possible, returning last "broken"
    bytes.

    :param input_bytes: Encoded input text
    :param encoding: Target encoding
    :returns: tuple(encoded text, left-over bytes)
    """
    decoder = codecs.getincrementaldecoder(encoding)(errors="ignore")
    text = decoder.decode(input_bytes, final=False)
    leftover, _ = decoder.getstate()
    return text, leftover

[ldoktor]

(and ideally in utils.astring as it's string manipulation)

[ldoktor]

Thanks and kudos for the analysis. I'd suggest using the incremental decoder instead but apart from that it's really welcome bugfix. Please include a test for various cases.

[christian-intra2net]

You are completely right, I did not know about the IncrementalDecoder, so I basically re-implemented its functionality. Using such an object, we can get rid of much more of my code, I will make the modifications early next week.

[christian-intra2net]

I'd suggest using the incremental decoder instead

I did that, reduces the code changes a lot

Please include a test for various cases.

I created a stand-alone test-script for development and testing but I guess you'd prefer some kind of unittest. It would have to spawn some process that produces byte-output, at least for linux I could piece that together.

[christian-intra2net]

I have converted my stand-alone test script to a unittest. It tests both code locations that were changed, filling the buffers of Tail and ShellSession with multibyte character strings of different alignments to force the issue of incomplete reads. In my tests, these tests succeed on the current branch but fail on main (after replacing the new buffer size const with a literal 1024).

Thanks @pevogam for pushing this PR

[ldoktor]

To be consistent this should return data.decode(self.encoding, "ignore").

[ldoktor]

Thanks for the selftests, originally I though about simple unit-tests, but this is probably better/safer, although I have to say it looks quite complex. How about something like this:

diff --git a/aexpect/client.py b/aexpect/client.py
index be3bcde..e2d7752 100644
--- a/aexpect/client.py
+++ b/aexpect/client.py
@@ -913,7 +913,7 @@ class Expect(Tail):
             try:
                 poll_status = poller.poll(internal_timeout)
             except select.error:
-                return read, data
+                return read, data.decode(self.encoding, "ignore")
             if poll_status:
                 raw_data = os.read(expect_pipe, READ_BUFFER_SIZE)
                 if not raw_data:
diff --git a/tests/test_client.py b/tests/test_client.py
index e2aa52a..8055a40 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -197,153 +197,84 @@ class CommandsTests(unittest.TestCase):
 
 class EncodingTest(unittest.TestCase):
 
-    DEBUG = False
-
-    # Encoding used to translate between Unicode text and bytes
-    ENCODING = "utf-8"
-
-    # text whose characters decode to multiple byte
     TEXT = "嗨😀"
-
-    REPETITIONS_FOR_TAIL = 3
-
     MAX_OFFSET = 10
 
-    def analyze_output(self, offset, new_output):
-        """Helper; Compare output to expectation"""
-        # remove the leading offset whitespace
-        idx = 0
-        for idx, char in enumerate(new_output):
-            if char.isspace():
-                continue
-            if char == self.TEXT[0]:
-                break
-            self.fail(
-                f"Unexpected char found at {idx=}: {char!r} ({char.encode(self.ENCODING)}). "
-                f"Line start: {new_output[:50]}, line length: {len(new_output)}"
-            )
-        if idx == len(new_output):
-            self.fail("Test text not found!")
-        if idx > 0:
-            new_output = new_output[idx:]
-        if self.DEBUG:
-            print(f"Skipping {idx} whitespace chars at start")
-
-        # print start and end, count chars
-        n_chars = len(new_output)
-        if self.DEBUG:
-            print(f"Output for offset {offset}: len={n_chars}.")
-            for idx, char in enumerate(new_output[:3]):
-                print_char = chr(0x21B2) if char == "\n" else char
-                print(
-                    f"char {idx}: {print_char} ({char.encode(self.ENCODING)})",
-                    end="; ",
-                )
-            print("...", end="")
-            for idx, char in enumerate(new_output[-3:]):
-                print_char = chr(0x21B2) if char == "\n" else char
-                print(
-                    f"char {n_chars-3+idx}: {print_char} ({char.encode(self.ENCODING)})",
-                    end="; ",
-                )
-            print()
-        return n_chars
-
-    def analyze_results(self, all_lengths: list):
-        """Helper: compare results, decide whether test was successful"""
-        if not all_lengths:
-            self.fail("no successful output analyses")
-        expect = all_lengths[0]
-        if any(curr_length != expect for curr_length in all_lengths[1:]):
-            self.fail("There were differences in encoded output lengths")
-        elif self.DEBUG:
-            print("SUCCESS")
+    def _multibyte_write_cmd(self, offset, count=1):
+        """Build a Python command that writes multibyte text to stdout."""
+        encoded = self.TEXT.encode("utf-8")
+        reps = 1024 // len(encoded) + 1
+        writes = "; ".join(["f.write(t); f.flush()"] * count)
+        return (
+            f"import os,sys; t=b' '*{offset}+{encoded!r}*{reps}+b'\\n'; "
+            f"f=os.fdopen(sys.stdout.fileno(),'wb',closefd=False); {writes}"
+        )
 
     @unittest.skipUnless(os.name == "posix", "Unix/Linux/macOS only")
     def test_shell(self):
-        """
-        Tests correct decoding of multibyte characters in ShellSession.
-
-        Even if reading is interrupted with incomplete characters, we
-        expect correct output.
-
-        Spawns a python session that produces multibyte output
-        with various single-byte offsets.
-        """
+        """Test multibyte decoding in ShellSession across buffer boundaries."""
         sess = client.ShellSession("/bin/sh")
-        sess.cmd_output(
-            "echo 'Just removing potential initial prompt from output'"
-        )
-        all_lengths = []
-        output = self.TEXT.encode(self.ENCODING)
-        repetitions = 1024 // len(output) + 1
+        sess.cmd_output("echo init")
+        lengths = []
         for offset in range(self.MAX_OFFSET):
-            if self.DEBUG:
-                print(f"Start testing with shell and offset {offset}")
-            cmd = (
-                f"import os; import sys; t=b' '*{offset}+{output!r}*{repetitions}+b'\\n'; "
-                f"f=os.fdopen(sys.stdout.fileno(), 'wb', closefd=False); f.write(t); f.flush()"
+            cmd = self._multibyte_write_cmd(offset)
+            result = sess.cmd_output(
+                f'{sys.executable} -c "{cmd}"'
+            ).lstrip()
+            self.assertTrue(
+                result.startswith(self.TEXT),
+                f"offset {offset}: unexpected start: {result[:20]!r}",
             )
-            new_output = sess.cmd_output(f'{sys.executable} -c "{cmd}"')
-            all_lengths.append(self.analyze_output(offset, new_output))
+            lengths.append(len(result))
         sess.close()
-        self.analyze_results(all_lengths)
+        self.assertTrue(lengths, "No output collected")
+        self.assertEqual(
+            len(set(lengths)), 1,
+            f"Output lengths vary across offsets: {lengths}",
+        )
 
+    @unittest.skipUnless(os.name == "posix", "Unix/Linux/macOS only")
     def test_tail(self):
-        """
-        Tests correct decoding of multibyte characters in Tail.
-
-        Like test_shell, but using a Tail and repeating the output to get
-        multiple lines of output. Requires custom output gatherer and
-        termination function
-        """
-        output_buffer = []
-        terminated = False
+        """Test multibyte decoding in Tail across buffer boundaries."""
+        tail_lines = 3
+        lengths = []
+        for offset in range(self.MAX_OFFSET):
+            output_buffer = []
+            terminated = False
 
-        def remember_output(new_output):
-            nonlocal output_buffer
-            output_buffer.append(new_output)
+            def on_output(text):
+                nonlocal output_buffer
+                output_buffer.append(text)
 
-        def termination_func(_status):
-            nonlocal terminated
-            terminated = True
+            def on_terminate(_status):
+                nonlocal terminated
+                terminated = True
 
-        output = self.TEXT.encode(self.ENCODING)
-        repetitions = 1024 // len(output) + 1
-        all_lengths = []
-        for offset in range(self.MAX_OFFSET):
-            terminated = False
-            output_buffer = []
-            cmd = (
-                f"import os; import sys; t=b' '*{offset}+{output!r}*{repetitions}+b'\\n';"
-                f"f=os.fdopen(sys.stdout.fileno(), 'wb', closefd=False); f.write(t); f.flush()"
-            )
-            for _ in range(self.REPETITIONS_FOR_TAIL - 1):
-                cmd += "; f.write(t); f.flush()"
-            if self.DEBUG:
-                print("Spawning Tail")
-            python = client.Tail(
+            cmd = self._multibyte_write_cmd(offset, count=tail_lines)
+            tail = client.Tail(
                 f'{sys.executable} -c "{cmd}"',
-                output_func=remember_output,
-                termination_func=termination_func,
+                output_func=on_output,
+                termination_func=on_terminate,
             )
-            if self.DEBUG:
-                print(f"Listening for subproc {python.get_pid()}")
             for _ in range(1000):
                 if terminated:
                     break
-                if self.DEBUG:
-                    print(".", end="", flush=True)
                 time.sleep(0.01)
-            if self.DEBUG:
-                print("\nDone")
-            python.close()
+            tail.close()
             for line in output_buffer:
                 if line.startswith("(Process terminated "):
                     continue
-                all_lengths.append(self.analyze_output(offset, line))
-
-        self.analyze_results(all_lengths)
+                stripped = line.lstrip()
+                self.assertTrue(
+                    stripped.startswith(self.TEXT),
+                    f"offset {offset}: unexpected start: {stripped[:20]!r}",
+                )
+                lengths.append(len(stripped))
+        self.assertTrue(lengths, "No output collected")
+        self.assertEqual(
+            len(set(lengths)), 1,
+            f"Output lengths vary across offsets: {lengths}",
+        )
 
 
 if __name__ == "__main__":