grass.app: Move mapset locking to the library (#4158)

This moves the lock_mapset function to the library. It introduces only small changes to the code, i.e., proper refactoring and de-duplication in relation to the code elsewhere is still needed.

However, this unifies the error handling to always raising a custom exception. It also fixes the reported user using the mapset (before always the current user, now, the owner of the lock file; for that, lock file is removed only after getting its owner). This also removes the debug message which I don't find particularly useful, for example it is currently misleading on Windows and it is relatively easy to confirm the actual existence in process manager when debugging.

Change message wording to add clarity. Fix translatable message are fixed. Uses format function everywhere.

Gets install path (GISBASE) automatically (requires caller to have environment set up which is likely a reasonable requirement at this point).

Fixes order of reporting owner and deleting the file. Cleans up order of paths.

lib/init/grass.py

@@ -1281,65 +1281,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"""
@@ -2421,14 +2362,16 @@ 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,
-            user=user,
-            force_gislock_removal=params.force_gislock_removal,
+            mapset_path=mapset_settings.full_mapset,
+            force_lock_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,69 @@ 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_lock_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 because of existing lock or due to insufficient permissions.
+    A corresponding localized message is given in the exception.
+
+    A *message_callback* is a function which will be called to report messages about
+    certain states. Specifically, the function is called when forcibly unlocking the
+    mapset.
+
+    Assumes that the runtime is set up (specifically that GISBASE is in
+    the environment).
+    """
+    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")
+    locker_path = os.path.join(os.environ["GISBASE"], "etc", "lock")
+    ret = subprocess.run(
+        [locker_path, lockfile, "%d" % os.getpid()], check=False
+    ).returncode
+    msg = None
+    if ret == 2:
+        if not force_lock_removal:
+            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=Path(lockfile).owner(), file=lockfile)
+        else:
+            message_callback(
+                _(
+                    "{user} is currently running GRASS in selected mapset"
+                    " (file {file} found), but forcing to launch GRASS anyway..."
+                ).format(user=Path(lockfile).owner(), file=lockfile)
+            )
+            gs.try_remove(lockfile)
+    elif ret != 0:
+        msg = _(
+            "Unable to properly access lock file '{name}'.\n"
+            "Please resolve this with your system administrator."
+        ).format(name=lockfile)
+
+    if msg:
+        raise MapsetLockingException(msg)
+    return lockfile