grass.script: Allow init to modify only specified environment

python/grass/experimental/tests/conftest.py

@@ -1,11 +1,11 @@
 """Fixtures for grass.script"""
 
 import uuid
+import os
 
 import pytest
 
 import grass.script as gs
-import grass.script.setup as grass_setup
 import grass.experimental as experimental
 
 
@@ -14,7 +14,7 @@ def xy_session(tmp_path):
     """Active session in an XY location (scope: function)"""
     location = "xy_test"
     gs.core._create_location_xy(tmp_path, location)  # pylint: disable=protected-access
-    with grass_setup.init(tmp_path / location) as session:
+    with gs.setup.init(tmp_path / location, env=os.environ.copy()) as session:
         yield session
 
 
@@ -29,7 +29,7 @@ def xy_session_for_module(tmp_path_factory):
     tmp_path = tmp_path_factory.mktemp("xy_session_for_module")
     location = "xy_test"
     gs.core._create_location_xy(tmp_path, location)  # pylint: disable=protected-access
-    with grass_setup.init(tmp_path / location) as session:
+    with gs.setup.init(tmp_path / location, env=os.environ.copy()) as session:
         yield session
 
 

python/grass/experimental/tests/grass_script_mapset_session_test.py

@@ -59,7 +59,7 @@ def test_create_overwrite(xy_session):
     """Session creates and creates again with overwrite"""
     name = "test_mapset_1"
     session_file = xy_session.env["GISRC"]
-    with experimental.MapsetSession(name, create=True) as session:
+    with experimental.MapsetSession(name, create=True, env=xy_session.env) as session:
         session_mapset = gs.read_command("g.mapset", flags="p", env=session.env).strip()
         assert name == session_mapset
         gs.run_command("r.mapcalc", expression="a = 1", env=session.env)
@@ -69,7 +69,9 @@ def test_create_overwrite(xy_session):
             .split()
         )
         assert len(rasters) == 1 and rasters[0] == "a"
-    with experimental.MapsetSession(name, create=True, overwrite=True) as session:
+    with experimental.MapsetSession(
+        name, create=True, overwrite=True, env=xy_session.env
+    ) as session:
         session_mapset = gs.read_command("g.mapset", flags="p", env=session.env).strip()
         assert name == session_mapset
         rasters = (
@@ -92,7 +94,7 @@ def test_ensure(xy_session):
     """Session ensures and does not delete"""
     name = "test_mapset_1"
     session_file = xy_session.env["GISRC"]
-    with experimental.MapsetSession(name, ensure=True) as session:
+    with experimental.MapsetSession(name, ensure=True, env=xy_session.env) as session:
         session_mapset = gs.read_command("g.mapset", flags="p", env=session.env).strip()
         assert name == session_mapset
         gs.run_command("r.mapcalc", expression="a = 1", env=session.env)
@@ -102,7 +104,7 @@ def test_ensure(xy_session):
             .split()
         )
         assert len(rasters) == 1 and rasters[0] == "a"
