Repository: byllyfish/shellous (language: python)

Issue:
Consider use of incremental codec for decoding output
The current implementation use a block codec for decoding output (e.g. from utf-8). This can present a problem if the read boundary occurs within a multi-byte char.


Discussion / hints:

Fix patch:
diff --git a/shellous/runner.py b/shellous/runner.py
index 3e26e49d..87ec4cc0 100644
--- a/shellous/runner.py
+++ b/shellous/runner.py
@@ -206,7 +206,7 @@ def _setup_input(self, input_, close, encoding):
         else:
             if encoding is None:
                 raise TypeError("when encoding is None, input must be bytes")
-            input_bytes = input_.encode(encoding)
+            input_bytes = input_.encode(*encoding.split(maxsplit=1))
 
         return stdin, input_bytes
 


Test patch:
diff --git a/tests/test_shellous.py b/tests/test_shellous.py
index 47e19934..b5513ba9 100644
--- a/tests/test_shellous.py
+++ b/tests/test_shellous.py
@@ -433,6 +433,14 @@ async def test_task():
         await task
 
 
+async def test_encoding_utf8(cat_cmd):
+    "Test use of encoding option with bad utf-8 data."
+
+    cat = cat_cmd.set(encoding="utf-8")
+    with pytest.raises(UnicodeDecodeError, match="invalid start byte"):
+        await (b"\x81abc" | cat)
+
+
 async def test_encoding_utf8_strict(cat_cmd):
     "Test use of encoding option with bad utf-8 data."
 
@@ -449,6 +457,54 @@ async def test_encoding_utf8_replace(cat_cmd):
     assert result == "\ufffdabc"
 
 
+async def test_encoding_utf8_replace_str(cat_cmd):
+    "Test use of encoding option with utf-8 data."
+
+    cat = cat_cmd.set(encoding="utf-8 replace")
+    result = await ("abc" | cat)
+    assert result == "abc"
+
+
+async def test_encoding_ascii_str(cat_cmd):
+    "Test use of encoding option with ascii data."
+
+    cat = cat_cmd.set(encoding="ascii")
+    result = await ("abc" | cat)
+    assert result == "abc"
+
+
+async def test_encoding_utf8_split(cat_cmd):
+    "Test reconstitution of split utf-8 chars in output."
+
+    buf = io.StringIO()
+    cmd = CAPTURE | cat_cmd | buf
+
+    async with cmd.run() as run:
+        # Not split.
+        run.stdin.write(b"\xf0\x9f\x90\x9f")  # "\U0001F41F" in utf-8
+        await run.stdin.drain()
+        await asyncio.sleep(0.2)
+
+        # Split in half.
+        run.stdin.write(b"\xf0\x9f")
+        await run.stdin.drain()
+        await asyncio.sleep(0.1)
+
+        run.stdin.write(b"\x90\x9f")
+        await run.stdin.drain()
+        await asyncio.sleep(0.1)
+
+        # Split up one byte at a time.
+        for item in b"\xf0\x9f\x90\x9f":
+            run.stdin.write(bytes([item]))
+            await run.stdin.drain()
+            await asyncio.sleep(0.1)
+
+        run.stdin.close()
+
+    assert buf.getvalue() == "\U0001F41F" * 3
+
+
 async def test_many_short_programs_sequential(echo_cmd):
     "Test many short programs (sequential)."
     COUNT = 10