loop: add new on_close callback to Loop

Add a new signal like callback to the `Loop` class which will be
invoked before the actual loop device is closed, i.e. the loop
device has an open file descriptor to the device node and it is
being closed. Can be used to perform custom cleanup tasks.

osbuild/loop.py

@@ -5,6 +5,7 @@ import fcntl
 import os
 import stat
 import time
+from typing import Callable, Optional
 
 from .util import linux
 
@@ -108,6 +109,7 @@ class Loop:
 
         self.devname = f"loop{minor}"
         self.minor = minor
+        self.on_close: Optional[Callable[["Loop"], None]] = None
 
         with contextlib.ExitStack() as stack:
             if not dir_fd:
@@ -129,9 +131,11 @@ class Loop:
 
         No operations on this object are valid after this call.
         """
-        if self.fd >= 0:
-            os.close(self.fd)
-            self.fd = -1
+        fd, self.fd = self.fd, -1
+        if fd >= 0:
+            if callable(self.on_close):
+                self.on_close(self)  # pylint: disable=not-callable
+            os.close(fd)
             self.devname = "<closed>"
 
     def flock(self, op: int) -> None:

test/mod/test_loop.py

@@ -216,3 +216,40 @@ def test_lock(tempdir):
             f.close()
 
         ctl.close()
+
+
+@pytest.mark.skipif(not TestBase.can_bind_mount(), reason="root only")
+def test_on_close(tempdir):
+
+    path = os.path.join(tempdir, "test.img")
+    ctl = loop.LoopControl()
+
+    assert ctl
+
+    lo, f = None, None
+    invoked = False
+
+    def on_close(l):
+        nonlocal invoked
+        invoked = True
+
+        # check that this is a no-op
+        l.close()
+
+    try:
+        f = open(path, "wb+")
+        f.truncate(1024)
+        f.flush()
+        lo = ctl.loop_for_fd(f.fileno(), autoclear=True, lock=True)
+        assert lo
+
+        lo.on_close = on_close
+        lo.close()
+
+        assert invoked
+
+    finally:
+        if lo:
+            lo.close()
+
+        ctl.close()