[389-commits] [lib389] 01/01: Issue 27 - Add a module for working with dse.ldif file

[git repository hosting]

This is an automated email from the git hooks/post-receive script.

spichugi pushed a commit to branch master
in repository lib389.

commit 33bb816ecd0bdd6a4127da863f42ea656c8bcc88
Author: Simon Pichugin <spichugi(a)redhat.com&gt;
Date:   Wed May 3 16:03:38 2017 +0200

    Issue 27 - Add a module for working with dse.ldif file
    
    Description: For some tests we need a way to parse and edit dse.ldif
    file. For starters, it will be nice to have next operations support:
    - get - Return attribute values under a given entry;
    - add - Add an attribute under a given entry;
    - delete - Delete singlevalued or multivalued attributes under a given entry;
    - replace - Replace attribute values with a new one under a given entry.
    
    Add tests to lib389/tests/dseldif_test.py
    
    https://pagure.io/lib389/issue/27
    
    Reviewed by: wibrown (Thanks!)
---
 lib389/_constants.py         |   2 +
 lib389/dseldif.py            |  95 +++++++++++++++++++++++++++++++
 lib389/tests/dseldif_test.py | 129 +++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 226 insertions(+)

diff --git a/lib389/_constants.py b/lib389/_constants.py
index 96e5517..1aff17d 100644
--- a/lib389/_constants.py
+++ b/lib389/_constants.py
@@ -69,6 +69,8 @@ DN_DM = "cn=Directory Manager"
 PW_DM = "password"
 DN_CONFIG = "cn=config"
 DN_LDBM = "cn=ldbm database,cn=plugins,cn=config"
+DN_CONFIG_LDBM = "cn=config,cn=ldbm database,cn=plugins,cn=config"
+DN_USERROOT_LDBM = "cn=userRoot,cn=ldbm database,cn=plugins,cn=config"
 DN_SCHEMA = "cn=schema"
 DN_MONITOR = "cn=monitor"
 DN_MONITOR_SNMP = "cn=snmp,cn=monitor"
