Added base.lock() and base.islocked() functions; added corresponding unit tests

2 files changed, 128 insertions, 0 deletions

diff --git a/ipalib/base.py b/ipalib/base.py
index 1651b01d1..e427b747e 100644
--- a/ipalib/base.py
+++ b/ipalib/base.py
@@ -130,6 +130,61 @@ class ReadOnly(object):
         return object.__delattr__(self, name)
 
 
+def lock(instance):
+    """
+    Lock an instance of the `ReadOnly` class or similar.
+
+    This function can be used to lock instances of any class that implements
+    the same locking API as the `ReadOnly` class.  For example, this function
+    can lock instances of the `config.Env` class.
+
+    So that this function can be easily used within an assignment, ``instance``
+    is returned after it is locked.  For example:
+
+    >>> readonly = ReadOnly()
+    >>> readonly is lock(readonly)
+    True
+    >>> readonly.attr = 'This wont work'
+    Traceback (most recent call last):
+      ...
+    AttributeError: locked: cannot set ReadOnly.attr to 'This wont work'
+
+    Also see the `islocked()` function.
+
+    :param instance: The instance of `ReadOnly` (or similar) to lock.
+    """
+    assert instance.__islocked__() is False, 'already locked: %r' % instance
+    instance.__lock__()
+    assert instance.__islocked__() is True, 'failed to lock: %r' % instance
+    return instance
+
+
+def islocked(instance):
+    """
+    Return ``True`` if ``instance`` is locked.
+
+    This function can be used on an instance of the `ReadOnly` class or an
+    instance of any other class implemented the same locking API.
+
+    For example:
+
+    >>> readonly = ReadOnly()
+    >>> islocked(readonly)
+    False
+    >>> readonly.__lock__()
+    >>> islocked(readonly)
+    True
+
+    Also see the `lock()` function.
+
+    :param instance: The instance of `ReadOnly` (or similar) to interrogate.
+    """
+    assert (
+        hasattr(instance, '__lock__') and callable(instance.__lock__)
+    ), 'no __lock__() method: %r' % instance
+    return instance.__islocked__()
+
+
 def check_name(name):
     """
     Verify that ``name`` is suitable for a `NameSpace` member name.
diff --git a/tests/test_ipalib/test_base.py b/tests/test_ipalib/test_base.py
index 49d5f39df..87e4c063d 100644
--- a/tests/test_ipalib/test_base.py
+++ b/tests/test_ipalib/test_base.py
@@ -84,6 +84,79 @@ class test_ReadOnly(ClassChecker):
         assert o.attr2 == 'How are you?'
 
 
+def test_lock():
+    """
+    Test the `ipalib.base.lock` function
+    """
+    f = base.lock
+
+    # Test with ReadOnly instance:
+    o = base.ReadOnly()
+    assert o.__islocked__() is False
+    assert f(o) is o
+    assert o.__islocked__() is True
+    e = raises(AssertionError, f, o)
+    assert str(e) == 'already locked: %r' % o
+
+    # Test with another class implemented locking protocol:
+    class Lockable(object):
+        __locked = False
+        def __lock__(self):
+            self.__locked = True
+        def __islocked__(self):
+            return self.__locked
+    o = Lockable()
+    assert o.__islocked__() is False
+    assert f(o) is o
+    assert o.__islocked__() is True
+    e = raises(AssertionError, f, o)
+    assert str(e) == 'already locked: %r' % o
+
+    # Test with a class incorrectly implementing the locking protocol:
+    class Broken(object):
+        def __lock__(self):
+            pass
+        def __islocked__(self):
+            return False
+    o = Broken()
+    e = raises(AssertionError, f, o)
+    assert str(e) == 'failed to lock: %r' % o
+
+
+def test_islocked():
+    """
+    Test the `ipalib.base.islocked` function.
+    """
+    f = base.islocked
+
+    # Test with ReadOnly instance:
+    o = base.ReadOnly()
+    assert f(o) is False
+    o.__lock__()
+    assert f(o) is True
+
+    # Test with another class implemented locking protocol:
+    class Lockable(object):
+        __locked = False
+        def __lock__(self):
+            self.__locked = True
+        def __islocked__(self):
+            return self.__locked
+    o = Lockable()
+    assert f(o) is False
+    o.__lock__()
+    assert f(o) is True
+
+    # Test with a class incorrectly implementing the locking protocol:
+    class Broken(object):
+        __lock__ = False
+        def __islocked__(self):
+            return False
+    o = Broken()
+    e = raises(AssertionError, f, o)
+    assert str(e) == 'no __lock__() method: %r' % o
+
+
 def test_check_name():
     """
     Test the `ipalib.base.check_name` function.