grass.script: Use text=True by default for subprocesses

general/g.mapsets/tests/g_mapsets_list_format_test.py

@@ -14,10 +14,8 @@
 """Test parsing and structure of outputs from g.mapsets"""
 
 import json
-import sys
 import pytest
 import grass.script as gs
-from grass.script import utils as gutils
 
 SEPARATORS = ["newline", "space", "comma", "tab", "pipe", ","]
 
@@ -31,10 +29,6 @@ def _check_parsed_list(mapsets, text, sep="|"):
     assert text == sep.join(mapsets) + "\n"
 
 
-@pytest.mark.xfail(
-    sys.platform == "win32",
-    reason="universal_newlines or text subprocess option not used",
-)
 @pytest.mark.parametrize("separator", SEPARATORS)
 def test_plain_list_output(simple_dataset, separator):
     """Test that the separators are properly applied with list flag"""
@@ -46,13 +40,9 @@ def test_plain_list_output(simple_dataset, separator):
         flags="l",
         env=simple_dataset.session.env,
     )
-    _check_parsed_list(mapsets, text, gutils.separator(separator))
+    _check_parsed_list(mapsets, text, gs.separator(separator))
 
 
-@pytest.mark.xfail(
-    sys.platform == "win32",
-    reason="universal_newlines or text subprocess option not used",
-)
 @pytest.mark.parametrize("separator", SEPARATORS)
 def test_plain_print_output(simple_dataset, separator):
     """Test that the separators are properly applied with print flag"""
@@ -64,7 +54,7 @@ def test_plain_print_output(simple_dataset, separator):
         flags="p",
         env=simple_dataset.session.env,
     )
-    _check_parsed_list(mapsets, text, gutils.separator(separator))
+    _check_parsed_list(mapsets, text, gs.separator(separator))
 
 
 def test_json_list_output(simple_dataset):

python/grass/gunittest/gmodules.py

@@ -11,7 +11,7 @@
 
 import subprocess
 from grass.script.core import start_command
-from grass.script.utils import encode, decode
+from grass.script.utils import decode
 from grass.exceptions import CalledModuleError
 from grass.pygrass.modules import Module
 
@@ -134,7 +134,7 @@ def call_module(
     # for no stdout, output is None which is out interface
     # for stderr=STDOUT or no stderr, errors is None
     # which is fine for CalledModuleError
-    output, errors = process.communicate(input=encode(decode(stdin)) if stdin else None)
+    output, errors = process.communicate(input=decode(stdin) if stdin else None)
     returncode = process.poll()
     if returncode:
         raise CalledModuleError(module, kwargs, returncode, errors)

python/grass/jupyter/utils.py

@@ -72,15 +72,12 @@ def reproject_region(region, from_proj, to_proj):
         stdout=gs.PIPE,
         stderr=gs.PIPE,
     )
-    proc.stdin.write(gs.encode(proj_input))
-    proc.stdin.close()
-    proc.stdin = None
-    proj_output, stderr = proc.communicate()
+    proj_output, stderr = proc.communicate(proj_input)
     if proc.returncode:
         raise RuntimeError(
             _("Encountered error while running m.proj: {}").format(stderr)
         )
-    output = gs.decode(proj_output).splitlines()
+    output = proj_output.splitlines()
     # get the largest bbox
     latitude_list = []
     longitude_list = []

python/grass/pygrass/modules/interface/module.py

@@ -4,7 +4,7 @@
 
 from grass.exceptions import CalledModuleError, GrassError, ParameterError
 from grass.script.core import Popen, PIPE, use_temp_region, del_temp_region
-from grass.script.utils import encode, decode
+from grass.script.utils import decode
 from .docstring import docstring_property
 from .parameter import Parameter
 from .flag import Flag
