Add: vmdb.unmount

3 files changed, 156 insertions, 0 deletions

diff --git a/vmdb/__init__.py b/vmdb/__init__.py
index 3953188..7a2c584 100644
--- a/vmdb/__init__.py
+++ b/vmdb/__init__.py
@@ -31,6 +31,7 @@ from .runcmd import (
     progress,
     error,
 )
+from .unmount import unmount, NotMounted
 from .spec import (
     Spec,
     expand_templates,
diff --git a/vmdb/unmount.py b/vmdb/unmount.py
new file mode 100644
index 0000000..1d9aaa5
--- /dev/null
+++ b/vmdb/unmount.py
@@ -0,0 +1,77 @@
+# Copyright 2018  Lars Wirzenius
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# =*= License: GPL-3+ =*=
+
+
+# Unmount a directory, including any mount points under that
+# directory. If /mnt/foo is given, and /mnt/foo/bar is also mounted,
+# unmount /mnt/foo/bar first, and /mnt/foo then. Look for sub-mounts
+# in /proc/mounts.
+
+
+import vmdb
+
+
+def unmount(what, mounts=None, real_unmount=None):
+    if mounts is None:  # pragma: no cover
+        mounts = _read_proc_mounts()
+    if real_unmount is None:  # pragma: no cover
+        real_unmount = _real_unmount
+
+    mounts = _parse_proc_mounts(mounts)
+    dirnames = _find_what_to_unmount(mounts, what)
+    for dirname in dirnames:
+        real_unmount(dirname)
+
+
+def _read_proc_mounts():  # pragma: no cover
+    with open('/proc/mounts') as f:
+        return f.read()
+
+
+def _real_unmount(what):  # pragma: no cover
+    vmdb.runcmd(['umount', what])
+
+
+def _parse_proc_mounts(text):
+    return [
+        line.split()[:2]
+        for line in text.splitlines()
+    ]
+
+
+def _find_what_to_unmount(mounts, what):
+    dirname = _find_mount_point(mounts, what)
+    dirnameslash = dirname + '/'
+    to_unmount = [
+        point
+        for dev, point in mounts
+        if point == dirname or point.startswith(dirnameslash)
+    ]
+    return list(reversed(sorted(to_unmount)))
+
+
+def _find_mount_point(mounts, what):
+    for dev, point in mounts:
+        if what in (dev, point):
+            return point
+    raise NotMounted(what)
+
+
+class NotMounted(Exception):
+
+    def __init__(self, what):
+        super().__init__('Not mounted: {}'.format(what))
diff --git a/vmdb/unmount_tests.py b/vmdb/unmount_tests.py
new file mode 100644
index 0000000..1ea3662
--- /dev/null
+++ b/vmdb/unmount_tests.py
@@ -0,0 +1,78 @@
+# Copyright 2018  Lars Wirzenius
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+# =*= License: GPL-3+ =*=
+
+
+import unittest
+
+
+import vmdb
+
+
+class UnmountTests(unittest.TestCase):
+
+    def setUp(self):
+        self.mounts = ProcMounts()
+
+    def unmount(self, what):
+        vmdb.unmount(
+            what,
+            mounts=str(self.mounts),
+            real_unmount=self.mounts.unmount)
+
+    def test_raises_error_if_not_mounted(self):
+        with self.assertRaises(vmdb.NotMounted):
+            self.unmount('/foo')
+
+    def test_unmounts_mounted_dir(self):
+        self.mounts.mount('/dev/foo', '/foo')
+        self.unmount('/foo')
+        self.assertFalse(self.mounts.is_mounted('/foo'))
+
+    def test_unmounts_mounted_dir_with_submounts(self):
+        self.mounts.mount('/dev/foo', '/foo')
+        self.mounts.mount('/dev/bar', '/foo/bar')
+        self.unmount('/foo')
+        self.assertFalse(self.mounts.is_mounted('/foo'))
+        self.assertFalse(self.mounts.is_mounted('/foo/bar'))
+
+
+class ProcMounts:
+
+    def __init__(self):
+        self.mounts = []
+
+    def is_mounted(self, what):
+        return any(what in mount for mount in self.mounts)
+
+    def mount(self, device, point):
+        self.mounts.append((device, point))
+
+    def unmount(self, what):
+        self.mounts = [
+            mount
+            for mount in self.mounts
+            if what not in mount
+        ]
+
+    def __str__(self):
+        return ''.join(
+            '{}\n'.format(self.mount_line(mount))
+            for mount in self.mounts
+        )
+
+    def mount_line(self, mount):
+        return '{} {} fstype options 0 0'.format(mount[0], mount[1])