util/mnt: extract MountGuard into new module

Extract the `MountGuard` class from all stages that defined it into
a new `mnt` utility module.

osbuild/util/mnt.py

@@ -0,0 +1,40 @@
+"""Mount utilities
+"""
+
+import contextlib
+import subprocess
+
+
+class MountGuard(contextlib.AbstractContextManager):
+    def __init__(self):
+        self.mounts = []
+
+    def mount(self, source, target, bind=True, ro=False, mode="0755"):
+        options = []
+        if bind:
+            options += ["bind"]
+        if ro:
+            options += ["ro"]
+        if mode:
+            options += [mode]
+
+        args = ["--make-private"]
+        if options:
+            args += ["-o", ",".join(options)]
+
+        subprocess.run(["mount"] + args + [source, target], check=True)
+        self.mounts += [{"source": source, "target": target}]
+
+    def umount(self):
+
+        while self.mounts:
+            mount = self.mounts.pop()  # FILO: get the last mount
+            target = mount["target"]
+            # The sync should in theory not be needed but in rare
+            # cases `target is busy` error has been spotted.
+            # Calling  `sync` does not hurt so we keep it for now.
+            subprocess.run(["sync", "-f", target], check=True)
+            subprocess.run(["umount", target], check=True)
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.umount()

stages/org.osbuild.ostree

@@ -21,14 +21,13 @@ the sysroot and the deployments. Additional kernel options can be passed via
 """
 
 
-import contextlib
 import os
 import sys
 import subprocess
 
 import osbuild.api
 from osbuild.util import selinux
-
+from osbuild.util.mnt import MountGuard
 
 SCHEMA_2 = """
 "inputs": {
@@ -153,40 +152,6 @@ def ostree(*args, _input=None, **kwargs):
                    check=True)
 
 
-class MountGuard(contextlib.AbstractContextManager):
-    def __init__(self):
-        self.mounts = []
-
-    def mount(self, source, target, bind=True, ro=False, mode="0755"):
-        options = []
-        if bind:
-            options += ["bind"]
-        if ro:
-            options += ["ro"]
-        if mode:
-            options += [mode]
-
-        args = ["--make-private"]
-        if options:
-            args += ["-o", ",".join(options)]
-
-        subprocess.run(["mount"] + args + [source, target], check=True)
-        self.mounts += [{"source": source, "target": target}]
-
-        subprocess.run(["mount"] + args + [source, target], check=True)
-
-    def unmount(self):
-
-        while self.mounts:
-            mount = self.mounts.pop()  # FILO: get the last mount
-            target = mount["target"]
-            subprocess.run(["umount", "--lazy", target],
-                           check=True)
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.unmount()
-
-
 def make_fs_identifier(desc):
     for key in ["uuid", "label"]:
         val = desc.get(key)

stages/org.osbuild.ostree.deploy

@@ -19,12 +19,12 @@ the sysroot and the deployments. Additional kernel options can be passed via
 [1] https://ostree.readthedocs.io/en/latest/manual/deployment/
 """
 
-import contextlib
 import os
 import sys
 import subprocess
 
 import osbuild.api
+from osbuild.util.mnt import MountGuard
 
 CAPABILITIES = ["CAP_MAC_ADMIN"]
 
@@ -92,41 +92,6 @@ def ostree(*args, _input=None, **kwargs):
                    check=True)
 
 
-class MountGuard(contextlib.AbstractContextManager):
-    def __init__(self):
-        self.mounts = []
-
-    def mount(self, source, target, bind=True, ro=False, mode="0755"):
-        options = []
-        if bind:
-            options += ["bind"]
-        if ro:
-            options += ["ro"]
-        if mode:
-            options += [mode]
-
-        args = ["--make-private"]
-        if options:
-            args += ["-o", ",".join(options)]
-
-        subprocess.run(["mount"] + args + [source, target], check=True)
-        self.mounts += [{"source": source, "target": target}]
-
-    def umount(self):
-
-        while self.mounts:
-            mount = self.mounts.pop()  # FILO: get the last mount
-            target = mount["target"]
-            # The sync should in theory not be needed but in rare
-            # cases `target is busy` error has been spotted.
-            # Calling  `sync` does not hurt so we keep it for now.
-            subprocess.run(["sync", "-f", target], check=True)
-            subprocess.run(["umount", target], check=True)
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.umount()
-
-
 def make_fs_identifier(desc):
     for key in ["uuid", "label"]:
         val = desc.get(key)

stages/org.osbuild.ostree.fillvar

@@ -4,12 +4,12 @@ Pre-populate /var directory for a given stateroot.
 """
 
 
-import contextlib
 import os
 import sys
 import subprocess
 
 import osbuild.api
+from osbuild.util.mnt import MountGuard
 from osbuild.util import ostree
 
 
@@ -40,40 +40,6 @@ SCHEMA = """
 """
 
 
-class MountGuard(contextlib.AbstractContextManager):
-    def __init__(self):
-        self.mounts = []
-
-    def mount(self, source, target, bind=True, ro=False, mode="0755"):
-        options = []
-        if bind:
-            options += ["bind"]
-        if ro:
-            options += ["ro"]
-        if mode:
-            options += [mode]
-
-        args = ["--make-private"]
-        if options:
-            args += ["-o", ",".join(options)]
-
-        subprocess.run(["mount"] + args + [source, target], check=True)
-        self.mounts += [{"source": source, "target": target}]
-
-        subprocess.run(["mount"] + args + [source, target], check=True)
-
-    def unmount(self):
-
-        while self.mounts:
-            mount = self.mounts.pop()  # FILO: get the last mount
-            target = mount["target"]
-            subprocess.run(["umount", "--lazy", target],
-                           check=True)
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.unmount()
-
-
 def populate_var(sysroot):
     # Like anaconda[1] and Fedora CoreOS dracut[2]
     # [1] pyanaconda/payload/rpmostreepayload.py