Repository: LUMC/pytest-workflow (language: python)

Issue:
In rare cases, pytest-workflow crashes on log files with special characters
For certain log files with special characters, pytest-workflow crashes due to the seek method that is used output the last X bytes of the log file in case of an error, if it happens to land half way through a special character. I've attached an example log file [here](https://github.com/LUMC/pytest-workflow/files/10399455/log.zip)



The error from pytest workflow
```python
INTERNALERROR> Traceback (most recent call last):
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/main.py", line 268, in wrap_session
INTERNALERROR>     session.exitstatus = doit(config, session) or 0
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/main.py", line 322, in _main
INTERNALERROR>     config.hook.pytest_runtestloop(session=session)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_hooks.py", line 265, in __call__
INTERNALERROR>     return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_manager.py", line 80, in _hookexec
INTERNALERROR>     return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_callers.py", line 60, in _multicall
INTERNALERROR>     return outcome.get_result()
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_result.py", line 60, in get_result
INTERNALERROR>     raise ex[1].with_traceback(ex[2])
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_callers.py", line 39, in _multicall
INTERNALERROR>     res = hook_impl.function(*args)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/main.py", line 347, in pytest_runtestloop
INTERNALERROR>     item.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_hooks.py", line 265, in __call__
INTERNALERROR>     return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_manager.py", line 80, in _hookexec
INTERNALERROR>     return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_callers.py", line 60, in _multicall
INTERNALERROR>     return outcome.get_result()
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_result.py", line 60, in get_result
INTERNALERROR>     raise ex[1].with_traceback(ex[2])
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_callers.py", line 39, in _multicall
INTERNALERROR>     res = hook_impl.function(*args)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/runner.py", line 111, in pytest_runtest_protocol
INTERNALERROR>     runtestprotocol(item, nextitem=nextitem)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/runner.py", line 130, in runtestprotocol
INTERNALERROR>     reports.append(call_and_report(item, "call", log))
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/runner.py", line 221, in call_and_report
INTERNALERROR>     report: TestReport = hook.pytest_runtest_makereport(item=item, call=call)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_hooks.py", line 265, in __call__
INTERNALERROR>     return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_manager.py", line 80, in _hookexec
INTERNALERROR>     return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_callers.py", line 55, in _multicall
INTERNALERROR>     gen.send(outcome)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/skipping.py", line 265, in pytest_runtest_makereport
INTERNALERROR>     rep = outcome.get_result()
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_result.py", line 60, in get_result
INTERNALERROR>     raise ex[1].with_traceback(ex[2])
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pluggy/_callers.py", line 39, in _multicall
INTERNALERROR>     res = hook_impl.function(*args)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/runner.py", line 365, in pytest_runtest_makereport
INTERNALERROR>     return TestReport.from_item_and_call(item, call)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/_pytest/reports.py", line 345, in from_item_and_call
INTERNALERROR>     longrepr = item.repr_failure(excinfo)
INTERNALERROR>   File "/home/user/miniconda3/envs/HAMLET/lib/python3.10/site-packages/pytest_workflow/plugin.py", line 503, in repr_failure
INTERNALERROR>     f"{standout_file.read().strip().decode('utf-8')}")
INTERNALERROR> UnicodeDecodeError: 'utf-8' codec can't decode byte 0x94 in position 0: invalid start byte
 Keeping temporary directories and logs. Use '--kwd' or '--keep-workflow-wd' to disable this behaviour.
```

Script to trigger the bug on the log file, notice that the crash does not happen if you remove the seek
```python
#!/usr/bin/env python3

import sys

with open(sys.argv[1], "rb") as fin:
    fin.seek(-1000, 2)
    print(fin.tell())
    print(fin.read().strip().decode('utf-8'))
```


Discussion / hints:
Good find. Pytest-workflow now also saves the encoding of the file so let me see if this can be fixed.

As a workaround you can set --stderr-bytes to some very high value. The whole log will be printed in that case, but the problem should be averted. Similarly stderr-bytes can be set to 0. In which case the log has to be looked up again, but also the problem is worked around.

