Merge pull request #342 from ales-erjavec/fixes/subprocess-encoding-error

[FIX] Addons: Install encoding error

Author: janezd

i18n/si/msgs.jaml

@@ -1326,6 +1326,15 @@ application/utils/addons.py:
     def `run_command`:
         Running %s: false
         ' ': false
+        env: false
+        PYTHONIOENCODING: false
+        utf-8: false
+        PYTHONUTF8: false
+        1: false
+        encoding: false
+        errors: false
+        backslashreplace: false
+        universal_newlines: false
         python: false
     def `run_process`:
         subprocess.Popen: false
@@ -3780,6 +3789,7 @@ utils/shtools.py:
         nt: false
     def `temp_named_file`:
         utf-8: false
+        wb: false
         wt: false
 utils/localization/__init__.py:
     import 'orangecanvas.localization', not 'orangecanvas.utils.localization': false

orangecanvas/application/tests/test_addons_utils.py

@@ -1,6 +1,8 @@
+import io
 import os
 import stat
 import unittest
+from shutil import which
 from tempfile import mkdtemp
 
 from requests import Session
@@ -14,9 +16,10 @@
     installable_from_json_response,
     installable_items,
     is_updatable,
-    prettify_name, _session,
+    prettify_name, _session, run_command,
 )
 from orangecanvas.application.tests.test_addons import FakeDistribution
+from orangecanvas.utils.shtools import temp_named_file
 
 
 class TestUtils(unittest.TestCase):
@@ -112,6 +115,13 @@ def test_session(self):
         os.chmod(temp_dir, stat.S_IRUSR)
         self.assertIsInstance(_session(temp_dir), Session)
 
+    @unittest.skipIf(which("cat") is None, "'cat' not present")
+    def test_run_command_decode_errors(self):
+        f = io.StringIO()
+        with temp_named_file(b'\x00' * 16, encoding=None) as fname:
+            run_command(["cat", fname], file=f)
+        self.assertEqual("\x00" * 16, f.getvalue())
+
 
 if __name__ == "__main__":
     unittest.main()

orangecanvas/application/utils/addons.py

@@ -639,20 +639,30 @@ def __bool__(self):
         return bool(self.conda)
 
 
-def run_command(command, raise_on_fail=True, **kwargs):
-    # type: (List[str], bool, Any) -> Tuple[int, List[AnyStr]]
+def run_command(
+        command: List[str], raise_on_fail: bool = True,/, file=None,  **kwargs
+) -> Tuple[int, List[str]]:
     """
     Run command in a subprocess.
 
     Return `process` return code and output once it completes.
     """
     log.info("Running %s", " ".join(command))
 
+    env = kwargs.pop("env", os.environ).copy()
+    env["PYTHONIOENCODING"] = "utf-8"
+    env["PYTHONUTF8"] = "1"
+    kwargs["env"] = env
+
+    kwargs.setdefault("encoding", "utf-8")
+    kwargs.setdefault("errors", "backslashreplace")
+    kwargs.setdefault("universal_newlines", True)
+
     if command[0] == "python":
         process = python_process(command[1:], **kwargs)
     else:
         process = create_process(command, **kwargs)
-    rcode, output = run_process(process, file=sys.stdout)
+    rcode, output = run_process(process, file=file)
     if rcode != 0 and raise_on_fail:
         raise CommandFailed(command, rcode, output)
     else:

orangecanvas/utils/shtools.py

@@ -81,7 +81,6 @@ def create_process(
         executable: Optional[str] = None,
         stderr=subprocess.STDOUT,
         stdout=subprocess.PIPE,
-        universal_newlines=True,
         **kwargs
 ) -> subprocess.Popen:
     """
@@ -99,14 +98,13 @@ def create_process(
         executable=executable,
         stderr=stderr,
         stdout=stdout,
-        universal_newlines=universal_newlines,
         **kwargs
     )
 
 
 @contextmanager
 def temp_named_file(
-        content: str, encoding="utf-8",
+        content: str | bytes, encoding: str | None = "utf-8",
         suffix: Optional[str] = None,
         prefix: Optional[str] = None,
         dir: Optional[str] = None,
@@ -118,7 +116,7 @@ def temp_named_file(
 
     Parameters
     ----------
-    content: str
+    content: str | bytes
         The contents to write into the temp file
     encoding: str
         Encoding
@@ -141,7 +139,8 @@ def temp_named_file(
     tempfile.mkstemp
     """
     fd, name = tempfile.mkstemp(suffix, prefix, dir=dir, text=True)
-    file = os.fdopen(fd, mode="wt", encoding=encoding,)
+    mode = "wb" if encoding is None else "wt"
+    file = os.fdopen(fd, mode=mode, encoding=encoding,)
     file.write(content)
     file.close()
     try:

orangecanvas/utils/tests/test_shtools.py

@@ -7,7 +7,7 @@
 
 class Test(unittest.TestCase):
     def test_python_process(self):
-        p = python_process(["-c", "print('Hello')"])
+        p = python_process(["-c", "print('Hello')"], encoding="utf-8")
         out, _ = p.communicate()
         self.assertEqual(out.strip(), "Hello")
         self.assertEqual(p.wait(), 0)
@@ -16,10 +16,12 @@ def test_temp_named_file(self):
         cases = [
             ("Hello", "utf-8"),
             ("Hello", "utf-16"),
+            (b"Hello", None)
         ]
         for content, encoding in cases:
             with temp_named_file(content, encoding=encoding) as fname:
-                with open(fname, "r", encoding=encoding) as f:
+                with open(fname, "rb" if encoding is None else "r",
+                          encoding=encoding) as f:
                     c = f.read()
                     self.assertEqual(c, content)
             self.assertFalse(os.path.exists(fname))