lib/ldb/tests/python/api.py

   1 #!/usr/bin/env python
   2 # Simple tests for the ldb python bindings.
   3 # Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
   4
   5 import os
   6 from unittest import TestCase
   7 import sys
   8 import gc
   9 import time
  10 import ldb
  11 import shutil
  12
  13 PY3 = sys.version_info > (3, 0)
  14
  15
  16 def tempdir():
  17     import tempfile
  18     try:
  19         dir_prefix = os.path.join(os.environ["SELFTEST_PREFIX"], "tmp")
  20     except KeyError:
  21         dir_prefix = None
  22     return tempfile.mkdtemp(dir=dir_prefix)
  23
  24
  25 class NoContextTests(TestCase):
  26
  27     def test_valid_attr_name(self):
  28         self.assertTrue(ldb.valid_attr_name("foo"))
  29         self.assertFalse(ldb.valid_attr_name("24foo"))
  30
  31     def test_timestring(self):
  32         self.assertEqual("19700101000000.0Z", ldb.timestring(0))
  33         self.assertEqual("20071119191012.0Z", ldb.timestring(1195499412))
  34
  35     def test_string_to_time(self):
  36         self.assertEqual(0, ldb.string_to_time("19700101000000.0Z"))
  37         self.assertEqual(1195499412, ldb.string_to_time("20071119191012.0Z"))
  38
  39     def test_binary_encode(self):
  40         encoded = ldb.binary_encode(b'test\\x')
  41         decoded = ldb.binary_decode(encoded)
  42         self.assertEqual(decoded, b'test\\x')
  43
  44         encoded2 = ldb.binary_encode('test\\x')
  45         self.assertEqual(encoded2, encoded)
  46
  47 class SimpleLdb(TestCase):
  48
  49     def setUp(self):
  50         super(SimpleLdb, self).setUp()
  51         self.testdir = tempdir()
  52         self.filename = os.path.join(self.testdir, "test.ldb")
  53         self.ldb = ldb.Ldb(self.filename)
  54
  55     def tearDown(self):
  56         shutil.rmtree(self.testdir)
  57         super(SimpleLdb, self).tearDown()
  58
  59     def test_connect(self):
  60         ldb.Ldb(self.filename)
  61
  62     def test_connect_none(self):
  63         ldb.Ldb()
  64
  65     def test_connect_later(self):
  66         x = ldb.Ldb()
  67         x.connect(self.filename)
  68
  69     def test_repr(self):
  70         x = ldb.Ldb()
  71         self.assertTrue(repr(x).startswith("<ldb connection"))
  72
  73     def test_set_create_perms(self):
  74         x = ldb.Ldb()
  75         x.set_create_perms(0o600)
  76
  77     def test_modules_none(self):
  78         x = ldb.Ldb()
  79         self.assertEqual([], x.modules())
  80
  81     def test_modules_tdb(self):
  82         x = ldb.Ldb(self.filename)
  83         self.assertEqual("[<ldb module 'tdb'>]", repr(x.modules()))
  84
  85     def test_firstmodule_none(self):
  86         x = ldb.Ldb()
  87         self.assertEqual(x.firstmodule, None)
  88
  89     def test_firstmodule_tdb(self):
  90         x = ldb.Ldb(self.filename)
  91         mod = x.firstmodule
  92         self.assertEqual(repr(mod), "<ldb module 'tdb'>")
  93
  94     def test_search(self):
  95         l = ldb.Ldb(self.filename)
  96         self.assertEqual(len(l.search()), 0)
  97
  98     def test_search_controls(self):
  99         l = ldb.Ldb(self.filename)
 100         self.assertEqual(len(l.search(controls=["paged_results:0:5"])), 0)
 101
 102     def test_search_attrs(self):
 103         l = ldb.Ldb(self.filename)
 104         self.assertEqual(len(l.search(ldb.Dn(l, ""), ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
 105
 106     def test_search_string_dn(self):
 107         l = ldb.Ldb(self.filename)
 108         self.assertEqual(len(l.search("", ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
 109
 110     def test_search_attr_string(self):
 111         l = ldb.Ldb(self.filename)
 112         self.assertRaises(TypeError, l.search, attrs="dc")
 113         self.assertRaises(TypeError, l.search, attrs=b"dc")
 114
 115     def test_opaque(self):
 116         l = ldb.Ldb(self.filename)
 117         l.set_opaque("my_opaque", l)
 118         self.assertTrue(l.get_opaque("my_opaque") is not None)
 119         self.assertEqual(None, l.get_opaque("unknown"))
 120
 121     def test_search_scope_base_empty_db(self):
 122         l = ldb.Ldb(self.filename)
 123         self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
 124                           ldb.SCOPE_BASE)), 0)
 125
 126     def test_search_scope_onelevel_empty_db(self):
 127         l = ldb.Ldb(self.filename)
 128         self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
 129                           ldb.SCOPE_ONELEVEL)), 0)
 130
 131     def test_delete(self):
 132         l = ldb.Ldb(self.filename)
 133         self.assertRaises(ldb.LdbError, lambda: l.delete(ldb.Dn(l, "dc=foo2")))
 134
 135     def test_delete_w_unhandled_ctrl(self):
 136         l = ldb.Ldb(self.filename)
 137         m = ldb.Message()
 138         m.dn = ldb.Dn(l, "dc=foo1")
 139         m["b"] = [b"a"]
 140         l.add(m)
 141         self.assertRaises(ldb.LdbError, lambda: l.delete(m.dn, ["search_options:1:2"]))
 142         l.delete(m.dn)
 143
 144     def test_contains(self):
 145         name = self.filename
 146         l = ldb.Ldb(name)
 147         self.assertFalse(ldb.Dn(l, "dc=foo3") in l)
 148         l = ldb.Ldb(name)
 149         m = ldb.Message()
 150         m.dn = ldb.Dn(l, "dc=foo3")
 151         m["b"] = ["a"]
 152         l.add(m)
 153         try:
 154             self.assertTrue(ldb.Dn(l, "dc=foo3") in l)
 155             self.assertFalse(ldb.Dn(l, "dc=foo4") in l)
 156         finally:
 157             l.delete(m.dn)
 158
 159     def test_get_config_basedn(self):
 160         l = ldb.Ldb(self.filename)
 161         self.assertEqual(None, l.get_config_basedn())
 162
 163     def test_get_root_basedn(self):
 164         l = ldb.Ldb(self.filename)
 165         self.assertEqual(None, l.get_root_basedn())
 166
 167     def test_get_schema_basedn(self):
 168         l = ldb.Ldb(self.filename)
 169         self.assertEqual(None, l.get_schema_basedn())
 170
 171     def test_get_default_basedn(self):
 172         l = ldb.Ldb(self.filename)
 173         self.assertEqual(None, l.get_default_basedn())
 174
 175     def test_add(self):
 176         l = ldb.Ldb(self.filename)
 177         m = ldb.Message()
 178         m.dn = ldb.Dn(l, "dc=foo4")
 179         m["bla"] = b"bla"
 180         self.assertEqual(len(l.search()), 0)
 181         l.add(m)
 182         try:
 183             self.assertEqual(len(l.search()), 1)
 184         finally:
 185             l.delete(ldb.Dn(l, "dc=foo4"))
 186
 187     def test_search_iterator(self):
 188         l = ldb.Ldb(self.filename)
 189         s = l.search_iterator()
 190         s.abandon()
 191         try:
 192             for me in s:
 193                 self.fail()
 194             self.fail()
 195         except RuntimeError as re:
 196             pass
 197         try:
 198             s.abandon()
 199             self.fail()
 200         except RuntimeError as re:
 201             pass
 202         try:
 203             s.result()
 204             self.fail()
 205         except RuntimeError as re:
 206             pass
 207
 208         s = l.search_iterator()
 209         count = 0
 210         for me in s:
 211             self.assertTrue(isinstance(me, ldb.Message))
 212             count += 1
 213         r = s.result()
 214         self.assertEqual(len(r), 0)
 215         self.assertEqual(count, 0)
 216
 217         m1 = ldb.Message()
 218         m1.dn = ldb.Dn(l, "dc=foo4")
 219         m1["bla"] = b"bla"
 220         l.add(m1)
 221         try:
 222             s = l.search_iterator()
 223             msgs = []
 224             for me in s:
 225                 self.assertTrue(isinstance(me, ldb.Message))
 226                 count += 1
 227                 msgs.append(me)
 228             r = s.result()
 229             self.assertEqual(len(r), 0)
 230             self.assertEqual(len(msgs), 1)
 231             self.assertEqual(msgs[0].dn, m1.dn)
 232
 233             m2 = ldb.Message()
 234             m2.dn = ldb.Dn(l, "dc=foo5")
 235             m2["bla"] = b"bla"
 236             l.add(m2)
 237
 238             s = l.search_iterator()
 239             msgs = []
 240             for me in s:
 241                 self.assertTrue(isinstance(me, ldb.Message))
 242                 count += 1
 243                 msgs.append(me)
 244             r = s.result()
 245             self.assertEqual(len(r), 0)
 246             self.assertEqual(len(msgs), 2)
 247             if msgs[0].dn == m1.dn:
 248                 self.assertEqual(msgs[0].dn, m1.dn)
 249                 self.assertEqual(msgs[1].dn, m2.dn)
 250             else:
 251                 self.assertEqual(msgs[0].dn, m2.dn)
 252                 self.assertEqual(msgs[1].dn, m1.dn)
 253
 254             s = l.search_iterator()
 255             msgs = []
 256             for me in s:
 257                 self.assertTrue(isinstance(me, ldb.Message))
 258                 count += 1
 259                 msgs.append(me)
 260                 break
 261             try:
 262                 s.result()
 263                 self.fail()
 264             except RuntimeError as re:
 265                 pass
 266             for me in s:
 267                 self.assertTrue(isinstance(me, ldb.Message))
 268                 count += 1
 269                 msgs.append(me)
 270                 break
 271             for me in s:
 272                 self.fail()
 273
 274             r = s.result()
 275             self.assertEqual(len(r), 0)
 276             self.assertEqual(len(msgs), 2)
 277             if msgs[0].dn == m1.dn:
 278                 self.assertEqual(msgs[0].dn, m1.dn)
 279                 self.assertEqual(msgs[1].dn, m2.dn)
 280             else:
 281                 self.assertEqual(msgs[0].dn, m2.dn)
 282                 self.assertEqual(msgs[1].dn, m1.dn)
 283         finally:
 284             l.delete(ldb.Dn(l, "dc=foo4"))
 285             l.delete(ldb.Dn(l, "dc=foo5"))
 286
 287     def test_add_text(self):
 288         l = ldb.Ldb(self.filename)
 289         m = ldb.Message()
 290         m.dn = ldb.Dn(l, "dc=foo4")
 291         m["bla"] = "bla"
 292         self.assertEqual(len(l.search()), 0)
 293         l.add(m)
 294         try:
 295             self.assertEqual(len(l.search()), 1)
 296         finally:
 297             l.delete(ldb.Dn(l, "dc=foo4"))
 298
 299     def test_add_w_unhandled_ctrl(self):
 300         l = ldb.Ldb(self.filename)
 301         m = ldb.Message()
 302         m.dn = ldb.Dn(l, "dc=foo4")
 303         m["bla"] = b"bla"
 304         self.assertEqual(len(l.search()), 0)
 305         self.assertRaises(ldb.LdbError, lambda: l.add(m,["search_options:1:2"]))
 306
 307     def test_add_dict(self):
 308         l = ldb.Ldb(self.filename)
 309         m = {"dn": ldb.Dn(l, "dc=foo5"),
 310              "bla": b"bla"}
 311         self.assertEqual(len(l.search()), 0)
 312         l.add(m)
 313         try:
 314             self.assertEqual(len(l.search()), 1)
 315         finally:
 316             l.delete(ldb.Dn(l, "dc=foo5"))
 317
 318     def test_add_dict_text(self):
 319         l = ldb.Ldb(self.filename)
 320         m = {"dn": ldb.Dn(l, "dc=foo5"),
 321              "bla": "bla"}
 322         self.assertEqual(len(l.search()), 0)
 323         l.add(m)
 324         try:
 325             self.assertEqual(len(l.search()), 1)
 326         finally:
 327             l.delete(ldb.Dn(l, "dc=foo5"))
 328
 329     def test_add_dict_string_dn(self):
 330         l = ldb.Ldb(self.filename)
 331         m = {"dn": "dc=foo6", "bla": b"bla"}
 332         self.assertEqual(len(l.search()), 0)
 333         l.add(m)
 334         try:
 335             self.assertEqual(len(l.search()), 1)
 336         finally:
 337             l.delete(ldb.Dn(l, "dc=foo6"))
 338
 339     def test_add_dict_bytes_dn(self):
 340         l = ldb.Ldb(self.filename)
 341         m = {"dn": b"dc=foo6", "bla": b"bla"}
 342         self.assertEqual(len(l.search()), 0)
 343         l.add(m)
 344         try:
 345             self.assertEqual(len(l.search()), 1)
 346         finally:
 347             l.delete(ldb.Dn(l, "dc=foo6"))
 348
 349     def test_rename(self):
 350         l = ldb.Ldb(self.filename)
 351         m = ldb.Message()
 352         m.dn = ldb.Dn(l, "dc=foo7")
 353         m["bla"] = b"bla"
 354         self.assertEqual(len(l.search()), 0)
 355         l.add(m)
 356         try:
 357             l.rename(ldb.Dn(l, "dc=foo7"), ldb.Dn(l, "dc=bar"))
 358             self.assertEqual(len(l.search()), 1)
 359         finally:
 360             l.delete(ldb.Dn(l, "dc=bar"))
 361
 362     def test_rename_string_dns(self):
 363         l = ldb.Ldb(self.filename)
 364         m = ldb.Message()
 365         m.dn = ldb.Dn(l, "dc=foo8")
 366         m["bla"] = b"bla"
 367         self.assertEqual(len(l.search()), 0)
 368         l.add(m)
 369         self.assertEqual(len(l.search()), 1)
 370         try:
 371             l.rename("dc=foo8", "dc=bar")
 372             self.assertEqual(len(l.search()), 1)
 373         finally:
 374             l.delete(ldb.Dn(l, "dc=bar"))
 375
 376     def test_empty_dn(self):
 377         l = ldb.Ldb(self.filename)
 378         self.assertEqual(0, len(l.search()))
 379         m = ldb.Message()
 380         m.dn = ldb.Dn(l, "dc=empty")
 381         l.add(m)
 382         rm = l.search()
 383         self.assertEqual(1, len(rm))
 384         self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
 385
 386         rm = l.search(m.dn)
 387         self.assertEqual(1, len(rm))
 388         self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
 389         rm = l.search(m.dn, attrs=["blah"])
 390         self.assertEqual(1, len(rm))
 391         self.assertEqual(0, len(rm[0]))
 392
 393     def test_modify_delete(self):
 394         l = ldb.Ldb(self.filename)
 395         m = ldb.Message()
 396         m.dn = ldb.Dn(l, "dc=modifydelete")
 397         m["bla"] = [b"1234"]
 398         l.add(m)
 399         rm = l.search(m.dn)[0]
 400         self.assertEqual([b"1234"], list(rm["bla"]))
 401         try:
 402             m = ldb.Message()
 403             m.dn = ldb.Dn(l, "dc=modifydelete")
 404             m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
 405             self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
 406             l.modify(m)
 407             rm = l.search(m.dn)
 408             self.assertEqual(1, len(rm))
 409             self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
 410             rm = l.search(m.dn, attrs=["bla"])
 411             self.assertEqual(1, len(rm))
 412             self.assertEqual(0, len(rm[0]))
 413         finally:
 414             l.delete(ldb.Dn(l, "dc=modifydelete"))
 415
 416     def test_modify_delete_text(self):
 417         l = ldb.Ldb(self.filename)
 418         m = ldb.Message()
 419         m.dn = ldb.Dn(l, "dc=modifydelete")
 420         m.text["bla"] = ["1234"]
 421         l.add(m)
 422         rm = l.search(m.dn)[0]
 423         self.assertEqual(["1234"], list(rm.text["bla"]))
 424         try:
 425             m = ldb.Message()
 426             m.dn = ldb.Dn(l, "dc=modifydelete")
 427             m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
 428             self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
 429             l.modify(m)
 430             rm = l.search(m.dn)
 431             self.assertEqual(1, len(rm))
 432             self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
 433             rm = l.search(m.dn, attrs=["bla"])
 434             self.assertEqual(1, len(rm))
 435             self.assertEqual(0, len(rm[0]))
 436         finally:
 437             l.delete(ldb.Dn(l, "dc=modifydelete"))
 438
 439     def test_modify_add(self):
 440         l = ldb.Ldb(self.filename)
 441         m = ldb.Message()
 442         m.dn = ldb.Dn(l, "dc=add")
 443         m["bla"] = [b"1234"]
 444         l.add(m)
 445         try:
 446             m = ldb.Message()
 447             m.dn = ldb.Dn(l, "dc=add")
 448             m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
 449             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
 450             l.modify(m)
 451             rm = l.search(m.dn)[0]
 452             self.assertEqual(2, len(rm))
 453             self.assertEqual([b"1234", b"456"], list(rm["bla"]))
 454         finally:
 455             l.delete(ldb.Dn(l, "dc=add"))
 456
 457     def test_modify_add_text(self):
 458         l = ldb.Ldb(self.filename)
 459         m = ldb.Message()
 460         m.dn = ldb.Dn(l, "dc=add")
 461         m.text["bla"] = ["1234"]
 462         l.add(m)
 463         try:
 464             m = ldb.Message()
 465             m.dn = ldb.Dn(l, "dc=add")
 466             m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
 467             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
 468             l.modify(m)
 469             rm = l.search(m.dn)[0]
 470             self.assertEqual(2, len(rm))
 471             self.assertEqual(["1234", "456"], list(rm.text["bla"]))
 472         finally:
 473             l.delete(ldb.Dn(l, "dc=add"))
 474
 475     def test_modify_replace(self):
 476         l = ldb.Ldb(self.filename)
 477         m = ldb.Message()
 478         m.dn = ldb.Dn(l, "dc=modify2")
 479         m["bla"] = [b"1234", b"456"]
 480         l.add(m)
 481         try:
 482             m = ldb.Message()
 483             m.dn = ldb.Dn(l, "dc=modify2")
 484             m["bla"] = ldb.MessageElement([b"789"], ldb.FLAG_MOD_REPLACE, "bla")
 485             self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
 486             l.modify(m)
 487             rm = l.search(m.dn)[0]
 488             self.assertEqual(2, len(rm))
 489             self.assertEqual([b"789"], list(rm["bla"]))
 490             rm = l.search(m.dn, attrs=["bla"])[0]
 491             self.assertEqual(1, len(rm))
 492         finally:
 493             l.delete(ldb.Dn(l, "dc=modify2"))
 494
 495     def test_modify_replace_text(self):
 496         l = ldb.Ldb(self.filename)
 497         m = ldb.Message()
 498         m.dn = ldb.Dn(l, "dc=modify2")
 499         m.text["bla"] = ["1234", "456"]
 500         l.add(m)
 501         try:
 502             m = ldb.Message()
 503             m.dn = ldb.Dn(l, "dc=modify2")
 504             m["bla"] = ldb.MessageElement(["789"], ldb.FLAG_MOD_REPLACE, "bla")
 505             self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
 506             l.modify(m)
 507             rm = l.search(m.dn)[0]
 508             self.assertEqual(2, len(rm))
 509             self.assertEqual(["789"], list(rm.text["bla"]))
 510             rm = l.search(m.dn, attrs=["bla"])[0]
 511             self.assertEqual(1, len(rm))
 512         finally:
 513             l.delete(ldb.Dn(l, "dc=modify2"))
 514
 515     def test_modify_flags_change(self):
 516         l = ldb.Ldb(self.filename)
 517         m = ldb.Message()
 518         m.dn = ldb.Dn(l, "dc=add")
 519         m["bla"] = [b"1234"]
 520         l.add(m)
 521         try:
 522             m = ldb.Message()
 523             m.dn = ldb.Dn(l, "dc=add")
 524             m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
 525             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
 526             l.modify(m)
 527             rm = l.search(m.dn)[0]
 528             self.assertEqual(2, len(rm))
 529             self.assertEqual([b"1234", b"456"], list(rm["bla"]))
 530
 531             # Now create another modify, but switch the flags before we do it
 532             m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
 533             m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
 534             l.modify(m)
 535             rm = l.search(m.dn, attrs=["bla"])[0]
 536             self.assertEqual(1, len(rm))
 537             self.assertEqual([b"1234"], list(rm["bla"]))
 538         finally:
 539             l.delete(ldb.Dn(l, "dc=add"))
 540
 541     def test_modify_flags_change_text(self):
 542         l = ldb.Ldb(self.filename)
 543         m = ldb.Message()
 544         m.dn = ldb.Dn(l, "dc=add")
 545         m.text["bla"] = ["1234"]
 546         l.add(m)
 547         try:
 548             m = ldb.Message()
 549             m.dn = ldb.Dn(l, "dc=add")
 550             m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
 551             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
 552             l.modify(m)
 553             rm = l.search(m.dn)[0]
 554             self.assertEqual(2, len(rm))
 555             self.assertEqual(["1234", "456"], list(rm.text["bla"]))
 556
 557             # Now create another modify, but switch the flags before we do it
 558             m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
 559             m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
 560             l.modify(m)
 561             rm = l.search(m.dn, attrs=["bla"])[0]
 562             self.assertEqual(1, len(rm))
 563             self.assertEqual(["1234"], list(rm.text["bla"]))
 564         finally:
 565             l.delete(ldb.Dn(l, "dc=add"))
 566
 567     def test_transaction_commit(self):
 568         l = ldb.Ldb(self.filename)
 569         l.transaction_start()
 570         m = ldb.Message(ldb.Dn(l, "dc=foo9"))
 571         m["foo"] = [b"bar"]
 572         l.add(m)
 573         l.transaction_commit()
 574         l.delete(m.dn)
 575
 576     def test_transaction_cancel(self):
 577         l = ldb.Ldb(self.filename)
 578         l.transaction_start()
 579         m = ldb.Message(ldb.Dn(l, "dc=foo10"))
 580         m["foo"] = [b"bar"]
 581         l.add(m)
 582         l.transaction_cancel()
 583         self.assertEqual(0, len(l.search(ldb.Dn(l, "dc=foo10"))))
 584
 585     def test_set_debug(self):
 586         def my_report_fn(level, text):
 587             pass
 588         l = ldb.Ldb(self.filename)
 589         l.set_debug(my_report_fn)
 590
 591     def test_zero_byte_string(self):
 592         """Testing we do not get trapped in the \0 byte in a property string."""
 593         l = ldb.Ldb(self.filename)
 594         l.add({
 595             "dn" : b"dc=somedn",
 596             "objectclass" : b"user",
 597             "cN" : b"LDAPtestUSER",
 598             "givenname" : b"ldap",
 599             "displayname" : b"foo\0bar",
 600         })
 601         res = l.search(expression="(dn=dc=somedn)")
 602         self.assertEqual(b"foo\0bar", res[0]["displayname"][0])
 603
 604     def test_no_crash_broken_expr(self):
 605         l = ldb.Ldb(self.filename)
 606         self.assertRaises(ldb.LdbError,lambda: l.search("", ldb.SCOPE_SUBTREE, "&(dc=*)(dn=*)", ["dc"]))
 607
 608 class SearchTests(TestCase):
 609     def tearDown(self):
 610         shutil.rmtree(self.testdir)
 611         super(SearchTests, self).tearDown()
 612
 613     def setUp(self):
 614         super(SearchTests, self).setUp()
 615         self.testdir = tempdir()
 616         self.filename = os.path.join(self.testdir, "search_test.ldb")
 617         self.l = ldb.Ldb(self.filename, options=["modules:rdn_name"])
 618
 619         self.l.add({"dn": "@ATTRIBUTES",
 620                     "DC": "CASE_INSENSITIVE"})
 621
 622         # Note that we can't use the name objectGUID here, as we
 623         # want to stay clear of the objectGUID handler in LDB and
 624         # instead use just the 16 bytes raw, which we just keep
 625         # to printable chars here for ease of handling.
 626
 627         self.l.add({"dn": "DC=SAMBA,DC=ORG",
 628                     "name": b"samba.org",
 629                     "objectUUID": b"0123456789abcddf"})
 630         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
 631                     "name": b"Admins",
 632                     "x": "z", "y": "a",
 633                     "objectUUID": b"0123456789abcde1"})
 634         self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG",
 635                     "name": b"Users",
 636                     "x": "z", "y": "a",
 637                     "objectUUID": b"0123456789abcde2"})
 638         self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG",
 639                     "name": b"OU #1",
 640                     "x": "y", "y": "a",
 641                     "objectUUID": b"0123456789abcde3"})
 642         self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG",
 643                     "name": b"OU #2",
 644                     "x": "y", "y": "a",
 645                     "objectUUID": b"0123456789abcde4"})
 646         self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG",
 647                     "name": b"OU #3",
 648                     "x": "y", "y": "a",
 649                     "objectUUID": b"0123456789abcde5"})
 650         self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG",
 651                     "name": b"OU #4",
 652                     "x": "y", "y": "a",
 653                     "objectUUID": b"0123456789abcde6"})
 654         self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG",
 655                     "name": b"OU #5",
 656                     "x": "y", "y": "a",
 657                     "objectUUID": b"0123456789abcde7"})
 658         self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG",
 659                     "name": b"OU #6",
 660                     "x": "y", "y": "a",
 661                     "objectUUID": b"0123456789abcde8"})
 662         self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG",
 663                     "name": b"OU #7",
 664                     "x": "y", "y": "a",
 665                     "objectUUID": b"0123456789abcde9"})
 666         self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG",
 667                     "name": b"OU #8",
 668                     "x": "y", "y": "a",
 669                     "objectUUID": b"0123456789abcde0"})
 670         self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG",
 671                     "name": b"OU #9",
 672                     "x": "y", "y": "a",
 673                     "objectUUID": b"0123456789abcdea"})
 674         self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG",
 675                     "name": b"OU #10",
 676                     "x": "y", "y": "a",
 677                     "objectUUID": b"0123456789abcdeb"})
 678         self.l.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
 679                     "name": b"OU #10",
 680                     "x": "y", "y": "a",
 681                     "objectUUID": b"0123456789abcdec"})
 682         self.l.add({"dn": "OU=OU12,DC=SAMBA,DC=ORG",
 683                     "name": b"OU #10",
 684                     "x": "y", "y": "b",
 685                     "objectUUID": b"0123456789abcded"})
 686         self.l.add({"dn": "OU=OU13,DC=SAMBA,DC=ORG",
 687                     "name": b"OU #10",
 688                     "x": "x", "y": "b",
 689                     "objectUUID": b"0123456789abcdee"})
 690         self.l.add({"dn": "OU=OU14,DC=SAMBA,DC=ORG",
 691                     "name": b"OU #10",
 692                     "x": "x", "y": "b",
 693                     "objectUUID": b"0123456789abcd01"})
 694         self.l.add({"dn": "OU=OU15,DC=SAMBA,DC=ORG",
 695                     "name": b"OU #10",
 696                     "x": "x", "y": "b",
 697                     "objectUUID": b"0123456789abcd02"})
 698         self.l.add({"dn": "OU=OU16,DC=SAMBA,DC=ORG",
 699                     "name": b"OU #10",
 700                     "x": "x", "y": "b",
 701                     "objectUUID": b"0123456789abcd03"})
 702         self.l.add({"dn": "OU=OU17,DC=SAMBA,DC=ORG",
 703                     "name": b"OU #10",
 704                     "x": "x", "y": "b",
 705                     "objectUUID": b"0123456789abcd04"})
 706         self.l.add({"dn": "OU=OU18,DC=SAMBA,DC=ORG",
 707                     "name": b"OU #10",
 708                     "x": "x", "y": "b",
 709                     "objectUUID": b"0123456789abcd05"})
 710         self.l.add({"dn": "OU=OU19,DC=SAMBA,DC=ORG",
 711                     "name": b"OU #10",
 712                     "x": "x", "y": "b",
 713                     "objectUUID": b"0123456789abcd06"})
 714         self.l.add({"dn": "OU=OU20,DC=SAMBA,DC=ORG",
 715                     "name": b"OU #10",
 716                     "x": "x", "y": "b",
 717                     "objectUUID": b"0123456789abcd07"})
 718         self.l.add({"dn": "OU=OU21,DC=SAMBA,DC=ORG",
 719                     "name": b"OU #10",
 720                     "x": "x", "y": "c",
 721                     "objectUUID": b"0123456789abcd08"})
 722         self.l.add({"dn": "OU=OU22,DC=SAMBA,DC=ORG",
 723                     "name": b"OU #10",
 724                     "x": "x", "y": "c",
 725                     "objectUUID": b"0123456789abcd09"})
 726
 727     def test_base(self):
 728         """Testing a search"""
 729
 730         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
 731                               scope=ldb.SCOPE_BASE)
 732         self.assertEqual(len(res11), 1)
 733
 734     def test_base_lower(self):
 735         """Testing a search"""
 736
 737         res11 = self.l.search(base="OU=OU11,DC=samba,DC=org",
 738                               scope=ldb.SCOPE_BASE)
 739         self.assertEqual(len(res11), 1)
 740
 741     def test_base_or(self):
 742         """Testing a search"""
 743
 744         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
 745                               scope=ldb.SCOPE_BASE,
 746                               expression="(|(ou=ou11)(ou=ou12))")
 747         self.assertEqual(len(res11), 1)
 748
 749     def test_base_or2(self):
 750         """Testing a search"""
 751
 752         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
 753                               scope=ldb.SCOPE_BASE,
 754                               expression="(|(x=y)(y=b))")
 755         self.assertEqual(len(res11), 1)
 756
 757     def test_base_and(self):
 758         """Testing a search"""
 759
 760         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
 761                               scope=ldb.SCOPE_BASE,
 762                               expression="(&(ou=ou11)(ou=ou12))")
 763         self.assertEqual(len(res11), 0)
 764
 765     def test_base_and2(self):
 766         """Testing a search"""
 767
 768         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
 769                               scope=ldb.SCOPE_BASE,
 770                               expression="(&(x=y)(y=a))")
 771         self.assertEqual(len(res11), 1)
 772
 773     def test_base_false(self):
 774         """Testing a search"""
 775
 776         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
 777                               scope=ldb.SCOPE_BASE,
 778                               expression="(|(ou=ou13)(ou=ou12))")
 779         self.assertEqual(len(res11), 0)
 780
 781     def test_check_base_false(self):
 782         """Testing a search"""
 783         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
 784                               scope=ldb.SCOPE_BASE,
 785                               expression="(|(ou=ou13)(ou=ou12))")
 786         self.assertEqual(len(res11), 0)
 787
 788     def test_check_base_error(self):
 789         """Testing a search"""
 790         self.l.add({"dn": "@OPTIONS", "checkBaseOnSearch": b"TRUE"})
 791
 792         try:
 793             res11 = self.l.search(base="OU=OU11x,DC=SAMBA,DC=ORG",
 794                                   scope=ldb.SCOPE_BASE,
 795                                   expression="(|(ou=ou13)(ou=ou12))")
 796             self.fail("Should have failed on missing base")
 797         except ldb.LdbError as err:
 798             enum = err.args[0]
 799             self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
 800
 801     def test_subtree_and(self):
 802         """Testing a search"""
 803
 804         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 805                               scope=ldb.SCOPE_SUBTREE,
 806                               expression="(&(ou=ou11)(ou=ou12))")
 807         self.assertEqual(len(res11), 0)
 808
 809     def test_subtree_and2(self):
 810         """Testing a search"""
 811
 812         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 813                               scope=ldb.SCOPE_SUBTREE,
 814                               expression="(&(x=y)(|(y=b)(y=c)))")
 815         self.assertEqual(len(res11), 1)
 816
 817     def test_subtree_and2_lower(self):
 818         """Testing a search"""
 819
 820         res11 = self.l.search(base="DC=samba,DC=org",
 821                               scope=ldb.SCOPE_SUBTREE,
 822                               expression="(&(x=y)(|(y=b)(y=c)))")
 823         self.assertEqual(len(res11), 1)
 824
 825     def test_subtree_or(self):
 826         """Testing a search"""
 827
 828         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 829                               scope=ldb.SCOPE_SUBTREE,
 830                               expression="(|(ou=ou11)(ou=ou12))")
 831         self.assertEqual(len(res11), 2)
 832
 833     def test_subtree_or2(self):
 834         """Testing a search"""
 835
 836         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 837                               scope=ldb.SCOPE_SUBTREE,
 838                               expression="(|(x=y)(y=b))")
 839         self.assertEqual(len(res11), 20)
 840
 841     def test_subtree_or3(self):
 842         """Testing a search"""
 843
 844         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 845                               scope=ldb.SCOPE_SUBTREE,
 846                               expression="(|(x=y)(y=b)(y=c))")
 847         self.assertEqual(len(res11), 22)
 848
 849     def test_one_and(self):
 850         """Testing a search"""
 851
 852         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 853                               scope=ldb.SCOPE_ONELEVEL,
 854                               expression="(&(ou=ou11)(ou=ou12))")
 855         self.assertEqual(len(res11), 0)
 856
 857     def test_one_and2(self):
 858         """Testing a search"""
 859
 860         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 861                               scope=ldb.SCOPE_ONELEVEL,
 862                               expression="(&(x=y)(y=b))")
 863         self.assertEqual(len(res11), 1)
 864
 865     def test_one_or(self):
 866         """Testing a search"""
 867
 868         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 869                               scope=ldb.SCOPE_ONELEVEL,
 870                               expression="(|(ou=ou11)(ou=ou12))")
 871         self.assertEqual(len(res11), 2)
 872
 873     def test_one_or2(self):
 874         """Testing a search"""
 875
 876         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 877                               scope=ldb.SCOPE_ONELEVEL,
 878                               expression="(|(x=y)(y=b))")
 879         self.assertEqual(len(res11), 20)
 880
 881     def test_one_or2_lower(self):
 882         """Testing a search"""
 883
 884         res11 = self.l.search(base="DC=samba,DC=org",
 885                               scope=ldb.SCOPE_ONELEVEL,
 886                               expression="(|(x=y)(y=b))")
 887         self.assertEqual(len(res11), 20)
 888
 889     def test_subtree_and_or(self):
 890         """Testing a search"""
 891
 892         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 893                               scope=ldb.SCOPE_SUBTREE,
 894                               expression="(&(|(x=z)(y=b))(x=x)(y=c))")
 895         self.assertEqual(len(res11), 0)
 896
 897     def test_subtree_and_or2(self):
 898         """Testing a search"""
 899
 900         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 901                               scope=ldb.SCOPE_SUBTREE,
 902                               expression="(&(x=x)(y=c)(|(x=z)(y=b)))")
 903         self.assertEqual(len(res11), 0)
 904
 905     def test_subtree_and_or3(self):
 906         """Testing a search"""
 907
 908         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 909                               scope=ldb.SCOPE_SUBTREE,
 910                               expression="(&(|(ou=ou11)(ou=ou10))(|(x=y)(y=b)(y=c)))")
 911         self.assertEqual(len(res11), 2)
 912
 913     def test_subtree_and_or4(self):
 914         """Testing a search"""
 915
 916         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 917                               scope=ldb.SCOPE_SUBTREE,
 918                               expression="(&(|(x=y)(y=b)(y=c))(|(ou=ou11)(ou=ou10)))")
 919         self.assertEqual(len(res11), 2)
 920
 921     def test_subtree_and_or5(self):
 922         """Testing a search"""
 923
 924         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 925                               scope=ldb.SCOPE_SUBTREE,
 926                               expression="(&(|(x=y)(y=b)(y=c))(ou=ou11))")
 927         self.assertEqual(len(res11), 1)
 928
 929     def test_subtree_or_and(self):
 930         """Testing a search"""
 931
 932         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 933                               scope=ldb.SCOPE_SUBTREE,
 934                               expression="(|(x=x)(y=c)(&(x=z)(y=b)))")
 935         self.assertEqual(len(res11), 10)
 936
 937     def test_subtree_large_and_unique(self):
 938         """Testing a search"""
 939
 940         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 941                               scope=ldb.SCOPE_SUBTREE,
 942                               expression="(&(ou=ou10)(y=a))")
 943         self.assertEqual(len(res11), 1)
 944
 945     def test_subtree_and_none(self):
 946         """Testing a search"""
 947
 948         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
 949                               scope=ldb.SCOPE_SUBTREE,
 950                               expression="(&(ou=ouX)(y=a))")
 951         self.assertEqual(len(res11), 0)
 952
 953
 954 class IndexedSearchTests(SearchTests):
 955     """Test searches using the index, to ensure the index doesn't
 956        break things"""
 957     def setUp(self):
 958         super(IndexedSearchTests, self).setUp()
 959         self.l.add({"dn": "@INDEXLIST",
 960                     "@IDXATTR": [b"x", b"y", b"ou"],
 961                     "@IDXONE": [b"1"]})
 962
 963 class GUIDIndexedSearchTests(SearchTests):
 964     """Test searches using the index, to ensure the index doesn't
 965        break things"""
 966     def setUp(self):
 967         super(GUIDIndexedSearchTests, self).setUp()
 968         self.l.add({"dn": "@INDEXLIST",
 969                     "@IDXATTR": [b"x", b"y", b"ou"],
 970                     "@IDXONE": [b"1"],
 971                     "@IDXGUID": [b"objectUUID"],
 972                     "@IDX_DN_GUID": [b"GUID"]})
 973
 974 class AddModifyTests(TestCase):
 975     def tearDown(self):
 976         shutil.rmtree(self.testdir)
 977         super(AddModifyTests, self).tearDown()
 978
 979     def setUp(self):
 980         super(AddModifyTests, self).setUp()
 981         self.testdir = tempdir()
 982         self.filename = os.path.join(self.testdir, "add_test.ldb")
 983         self.l = ldb.Ldb(self.filename, options=["modules:rdn_name"])
 984         self.l.add({"dn": "DC=SAMBA,DC=ORG",
 985                     "name": b"samba.org",
 986                     "objectUUID": b"0123456789abcdef"})
 987         self.l.add({"dn": "@ATTRIBUTES",
 988                     "objectUUID": "UNIQUE_INDEX"})
 989
 990     def test_add_dup(self):
 991         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
 992                     "name": b"Admins",
 993                     "x": "z", "y": "a",
 994                     "objectUUID": b"0123456789abcde1"})
 995         try:
 996             self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
 997                         "name": b"Admins",
 998                         "x": "z", "y": "a",
 999                         "objectUUID": b"0123456789abcde2"})
