tests: Use string and subprocess for isolation

lib/gis/tests/lib_gis_env_test.py

@@ -1,34 +1,67 @@
 """Test environment and GIS environment functions"""
 
-import multiprocessing
+import os
+import subprocess
+import sys
+import json
+from textwrap import dedent
 
 import pytest
 
 import grass.script as gs
 
-xfail_mp_spawn = pytest.mark.xfail(
-    multiprocessing.get_start_method() == "spawn",
-    reason="Multiprocessing using 'spawn' start method requires pickable functions",
-    raises=AttributeError,
-    strict=True,
-)
-
 
-def run_in_subprocess(function):
-    """Run function in a separate process
+def run_in_subprocess(code, tmp_path, env):
+    """Code as a script in a separate process and return parsed JSON result
 
-    The function must take a Queue and put its result there.
-    The result is then returned from this function.
+    This is useful when we want to ensure that function like init does
+    not change the global environment for other tests.
+    Some effort is made to remove a current if any, but it does not touch
+    runtime variables on purpose to enable dynamic library loading based on
+    a path variable.
     """
-    queue = multiprocessing.Queue()
-    process = multiprocessing.Process(target=function, args=(queue,))
-    process.start()
-    result = queue.get()
-    process.join()
-    return result
+    source_file = tmp_path / "test.py"
+    source_file.write_text(dedent(code))
+    env = env.copy()
+    for variable in ("GISRC", "GIS_LOCK"):
+        if variable in env:
+            del env[variable]
+    result = subprocess.run(
+        [sys.executable, os.fspath(source_file)],
+        capture_output=True,
+        text=True,
+        check=False,
+        env=env,
+    )
+    if result.returncode != 0:
+        if result.stderr:
+            msg = (
+                "Execution of code in subprocess failed, "
+                f"captured stderr from subprocess:\n{result.stderr}\n"
+            )
+        else:
+            msg = (
+                f"Execution of code in subprocess gave return code {result.returncode}"
+                "but there was no stderr"
+            )
+        raise RuntimeError(msg)
+    if not result.stdout:
+        msg = "Empty result from subprocess running code"
+        raise ValueError(msg)
+    try:
+        return json.loads(result.stdout)
+    except json.JSONDecodeError as error:
+        msg = f"Invalid JSON: {result.stdout}"
+        raise ValueError(msg) from error
 
 
-@xfail_mp_spawn
+# To read the new variables on Windows, another subprocess would be needed,
+# but that would not allow for the dynamic changes of the session variables within
+# a single process which is what this test is testing.
+@pytest.mark.skipif(
+    sys.platform.startswith("win"),
+    reason="On Windows, C has a cached environment, so the libraries don't know about the session.",
+)
 def test_reading_respects_change_of_session(tmp_path):
     """Check new session file path is retrieved and the file is read"""
 
@@ -37,23 +70,27 @@ def test_reading_respects_change_of_session(tmp_path):
     # process-separated otherwise other tests will be influenced by the loaded
     # libraries and initialized data structures.
 
-    def switch_through_locations(queue):
-        """Switches through a list of locations"""
-        # Just to be sure we don't influence other tests.
-        # pylint: disable=import-outside-toplevel
+    code = f"""
+        # Switches through a list of projects
+        import json
+        from pathlib import Path
+        import grass.script as gs
         import grass.pygrass.utils as pygrass_utils
         import grass.lib.gis as libgis
 
         names = []
         for project_name in ["test1", "test2", "abc"]:
-            gs.create_project(tmp_path / project_name)
-            with gs.setup.init(tmp_path / project_name):
+            gs.create_project(Path(r"{tmp_path}") / project_name)
+            with gs.setup.init(Path(r"{tmp_path}") / project_name):
                 libgis.G__read_gisrc_path()
                 libgis.G__read_gisrc_env()
                 names.append((pygrass_utils.getenv("LOCATION_NAME"), project_name))
-        queue.put(names)
+        print(json.dumps(names))
+        """
 
-    names = run_in_subprocess(switch_through_locations)
+    gs.create_project(tmp_path / "base")
+    with gs.setup.init(tmp_path / "base", env=os.environ.copy()) as session:
+        names = run_in_subprocess(code, tmp_path=tmp_path, env=session.env)
 
     for getenv_name, expected_name in names:
         assert getenv_name == expected_name, f"All recorded names: {names}"

python/grass/script/tests/conftest.py

@@ -20,8 +20,12 @@ def test_session_handling():
 
     There may or may not be a session in the background (we don't check either way).
     """
+    # Session
     monkeypatch.delenv("GISRC", raising=False)
+    monkeypatch.delenv("GIS_LOCK", raising=False)
+    # Runtime
     monkeypatch.delenv("GISBASE", raising=False)
+    monkeypatch.delenv("GRASS_PREFIX", raising=False)
 
 
 @pytest.fixture

python/grass/script/tests/grass_script_core_location_test.py

@@ -1,6 +1,5 @@
 """Test functions in grass.script.setup"""
 
-import multiprocessing
 import os
 
 import pytest
@@ -9,32 +8,15 @@
 from grass.exceptions import ScriptError
 from grass.tools import Tools
 
-xfail_mp_spawn = pytest.mark.xfail(
-    multiprocessing.get_start_method() == "spawn",
-    reason="Multiprocessing using 'spawn' start method requires pickable functions",
-    raises=AttributeError,
-    strict=True,
-)
 
+def test_with_same_path(tmp_path):
+    """Check correct EPSG is created with same path as the current one.
 
-# This is useful when we want to ensure that function like init does
-# not change the global environment.
-def run_in_subprocess(function):
-    """Run function in a separate process
-
-    The function must take a Queue and put its result there.
-    The result is then returned from this function.
+    Creates two projects, a XY bootstrap one and a desired project using
+    the same path (build for the case when only XY project can be created
+    which is no longer the case, but still it makes sense to test that as
+    a possible situation).
     """
-    queue = multiprocessing.Queue()
-    process = multiprocessing.Process(target=function, args=(queue,))
-    process.start()
-    result = queue.get()
-    process.join()
-    return result
-
-
-def create_and_get_srid(tmp_path):
-    """Create location on the same path as the current one"""
     bootstrap = "bootstrap"
     desired = "desired"
     gs.create_project(tmp_path / bootstrap)
@@ -46,29 +28,12 @@ def create_and_get_srid(tmp_path):
         gs.run_command("g.gisenv", set=f"GISDBASE={tmp_path}", env=session.env)
         gs.run_command("g.gisenv", set=f"LOCATION_NAME={desired}", env=session.env)
         gs.run_command("g.gisenv", set="MAPSET=PERMANENT", env=session.env)
-        return gs.parse_command("g.proj", flags="p", format="shell", env=session.env)[
+        srid = gs.parse_command("g.proj", flags="p", format="shell", env=session.env)[
             "srid"
         ]
-
-
-def test_with_same_path(tmp_path):
-    """Check correct EPSG is created with same path as the current one"""
-    srid = create_and_get_srid(tmp_path)
     assert srid == "EPSG:3358"
 
 
-@xfail_mp_spawn
-def test_with_init_in_subprocess(tmp_path):
-    """Check creation when running in a subprocess"""
-
-    def workload(queue):
-        """Transfer the return value using queue"""
-        queue.put(create_and_get_srid(tmp_path))
-
-    epsg = run_in_subprocess(workload)
-    assert epsg == "EPSG:3358"
-
-
 @pytest.mark.usefixtures("mock_no_session")
 def test_without_session(tmp_path):
     """Check that creation works outside of session.