lib/ldb/tests/python/repack.py

   1 import os
   2 from unittest import TestCase
   3 import shutil
   4 from subprocess import check_output
   5 import ldb
   6
   7 TDB_PREFIX = "tdb://"
   8 MDB_PREFIX = "mdb://"
   9
  10 def tempdir():
  11     import tempfile
  12     try:
  13         dir_prefix = os.path.join(os.environ["SELFTEST_PREFIX"], "tmp")
  14     except KeyError:
  15         dir_prefix = None
  16     return tempfile.mkdtemp(dir=dir_prefix)
  17
  18
  19 # Check enabling and disabling GUID indexing works and that the database is
  20 # repacked at version 2 if GUID indexing is enabled, or version 1 if disabled.
  21 class GUIDIndexAndPackFormatTests(TestCase):
  22     prefix = TDB_PREFIX
  23
  24     def setup_newdb(self):
  25         self.testdir = tempdir()
  26         self.filename = os.path.join(self.testdir,
  27                                      "guidpackformattest.ldb")
  28         url = self.prefix + self.filename
  29         self.l = ldb.Ldb(url, options=["modules:"])
  30
  31         self.num_recs_added = 0
  32
  33     #guidindexpackv1.ldb is a pre-made database packed with version 1 format
  34     #but with GUID indexing enabled, which is not allowed, so Samba should
  35     #repack the database on the first transaction.
  36     def setup_premade_v1_db(self):
  37         db_name = "guidindexpackv1.ldb"
  38         this_file_dir = os.path.dirname(os.path.abspath(__file__))
  39         db_path = os.path.join(this_file_dir, "../", db_name)
  40         self.testdir = tempdir()
  41         self.filename = os.path.join(self.testdir, db_name)
  42
  43         shutil.copy(db_path, self.filename)
  44
  45         url = self.prefix + self.filename
  46         self.l = ldb.Ldb(url, options=["modules:"])
  47         self.num_recs_added = 10
  48
  49     def tearDown(self):
  50         if hasattr(self, 'testdir'):
  51             shutil.rmtree(self.testdir)
  52
  53     def add_one_rec(self):
  54         ouuid = 0x0123456789abcdef + self.num_recs_added
  55         ouuid_s = '0' + hex(ouuid)[2:]
  56         dn = "OU=GUIDPFTEST{},DC=SAMBA,DC=ORG".format(self.num_recs_added)
  57         rec = {"dn": dn, "objectUUID": ouuid_s, "distinguishedName": dn}
  58         self.l.add(rec)
  59         self.num_recs_added += 1
  60
  61         # Turn GUID back into a str for easier comparisons
  62         return rec
  63
  64     def set_guid_indexing(self, enable=True):
  65         modmsg = ldb.Message()
  66         modmsg.dn = ldb.Dn(self.l, '@INDEXLIST')
  67
  68         attrs = {"@IDXGUID": [b"objectUUID"],
  69                  "@IDX_DN_GUID": [b"GUID"]}
  70         for attr, val in attrs.items():
  71             replace = ldb.FLAG_MOD_REPLACE
  72             el = val if enable else []
  73             el = ldb.MessageElement(elements=el, flags=replace, name=attr)
  74             modmsg.add(el)
  75
  76         self.l.modify(modmsg)
  77
  78     # Parse out the comments above each record that ldbdump produces
  79     # containing pack format version and KV level key for each record.
  80     # Return all GUID index keys and the set of all unique pack formats.
  81     def ldbdump_guid_keys_pack_formats(self):
  82         dump = check_output(["bin/ldbdump", "-i", self.filename])
  83         dump = dump.decode("utf-8")
  84         dump = dump.split("\n")
  85
  86         comments = [s for s in dump if s.startswith("#")]
  87
  88         guid_key_tag = "# key: GUID="
  89         guid_keys = {c[len(guid_key_tag):] for c in comments
  90                          if c.startswith(guid_key_tag)}
  91
  92         pack_format_tag = "# pack format: "
  93         pack_formats = {c[len(pack_format_tag):] for c in comments
  94                         if c.startswith(pack_format_tag)}
  95         pack_formats = [int(s, 16) for s in pack_formats]
  96
  97         return guid_keys, pack_formats
  98
  99     # Put the whole database in a dict so we can easily check the database
 100     # hasn't changed
 101     def get_database(self):
 102         recs = self.l.search(base="", scope=ldb.SCOPE_SUBTREE, expression="")
 103         db = dict()
 104         for r in recs:
 105             dn = str(r.dn)
 106             self.assertNotIn(dn, db)
 107             db[dn] = dict()
 108             for k in r.keys():
 109                 k = str(k)
 110                 db[dn][k] = str(r.get(k))
 111         return db
 112
 113     # Toggle GUID indexing on and off a few times, and check that when GUID
 114     # indexing is enabled, the database is repacked to pack format V2, and
 115     # when GUID indexing is disabled again, the database is repacked with
 116     # pack format V1.
 117     def toggle_guidindex_check_pack(self):
 118         expect_db = self.get_database()
 119
 120         for enable in [False, False, True, False, True, True, False]:
 121             pf = ldb.PACKING_FORMAT_V2 if enable else ldb.PACKING_FORMAT
 122
 123             self.set_guid_indexing(enable=enable)
 124
 125             guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 126             num_guid_keys = self.num_recs_added if enable else 0
 127             self.assertEqual(len(guid_keys), num_guid_keys)
 128             self.assertEqual(pack_formats, [pf])
 129             self.assertEqual(self.get_database(), expect_db)
 130
 131             rec = self.add_one_rec()
 132             expect_db[rec['dn']] = rec
 133
 134             guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 135             num_guid_keys = self.num_recs_added if enable else 0
 136             self.assertEqual(len(guid_keys), num_guid_keys)
 137             self.assertEqual(pack_formats, [pf])
 138             self.assertEqual(self.get_database(), expect_db)
 139
 140     # Check a newly created database is initially packed at V1, then is
 141     # repacked at V2 when GUID indexing is enabled.
 142     def test_repack(self):
 143         self.setup_newdb()
 144
 145         guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 146         self.assertEqual(len(guid_keys), 0)
 147         self.assertEqual(pack_formats, [ldb.PACKING_FORMAT])
 148         self.assertEqual(self.get_database(), {})
 149
 150         self.l.add({"dn": "@ATTRIBUTES"})
 151
 152         guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 153         self.assertEqual(len(guid_keys), 0)
 154         self.assertEqual(pack_formats, [ldb.PACKING_FORMAT])
 155         self.assertEqual(self.get_database(), {})
 156
 157         self.l.add({"dn": "@INDEXLIST",
 158                     "@IDXONE": [b"1"],
 159                     "@IDXGUID": [b"objectUUID"],
 160                     "@IDX_DN_GUID": [b"GUID"]})
 161
 162         guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 163         self.assertEqual(len(guid_keys), 0)
 164         self.assertEqual(pack_formats, [ldb.PACKING_FORMAT_V2])
 165         self.assertEqual(self.get_database(), {})
 166
 167         rec = self.add_one_rec()
 168         expect_db = {rec["dn"]: rec}
 169
 170         guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 171         self.assertEqual(len(guid_keys), 1)
 172         self.assertEqual(pack_formats, [ldb.PACKING_FORMAT_V2])
 173         self.assertEqual(self.get_database(), expect_db)
 174
 175         self.toggle_guidindex_check_pack()
 176
 177     # Check a database with V1 format with GUID indexing enabled is repacked
 178     # with version 2 format.
 179     def test_guid_indexed_v1_db(self):
 180         self.setup_premade_v1_db()
 181
 182         expect_db = self.get_database()
 183
 184         guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 185         self.assertEqual(len(guid_keys), self.num_recs_added)
 186         self.assertEqual(pack_formats, [ldb.PACKING_FORMAT])
 187         self.assertEqual(self.get_database(), expect_db)
 188
 189         rec = self.add_one_rec()
 190         expect_db[rec['dn']] = rec
 191
 192         guid_keys, pack_formats = self.ldbdump_guid_keys_pack_formats()
 193         self.assertEqual(len(guid_keys), self.num_recs_added)
 194         self.assertEqual(pack_formats, [ldb.PACKING_FORMAT_V2])
 195         self.assertEqual(self.get_database(), expect_db)
 196
 197         self.toggle_guidindex_check_pack()
 198
 199
 200 if __name__ == '__main__':
 201     import unittest
 202
 203
 204     unittest.TestProgram()