Add is_paused, pause_aware_restart_on_change

lib/swift_utils.py

@@ -42,6 +42,7 @@ from charmhelpers.core.hookenv import (
     relation_set,
     relation_ids,
     related_units,
+    status_get
 )
 from charmhelpers.fetch import (
     apt_update,
@@ -50,7 +51,8 @@ from charmhelpers.fetch import (
     add_source
 )
 from charmhelpers.core.host import (
-    lsb_release
+    lsb_release,
+    restart_on_change,
 )
 from charmhelpers.contrib.network.ip import (
     format_ipv6_addr,
@@ -994,3 +996,19 @@ def get_hostaddr():
         return get_ipv6_addr(exc_list=[config('vip')])[0]
 
     return unit_get('private-address')
+
+
+def is_paused(status_get=status_get):
+    """Is the unit paused?"""
+    status, message = status_get()
+    return status == "maintenance" and message.startswith("Paused")
+
+
+def pause_aware_restart_on_change(restart_map):
+    """Avoids restarting services if config changes when unit is paused."""
+    def wrapper(f):
+        if is_paused():
+            return f
+        else:
+            return restart_on_change(restart_map)(f)
+    return wrapper

unit_tests/test_swift_utils.py

@@ -222,3 +222,15 @@ class SwiftUtilsTestCase(unittest.TestCase):
 
         rsps = []
         self.assertIsNone(swift_utils.get_first_available_value(rsps, 'key3'))
+
+    def test_is_paused_unknown(self):
+        fake_status_get = lambda: ("unknown", "")
+        self.assertFalse(swift_utils.is_paused(status_get=fake_status_get))
+
+    def test_is_paused_paused(self):
+        fake_status_get = lambda: ("maintenance", "Paused")
+        self.assertTrue(swift_utils.is_paused(status_get=fake_status_get))
+
+    def test_is_paused_other_maintenance(self):
+        fake_status_get = lambda: ("maintenance", "Hook")
+        self.assertFalse(swift_utils.is_paused(status_get=fake_status_get))