-    with experimental.MapsetSession(name, ensure=True) as session:
+    with experimental.MapsetSession(name, ensure=True, env=xy_session.env) as session:
         session_mapset = gs.read_command("g.mapset", flags="p", env=session.env).strip()
         assert name == session_mapset
         rasters = (

python/grass/script/core.py

@@ -652,16 +652,16 @@ def exec_command(
 # interface to g.message
 
 
-def message(msg, flag=None):
+def message(msg, flag=None, env=None):
     """Display a message using `g.message`
 
     :param str msg: message to be displayed
     :param str flag: flags (given as string)
     """
-    run_command("g.message", flags=flag, message=msg, errors="ignore")
+    run_command("g.message", flags=flag, message=msg, errors="ignore", env=env)
 
 
-def debug(msg, debug=1):
+def debug(msg, debug=1, env=None):
     """Display a debugging message using `g.message -d`.
 
     The visibility of a debug message at runtime is controlled by
@@ -679,26 +679,26 @@ def debug(msg, debug=1):
         if sys.platform == "win32":
             msg = msg.replace("&", "^&")
 
-        run_command("g.message", flags="d", message=msg, debug=debug)
+        run_command("g.message", flags="d", message=msg, debug=debug, env=env)
 
 
-def verbose(msg):
+def verbose(msg, env=None):
     """Display a verbose message using `g.message -v`
 
     :param str msg: verbose message to be displayed
     """
-    message(msg, flag="v")
+    message(msg, flag="v", env=env)
 
 
-def info(msg):
+def info(msg, env=None):
     """Display an informational message using `g.message -i`
 
     :param str msg: informational message to be displayed
     """
-    message(msg, flag="i")
+    message(msg, flag="i", env=env)
 
 
-def percent(i, n, s):
+def percent(i, n, s, env=None):
     """Display a progress info message using `g.message -p`
 
     ::
@@ -713,18 +713,18 @@ def percent(i, n, s):
     :param int n: total number of items
     :param int s: increment size
     """
-    message("%d %d %d" % (i, n, s), flag="p")
+    message("%d %d %d" % (i, n, s), flag="p", env=env)
 
 
-def warning(msg):
+def warning(msg, env=None):
     """Display a warning message using `g.message -w`
 
     :param str msg: warning message to be displayed
     """
-    message(msg, flag="w")
+    message(msg, flag="w", env=env)
 
 
-def error(msg):
+def error(msg, env=None):
     """Display an error message using `g.message -e`
 
     This function does not end the execution of the program.
@@ -733,10 +733,10 @@ def error(msg):
 
     :param str msg: error message to be displayed
     """
-    message(msg, flag="e")
+    message(msg, flag="e", env=env)
 
 
-def fatal(msg):
+def fatal(msg, env=None):
     """Display an error message using `g.message -e`, then abort or raise
 
     Raises exception when module global raise_on_error is 'True', abort
@@ -749,7 +749,7 @@ def fatal(msg):
     if raise_on_error:
         raise ScriptError(msg)
 
-    error(msg)
+    error(msg, env=None)
     sys.exit(1)
 
 

python/grass/script/setup.py

@@ -268,7 +268,7 @@ def setup_runtime_env(gisbase=None, *, env=None):
     env["PYTHONPATH"] = path
 
 
-def init(path, location=None, mapset=None, grass_path=None):
+def init(path, location=None, mapset=None, *, grass_path=None, env=None):
     """Initialize system variables to run GRASS modules
 
     This function is for running GRASS GIS without starting it with the
@@ -320,6 +320,7 @@ def init(path, location=None, mapset=None, grass_path=None):
 
     :returns: reference to a session handle object which is a context manager
     """
+    # The path heuristic always uses the global environment.
     grass_path = get_install_path(grass_path)
     if not grass_path:
         raise ValueError(
@@ -351,15 +352,18 @@ def init(path, location=None, mapset=None, grass_path=None):
             )
         )
 
-    setup_runtime_env(grass_path)
+    # If environment is not provided, use the global one.
+    if not env:
+        env = os.environ
+    setup_runtime_env(grass_path, env=env)
 
     # TODO: lock the mapset?
-    os.environ["GIS_LOCK"] = str(os.getpid())
+    env["GIS_LOCK"] = str(os.getpid())
 
-    os.environ["GISRC"] = write_gisrc(
+    env["GISRC"] = write_gisrc(
         mapset_path.directory, mapset_path.location, mapset_path.mapset
     )
-    return SessionHandle()
+    return SessionHandle(env=env)
 
 
 class SessionHandle:
@@ -390,9 +394,27 @@ class SessionHandle:
         with gs.setup.init("~/grassdata/nc_spm_08/user1"):
             # ... use GRASS modules here
         # session ends automatically here
+
+    The example above is modifying the global, process environment (`os.environ`).
+    If you don't want to modify the global environment, use the _env_ parameter
+    for the _init_ function to modify the provided environment instead.
+    This environment is then available as an attribute of the session object.
+    The attribute than needs to be passed to all calls of GRASS
+    tools and function which wrap them.
+    Context manager usage with custom environment::
+
+        # ... setup sys.path before import as needed
+
+        import grass.script as gs
+
+        with gs.setup.init("~/grassdata/nc_spm_08/user1", env=os.environ.copy()):
+            # ... use GRASS modules here with env parameter
+            gs.run_command("g.region", flags="p", env=session.env)
+        # session ends automatically here, global environment was never modifed
     """
 
-    def __init__(self, active=True):
+    def __init__(self, *, env, active=True):
+        self._env = env
         self._active = active
         self._start_time = datetime.datetime.now(datetime.timezone.utc)
 
@@ -403,7 +425,7 @@ def active(self):
 
     @property
     def env(self):
-        return os.environ
+        return self._env
 
     def __enter__(self):
         """Enter the context manager context.
@@ -434,14 +456,14 @@ def finish(self):
         if not self.active:
             raise ValueError("Attempt to finish an already finished session")
         self._active = False
-        finish(start_time=self._start_time)
+        finish(env=self._env, start_time=self._start_time)
 
 
 # clean-up functions when terminating a GRASS session
 # these fns can only be called within a valid GRASS session
 
 
-def clean_default_db(*, modified_after=None):
+def clean_default_db(*, modified_after=None, env=None):
     """Clean (vacuum) the default db if it is SQLite
 
     When *modified_after* is set, database is cleaned only when it was modified
@@ -451,11 +473,11 @@ def clean_default_db(*, modified_after=None):
     # pylint: disable=import-outside-toplevel
     import grass.script as gs
 
-    conn = gs.db_connection()
+    conn = gs.db_connection(env=env)
     if not conn or conn["driver"] != "sqlite":
         return
     # check if db exists
-    gis_env = gs.gisenv()
+    gis_env = gs.gisenv(env=env)
     database = conn["database"]
     database = database.replace("$GISDBASE", gis_env["GISDBASE"])
     database = database.replace("$LOCATION_NAME", gis_env["LOCATION_NAME"])
@@ -477,8 +499,8 @@ def clean_default_db(*, modified_after=None):
     # Start the vacuum process, then show the message in parallel while
     # the vacuum is running. Finally, wait for the vacuum process to finish.
     # Error handling is the same as errors="ignore".
-    process = gs.start_command("db.execute", sql="VACUUM")
-    gs.verbose(_("Cleaning up default SQLite database..."))
+    process = gs.start_command("db.execute", sql="VACUUM", env=env)
+    gs.verbose(_("Cleaning up SQLite attribute database..."), env=env)
     process.wait()
 
 
@@ -489,18 +511,23 @@ def call(cmd, **kwargs):
     return subprocess.call(cmd, **kwargs)
 
 
-def clean_temp():
+def clean_temp(env=None):
     """Clean mapset temporary directory"""
     # Lazy-importing to reduce dependencies (this can be eventually removed).
     # pylint: disable=import-outside-toplevel
     import grass.script as gs
 
-    gs.verbose(_("Cleaning up temporary files..."))
-    gisbase = os.environ["GISBASE"]
-    call([os.path.join(gisbase, "etc", "clean_temp")], stdout=subprocess.DEVNULL)
+    if not env:
+        env = os.environ
+
+    gs.verbose(_("Cleaning up temporary files..."), env=env)
+    gisbase = env["GISBASE"]
+    call(
+        [os.path.join(gisbase, "etc", "clean_temp")], stdout=subprocess.DEVNULL, env=env
+    )
 
 
-def finish(*, start_time=None):
+def finish(*, env=None, start_time=None):
     """Terminate the GRASS session and clean up
 
     GRASS commands can no longer be used after this function has been
@@ -518,13 +545,16 @@ def finish(*, start_time=None):
     Currently, it is used to do SQLite database vacuum only when database was modified
     since the session started.
     """
-    clean_default_db(modified_after=start_time)
-    clean_temp()
+    if not env:
+        env = os.environ
+
+    clean_default_db(modified_after=start_time, env=env)
+    clean_temp(env=env)
     # TODO: unlock the mapset?
     # unset the GISRC and delete the file
     from grass.script import utils as gutils
 
-    gutils.try_remove(os.environ["GISRC"])
-    os.environ.pop("GISRC")
+    gutils.try_remove(env["GISRC"])
+    del env["GISRC"]
     # remove gislock env var (not the gislock itself
-    os.environ.pop("GIS_LOCK")
+    del env["GIS_LOCK"]

python/grass/script/tests/conftest.py

@@ -0,0 +1,20 @@
+"""Fixtures for grass.script"""
+
+import pytest
+
+
+@pytest.fixture
+def mock_no_session(monkeypatch):
+    """Set the environment variables as if there would be no background session.
+
+    Use with usefixtures (not as a paramter) to avoid warnings about an unused
+    parameter::
+
+        @pytest.mark.usefixtures("mock_no_session")
+        def test_session_handling():
+            pass
+
+    There may or may not be a session in the background (we don't check either way).
+    """
+    monkeypatch.delenv("GISRC", raising=False)
+    monkeypatch.delenv("GISBASE", raising=False)