diff --git a/lib389/dseldif.py b/lib389/dseldif.py
new file mode 100644
index 0000000..39e35b2
--- /dev/null
+++ b/lib389/dseldif.py
@@ -0,0 +1,95 @@
+# --- BEGIN COPYRIGHT BLOCK ---
+# Copyright (C) 2017 Red Hat, Inc.
+# All rights reserved.
+#
+# License: GPL (version 3 or any later version).
+# See LICENSE for details.
+# --- END COPYRIGHT BLOCK ---
+#
+import os
+from lib389.paths import Paths
+
+
+class DSEldif(object):
+    """A class for working with dse.ldif file"""
+
+    def __init__(self, instance):
+        self._instance = instance
+
+        ds_paths = Paths(self._instance.serverid, self._instance)
+        self.path = os.path.join(ds_paths.config_dir, 'dse.ldif')
+
+        with open(self.path, 'r') as file_dse:
+            self._contents = file_dse.readlines()
+
+    def _update(self):
+        """Update the dse.ldif with a new contents"""
+
+        with open(self.path, "w") as file_dse:
+            file_dse.write("".join(self._contents))
+
+    def _find_attr(self, entry_dn, attr):
+        """Find all attribute values and indexes under a given entry
+
+        Returns entry dn index and attribute data dict:
+        relative attribute indexes and the attribute value
+        """
+
+        entry_dn_i = self._contents.index("dn: {}\n".format(entry_dn))
+        attr_data = {}
+
+        # Find where the entry ends
+        try:
+            dn_end_i = self._contents[entry_dn_i:].index("\n")
+        except ValueError:
+            # We are in the end of the list
+            dn_end_i = len(self._contents)
+
+        entry_slice = self._contents[entry_dn_i:entry_dn_i + dn_end_i]
+
+        # Find the attribute
+        for line in entry_slice:
+            if line.startswith(attr):
+                attr_value = line.split(" ", 1)[1][:-1]
+                attr_data.update({entry_slice.index(line): attr_value})
+
+        if not attr_data:
+            raise ValueError("Attribute {} wasn't found under dn:
{}".format(attr, entry_dn))
+
+        return entry_dn_i, attr_data
+
+    def get(self, entry_dn, attr):
+        """Return attribute values under a given entry"""
+
+        _, attr_data = self._find_attr(entry_dn, attr)
+
+        return attr_data.values()
+
+    def add(self, entry_dn, attr, value):
+        """Add an attribute under a given entry"""
+
+        entry_dn_i = self._contents.index("dn: {}\n".format(entry_dn))
+        self._contents.insert(entry_dn_i+1, "{}: {}\n".format(attr, value))
+        self._update()
+
+    def delete(self, entry_dn, attr, value=None):
+        """Delete attributes under a given entry"""
+
+        entry_dn_i, attr_data = self._find_attr(entry_dn, attr)
+
+        if value is not None:
+            for attr_i, attr_value in attr_data.items():
+                if attr_value == value:
+                    del self._contents[entry_dn_i + attr_i]
+        else:
+            for attr_i in sorted(attr_data.keys(), reverse=True):
+                del self._contents[entry_dn_i + attr_i]
+        self._update()
+
+    def replace(self, entry_dn, attr, value):
+        """Replace attribute values with a new one under a given
entry"""
+
+        self.delete(entry_dn, attr)
+        self.add(entry_dn, attr, value)
+        self._update()
+
diff --git a/lib389/tests/dseldif_test.py b/lib389/tests/dseldif_test.py
new file mode 100644
index 0000000..25eef77
--- /dev/null
+++ b/lib389/tests/dseldif_test.py
@@ -0,0 +1,129 @@
+# --- BEGIN COPYRIGHT BLOCK ---
+# Copyright (C) 2017 Red Hat, Inc.
+# All rights reserved.
+#
+# License: GPL (version 3 or any later version).
+# See LICENSE for details.
+# --- END COPYRIGHT BLOCK ---
+#
+import logging
+import pytest
+
+from lib389._constants import *
+from lib389.dseldif import DSEldif
+from lib389.topologies import topology_st as topo
+
+DEBUGGING = os.getenv('DEBUGGING', False)
+
+if DEBUGGING:
+    logging.getLogger(__name__).setLevel(logging.DEBUG)
+else:
+    logging.getLogger(__name__).setLevel(logging.INFO)
+
+log = logging.getLogger(__name__)
+
+
+(a)pytest.mark.parametrize(&quot;entry_dn&quot;, (DN_CONFIG,
+                                      DN_CONFIG_LDBM))
+def test_get_singlevalue(topo, entry_dn):
+    """Check that we can get an attribute value under different
suffixes"""
+
+    dse_ldif = DSEldif(topo.standalone)
+
+    log.info("Get 'cn' attr from {}".format(entry_dn))
+    attr_values = dse_ldif.get(entry_dn, "cn")
+    assert attr_values == ["config"]
+
+
+def test_get_multivalue(topo):
+    """Check that we can get attribute values"""
+
+    dse_ldif = DSEldif(topo.standalone)
+
+    log.info("Get objectClass from {}".format(DN_CONFIG))
+    attr_values = dse_ldif.get(DN_CONFIG, "objectClass")
+    assert len(attr_values) == 3
+    assert "top" in attr_values
+    assert "extensibleObject" in attr_values
+    assert "nsslapdConfig" in attr_values
+
+
+(a)pytest.mark.parametrize(&quot;fake_attr_value&quot;, ("fake value",
+                                             "fakevalue"))
+def test_add(topo, fake_attr_value):
+    """Check that we can add an attribute to a given
suffix"""
+
+    dse_ldif = DSEldif(topo.standalone)
+    fake_attr = "fakeAttr"
+
+    log.info("Add {} to {}".format(fake_attr, DN_CONFIG))
+    dse_ldif.add(DN_CONFIG, fake_attr, fake_attr_value)
+    attr_values = dse_ldif.get(DN_CONFIG, fake_attr)
+    assert attr_values == [fake_attr_value]
+
+    log.info("Clean up")
+    dse_ldif.delete(DN_CONFIG, fake_attr)
+    with pytest.raises(ValueError):
+        dse_ldif.get(DN_CONFIG, fake_attr)
+
+
+def test_replace(topo):
+    """Check that we can replace an attribute to a given
suffix"""
+
+    dse_ldif = DSEldif(topo.standalone)
+    port_attr = "nsslapd-port"
+    port_value = "390"
+
+    log.info("Get default value of {}".format(port_attr))
+    default_value = dse_ldif.get(DN_CONFIG, port_attr)[0]
+
+    log.info("Replace {} with {}".format(port_attr, port_value))
+    dse_ldif.replace(DN_CONFIG, port_attr, port_value)
+    attr_values = dse_ldif.get(DN_CONFIG, port_attr)
+    assert attr_values == [port_value]
+
+    log.info("Restore default value")
+    dse_ldif.replace(DN_CONFIG, port_attr, default_value)
+
+
+def test_delete_singlevalue(topo):
+    """Check that we can delete an attribute from a given
suffix"""
+
+    dse_ldif = DSEldif(topo.standalone)
+    fake_attr = "fakeAttr"
+    fake_attr_values = ["fake1", "fake2", "fake3"]
+
+    log.info("Add multivalued {} to {}".format(fake_attr, DN_CONFIG))
+    for value in fake_attr_values:
+        dse_ldif.add(DN_CONFIG, fake_attr, value)
+
+    log.info("Delete {}".format(fake_attr_values[0]))
+    dse_ldif.delete(DN_CONFIG, fake_attr, fake_attr_values[0])
+    attr_values = dse_ldif.get(DN_CONFIG, fake_attr)
+    assert len(attr_values) == 2
+    assert fake_attr_values[0] not in attr_values
+    assert fake_attr_values[1] in attr_values
+    assert fake_attr_values[2] in attr_values
+
+    log.info("Clean up")
+    dse_ldif.delete(DN_CONFIG, fake_attr)
+    with pytest.raises(ValueError):
+        dse_ldif.get(DN_CONFIG, fake_attr)
+
+
+def test_delete_multivalue(topo):
+    """Check that we can delete attributes from a given
suffix"""
+
+    dse_ldif = DSEldif(topo.standalone)
+    fake_attr = "fakeAttr"
+    fake_attr_values = ["fake1", "fake2", "fake3"]
+
+    log.info("Add multivalued {} to {}".format(fake_attr, DN_CONFIG))
+    for value in fake_attr_values:
+        dse_ldif.add(DN_CONFIG, fake_attr, value)
+
+    log.info("Delete all values of {}".format(fake_attr))
+    dse_ldif.delete(DN_CONFIG, fake_attr)
+    with pytest.raises(ValueError):
+        dse_ldif.get(DN_CONFIG, fake_attr)
+

-- 
To stop receiving notification emails like this one, please contact
the administrator of this repository.