ldb: Add test to show a reindex failure must not leave the DB corrupt

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13335

Signed-off-by: Andrew Bartlett <abartlet@samba.org>
Reviewed-by: Gary Lockyer <gary@catalyst.net.nz>
Autobuild-User(master): Andrew Bartlett <abartlet@samba.org>
Autobuild-Date(master): Thu Apr  5 07:53:10 CEST 2018 on sn-devel-144

diff --git a/lib/ldb/tests/python/api.py b/lib/ldb/tests/python/api.py
index 51c9c592f78caa4abb9f82eb58799f81caed2aa2..12409a8c9917e1ec15e1e90971384bc90d70add7 100755 (executable)
--- a/lib/ldb/tests/python/api.py
+++ b/lib/ldb/tests/python/api.py
@@ -1408,6 +1408,166 @@ class TransIndexedAddModifyTests(IndexedAddModifyTests):
         super(TransIndexedAddModifyTests, self).tearDown()
 
 
+class BadIndexTests(LdbBaseTest):
+    def setUp(self):
+        super(BadIndexTests, self).setUp()
+        self.testdir = tempdir()
+        self.filename = os.path.join(self.testdir, "test.ldb")
+        self.ldb = ldb.Ldb(self.url(), flags=self.flags())
+        if hasattr(self, 'IDXGUID'):
+            self.ldb.add({"dn": "@INDEXLIST",
+                          "@IDXATTR": [b"x", b"y", b"ou"],
+                          "@IDXGUID": [b"objectUUID"],
+                          "@IDX_DN_GUID": [b"GUID"]})
+        else:
+            self.ldb.add({"dn": "@INDEXLIST",
+                          "@IDXATTR": [b"x", b"y", b"ou"]})
+
+        super(BadIndexTests, self).setUp()
+
+    def test_unique(self):
+        self.ldb.add({"dn": "x=x,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde1",
+                      "y": "1"})
+        self.ldb.add({"dn": "x=y,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde2",
+                      "y": "1"})
+        self.ldb.add({"dn": "x=z,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde3",
+                      "y": "1"})
+
+        res = self.ldb.search(expression="(y=1)",
+                              base="dc=samba,dc=org")
+        self.assertEquals(len(res), 3)
+
+        # Now set this to unique index, but forget to check the result
+        try:
+            self.ldb.add({"dn": "@ATTRIBUTES",
+                        "y": "UNIQUE_INDEX"})
+            self.fail()
+        except ldb.LdbError:
+            pass
+
+        # We must still have a working index
+        res = self.ldb.search(expression="(y=1)",
+                              base="dc=samba,dc=org")
+        self.assertEquals(len(res), 3)
+
+    def test_unique_transaction(self):
+        self.ldb.add({"dn": "x=x,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde1",
+                      "y": "1"})
+        self.ldb.add({"dn": "x=y,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde2",
+                      "y": "1"})
+        self.ldb.add({"dn": "x=z,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde3",
+                      "y": "1"})
+
+        res = self.ldb.search(expression="(y=1)",
+                              base="dc=samba,dc=org")
+        self.assertEquals(len(res), 3)
+
+        self.ldb.transaction_start()
+
+        # Now set this to unique index, but forget to check the result
+        try:
+            self.ldb.add({"dn": "@ATTRIBUTES",
+                        "y": "UNIQUE_INDEX"})
+        except ldb.LdbError:
+            pass
+
+        try:
+            self.ldb.transaction_commit()
+            self.fail()
+
+        except ldb.LdbError as err:
+            enum = err.args[0]
+            self.assertEqual(enum, ldb.ERR_OPERATIONS_ERROR)
+
+        # We must still have a working index
+        res = self.ldb.search(expression="(y=1)",
+                              base="dc=samba,dc=org")
+
+        self.assertEquals(len(res), 3)
+
+    def test_casefold(self):
+        self.ldb.add({"dn": "x=x,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde1",
+                      "y": "a"})
+        self.ldb.add({"dn": "x=y,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde2",
+                      "y": "A"})
+        self.ldb.add({"dn": "x=z,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde3",
+                      "y": ["a", "A"]})
+
+        res = self.ldb.search(expression="(y=a)",
+                              base="dc=samba,dc=org")
+        self.assertEquals(len(res), 2)
+
+        self.ldb.add({"dn": "@ATTRIBUTES",
+                      "y": "CASE_INSENSITIVE"})
+
+        # We must still have a working index
+        res = self.ldb.search(expression="(y=a)",
+                              base="dc=samba,dc=org")
+
+        if hasattr(self, 'IDXGUID'):
+            self.assertEquals(len(res), 3)
+        else:
+            # We should not return this entry twice, but sadly
+            # we have not yet fixed
+            # https://bugzilla.samba.org/show_bug.cgi?id=13361
+            self.assertEquals(len(res), 4)
+
+    def test_casefold_transaction(self):
+        self.ldb.add({"dn": "x=x,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde1",
+                      "y": "a"})
+        self.ldb.add({"dn": "x=y,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde2",
+                      "y": "A"})
+        self.ldb.add({"dn": "x=z,dc=samba,dc=org",
+                      "objectUUID": b"0123456789abcde3",
+                      "y": ["a", "A"]})
+
+        res = self.ldb.search(expression="(y=a)",
+                              base="dc=samba,dc=org")
+        self.assertEquals(len(res), 2)
+
+        self.ldb.transaction_start()
+
+        self.ldb.add({"dn": "@ATTRIBUTES",
+                      "y": "CASE_INSENSITIVE"})
+
+        self.ldb.transaction_commit()
+
+        # We must still have a working index
+        res = self.ldb.search(expression="(y=a)",
+                              base="dc=samba,dc=org")
+
+        if hasattr(self, 'IDXGUID'):
+            self.assertEquals(len(res), 3)
+        else:
+            # We should not return this entry twice, but sadly
+            # we have not yet fixed
+            # https://bugzilla.samba.org/show_bug.cgi?id=13361
+            self.assertEquals(len(res), 4)
+
+
+    def tearDown(self):
+        super(BadIndexTests, self).tearDown()
+
+
+class GUIDBadIndexTests(BadIndexTests):
+    """Test Bad index things with GUID index mode"""
+    def setUp(self):
+        self.IDXGUID = True
+
+        super(GUIDBadIndexTests, self).setUp()
+
+
 class DnTests(TestCase):
 
     def setUp(self):