1000             self.fail("Should have failed adding dupliate entry")
1001         except ldb.LdbError as err:
1002             enum = err.args[0]
1003             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1004
1005     def test_add_del_add(self):
1006         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1007                     "name": b"Admins",
1008                     "x": "z", "y": "a",
1009                     "objectUUID": b"0123456789abcde1"})
1010         self.l.delete("OU=DUP,DC=SAMBA,DC=ORG")
1011         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1012                     "name": b"Admins",
1013                     "x": "z", "y": "a",
1014                     "objectUUID": b"0123456789abcde2"})
1015
1016     def test_add_move_add(self):
1017         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1018                     "name": b"Admins",
1019                     "x": "z", "y": "a",
1020                     "objectUUID": b"0123456789abcde1"})
1021         self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1022                       "OU=DUP2,DC=SAMBA,DC=ORG")
1023         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1024                     "name": b"Admins",
1025                     "x": "z", "y": "a",
1026                     "objectUUID": b"0123456789abcde2"})
1027
1028     def test_add_move_fail_move_move(self):
1029         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1030                     "name": b"Admins",
1031                     "x": "z", "y": "a",
1032                     "objectUUID": b"0123456789abcde1"})
1033         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1034                     "name": b"Admins",
1035                     "x": "z", "y": "a",
1036                     "objectUUID": b"0123456789abcde2"})
1037         try:
1038             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1039                           "OU=DUP2,DC=SAMBA,DC=ORG")
1040             self.fail("Should have failed on duplicate DN")
1041         except ldb.LdbError as err:
1042             enum = err.args[0]
1043             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1044
1045         self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1046                       "OU=DUP3,DC=SAMBA,DC=ORG")
1047
1048         self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1049                       "OU=DUP2,DC=SAMBA,DC=ORG")
1050
1051         res2 = self.l.search(base="DC=SAMBA,DC=ORG",
1052                              scope=ldb.SCOPE_SUBTREE,
1053                              expression="(objectUUID=0123456789abcde1)")
1054         self.assertEqual(len(res2), 1)
1055         self.assertEqual(str(res2[0].dn), "OU=DUP2,DC=SAMBA,DC=ORG")
1056
1057         res3 = self.l.search(base="DC=SAMBA,DC=ORG",
1058                              scope=ldb.SCOPE_SUBTREE,
1059                              expression="(objectUUID=0123456789abcde2)")
1060         self.assertEqual(len(res3), 1)
1061         self.assertEqual(str(res3[0].dn), "OU=DUP3,DC=SAMBA,DC=ORG")
1062
1063     def test_move_missing(self):
1064         try:
1065             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1066                           "OU=DUP2,DC=SAMBA,DC=ORG")
1067             self.fail("Should have failed on missing")
1068         except ldb.LdbError as err:
1069             enum = err.args[0]
1070             self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1071
1072     def test_move_missing2(self):
1073         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1074                     "name": b"Admins",
1075                     "x": "z", "y": "a",
1076                     "objectUUID": b"0123456789abcde2"})
1077
1078         try:
1079             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1080                           "OU=DUP2,DC=SAMBA,DC=ORG")
1081             self.fail("Should have failed on missing")
1082         except ldb.LdbError as err:
1083             enum = err.args[0]
1084             self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1085
1086     def test_move_fail_move_add(self):
1087         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1088                     "name": b"Admins",
1089                     "x": "z", "y": "a",
1090                     "objectUUID": b"0123456789abcde1"})
1091         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1092                     "name": b"Admins",
1093                     "x": "z", "y": "a",
1094                     "objectUUID": b"0123456789abcde2"})
1095         try:
1096             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1097                           "OU=DUP2,DC=SAMBA,DC=ORG")
1098             self.fail("Should have failed on duplicate DN")
1099         except ldb.LdbError as err:
1100             enum = err.args[0]
1101             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1102
1103         self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1104                       "OU=DUP3,DC=SAMBA,DC=ORG")
1105
1106         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1107                     "name": b"Admins",
1108                     "x": "z", "y": "a",
1109                     "objectUUID": b"0123456789abcde3"})
1110
1111 class IndexedAddModifyTests(AddModifyTests):
1112     """Test searches using the index, to ensure the index doesn't
1113        break things"""
1114     def setUp(self):
1115         super(IndexedAddModifyTests, self).setUp()
1116         self.l.add({"dn": "@INDEXLIST",
1117                     "@IDXATTR": [b"x", b"y", b"ou", b"objectUUID"],
1118                     "@IDXONE": [b"1"]})
1119
1120     def test_duplicate_GUID(self):
1121         try:
1122             self.l.add({"dn": "OU=DUPGUID,DC=SAMBA,DC=ORG",
1123                         "name": b"Admins",
1124                         "x": "z", "y": "a",
1125                         "objectUUID": b"0123456789abcdef"})
1126             self.fail("Should have failed adding dupliate GUID")
1127         except ldb.LdbError as err:
1128             enum = err.args[0]
1129             self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1130
1131     def test_duplicate_name_dup_GUID(self):
1132         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1133                     "name": b"Admins",
1134                     "x": "z", "y": "a",
1135                     "objectUUID": b"a123456789abcdef"})
1136         try:
1137             self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1138                         "name": b"Admins",
1139                         "x": "z", "y": "a",
1140                         "objectUUID": b"a123456789abcdef"})
1141             self.fail("Should have failed adding dupliate GUID")
1142         except ldb.LdbError as err:
1143             enum = err.args[0]
1144             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1145
1146     def test_duplicate_name_dup_GUID2(self):
1147         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1148                     "name": b"Admins",
1149                     "x": "z", "y": "a",
1150                     "objectUUID": b"abc3456789abcdef"})
1151         try:
1152             self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1153                         "name": b"Admins",
1154                         "x": "z", "y": "a",
1155                         "objectUUID": b"aaa3456789abcdef"})
1156             self.fail("Should have failed adding dupliate DN")
1157         except ldb.LdbError as err:
1158             enum = err.args[0]
1159             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1160
1161         # Checking the GUID didn't stick in the index
1162         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1163                     "name": b"Admins",
1164                     "x": "z", "y": "a",
1165                     "objectUUID": b"aaa3456789abcdef"})
1166
1167     def test_add_dup_guid_add(self):
1168         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1169                     "name": b"Admins",
1170                     "x": "z", "y": "a",
1171                     "objectUUID": b"0123456789abcde1"})
1172         try:
1173             self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1174                         "name": b"Admins",
1175                         "x": "z", "y": "a",
1176                         "objectUUID": b"0123456789abcde1"})
1177             self.fail("Should have failed on duplicate GUID")
1178
1179         except ldb.LdbError as err:
1180             enum = err.args[0]
1181             self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1182
1183         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1184                     "name": b"Admins",
1185                     "x": "z", "y": "a",
1186                     "objectUUID": b"0123456789abcde2"})
1187
1188 class GUIDIndexedAddModifyTests(IndexedAddModifyTests):
1189     """Test searches using the index, to ensure the index doesn't
1190        break things"""
1191     def setUp(self):
1192         super(GUIDIndexedAddModifyTests, self).setUp()
1193         indexlist = {"dn": "@INDEXLIST",
1194                      "@IDXATTR": [b"x", b"y", b"ou"],
1195                      "@IDXONE": [b"1"],
1196                      "@IDXGUID": [b"objectUUID"],
1197                      "@IDX_DN_GUID": [b"GUID"]}
1198         m = ldb.Message.from_dict(self.l, indexlist, ldb.FLAG_MOD_REPLACE)
1199         self.l.modify(m)
1200
1201
1202 class GUIDTransIndexedAddModifyTests(GUIDIndexedAddModifyTests):
1203     """Test GUID index behaviour insdie the transaction"""
1204     def setUp(self):
1205         super(GUIDTransIndexedAddModifyTests, self).setUp()
1206         self.l.transaction_start()
1207
1208     def tearDown(self):
1209         self.l.transaction_commit()
1210         super(GUIDTransIndexedAddModifyTests, self).tearDown()
1211
1212 class TransIndexedAddModifyTests(IndexedAddModifyTests):
1213     """Test index behaviour insdie the transaction"""
1214     def setUp(self):
1215         super(TransIndexedAddModifyTests, self).setUp()
1216         self.l.transaction_start()
1217
1218     def tearDown(self):
1219         self.l.transaction_commit()
1220         super(TransIndexedAddModifyTests, self).tearDown()
1221
1222
1223
1224 class DnTests(TestCase):
1225
1226     def setUp(self):
1227         super(DnTests, self).setUp()
1228         self.testdir = tempdir()
1229         self.filename = os.path.join(self.testdir, "test.ldb")
1230         self.ldb = ldb.Ldb(self.filename)
1231
1232     def tearDown(self):
1233         shutil.rmtree(self.testdir)
1234         super(DnTests, self).tearDown()
1235
1236     def test_set_dn_invalid(self):
1237         x = ldb.Message()
1238         def assign():
1239             x.dn = "astring"
1240         self.assertRaises(TypeError, assign)
1241
1242     def test_eq(self):
1243         x = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
1244         y = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
1245         self.assertEqual(x, y)
1246         y = ldb.Dn(self.ldb, "dc=foo11,bar=blie")
1247         self.assertNotEqual(x, y)
1248
1249     def test_str(self):
1250         x = ldb.Dn(self.ldb, "dc=foo12,bar=bloe")
1251         self.assertEqual(x.__str__(), "dc=foo12,bar=bloe")
1252
1253     def test_repr(self):
1254         x = ldb.Dn(self.ldb, "dc=foo13,bla=blie")
1255         self.assertEqual(x.__repr__(), "Dn('dc=foo13,bla=blie')")
1256
1257     def test_get_casefold(self):
1258         x = ldb.Dn(self.ldb, "dc=foo14,bar=bloe")
1259         self.assertEqual(x.get_casefold(), "DC=FOO14,BAR=bloe")
1260
1261     def test_validate(self):
1262         x = ldb.Dn(self.ldb, "dc=foo15,bar=bloe")
1263         self.assertTrue(x.validate())
1264
1265     def test_parent(self):
1266         x = ldb.Dn(self.ldb, "dc=foo16,bar=bloe")
1267         self.assertEqual("bar=bloe", x.parent().__str__())
1268
1269     def test_parent_nonexistent(self):
1270         x = ldb.Dn(self.ldb, "@BLA")
1271         self.assertEqual(None, x.parent())
1272
1273     def test_is_valid(self):
1274         x = ldb.Dn(self.ldb, "dc=foo18,dc=bloe")
1275         self.assertTrue(x.is_valid())
1276         x = ldb.Dn(self.ldb, "")
1277         self.assertTrue(x.is_valid())
1278
1279     def test_is_special(self):
1280         x = ldb.Dn(self.ldb, "dc=foo19,bar=bloe")
1281         self.assertFalse(x.is_special())
1282         x = ldb.Dn(self.ldb, "@FOOBAR")
1283         self.assertTrue(x.is_special())
1284
1285     def test_check_special(self):
1286         x = ldb.Dn(self.ldb, "dc=foo20,bar=bloe")
1287         self.assertFalse(x.check_special("FOOBAR"))
1288         x = ldb.Dn(self.ldb, "@FOOBAR")
1289         self.assertTrue(x.check_special("@FOOBAR"))
1290
1291     def test_len(self):
1292         x = ldb.Dn(self.ldb, "dc=foo21,bar=bloe")
1293         self.assertEqual(2, len(x))
1294         x = ldb.Dn(self.ldb, "dc=foo21")
1295         self.assertEqual(1, len(x))
1296
1297     def test_add_child(self):
1298         x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
1299         self.assertTrue(x.add_child(ldb.Dn(self.ldb, "bla=bloe")))
1300         self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
1301
1302     def test_add_base(self):
1303         x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
1304         base = ldb.Dn(self.ldb, "bla=bloe")
1305         self.assertTrue(x.add_base(base))
1306         self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
1307
1308     def test_add_child_str(self):
1309         x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
1310         self.assertTrue(x.add_child("bla=bloe"))
1311         self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
1312
1313     def test_add_base_str(self):
1314         x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
1315         base = "bla=bloe"
1316         self.assertTrue(x.add_base(base))
1317         self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
1318
1319     def test_add(self):
1320         x = ldb.Dn(self.ldb, "dc=foo24")
1321         y = ldb.Dn(self.ldb, "bar=bla")
1322         self.assertEqual("dc=foo24,bar=bla", str(x + y))
1323
1324     def test_remove_base_components(self):
1325         x = ldb.Dn(self.ldb, "dc=foo24,dc=samba,dc=org")
1326         x.remove_base_components(len(x)-1)
1327         self.assertEqual("dc=foo24", str(x))
1328
1329     def test_parse_ldif(self):
1330         msgs = self.ldb.parse_ldif("dn: foo=bar\n")
1331         msg = next(msgs)
1332         self.assertEqual("foo=bar", str(msg[1].dn))
1333         self.assertTrue(isinstance(msg[1], ldb.Message))
1334         ldif = self.ldb.write_ldif(msg[1], ldb.CHANGETYPE_NONE)
1335         self.assertEqual("dn: foo=bar\n\n", ldif)
1336
1337     def test_parse_ldif_more(self):
1338         msgs = self.ldb.parse_ldif("dn: foo=bar\n\n\ndn: bar=bar")
1339         msg = next(msgs)
1340         self.assertEqual("foo=bar", str(msg[1].dn))
1341         msg = next(msgs)
1342         self.assertEqual("bar=bar", str(msg[1].dn))
1343
1344     def test_canonical_string(self):
1345         x = ldb.Dn(self.ldb, "dc=foo25,bar=bloe")
1346         self.assertEqual("/bloe/foo25", x.canonical_str())
1347
1348     def test_canonical_ex_string(self):
1349         x = ldb.Dn(self.ldb, "dc=foo26,bar=bloe")
1350         self.assertEqual("/bloe\nfoo26", x.canonical_ex_str())
1351
1352     def test_ldb_is_child_of(self):
1353         """Testing ldb_dn_compare_dn"""
1354         dn1 = ldb.Dn(self.ldb, "dc=base")
1355         dn2 = ldb.Dn(self.ldb, "cn=foo,dc=base")
1356         dn3 = ldb.Dn(self.ldb, "cn=bar,dc=base")
1357         dn4 = ldb.Dn(self.ldb, "cn=baz,cn=bar,dc=base")
1358
1359         self.assertTrue(dn2.is_child_of(dn1))
1360         self.assertTrue(dn4.is_child_of(dn1))
1361         self.assertTrue(dn4.is_child_of(dn3))
1362         self.assertFalse(dn3.is_child_of(dn2))
1363         self.assertFalse(dn1.is_child_of(dn4))
1364
1365     def test_ldb_is_child_of_str(self):
1366         """Testing ldb_dn_compare_dn"""
1367         dn1_str = "dc=base"
1368         dn2_str = "cn=foo,dc=base"
1369         dn3_str = "cn=bar,dc=base"
1370         dn4_str = "cn=baz,cn=bar,dc=base"
1371
1372         dn1 = ldb.Dn(self.ldb, dn1_str)
1373         dn2 = ldb.Dn(self.ldb, dn2_str)
1374         dn3 = ldb.Dn(self.ldb, dn3_str)
1375         dn4 = ldb.Dn(self.ldb, dn4_str)
1376
1377         self.assertTrue(dn2.is_child_of(dn1_str))
1378         self.assertTrue(dn4.is_child_of(dn1_str))
1379         self.assertTrue(dn4.is_child_of(dn3_str))
1380         self.assertFalse(dn3.is_child_of(dn2_str))
1381         self.assertFalse(dn1.is_child_of(dn4_str))
1382
1383     def test_get_component_name(self):
1384         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1385         self.assertEqual(dn.get_component_name(0), 'cn')
1386         self.assertEqual(dn.get_component_name(1), 'dc')
1387         self.assertEqual(dn.get_component_name(2), None)
1388         self.assertEqual(dn.get_component_name(-1), None)
1389
1390     def test_get_component_value(self):
1391         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1392         self.assertEqual(dn.get_component_value(0), 'foo')
1393         self.assertEqual(dn.get_component_value(1), 'base')
1394         self.assertEqual(dn.get_component_name(2), None)
1395         self.assertEqual(dn.get_component_name(-1), None)
1396
1397     def test_set_component(self):
1398         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1399         dn.set_component(0, 'cn', 'bar')
1400         self.assertEqual(str(dn), "cn=bar,dc=base")
1401         dn.set_component(1, 'o', 'asep')
1402         self.assertEqual(str(dn), "cn=bar,o=asep")
1403         self.assertRaises(TypeError, dn.set_component, 2, 'dc', 'base')
1404         self.assertEqual(str(dn), "cn=bar,o=asep")
1405         dn.set_component(1, 'o', 'a,b+c')
1406         self.assertEqual(str(dn), r"cn=bar,o=a\,b\+c")
1407
1408     def test_set_component_bytes(self):
1409         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1410         dn.set_component(0, 'cn', b'bar')
1411         self.assertEqual(str(dn), "cn=bar,dc=base")
1412         dn.set_component(1, 'o', b'asep')
1413         self.assertEqual(str(dn), "cn=bar,o=asep")
1414
1415     def test_set_component_none(self):
1416         dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
1417         self.assertRaises(TypeError, dn.set_component, 1, 'cn', None)
1418
1419     def test_get_extended_component_null(self):
1420         dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
1421         self.assertEqual(dn.get_extended_component("TEST"), None)
1422
1423     def test_get_extended_component(self):
1424         self.ldb._register_test_extensions()
1425         dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
1426         self.assertEqual(dn.get_extended_component("TEST"), b"foo")
1427
1428     def test_set_extended_component(self):
1429         self.ldb._register_test_extensions()
1430         dn = ldb.Dn(self.ldb, "dc=base")
1431         dn.set_extended_component("TEST", "foo")
1432         self.assertEqual(dn.get_extended_component("TEST"), b"foo")
1433         dn.set_extended_component("TEST", b"bar")
1434         self.assertEqual(dn.get_extended_component("TEST"), b"bar")
1435
1436     def test_extended_str(self):
1437         self.ldb._register_test_extensions()
1438         dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
1439         self.assertEqual(dn.extended_str(), "<TEST=foo>;cn=bar,dc=base")
1440
1441     def test_get_rdn_name(self):
1442         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1443         self.assertEqual(dn.get_rdn_name(), 'cn')
1444
1445     def test_get_rdn_value(self):
1446         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1447         self.assertEqual(dn.get_rdn_value(), 'foo')
1448
1449     def test_get_casefold(self):
1450         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1451         self.assertEqual(dn.get_casefold(), 'CN=FOO,DC=BASE')
1452
1453     def test_get_linearized(self):
1454         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1455         self.assertEqual(dn.get_linearized(), 'cn=foo,dc=base')
1456
1457     def test_is_null(self):
1458         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
1459         self.assertFalse(dn.is_null())
1460
1461         dn = ldb.Dn(self.ldb, '')
1462         self.assertTrue(dn.is_null())
1463
1464 class LdbMsgTests(TestCase):
1465
1466     def setUp(self):
1467         super(LdbMsgTests, self).setUp()
1468         self.msg = ldb.Message()
1469         self.testdir = tempdir()
1470         self.filename = os.path.join(self.testdir, "test.ldb")
1471
1472     def tearDown(self):
1473         shutil.rmtree(self.testdir)
1474         super(LdbMsgTests, self).tearDown()
1475
1476     def test_init_dn(self):
1477         self.msg = ldb.Message(ldb.Dn(ldb.Ldb(), "dc=foo27"))
1478         self.assertEqual("dc=foo27", str(self.msg.dn))
1479
1480     def test_iter_items(self):
1481         self.assertEqual(0, len(self.msg.items()))
1482         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "dc=foo28")
1483         self.assertEqual(1, len(self.msg.items()))
1484
1485     def test_repr(self):
1486         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "dc=foo29")
1487         self.msg["dc"] = b"foo"
1488         if PY3:
1489             self.assertIn(repr(self.msg), [
1490                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])})",
1491                 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')})",
1492             ])
1493             self.assertIn(repr(self.msg.text), [
1494                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])}).text",
1495                 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')}).text",
1496             ])
1497         else:
1498             self.assertEqual(
1499                 repr(self.msg),
1500                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])})")
1501             self.assertEqual(
1502                 repr(self.msg.text),
1503                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])}).text")
1504
1505     def test_len(self):
1506         self.assertEqual(0, len(self.msg))
1507
1508     def test_notpresent(self):
1509         self.assertRaises(KeyError, lambda: self.msg["foo"])
1510
1511     def test_del(self):
1512         del self.msg["foo"]
1513
1514     def test_add(self):
1515         self.msg.add(ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla"))
1516
1517     def test_add_text(self):
1518         self.msg.add(ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla"))
1519
1520     def test_elements_empty(self):
1521         self.assertEqual([], self.msg.elements())
1522
1523     def test_elements(self):
1524         el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
1525         self.msg.add(el)
1526         self.assertEqual([el], self.msg.elements())
1527         self.assertEqual([el.text], self.msg.text.elements())
1528
1529     def test_add_value(self):
1530         self.assertEqual(0, len(self.msg))
1531         self.msg["foo"] = [b"foo"]
1532         self.assertEqual(1, len(self.msg))
1533
1534     def test_add_value_text(self):
1535         self.assertEqual(0, len(self.msg))
1536         self.msg["foo"] = ["foo"]
1537         self.assertEqual(1, len(self.msg))
1538
1539     def test_add_value_multiple(self):
1540         self.assertEqual(0, len(self.msg))
1541         self.msg["foo"] = [b"foo", b"bla"]
1542         self.assertEqual(1, len(self.msg))
1543         self.assertEqual([b"foo", b"bla"], list(self.msg["foo"]))
1544
1545     def test_add_value_multiple_text(self):
1546         self.assertEqual(0, len(self.msg))
1547         self.msg["foo"] = ["foo", "bla"]
1548         self.assertEqual(1, len(self.msg))
1549         self.assertEqual(["foo", "bla"], list(self.msg.text["foo"]))
1550
1551     def test_set_value(self):
1552         self.msg["foo"] = [b"fool"]
1553         self.assertEqual([b"fool"], list(self.msg["foo"]))
1554         self.msg["foo"] = [b"bar"]
1555         self.assertEqual([b"bar"], list(self.msg["foo"]))
1556
1557     def test_set_value_text(self):
1558         self.msg["foo"] = ["fool"]
1559         self.assertEqual(["fool"], list(self.msg.text["foo"]))
1560         self.msg["foo"] = ["bar"]
1561         self.assertEqual(["bar"], list(self.msg.text["foo"]))
1562
1563     def test_keys(self):
1564         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "@BASEINFO")
1565         self.msg["foo"] = [b"bla"]
1566         self.msg["bar"] = [b"bla"]
1567         self.assertEqual(["dn", "foo", "bar"], self.msg.keys())
1568
1569     def test_keys_text(self):
1570         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "@BASEINFO")
1571         self.msg["foo"] = ["bla"]
1572         self.msg["bar"] = ["bla"]
1573         self.assertEqual(["dn", "foo", "bar"], self.msg.text.keys())
1574
1575     def test_dn(self):
1576         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "@BASEINFO")
1577         self.assertEqual("@BASEINFO", self.msg.dn.__str__())
1578
1579     def test_get_dn(self):
1580         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "@BASEINFO")
1581         self.assertEqual("@BASEINFO", self.msg.get("dn").__str__())
1582
1583     def test_dn_text(self):
1584         self.msg.text.dn = ldb.Dn(ldb.Ldb(self.filename), "@BASEINFO")
1585         self.assertEqual("@BASEINFO", str(self.msg.dn))
1586         self.assertEqual("@BASEINFO", str(self.msg.text.dn))
1587
1588     def test_get_dn_text(self):
1589         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "@BASEINFO")
1590         self.assertEqual("@BASEINFO", str(self.msg.get("dn")))
1591         self.assertEqual("@BASEINFO", str(self.msg.text.get("dn")))
1592
1593     def test_get_invalid(self):
1594         self.msg.dn = ldb.Dn(ldb.Ldb(self.filename), "@BASEINFO")
1595         self.assertRaises(TypeError, self.msg.get, 42)
1596
1597     def test_get_other(self):
1598         self.msg["foo"] = [b"bar"]
1599         self.assertEqual(b"bar", self.msg.get("foo")[0])
1600         self.assertEqual(b"bar", self.msg.get("foo", idx=0))
1601         self.assertEqual(None, self.msg.get("foo", idx=1))
1602         self.assertEqual("", self.msg.get("foo", default='', idx=1))
1603
1604     def test_get_other_text(self):
1605         self.msg["foo"] = ["bar"]
1606         self.assertEqual(["bar"], list(self.msg.text.get("foo")))
1607         self.assertEqual("bar", self.msg.text.get("foo")[0])
1608         self.assertEqual("bar", self.msg.text.get("foo", idx=0))
1609         self.assertEqual(None, self.msg.get("foo", idx=1))
1610         self.assertEqual("", self.msg.get("foo", default='', idx=1))
1611
1612     def test_get_default(self):
1613         self.assertEqual(None, self.msg.get("tatayoyo", idx=0))
1614         self.assertEqual("anniecordie", self.msg.get("tatayoyo", "anniecordie"))
1615
1616     def test_get_default_text(self):
1617         self.assertEqual(None, self.msg.text.get("tatayoyo", idx=0))
1618         self.assertEqual("anniecordie", self.msg.text.get("tatayoyo", "anniecordie"))
1619
1620     def test_get_unknown(self):
1621         self.assertEqual(None, self.msg.get("lalalala"))
1622
1623     def test_get_unknown_text(self):
1624         self.assertEqual(None, self.msg.text.get("lalalala"))
1625
1626     def test_msg_diff(self):
1627         l = ldb.Ldb()
1628         msgs = l.parse_ldif("dn: foo=bar\nfoo: bar\nbaz: do\n\ndn: foo=bar\nfoo: bar\nbaz: dont\n")
1629         msg1 = next(msgs)[1]
1630         msg2 = next(msgs)[1]
1631         msgdiff = l.msg_diff(msg1, msg2)
1632         self.assertEqual("foo=bar", msgdiff.get("dn").__str__())
1633         self.assertRaises(KeyError, lambda: msgdiff["foo"])
1634         self.assertEqual(1, len(msgdiff))
1635
1636     def test_equal_empty(self):
1637         msg1 = ldb.Message()
1638         msg2 = ldb.Message()
1639         self.assertEqual(msg1, msg2)
1640
1641     def test_equal_simplel(self):
1642         db = ldb.Ldb(self.filename)
1643         msg1 = ldb.Message()
1644         msg1.dn = ldb.Dn(db, "foo=bar")
1645         msg2 = ldb.Message()
1646         msg2.dn = ldb.Dn(db, "foo=bar")
1647         self.assertEqual(msg1, msg2)
1648         msg1['foo'] = b'bar'
1649         msg2['foo'] = b'bar'
1650         self.assertEqual(msg1, msg2)
1651         msg2['foo'] = b'blie'
1652         self.assertNotEqual(msg1, msg2)
1653         msg2['foo'] = b'blie'
1654
1655     def test_from_dict(self):
1656         rec = {"dn": "dc=fromdict",
1657                "a1": [b"a1-val1", b"a1-val1"]}
1658         l = ldb.Ldb()
1659         # check different types of input Flags
1660         for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
1661             m = ldb.Message.from_dict(l, rec, flags)
1662             self.assertEqual(rec["a1"], list(m["a1"]))
1663             self.assertEqual(flags, m["a1"].flags())
1664         # check input params
1665         self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
1666         self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
1667         self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
1668         # Message.from_dict expects dictionary with 'dn'
1669         err_rec = {"a1": [b"a1-val1", b"a1-val1"]}
1670         self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
1671
1672     def test_from_dict_text(self):
1673         rec = {"dn": "dc=fromdict",
1674                "a1": ["a1-val1", "a1-val1"]}
1675         l = ldb.Ldb()
1676         # check different types of input Flags
1677         for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
1678             m = ldb.Message.from_dict(l, rec, flags)
1679             self.assertEqual(rec["a1"], list(m.text["a1"]))
1680             self.assertEqual(flags, m.text["a1"].flags())
1681         # check input params
1682         self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
1683         self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
1684         self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
1685         # Message.from_dict expects dictionary with 'dn'
1686         err_rec = {"a1": ["a1-val1", "a1-val1"]}
1687         self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
1688
1689     def test_copy_add_message_element(self):
1690         m = ldb.Message()
1691         m["1"] = ldb.MessageElement([b"val 111"], ldb.FLAG_MOD_ADD, "1")
1692         m["2"] = ldb.MessageElement([b"val 222"], ldb.FLAG_MOD_ADD, "2")
1693         mto = ldb.Message()
1694         mto["1"] = m["1"]
1695         mto["2"] = m["2"]
1696         self.assertEqual(mto["1"], m["1"])
1697         self.assertEqual(mto["2"], m["2"])
1698         mto = ldb.Message()
1699         mto.add(m["1"])
1700         mto.add(m["2"])
1701         self.assertEqual(mto["1"], m["1"])
1702         self.assertEqual(mto["2"], m["2"])
1703
1704     def test_copy_add_message_element_text(self):
1705         m = ldb.Message()
1706         m["1"] = ldb.MessageElement(["val 111"], ldb.FLAG_MOD_ADD, "1")
1707         m["2"] = ldb.MessageElement(["val 222"], ldb.FLAG_MOD_ADD, "2")
1708         mto = ldb.Message()
1709         mto["1"] = m["1"]
1710         mto["2"] = m["2"]
1711         self.assertEqual(mto["1"], m.text["1"])
1712         self.assertEqual(mto["2"], m.text["2"])
1713         mto = ldb.Message()
1714         mto.add(m["1"])
1715         mto.add(m["2"])
1716         self.assertEqual(mto.text["1"], m.text["1"])
1717         self.assertEqual(mto.text["2"], m.text["2"])
1718         self.assertEqual(mto["1"], m["1"])
1719         self.assertEqual(mto["2"], m["2"])
1720
1721
1722 class MessageElementTests(TestCase):
1723
1724     def test_cmp_element(self):
1725         x = ldb.MessageElement([b"foo"])
1726         y = ldb.MessageElement([b"foo"])
1727         z = ldb.MessageElement([b"bzr"])
1728         self.assertEqual(x, y)
1729         self.assertNotEqual(x, z)
1730
1731     def test_cmp_element_text(self):
1732         x = ldb.MessageElement([b"foo"])
1733         y = ldb.MessageElement(["foo"])
1734         self.assertEqual(x, y)
1735
1736     def test_create_iterable(self):
1737         x = ldb.MessageElement([b"foo"])
1738         self.assertEqual([b"foo"], list(x))
1739         self.assertEqual(["foo"], list(x.text))
1740
1741     def test_repr(self):
1742         x = ldb.MessageElement([b"foo"])
1743         if PY3:
1744             self.assertEqual("MessageElement([b'foo'])", repr(x))
1745             self.assertEqual("MessageElement([b'foo']).text", repr(x.text))
1746         else:
1747             self.assertEqual("MessageElement(['foo'])", repr(x))
1748             self.assertEqual("MessageElement(['foo']).text", repr(x.text))
1749         x = ldb.MessageElement([b"foo", b"bla"])
1750         self.assertEqual(2, len(x))
1751         if PY3:
1752             self.assertEqual("MessageElement([b'foo',b'bla'])", repr(x))
1753             self.assertEqual("MessageElement([b'foo',b'bla']).text", repr(x.text))
1754         else:
1755             self.assertEqual("MessageElement(['foo','bla'])", repr(x))
1756             self.assertEqual("MessageElement(['foo','bla']).text", repr(x.text))
1757
1758     def test_get_item(self):
1759         x = ldb.MessageElement([b"foo", b"bar"])
1760         self.assertEqual(b"foo", x[0])
1761         self.assertEqual(b"bar", x[1])
1762         self.assertEqual(b"bar", x[-1])
1763         self.assertRaises(IndexError, lambda: x[45])
1764
1765     def test_get_item_text(self):
1766         x = ldb.MessageElement(["foo", "bar"])
1767         self.assertEqual("foo", x.text[0])
1768         self.assertEqual("bar", x.text[1])
1769         self.assertEqual("bar", x.text[-1])
1770         self.assertRaises(IndexError, lambda: x[45])
1771
1772     def test_len(self):
1773         x = ldb.MessageElement([b"foo", b"bar"])
1774         self.assertEqual(2, len(x))
1775
1776     def test_eq(self):
1777         x = ldb.MessageElement([b"foo", b"bar"])
1778         y = ldb.MessageElement([b"foo", b"bar"])
1779         self.assertEqual(y, x)
1780         x = ldb.MessageElement([b"foo"])
1781         self.assertNotEqual(y, x)
1782         y = ldb.MessageElement([b"foo"])
1783         self.assertEqual(y, x)
1784
1785     def test_extended(self):
1786         el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
1787         if PY3:
1788             self.assertEqual("MessageElement([b'456'])", repr(el))
1789             self.assertEqual("MessageElement([b'456']).text", repr(el.text))
1790         else:
1791             self.assertEqual("MessageElement(['456'])", repr(el))
1792             self.assertEqual("MessageElement(['456']).text", repr(el.text))
1793
1794     def test_bad_text(self):
1795         el = ldb.MessageElement(b'\xba\xdd')
1796         self.assertRaises(UnicodeDecodeError, el.text.__getitem__, 0)
1797
1798
1799 class ModuleTests(TestCase):
1800
1801     def setUp(self):
1802         super(ModuleTests, self).setUp()
1803         self.testdir = tempdir()
1804         self.filename = os.path.join(self.testdir, "test.ldb")
1805         self.ldb = ldb.Ldb(self.filename)
1806
1807     def tearDown(self):
1808         shutil.rmtree(self.testdir)
1809         super(ModuleTests, self).setUp()
1810
1811     def test_register_module(self):
1812         class ExampleModule:
1813             name = "example"
1814         ldb.register_module(ExampleModule)
1815
1816     def test_use_module(self):
1817         ops = []
1818         class ExampleModule:
1819             name = "bla"
1820
1821             def __init__(self, ldb, next):
1822                 ops.append("init")
1823                 self.next = next
1824
1825             def search(self, *args, **kwargs):
1826                 return self.next.search(*args, **kwargs)
1827
1828             def request(self, *args, **kwargs):
1829                 pass
1830
1831         ldb.register_module(ExampleModule)
1832         l = ldb.Ldb(self.filename)
1833         l.add({"dn": "@MODULES", "@LIST": "bla"})
1834         self.assertEqual([], ops)
1835         l = ldb.Ldb(self.filename)
1836         self.assertEqual(["init"], ops)
1837
1838 class LdbResultTests(TestCase):
1839
1840     def setUp(self):
1841         super(LdbResultTests, self).setUp()
1842         self.testdir = tempdir()
1843         self.filename = os.path.join(self.testdir, "test.ldb")
1844         self.l = ldb.Ldb(self.filename)
1845         self.l.add({"dn": "DC=SAMBA,DC=ORG", "name": b"samba.org"})
1846         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG", "name": b"Admins"})
1847         self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG", "name": b"Users"})
1848         self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG", "name": b"OU #1"})
1849         self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG", "name": b"OU #2"})
1850         self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG", "name": b"OU #3"})
1851         self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG", "name": b"OU #4"})
1852         self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG", "name": b"OU #5"})
1853         self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG", "name": b"OU #6"})
1854         self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG", "name": b"OU #7"})
1855         self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG", "name": b"OU #8"})
1856         self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG", "name": b"OU #9"})
1857         self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG", "name": b"OU #10"})
1858
1859     def tearDown(self):
1860         shutil.rmtree(self.testdir)
1861         super(LdbResultTests, self).tearDown()
1862
1863     def test_return_type(self):
1864         res = self.l.search()
1865         self.assertEqual(str(res), "<ldb result>")
1866
1867     def test_get_msgs(self):
1868         res = self.l.search()
1869         list = res.msgs
1870
1871     def test_get_controls(self):
1872         res = self.l.search()
1873         list = res.controls
1874
1875     def test_get_referals(self):
1876         res = self.l.search()
1877         list = res.referals
1878
1879     def test_iter_msgs(self):
1880         found = False
1881         for l in self.l.search().msgs:
1882             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
1883                 found = True
1884         self.assertTrue(found)
1885
1886     def test_iter_msgs_count(self):
1887         self.assertTrue(self.l.search().count > 0)
1888         # 13 objects has been added to the DC=SAMBA, DC=ORG
1889         self.assertEqual(self.l.search(base="DC=SAMBA,DC=ORG").count, 13)
1890
1891     def test_iter_controls(self):
1892         res = self.l.search().controls
1893         it = iter(res)
1894
1895     def test_create_control(self):
1896         self.assertRaises(ValueError, ldb.Control, self.l, "tatayoyo:0")
1897         c = ldb.Control(self.l, "relax:1")
1898         self.assertEqual(c.critical, True)
1899         self.assertEqual(c.oid, "1.3.6.1.4.1.4203.666.5.12")
1900
1901     def test_iter_refs(self):
1902         res = self.l.search().referals
1903         it = iter(res)
1904
1905     def test_search_sequence_msgs(self):
1906         found = False
1907         res = self.l.search().msgs
1908
1909         for i in range(0, len(res)):
1910             l = res[i]
1911             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
1912                 found = True
1913         self.assertTrue(found)
1914
1915     def test_search_as_iter(self):
1916         found = False
1917         res = self.l.search()
1918
1919         for l in res:
1920             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
1921                 found = True
1922         self.assertTrue(found)
1923
1924     def test_search_iter(self):
1925         found = False
1926         res = self.l.search_iterator()
1927
1928         for l in res:
1929             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
1930                 found = True
1931         self.assertTrue(found)
1932
1933
1934     # Show that search results can't see into a transaction
1935     def test_search_against_trans(self):
1936         found11 = False
1937
1938         (r1, w1) = os.pipe()
1939
1940         (r2, w2) = os.pipe()
1941
1942         # For the first element, fork a child that will
1943         # write to the DB
1944         pid = os.fork()
1945         if pid == 0:
1946             # In the child, re-open
1947             del(self.l)
1948             gc.collect()
1949
1950             child_ldb = ldb.Ldb(self.filename)
1951             # start a transaction
1952             child_ldb.transaction_start()
1953
1954             # write to it
1955             child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
1956                            "name": b"samba.org"})
1957
1958             os.write(w1, b"added")
1959
1960             # Now wait for the search to be done
1961             os.read(r2, 6)
1962
1963             # and commit
1964             try:
1965                 child_ldb.transaction_commit()
1966             except LdbError as err:
1967                 # We print this here to see what went wrong in the child
1968                 print(err)
1969                 os._exit(1)
1970
1971             os.write(w1, b"transaction")
1972             os._exit(0)
1973
1974         self.assertEqual(os.read(r1, 5), b"added")
1975
1976         # This should not turn up until the transaction is concluded
1977         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
1978                             scope=ldb.SCOPE_BASE)
1979         self.assertEqual(len(res11), 0)
1980
1981         os.write(w2, b"search")
1982
1983         # Now wait for the transaction to be done.  This should
1984         # deadlock, but the search doesn't hold a read lock for the
1985         # iterator lifetime currently.
1986         self.assertEqual(os.read(r1, 11), b"transaction")
1987
1988         # This should now turn up, as the transaction is over
1989         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
1990                             scope=ldb.SCOPE_BASE)
1991         self.assertEqual(len(res11), 1)
1992
1993         self.assertFalse(found11)
1994
1995         (got_pid, status) = os.waitpid(pid, 0)
1996         self.assertEqual(got_pid, pid)
1997
1998
1999     def test_search_iter_against_trans(self):
2000         found = False
2001         found11 = False
2002
2003         # We need to hold this iterator open to hold the all-record
2004         # lock
2005         res = self.l.search_iterator()
2006
2007         (r1, w1) = os.pipe()
2008
2009         (r2, w2) = os.pipe()
2010
2011         # For the first element, with the sequence open (which
2012         # means with ldb locks held), fork a child that will
2013         # write to the DB
2014         pid = os.fork()
2015         if pid == 0:
2016             # In the child, re-open
2017             del(res)
2018             del(self.l)
2019             gc.collect()
2020
2021             child_ldb = ldb.Ldb(self.filename)
2022             # start a transaction
2023             child_ldb.transaction_start()
2024
2025             # write to it
2026             child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
2027                            "name": b"samba.org"})
2028
2029             os.write(w1, b"added")
2030
2031             # Now wait for the search to be done
2032             os.read(r2, 6)
2033
2034             # and commit
2035             try:
2036                 child_ldb.transaction_commit()
2037             except LdbError as err:
2038                 # We print this here to see what went wrong in the child
2039                 print(err)
2040                 os._exit(1)
2041
2042             os.write(w1, b"transaction")
2043             os._exit(0)
2044
2045         self.assertEqual(os.read(r1, 5), b"added")
2046
2047         # This should not turn up until the transaction is concluded
2048         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2049                             scope=ldb.SCOPE_BASE)
2050         self.assertEqual(len(res11), 0)
2051
2052         os.write(w2, b"search")
2053
2054         # allow the transaction to start
2055         time.sleep(1)
2056
2057         # This should not turn up until the search finishes and
2058         # removed the read lock, but for ldb_tdb that happened as soon
2059         # as we called the first res.next()
2060         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2061                             scope=ldb.SCOPE_BASE)
2062         self.assertEqual(len(res11), 0)
2063
2064         # These results are all collected at the first next(res) call
2065         for l in res:
2066             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2067                 found = True
2068             if str(l.dn) == "OU=OU11,DC=SAMBA,DC=ORG":
2069                 found11 = True
2070
2071         # Now wait for the transaction to be done.
2072         self.assertEqual(os.read(r1, 11), b"transaction")
2073
2074         # This should now turn up, as the transaction is over and all
2075         # read locks are gone
2076         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2077                             scope=ldb.SCOPE_BASE)
2078         self.assertEqual(len(res11), 1)
2079
2080         self.assertTrue(found)
2081         self.assertFalse(found11)
2082
2083         (got_pid, status) = os.waitpid(pid, 0)
2084         self.assertEqual(got_pid, pid)
2085
2086
2087 class BadTypeTests(TestCase):
2088     def test_control(self):
2089         l = ldb.Ldb()
2090         self.assertRaises(TypeError, ldb.Control, '<bad type>', 'relax:1')
2091         self.assertRaises(TypeError, ldb.Control, ldb, 1234)
2092
2093     def test_modify(self):
2094         l = ldb.Ldb()
2095         dn = ldb.Dn(l, 'a=b')
2096         m = ldb.Message(dn)
2097         self.assertRaises(TypeError, l.modify, '<bad type>')
2098         self.assertRaises(TypeError, l.modify, m, '<bad type>')
2099
2100     def test_add(self):
2101         l = ldb.Ldb()
2102         dn = ldb.Dn(l, 'a=b')
2103         m = ldb.Message(dn)
2104         self.assertRaises(TypeError, l.add, '<bad type>')
2105         self.assertRaises(TypeError, l.add, m, '<bad type>')
2106
2107     def test_delete(self):
2108         l = ldb.Ldb()
2109         dn = ldb.Dn(l, 'a=b')
2110         self.assertRaises(TypeError, l.add, '<bad type>')
2111         self.assertRaises(TypeError, l.add, dn, '<bad type>')
2112
2113     def test_rename(self):
2114         l = ldb.Ldb()
2115         dn = ldb.Dn(l, 'a=b')
2116         self.assertRaises(TypeError, l.add, '<bad type>', dn)
2117         self.assertRaises(TypeError, l.add, dn, '<bad type>')
2118         self.assertRaises(TypeError, l.add, dn, dn, '<bad type>')
2119
2120     def test_search(self):
2121         l = ldb.Ldb()
2122         self.assertRaises(TypeError, l.search, base=1234)
2123         self.assertRaises(TypeError, l.search, scope='<bad type>')
2124         self.assertRaises(TypeError, l.search, expression=1234)
2125         self.assertRaises(TypeError, l.search, attrs='<bad type>')
2126         self.assertRaises(TypeError, l.search, controls='<bad type>')
2127
2128
2129 class VersionTests(TestCase):
2130
2131     def test_version(self):
2132         self.assertTrue(isinstance(ldb.__version__, str))
2133
2134
2135 if __name__ == '__main__':
2136     import unittest
2137     unittest.TestProgram()