Fix patch:
diff --git a/HISTORY.rst b/HISTORY.rst
index 79d9d35c..b1e2168f 100644
--- a/HISTORY.rst
+++ b/HISTORY.rst
@@ -7,6 +7,12 @@ Changelog
 .. This document is user facing. Please word the changes in such a way
 .. that users understand how the changes affect the new version.
 
+version 2.1.0-dev
+---------------------------
++ Fixed a bug where pytest-workflow would crash on logs that used non-ASCII
+  characters where the chunk of size ``--stderr-bytes`` did not properly align
+  with the used encoding.
+
 version 2.0.0
 ---------------------------
 This major release greatly cleans up the output of pytest-workflow in case of


Test patch:
diff --git a/src/pytest_workflow/plugin.py b/src/pytest_workflow/plugin.py
index 85cb148c..1e012d38 100644
--- a/src/pytest_workflow/plugin.py
+++ b/src/pytest_workflow/plugin.py
@@ -30,7 +30,8 @@
 from .content_tests import ContentTestCollector
 from .file_tests import FileTestCollector
 from .schema import WorkflowTest, workflow_tests_from_schema
-from .util import duplicate_tree, is_in_dir, replace_whitespace
+from .util import (decode_unaligned, duplicate_tree, is_in_dir,
+                   replace_whitespace)
 from .workflow import Workflow, WorkflowQueue
 
 
