lvm: Add a function to disable and enable LVM auto-activation

blivet/devicelibs/lvm.py

@@ -21,6 +21,7 @@
 #
 
 import math
+import os
 import re
 
 from collections import namedtuple
@@ -83,6 +84,7 @@
 
 
 LVM_DEVICES_FILE = "/etc/lvm/devices/system.devices"
+LVM_LOCAL_CONF = "/etc/lvm/lvmlocal.conf"
 
 if hasattr(blockdev.LVMTech, "CONFIG"):
     try:
@@ -289,3 +291,50 @@ def recommend_thpool_chunk_size(thpool_size):
 
 def is_valid_cache_md_size(md_size):
     return md_size >= LVM_CACHE_MIN_METADATA_SIZE and md_size <= LVM_CACHE_MAX_METADATA_SIZE
+
+
+def disable_lvm_autoactivation(lvmconf=LVM_LOCAL_CONF):
+    """ Disable LVM auto-activation *globally* by writing the configuration to
+        :attr:`lvmconf` (defaults to /etc/lvm/lvmlocal.conf)
+    """
+    if not os.path.exists(lvmconf):
+        raise RuntimeError("Cannot disable LVM auto-activation, configuration file %s does not exist" % lvmconf)
+
+    with open(lvmconf, "r") as f:
+        for line in f:
+            if "event_activation" in line and not line.strip().startswith("#"):
+                raise RuntimeError("LVM auto-activation is already configured in %s" % lvmconf)
+
+    with open(lvmconf, "a") as f:
+        f.write("global { event_activation = 0 }")
+        f.flush()
+
+    log.info("LVM auto-activation is now disabled in %s", lvmconf)
+
+    global AUTO_ACTIVATION
+    AUTO_ACTIVATION = False
+
+
+def reenable_lvm_autoactivation(lvmconf=LVM_LOCAL_CONF):
+    """ Enable LVM auto-activation previously disabled in :attr:`lvmconf`.
+        This function is intended to revert configuration done by
+        :func:`disable_lvm_autoactivation`.
+    """
+    if not os.path.exists(lvmconf):
+        raise RuntimeError("Cannot reenable LVM auto-activation, configuration file %s does not exist" % lvmconf)
+
+    with open(lvmconf, "r") as f:
+        lines = f.readlines()
+
+    if "global { event_activation = 0 }" not in lines:
+        raise RuntimeError("LVM auto-activation is not configured in %s" % lvmconf)
+
+    with open(lvmconf, "w+") as f:
+        for line in lines:
+            if line.strip("\n") != "global { event_activation = 0 }":
+                f.write(line)
+
+    log.info("LVM auto-activation configuration is now removed from %s", lvmconf)
+
+    global AUTO_ACTIVATION
+    AUTO_ACTIVATION = False

tests/unit_tests/devicelibs_test/__init__.py

@@ -1,4 +1,5 @@
 from .disk_test import *
 from .edd_test import *
+from .lvm_test import *
 from .mdraid_test import *
 from .raid_test import *

tests/unit_tests/devicelibs_test/lvm_test.py

@@ -0,0 +1,41 @@
+import unittest
+from unittest.mock import patch, mock_open
+
+import blivet.devicelibs.lvm as lvm
+
+
+class LVMTestCase(unittest.TestCase):
+
+    def test_lvm_autoactivation(self):
+        localconf = "global { event_activation = 0 }"
+
+        with patch("builtins.open", mock_open(read_data=localconf)):
+            # already disabled
+            with self.assertRaises(RuntimeError):
+                lvm.disable_lvm_autoactivation()
+
+        localconf = ""
+        with patch("builtins.open", mock_open(read_data=localconf)) as m:
+            lvm.disable_lvm_autoactivation()
+            m.assert_called_with("/etc/lvm/lvmlocal.conf", "a")
+            handle = m()
+            handle.write.assert_called_once_with("global { event_activation = 0 }")
+
+        localconf = "test\ntest"
+        with patch("builtins.open", mock_open(read_data=localconf)) as m:
+            # not disabled
+            with self.assertRaises(RuntimeError):
+                lvm.reenable_lvm_autoactivation()
+
+        localconf = "# global { event_activation = 0 }"
+        with patch("builtins.open", mock_open(read_data=localconf)) as m:
+            # not disabled
+            with self.assertRaises(RuntimeError):
+                lvm.reenable_lvm_autoactivation()
+
+        localconf = "test\nglobal { event_activation = 0 }"
+        with patch("builtins.open", mock_open(read_data=localconf)) as m:
+            lvm.reenable_lvm_autoactivation()
+            m.assert_called_with("/etc/lvm/lvmlocal.conf", "w+")
+            handle = m()
+            handle.write.assert_called_once_with("test\n")