grass.app: Move mapset locking to library

lib/init/grass.py

@@ -1301,64 +1301,6 @@ def set_language(grass_config_dir):
     gettext.install("grasslibs", gpath("locale"))
 
 
-def lock_mapset(mapset_path, force_gislock_removal, user):
-    """Lock the mapset and return name of the lock file
-
-    Behavior on error must be changed somehow; now it fatals but GUI case is
-    unresolved.
-    """
-    if not os.path.exists(mapset_path):
-        fatal(_("Path '%s' doesn't exist") % mapset_path)
-    if not os.access(mapset_path, os.W_OK):
-        error = _("Path '%s' not accessible.") % mapset_path
-        stat_info = os.stat(mapset_path)
-        mapset_uid = stat_info.st_uid
-        if mapset_uid != os.getuid():
-            # GTC %s is mapset's folder path
-            error = "%s\n%s" % (
-                error,
-                _("You are not the owner of '%s'.") % mapset_path,
-            )
-        fatal(error)
-    # Check for concurrent use
-    lockfile = os.path.join(mapset_path, ".gislock")
-    ret = call([gpath("etc", "lock"), lockfile, "%d" % os.getpid()])
-    msg = None
-    if ret == 2:
-        if not force_gislock_removal:
-            msg = _(
-                "%(user)s is currently running GRASS in selected mapset"
-                " (file %(file)s found). Concurrent use not allowed.\n"
-                "You can force launching GRASS using -f flag"
-                " (note that you need permission for this operation)."
-                " Have another look in the processor "
-                "manager just to be sure..." % {"user": user, "file": lockfile}
-            )
-        else:
-            try_remove(lockfile)
-            message(
-                _(
-                    "%(user)s is currently running GRASS in selected mapset"
-                    " (file %(file)s found). Forcing to launch GRASS..."
-                    % {"user": user, "file": lockfile}
-                )
-            )
-    elif ret != 0:
-        msg = (
-            _("Unable to properly access '%s'.\nPlease notify system personnel.")
-            % lockfile
-        )
-
-    if msg:
-        raise Exception(msg)
-    debug(
-        "Mapset <{mapset}> locked using '{lockfile}'".format(
-            mapset=mapset_path, lockfile=lockfile
-        )
-    )
-    return lockfile
-
-
 # TODO: the gisrcrc here does not make sense, remove it from load_gisrc
 def unlock_gisrc_mapset(gisrc, gisrcrc):
     """Unlock mapset from the gisrc file"""
@@ -2440,14 +2382,18 @@ def main():
 
     location = mapset_settings.full_mapset
 
+    from grass.app.data import lock_mapset, MapsetLockingException
+
     try:
         # check and create .gislock file
         lock_mapset(
-            mapset_settings.full_mapset,
+            install_path=GISBASE,
+            mapset_path=mapset_settings.full_mapset,
             user=user,
             force_gislock_removal=params.force_gislock_removal,
+            message_callback=message,
         )
-    except Exception as e:
+    except MapsetLockingException as e:
         fatal(e.args[0])
         sys.exit(_("Exiting..."))
 

python/grass/app/data.py

@@ -15,8 +15,12 @@
 import os
 import tempfile
 import getpass
+import subprocess
 import sys
 from shutil import copytree, ignore_patterns
+from pathlib import Path
+
+import grass.script as gs
 import grass.grassdb.config as cfg
 
 from grass.grassdb.checks import is_location_valid
@@ -162,3 +166,65 @@ def ensure_default_data_hierarchy():
     mapset_path = os.path.join(gisdbase, location, mapset)
 
     return gisdbase, location, mapset, mapset_path
+
+
+class MapsetLockingException(Exception):
+    pass
+
+
+def lock_mapset(mapset_path, force_gislock_removal, message_callback
+):
+    """Acquire a lock for a mapset and return name of new lock file
+
+    Raises MapsetLockingException when it is not possible to acquire a lock for the
+    given mapset either becuase of existing lock or due to insufficient permissions.
+    A corresponding localized message is given in the exception.
+
+    Assumes that the runtime is setup (GISBASE).
+    """
+    if not os.path.exists(mapset_path):
+        raise MapsetLockingException(_("Path '{}' doesn't exist").format(mapset_path))
+    if not os.access(mapset_path, os.W_OK):
+        error = _("Path '{}' not accessible.").format(mapset_path)
+        stat_info = os.stat(mapset_path)
+        mapset_uid = stat_info.st_uid
+        if mapset_uid != os.getuid():
+            error = "{error}\n{detail}".format(
+                error=error,
+                detail=_("You are not the owner of '{}'.").format(mapset_path),
+            )
+        raise MapsetLockingException(error)
+    # Check for concurrent use
+    lockfile = os.path.join(mapset_path, ".gislock")
+    install_path = Path(os.environ["GISBASE"])
+    ret = subprocess.run(
+        [install_path / "etc" / "lock", lockfile, "%d" % os.getpid()], check=False
+    ).returncode
+    msg = None
+    if ret == 2:
+        if not force_gislock_removal:
+            lockfile = Path(lockfile)
+            msg = _(
+                "{user} is currently running GRASS in selected mapset"
+                " (file {file} found). Concurrent use of one mapset not allowed.\n"
+                "You can force launching GRASS using -f flag"
+                " (assuming your have sufficient access permissions)."
+                " Confirm in a process manager "
+                "that there is no other process using the mapset."
+            ).format(user=lockfile.owner(), file=lockfile)
+        else:
+            gs.try_remove(lockfile)
+            message_callback(
+                _(
+                    "{user} is currently running GRASS in selected mapset"
+                    " (file {file} found), but forcing to launch GRASS anyway..."
+                ).format(user=lockfile.owner(), file=lockfile)
+            )
+    elif ret != 0:
+        msg = _(
+            "Unable to properly access '{}'.\nPlease notify your system administrator."
+        ).format(lockfile)
+
+    if msg:
+        raise MapsetLockingException(msg)
+    return lockfile