@@ -853,8 +853,6 @@ def wait(self):
         :return: A reference to this object
         """
         if self._finished is False:
-            if self.stdin:
-                self.stdin = encode(self.stdin)
             stdout, stderr = self._popen.communicate(input=self.stdin)
             self.outputs["stdout"].value = decode(stdout) if stdout else ""
             self.outputs["stderr"].value = decode(stderr) if stderr else ""

python/grass/script/core.py

@@ -37,7 +37,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, TypeVar
 
-from .utils import KeyValue, parse_key_val, basename, encode, decode, try_remove
+from .utils import KeyValue, parse_key_val, basename, decode, try_remove
 from grass.exceptions import ScriptError, CalledModuleError
 from grass.grassdb.manage import resolve_mapset_path
 
@@ -54,6 +54,40 @@
 class Popen(subprocess.Popen):
     _builtin_exts = {".com", ".exe", ".bat", ".cmd"}
 
+    class StdinWrapper:
+        """
+        Decodes bytes into str if writing failed and text mode was automatically set.
+
+        Remove for version 9
+        """
+
+        def __init__(self, stdin, implied_text):
+            self._stdin = stdin
+            self._implied_text = implied_text
+
+        def write(self, data):
+            try:
+                self._stdin.write(data)
+            except TypeError:
+                if self._implied_text:
+                    self._stdin.write(decode(data))
+                else:
+                    raise
+
+        def flush(self):
+            if self._stdin:
+                self._stdin.flush()
+
+        def close(self):
+            if self._stdin:
+                self._stdin.close()
+
+        def __getattr__(self, name):
+            # Forward everything else to the original stdin
+            if self._stdin:
+                return getattr(self._stdin, name)
+            return None
+
     @staticmethod
     def _escape_for_shell(arg):
         # TODO: what are cmd.exe's parsing rules?
@@ -66,6 +100,13 @@ def __init__(self, args, **kwargs):
         if cmd is None:
             raise OSError(_("Cannot find the executable {0}").format(args[0]))
         args = [cmd] + args[1:]
+
+        # Use text mode by default
+        self._implied_text = False
+        if "text" not in kwargs and "universal_newlines" not in kwargs:
+            kwargs["text"] = True
+            self._implied_text = True
+
         if (
             sys.platform == "win32"
             and isinstance(args, list)
@@ -85,6 +126,14 @@ def __init__(self, args, **kwargs):
 
         subprocess.Popen.__init__(self, args, **kwargs)
 
+    @property
+    def stdin(self):
+        return self._wrapped_stdin
+
+    @stdin.setter
+    def stdin(self, value):
+        self._wrapped_stdin = Popen.StdinWrapper(value, self._implied_text)
+
 
 PIPE = subprocess.PIPE
 STDOUT = subprocess.STDOUT
@@ -736,10 +785,6 @@ def write_command(*args, **kwargs):
         encoding = kwargs["encoding"]
     # TODO: should we delete it from kwargs?
     stdin = kwargs["stdin"]
-    if encoding is None or encoding == "default":
-        stdin = encode(stdin)
-    else:
-        stdin = encode(stdin, encoding=encoding)
     if _capture_stderr and "stderr" not in kwargs.keys():
         kwargs["stderr"] = PIPE
     process = feed_command(*args, **kwargs)
@@ -2040,7 +2085,7 @@ def local_env():
                 stdin=PIPE,
                 env=local_env(),
             )
-            stdin = encode(wkt)
+            stdin = wkt
     else:
         _create_location_xy(mapset_path.directory, mapset_path.location)
 

python/grass/script/raster.py

@@ -39,7 +39,6 @@
 )
 from grass.exceptions import CalledModuleError
 from .utils import (
-    encode,
     float_or_dms,
     parse_key_val,
     try_remove,
@@ -215,7 +214,7 @@ def mapcalc_start(
         verbose=verbose,
         overwrite=overwrite,
     )
-    p.stdin.write(encode(e))
+    p.stdin.write(e)
     p.stdin.close()
     return p
 

python/grass/script/task.py

@@ -414,13 +414,19 @@ def get_task(self):
 
 def convert_xml_to_utf8(xml_text):
     # enc = locale.getdefaultlocale()[1]
-
+    if xml_text is None:
+        return None
     # modify: fetch encoding from the interface description text(xml)
     # e.g. <?xml version="1.0" encoding="GBK"?>
     pattern = re.compile(rb'<\?xml[^>]*\Wencoding="([^"]*)"[^>]*\?>')
     m = re.match(pattern, xml_text)
     if m is None:
-        return xml_text.encode("utf-8") if xml_text else None
+        try:
+            xml_text.decode("utf-8", errors="strict")
+            return xml_text
+        except UnicodeDecodeError:
+            return decode(xml_text).encode("utf-8")
+
     enc = m.groups()[0]
 
     # modify: change the encoding to "utf-8", for correct parsing
@@ -441,7 +447,9 @@ def get_interface_description(cmd):
         When unable to fetch the interface description for a command.
     """
     try:
-        p = Popen([cmd, "--interface-description"], stdout=PIPE, stderr=PIPE)
+        p = Popen(
+            [cmd, "--interface-description"], stdout=PIPE, stderr=PIPE, text=False
+        )
         cmdout, cmderr = p.communicate()
 
         # TODO: do it better (?)
@@ -458,6 +466,7 @@ def get_interface_description(cmd):
                 [sys.executable, get_real_command(cmd), "--interface-description"],
                 stdout=PIPE,
                 stderr=PIPE,
+                text=False,
             )
             cmdout, cmderr = p.communicate()
 

python/grass/script/testsuite/test_start_command_functions.py

@@ -6,8 +6,7 @@
 from grass.gunittest.main import test
 from grass.gunittest.utils import xfail_windows
 
-from grass.script.core import start_command, PIPE, run_command, write_command
-from grass.script.core import read_command, find_program
+from grass.script import start_command, PIPE, run_command, write_command, read_command
 from grass.script.utils import encode
 
 