@@ -450,7 +451,10 @@ def collect(self):
         tests += [ExitCodeTest.from_parent(
             parent=self,
             workflow=workflow,
-            stderr_bytes=self.config.getoption("stderr_bytes"))]
+            stderr_bytes=self.config.getoption("stderr_bytes"),
+            stdout_encoding=self.workflow_test.stdout.encoding,
+            stderr_encoding=self.workflow_test.stderr.encoding,
+        )]
 
         tests += [
             FileTestCollector.from_parent(
@@ -476,11 +480,16 @@ def collect(self):
 
 class ExitCodeTest(pytest.Item):
     def __init__(self, parent: pytest.Collector,
-                 workflow: Workflow, stderr_bytes: int):
+                 workflow: Workflow,
+                 stderr_bytes: int,
+                 stdout_encoding: Optional[str] = None,
+                 stderr_encoding: Optional[str] = None):
         name = f"exit code should be {workflow.desired_exit_code}"
         super().__init__(name, parent=parent)
         self.stderr_bytes = stderr_bytes
         self.workflow = workflow
+        self.stdout_encoding = stdout_encoding
+        self.stderr_encoding = stderr_encoding
 
     def runtest(self):
         # workflow.exit_code waits for workflow to finish.
@@ -489,16 +498,21 @@ def runtest(self):
     def repr_failure(self, excinfo, style=None):
         standerr = self.workflow.stderr_file
         standout = self.workflow.stdout_file
-        with open(standout, "rb") as standout_file, \
-             open(standerr, "rb") as standerr_file:
-            if os.path.getsize(standerr) >= self.stderr_bytes:
-                standerr_file.seek(-self.stderr_bytes, os.SEEK_END)
+
+        with open(standout, "rb") as standout_file:
             if os.path.getsize(standout) >= self.stderr_bytes:
                 standout_file.seek(-self.stderr_bytes, os.SEEK_END)
-            message = (f"'{self.workflow.name}' exited with exit code " +
-                       f"'{self.workflow.exit_code}' instead of "
-                       f"'{self.workflow.desired_exit_code}'.\nstderr: "
-                       f"{standerr_file.read().strip().decode('utf-8')}"
-                       f"\nstdout: "
-                       f"{standout_file.read().strip().decode('utf-8')}")
-        return message
+            stdout_text = decode_unaligned(standout_file.read().strip(),
+                                           encoding=self.stdout_encoding)
+        with open(standerr, "rb") as standerr_file:
+            if os.path.getsize(standerr) >= self.stderr_bytes:
+                standerr_file.seek(-self.stderr_bytes, os.SEEK_END)
+            stderr_text = decode_unaligned(standerr_file.read().strip(),
+                                           encoding=self.stderr_encoding)
+
+        return (
+            f"'{self.workflow.name}' exited with exit code " +
+            f"'{self.workflow.exit_code}' instead of "
+            f"'{self.workflow.desired_exit_code}'.\n"
+            f"stderr: {stderr_text}\n"
+            f"stdout: {stdout_text}")
diff --git a/src/pytest_workflow/util.py b/src/pytest_workflow/util.py
index 211e3ab3..a7c91bc0 100644
--- a/src/pytest_workflow/util.py
+++ b/src/pytest_workflow/util.py
@@ -7,7 +7,7 @@
 import sys
 import warnings
 from pathlib import Path
-from typing import Callable, Iterator, List, Set, Tuple, Union
+from typing import Callable, Iterator, List, Optional, Set, Tuple, Union
 
 Filepath = Union[str, os.PathLike]
 
@@ -209,3 +209,17 @@ def file_md5sum(filepath: Path, block_size=64 * 1024) -> str:
         for block in iter(lambda: file_handler.read(block_size), b''):
             hasher.update(block)
     return hasher.hexdigest()
+
+
+def decode_unaligned(data: bytes, encoding: Optional[str] = None):
+    if encoding is None:
+        encoding = sys.getdefaultencoding()
+    for offset in range(4):
+        try:
+            decoded = data[offset:].decode(encoding=encoding, errors="strict")
+            return decoded
+        except UnicodeDecodeError:
+            continue
+    # When no return happens in the loop, decode again. This will throw an
+    # error that is not caught and shown to the user.
+    return data.decode(encoding=encoding)
diff --git a/tests/test_miscellaneous_crashes.py b/tests/test_miscellaneous_crashes.py
index 4c78f47e..5223f9dd 100644
--- a/tests/test_miscellaneous_crashes.py
+++ b/tests/test_miscellaneous_crashes.py
@@ -14,6 +14,10 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with pytest-workflow.  If not, see <https://www.gnu.org/licenses/
 
+import textwrap
+
+from pytest import ExitCode
+
 from .test_success_messages import SIMPLE_ECHO
 
 
@@ -27,3 +31,13 @@ def test_same_name_different_files(pytester):
     conflicting_message = (
         "Conflicting tests: test_b.yml::simple echo, test_a.yml::simple echo.")
     assert conflicting_message in result.stdout.str()
+
+
+def test_non_ascii_logs_stderr_bytes(pytester):
+    test = textwrap.dedent("""
+    - name: print non-ascii
+      command: bash -c 'printf èèèèèèèèè && exit 1'
+    """)
+    pytester.makefile(".yml", test_non_ascii=test)
+    result = pytester.runpytest("--stderr-bytes", "7")
+    assert result.ret == ExitCode.TESTS_FAILED
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 005d5a72..45f789c3 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -14,16 +14,18 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with pytest-workflow.  If not, see <https://www.gnu.org/licenses/
 import hashlib
+import itertools
 import os
 import shutil
 import subprocess
+import sys
 import tempfile
 from pathlib import Path
 
 import pytest
 
-from pytest_workflow.util import duplicate_tree, file_md5sum, \
-    git_check_submodules_cloned, git_root, \
+from pytest_workflow.util import decode_unaligned, duplicate_tree, \
+    file_md5sum, git_check_submodules_cloned, git_root, \
     is_in_dir, link_tree, replace_whitespace
 
 WHITESPACE_TESTS = [
@@ -227,3 +229,20 @@ def test_duplicate_git_tree_submodule_symlinks(git_repo_with_submodules):
     assert link.exists()
     assert link.is_symlink()
     assert link.resolve() == dest / "bird" / "sub"
+
+
+@pytest.mark.parametrize(["offset", "encoding"],
+                         list(itertools.product(
+                             range(4), (None, "utf-8", "utf-16", "utf-32"))
+                         ))
+def test_decode_unaligned(offset, encoding):
+    string = "èèèèèèèèèèè"
+    data = string.encode(encoding or sys.getdefaultencoding())
+    decoded = decode_unaligned(data[offset:], encoding)
+    assert string.endswith(decoded)
+
+
+def test_decode_unaligned_wrong_encoding_throws_error():
+    data = "hello".encode("utf-8")
+    with pytest.raises(UnicodeDecodeError):
+        decode_unaligned(data, "utf-32-le")