@@ -17,20 +16,20 @@ class TestPythonKeywordsInParameters(TestCase):
     It works the same for keywords, buildins and any names.
     """
 
-    raster = b"does_not_exist"
+    raster = "does_not_exist"
 
     def test_prefixed_underscore(self):
         proc = start_command("g.region", _raster=self.raster, stderr=PIPE)
         stderr = proc.communicate()[1]
-        self.assertNotIn(b"_raster", stderr)
+        self.assertNotIn("_raster", stderr)
         self.assertIn(
             self.raster, stderr, msg="Raster map name should appear in the error output"
         )
 
     def test_suffixed_underscore(self):
         proc = start_command("g.region", raster_=self.raster, stderr=PIPE)
         stderr = proc.communicate()[1]
-        self.assertNotIn(b"raster_", stderr)
+        self.assertNotIn("raster_", stderr)
         self.assertIn(
             self.raster, stderr, msg="Raster map name should appear in the error output"
         )
@@ -40,7 +39,7 @@ def test_multiple_underscores(self):
         stderr = proc.communicate()[1]
         returncode = proc.poll()
         self.assertEqual(returncode, 1)
-        self.assertIn(b"raster", stderr)
+        self.assertIn("raster", stderr)
 
 
 class TestPythonModuleWithUnicodeParameters(TestCase):
@@ -86,10 +85,8 @@ def setUpClass(cls):
     def tearDownClass(cls):
         cls.runModule("g.remove", type="raster", name=cls.raster, flags="f")
 
-    @xfail_windows
-    def test_write_labels_unicode(self):
-        """This tests if Python module works"""
-        find_program("ls", "--version")
+    def test_write_read_labels_str(self):
+        """This tests if standard string works"""
         write_command(
             "r.category",
             map=self.raster,
@@ -102,21 +99,18 @@ def test_write_labels_unicode(self):
         self.assertIsInstance(res, str)
 
     @xfail_windows
-    def test_write_labels_bytes(self):
-        """This tests if Python module works"""
+    def test_write_bytes_read_str(self):
+        """This test backwards compatibility when writing bytes"""
         write_command(
             "r.category",
             map=self.raster,
             rules="-",
-            stdin="1:kůň\n2:kráva\n3:ovečka\n4:býk",
+            stdin=encode("1:kůň\n2:kráva\n3:ovečka\n4:býk"),
             separator=":",
-            encoding=None,
         )
-        res = read_command(
-            "r.category", map=self.raster, separator=":", encoding=None
-        ).strip()
-        self.assertEqual(res, encode("1:kůň\n2:kráva\n3:ovečka\n4:býk"))
-        self.assertIsInstance(res, bytes)
+        res = read_command("r.category", map=self.raster, separator=":").strip()
+        self.assertEqual(res, "1:kůň\n2:kráva\n3:ovečka\n4:býk")
+        self.assertIsInstance(res, str)
 
 
 if __name__ == "__main__":

python/grass/script/testsuite/test_start_command_functions_nc.py

@@ -29,7 +29,7 @@ def test_prefixed_underscore(self):
         stderr = proc.communicate()[1]
         returncode = proc.poll()
         self.assertEqual(returncode, 0, msg="Underscore as prefix was not accepted")
-        self.assertNotIn(b"_raster", stderr)
+        self.assertNotIn("_raster", stderr)
 
     def test_suffixed_underscore(self):
         proc = start_command("g.region", raster_=self.raster, stderr=PIPE)
@@ -40,14 +40,14 @@ def test_suffixed_underscore(self):
             0,
             msg="Underscore as suffix was not accepted, stderr is:\n%s" % stderr,
         )
-        self.assertNotIn(b"raster_", stderr)
+        self.assertNotIn("raster_", stderr)
 
     def test_multiple_underscores(self):
         proc = start_command("g.region", _raster_=self.raster, stderr=PIPE)
         stderr = proc.communicate()[1]
         returncode = proc.poll()
         self.assertEqual(returncode, 1, msg="Underscore at both sides was accepted")
-        self.assertIn(b"raster", stderr)
+        self.assertIn("raster", stderr)
 
 
 class TestParseCommand(TestCase):

python/grass/semantic_label/reader.py

@@ -1,5 +1,4 @@
 import os
-import sys
 import json
 import glob
 import re
@@ -86,10 +85,8 @@ def print_kv(k, v, indent):
             print_kv(k, v, indent)
 
     def _print_label(self, semantic_label=None, tag=None):
-        sys.stdout.write(semantic_label)
-        if tag:
-            sys.stdout.write(" {}".format(tag))
-        sys.stdout.write(os.linesep)
+        tag_text = f" {tag}" if tag else ""
+        print(f"{semantic_label}{tag_text}")
 
     def print_info(self, shortcut=None, band=None, semantic_label=None, extended=False):
         """Prints semantic label information to stdout.