2 # Simple tests for the ldb python bindings.
3 # Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
6 from unittest import TestCase
13 PY3 = sys.version_info > (3, 0)
21 "@IDXGUID": [b"objectUUID"],
22 "@IDX_DN_GUID": [b"GUID"]
29 dir_prefix = os.path.join(os.environ["SELFTEST_PREFIX"], "tmp")
32 return tempfile.mkdtemp(dir=dir_prefix)
35 class NoContextTests(TestCase):
37 def test_valid_attr_name(self):
38 self.assertTrue(ldb.valid_attr_name("foo"))
39 self.assertFalse(ldb.valid_attr_name("24foo"))
41 def test_timestring(self):
42 self.assertEqual("19700101000000.0Z", ldb.timestring(0))
43 self.assertEqual("20071119191012.0Z", ldb.timestring(1195499412))
45 def test_string_to_time(self):
46 self.assertEqual(0, ldb.string_to_time("19700101000000.0Z"))
47 self.assertEqual(1195499412, ldb.string_to_time("20071119191012.0Z"))
49 def test_binary_encode(self):
50 encoded = ldb.binary_encode(b'test\\x')
51 decoded = ldb.binary_decode(encoded)
52 self.assertEqual(decoded, b'test\\x')
54 encoded2 = ldb.binary_encode('test\\x')
55 self.assertEqual(encoded2, encoded)
58 class LdbBaseTest(TestCase):
60 super(LdbBaseTest, self).setUp()
62 if self.prefix is None:
63 self.prefix = TDB_PREFIX
64 except AttributeError:
65 self.prefix = TDB_PREFIX
68 super(LdbBaseTest, self).tearDown()
71 return self.prefix + self.filename
74 if self.prefix == MDB_PREFIX:
80 class SimpleLdb(LdbBaseTest):
83 super(SimpleLdb, self).setUp()
84 self.testdir = tempdir()
85 self.filename = os.path.join(self.testdir, "test.ldb")
86 self.ldb = ldb.Ldb(self.url(), flags=self.flags())
88 self.ldb.add(self.index)
89 except AttributeError:
93 shutil.rmtree(self.testdir)
94 super(SimpleLdb, self).tearDown()
95 # Ensure the LDB is closed now, so we close the FD
98 def test_connect(self):
99 ldb.Ldb(self.url(), flags=self.flags())
101 def test_connect_none(self):
104 def test_connect_later(self):
106 x.connect(self.url(), flags=self.flags())
110 self.assertTrue(repr(x).startswith("<ldb connection"))
112 def test_set_create_perms(self):
114 x.set_create_perms(0o600)
116 def test_modules_none(self):
118 self.assertEqual([], x.modules())
120 def test_modules_tdb(self):
121 x = ldb.Ldb(self.url(), flags=self.flags())
122 self.assertEqual("[<ldb module 'tdb'>]", repr(x.modules()))
124 def test_firstmodule_none(self):
126 self.assertEqual(x.firstmodule, None)
128 def test_firstmodule_tdb(self):
129 x = ldb.Ldb(self.url(), flags=self.flags())
131 self.assertEqual(repr(mod), "<ldb module 'tdb'>")
133 def test_search(self):
134 l = ldb.Ldb(self.url(), flags=self.flags())
135 self.assertEqual(len(l.search()), 0)
137 def test_search_controls(self):
138 l = ldb.Ldb(self.url(), flags=self.flags())
139 self.assertEqual(len(l.search(controls=["paged_results:0:5"])), 0)
141 def test_search_attrs(self):
142 l = ldb.Ldb(self.url(), flags=self.flags())
143 self.assertEqual(len(l.search(ldb.Dn(l, ""), ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
145 def test_search_string_dn(self):
146 l = ldb.Ldb(self.url(), flags=self.flags())
147 self.assertEqual(len(l.search("", ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
149 def test_search_attr_string(self):
150 l = ldb.Ldb(self.url(), flags=self.flags())
151 self.assertRaises(TypeError, l.search, attrs="dc")
152 self.assertRaises(TypeError, l.search, attrs=b"dc")
154 def test_opaque(self):
155 l = ldb.Ldb(self.url(), flags=self.flags())
156 l.set_opaque("my_opaque", l)
157 self.assertTrue(l.get_opaque("my_opaque") is not None)
158 self.assertEqual(None, l.get_opaque("unknown"))
160 def test_search_scope_base_empty_db(self):
161 l = ldb.Ldb(self.url(), flags=self.flags())
162 self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
165 def test_search_scope_onelevel_empty_db(self):
166 l = ldb.Ldb(self.url(), flags=self.flags())
167 self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
168 ldb.SCOPE_ONELEVEL)), 0)
170 def test_delete(self):
171 l = ldb.Ldb(self.url(), flags=self.flags())
172 self.assertRaises(ldb.LdbError, lambda: l.delete(ldb.Dn(l, "dc=foo2")))
174 def test_delete_w_unhandled_ctrl(self):
175 l = ldb.Ldb(self.url(), flags=self.flags())
177 m.dn = ldb.Dn(l, "dc=foo1")
179 m["objectUUID"] = b"0123456789abcdef"
181 self.assertRaises(ldb.LdbError, lambda: l.delete(m.dn, ["search_options:1:2"]))
184 def test_contains(self):
186 l = ldb.Ldb(name, flags=self.flags())
187 self.assertFalse(ldb.Dn(l, "dc=foo3") in l)
188 l = ldb.Ldb(name, flags=self.flags())
190 m.dn = ldb.Dn(l, "dc=foo3")
192 m["objectUUID"] = b"0123456789abcdef"
195 self.assertTrue(ldb.Dn(l, "dc=foo3") in l)
196 self.assertFalse(ldb.Dn(l, "dc=foo4") in l)
200 def test_get_config_basedn(self):
201 l = ldb.Ldb(self.url(), flags=self.flags())
202 self.assertEqual(None, l.get_config_basedn())
204 def test_get_root_basedn(self):
205 l = ldb.Ldb(self.url(), flags=self.flags())
206 self.assertEqual(None, l.get_root_basedn())
208 def test_get_schema_basedn(self):
209 l = ldb.Ldb(self.url(), flags=self.flags())
210 self.assertEqual(None, l.get_schema_basedn())
212 def test_get_default_basedn(self):
213 l = ldb.Ldb(self.url(), flags=self.flags())
214 self.assertEqual(None, l.get_default_basedn())
217 l = ldb.Ldb(self.url(), flags=self.flags())
219 m.dn = ldb.Dn(l, "dc=foo4")
221 m["objectUUID"] = b"0123456789abcdef"
222 self.assertEqual(len(l.search()), 0)
225 self.assertEqual(len(l.search()), 1)
227 l.delete(ldb.Dn(l, "dc=foo4"))
229 def test_search_iterator(self):
230 l = ldb.Ldb(self.url(), flags=self.flags())
231 s = l.search_iterator()
237 except RuntimeError as re:
242 except RuntimeError as re:
247 except RuntimeError as re:
250 s = l.search_iterator()
253 self.assertTrue(isinstance(me, ldb.Message))
256 self.assertEqual(len(r), 0)
257 self.assertEqual(count, 0)
260 m1.dn = ldb.Dn(l, "dc=foo4")
262 m1["objectUUID"] = b"0123456789abcdef"
265 s = l.search_iterator()
268 self.assertTrue(isinstance(me, ldb.Message))
272 self.assertEqual(len(r), 0)
273 self.assertEqual(len(msgs), 1)
274 self.assertEqual(msgs[0].dn, m1.dn)
277 m2.dn = ldb.Dn(l, "dc=foo5")
279 m2["objectUUID"] = b"0123456789abcdee"
282 s = l.search_iterator()
285 self.assertTrue(isinstance(me, ldb.Message))
289 self.assertEqual(len(r), 0)
290 self.assertEqual(len(msgs), 2)
291 if msgs[0].dn == m1.dn:
292 self.assertEqual(msgs[0].dn, m1.dn)
293 self.assertEqual(msgs[1].dn, m2.dn)
295 self.assertEqual(msgs[0].dn, m2.dn)
296 self.assertEqual(msgs[1].dn, m1.dn)
298 s = l.search_iterator()
301 self.assertTrue(isinstance(me, ldb.Message))
308 except RuntimeError as re:
311 self.assertTrue(isinstance(me, ldb.Message))
319 self.assertEqual(len(r), 0)
320 self.assertEqual(len(msgs), 2)
321 if msgs[0].dn == m1.dn:
322 self.assertEqual(msgs[0].dn, m1.dn)
323 self.assertEqual(msgs[1].dn, m2.dn)
325 self.assertEqual(msgs[0].dn, m2.dn)
326 self.assertEqual(msgs[1].dn, m1.dn)
328 l.delete(ldb.Dn(l, "dc=foo4"))
329 l.delete(ldb.Dn(l, "dc=foo5"))
331 def test_add_text(self):
332 l = ldb.Ldb(self.url(), flags=self.flags())
334 m.dn = ldb.Dn(l, "dc=foo4")
336 m["objectUUID"] = b"0123456789abcdef"
337 self.assertEqual(len(l.search()), 0)
340 self.assertEqual(len(l.search()), 1)
342 l.delete(ldb.Dn(l, "dc=foo4"))
344 def test_add_w_unhandled_ctrl(self):
345 l = ldb.Ldb(self.url(), flags=self.flags())
347 m.dn = ldb.Dn(l, "dc=foo4")
349 self.assertEqual(len(l.search()), 0)
350 self.assertRaises(ldb.LdbError, lambda: l.add(m, ["search_options:1:2"]))
352 def test_add_dict(self):
353 l = ldb.Ldb(self.url(), flags=self.flags())
354 m = {"dn": ldb.Dn(l, "dc=foo5"),
356 "objectUUID": b"0123456789abcdef"}
357 self.assertEqual(len(l.search()), 0)
360 self.assertEqual(len(l.search()), 1)
362 l.delete(ldb.Dn(l, "dc=foo5"))
364 def test_add_dict_text(self):
365 l = ldb.Ldb(self.url(), flags=self.flags())
366 m = {"dn": ldb.Dn(l, "dc=foo5"),
368 "objectUUID": b"0123456789abcdef"}
369 self.assertEqual(len(l.search()), 0)
372 self.assertEqual(len(l.search()), 1)
374 l.delete(ldb.Dn(l, "dc=foo5"))
376 def test_add_dict_string_dn(self):
377 l = ldb.Ldb(self.url(), flags=self.flags())
378 m = {"dn": "dc=foo6", "bla": b"bla",
379 "objectUUID": b"0123456789abcdef"}
380 self.assertEqual(len(l.search()), 0)
383 self.assertEqual(len(l.search()), 1)
385 l.delete(ldb.Dn(l, "dc=foo6"))
387 def test_add_dict_bytes_dn(self):
388 l = ldb.Ldb(self.url(), flags=self.flags())
389 m = {"dn": b"dc=foo6", "bla": b"bla",
390 "objectUUID": b"0123456789abcdef"}
391 self.assertEqual(len(l.search()), 0)
394 self.assertEqual(len(l.search()), 1)
396 l.delete(ldb.Dn(l, "dc=foo6"))
398 def test_rename(self):
399 l = ldb.Ldb(self.url(), flags=self.flags())
401 m.dn = ldb.Dn(l, "dc=foo7")
403 m["objectUUID"] = b"0123456789abcdef"
404 self.assertEqual(len(l.search()), 0)
407 l.rename(ldb.Dn(l, "dc=foo7"), ldb.Dn(l, "dc=bar"))
408 self.assertEqual(len(l.search()), 1)
410 l.delete(ldb.Dn(l, "dc=bar"))
412 def test_rename_string_dns(self):
413 l = ldb.Ldb(self.url(), flags=self.flags())
415 m.dn = ldb.Dn(l, "dc=foo8")
417 m["objectUUID"] = b"0123456789abcdef"
418 self.assertEqual(len(l.search()), 0)
420 self.assertEqual(len(l.search()), 1)
422 l.rename("dc=foo8", "dc=bar")
423 self.assertEqual(len(l.search()), 1)
425 l.delete(ldb.Dn(l, "dc=bar"))
427 def test_rename_bad_string_dns(self):
428 l = ldb.Ldb(self.url(), flags=self.flags())
430 m.dn = ldb.Dn(l, "dc=foo8")
432 m["objectUUID"] = b"0123456789abcdef"
433 self.assertEqual(len(l.search()), 0)
435 self.assertEqual(len(l.search()), 1)
436 self.assertRaises(ldb.LdbError,lambda: l.rename("dcXfoo8", "dc=bar"))
437 self.assertRaises(ldb.LdbError,lambda: l.rename("dc=foo8", "dcXbar"))
438 l.delete(ldb.Dn(l, "dc=foo8"))
440 def test_empty_dn(self):
441 l = ldb.Ldb(self.url(), flags=self.flags())
442 self.assertEqual(0, len(l.search()))
444 m.dn = ldb.Dn(l, "dc=empty")
445 m["objectUUID"] = b"0123456789abcdef"
448 self.assertEqual(1, len(rm))
449 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
453 self.assertEqual(1, len(rm))
454 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
456 rm = l.search(m.dn, attrs=["blah"])
457 self.assertEqual(1, len(rm))
458 self.assertEqual(0, len(rm[0]))
460 def test_modify_delete(self):
461 l = ldb.Ldb(self.url(), flags=self.flags())
463 m.dn = ldb.Dn(l, "dc=modifydelete")
465 m["objectUUID"] = b"0123456789abcdef"
467 rm = l.search(m.dn)[0]
468 self.assertEqual([b"1234"], list(rm["bla"]))
471 m.dn = ldb.Dn(l, "dc=modifydelete")
472 m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
473 self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
476 self.assertEqual(1, len(rm))
477 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
479 rm = l.search(m.dn, attrs=["bla"])
480 self.assertEqual(1, len(rm))
481 self.assertEqual(0, len(rm[0]))
483 l.delete(ldb.Dn(l, "dc=modifydelete"))
485 def test_modify_delete_text(self):
486 l = ldb.Ldb(self.url(), flags=self.flags())
488 m.dn = ldb.Dn(l, "dc=modifydelete")
489 m.text["bla"] = ["1234"]
490 m["objectUUID"] = b"0123456789abcdef"
492 rm = l.search(m.dn)[0]
493 self.assertEqual(["1234"], list(rm.text["bla"]))
496 m.dn = ldb.Dn(l, "dc=modifydelete")
497 m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
498 self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
501 self.assertEqual(1, len(rm))
502 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
504 rm = l.search(m.dn, attrs=["bla"])
505 self.assertEqual(1, len(rm))
506 self.assertEqual(0, len(rm[0]))
508 l.delete(ldb.Dn(l, "dc=modifydelete"))
510 def test_modify_add(self):
511 l = ldb.Ldb(self.url(), flags=self.flags())
513 m.dn = ldb.Dn(l, "dc=add")
515 m["objectUUID"] = b"0123456789abcdef"
519 m.dn = ldb.Dn(l, "dc=add")
520 m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
521 self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
523 rm = l.search(m.dn)[0]
524 self.assertEqual(3, len(rm))
525 self.assertEqual([b"1234", b"456"], list(rm["bla"]))
527 l.delete(ldb.Dn(l, "dc=add"))
529 def test_modify_add_text(self):
530 l = ldb.Ldb(self.url(), flags=self.flags())
532 m.dn = ldb.Dn(l, "dc=add")
533 m.text["bla"] = ["1234"]
534 m["objectUUID"] = b"0123456789abcdef"
538 m.dn = ldb.Dn(l, "dc=add")
539 m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
540 self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
542 rm = l.search(m.dn)[0]
543 self.assertEqual(3, len(rm))
544 self.assertEqual(["1234", "456"], list(rm.text["bla"]))
546 l.delete(ldb.Dn(l, "dc=add"))
548 def test_modify_replace(self):
549 l = ldb.Ldb(self.url(), flags=self.flags())
551 m.dn = ldb.Dn(l, "dc=modify2")
552 m["bla"] = [b"1234", b"456"]
553 m["objectUUID"] = b"0123456789abcdef"
557 m.dn = ldb.Dn(l, "dc=modify2")
558 m["bla"] = ldb.MessageElement([b"789"], ldb.FLAG_MOD_REPLACE, "bla")
559 self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
561 rm = l.search(m.dn)[0]
562 self.assertEqual(3, len(rm))
563 self.assertEqual([b"789"], list(rm["bla"]))
564 rm = l.search(m.dn, attrs=["bla"])[0]
565 self.assertEqual(1, len(rm))
567 l.delete(ldb.Dn(l, "dc=modify2"))
569 def test_modify_replace_text(self):
570 l = ldb.Ldb(self.url(), flags=self.flags())
572 m.dn = ldb.Dn(l, "dc=modify2")
573 m.text["bla"] = ["1234", "456"]
574 m["objectUUID"] = b"0123456789abcdef"
578 m.dn = ldb.Dn(l, "dc=modify2")
579 m["bla"] = ldb.MessageElement(["789"], ldb.FLAG_MOD_REPLACE, "bla")
580 self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
582 rm = l.search(m.dn)[0]
583 self.assertEqual(3, len(rm))
584 self.assertEqual(["789"], list(rm.text["bla"]))
585 rm = l.search(m.dn, attrs=["bla"])[0]
586 self.assertEqual(1, len(rm))
588 l.delete(ldb.Dn(l, "dc=modify2"))
590 def test_modify_flags_change(self):
591 l = ldb.Ldb(self.url(), flags=self.flags())
593 m.dn = ldb.Dn(l, "dc=add")
595 m["objectUUID"] = b"0123456789abcdef"
599 m.dn = ldb.Dn(l, "dc=add")
600 m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
601 self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
603 rm = l.search(m.dn)[0]
604 self.assertEqual(3, len(rm))
605 self.assertEqual([b"1234", b"456"], list(rm["bla"]))
607 # Now create another modify, but switch the flags before we do it
608 m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
609 m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
611 rm = l.search(m.dn, attrs=["bla"])[0]
612 self.assertEqual(1, len(rm))
613 self.assertEqual([b"1234"], list(rm["bla"]))
615 l.delete(ldb.Dn(l, "dc=add"))
617 def test_modify_flags_change_text(self):
618 l = ldb.Ldb(self.url(), flags=self.flags())
620 m.dn = ldb.Dn(l, "dc=add")
621 m.text["bla"] = ["1234"]
622 m["objectUUID"] = b"0123456789abcdef"
626 m.dn = ldb.Dn(l, "dc=add")
627 m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
628 self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
630 rm = l.search(m.dn)[0]
631 self.assertEqual(3, len(rm))
632 self.assertEqual(["1234", "456"], list(rm.text["bla"]))
634 # Now create another modify, but switch the flags before we do it
635 m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
636 m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
638 rm = l.search(m.dn, attrs=["bla"])[0]
639 self.assertEqual(1, len(rm))
640 self.assertEqual(["1234"], list(rm.text["bla"]))
642 l.delete(ldb.Dn(l, "dc=add"))
644 def test_transaction_commit(self):
645 l = ldb.Ldb(self.url(), flags=self.flags())
646 l.transaction_start()
647 m = ldb.Message(ldb.Dn(l, "dc=foo9"))
649 m["objectUUID"] = b"0123456789abcdef"
651 l.transaction_commit()
654 def test_transaction_cancel(self):
655 l = ldb.Ldb(self.url(), flags=self.flags())
656 l.transaction_start()
657 m = ldb.Message(ldb.Dn(l, "dc=foo10"))
659 m["objectUUID"] = b"0123456789abcdee"
661 l.transaction_cancel()
662 self.assertEqual(0, len(l.search(ldb.Dn(l, "dc=foo10"))))
664 def test_set_debug(self):
665 def my_report_fn(level, text):
667 l = ldb.Ldb(self.url(), flags=self.flags())
668 l.set_debug(my_report_fn)
670 def test_zero_byte_string(self):
671 """Testing we do not get trapped in the \0 byte in a property string."""
672 l = ldb.Ldb(self.url(), flags=self.flags())
675 "objectclass": b"user",
676 "cN": b"LDAPtestUSER",
677 "givenname": b"ldap",
678 "displayname": b"foo\0bar",
679 "objectUUID": b"0123456789abcdef"
681 res = l.search(expression="(dn=dc=somedn)")
682 self.assertEqual(b"foo\0bar", res[0]["displayname"][0])
684 def test_no_crash_broken_expr(self):
685 l = ldb.Ldb(self.url(), flags=self.flags())
686 self.assertRaises(ldb.LdbError, lambda: l.search("", ldb.SCOPE_SUBTREE, "&(dc=*)(dn=*)", ["dc"]))
688 # Run the SimpleLdb tests against an lmdb backend
691 class SimpleLdbLmdb(SimpleLdb):
694 self.prefix = MDB_PREFIX
695 self.index = MDB_INDEX_OBJ
696 super(SimpleLdbLmdb, self).setUp()
699 super(SimpleLdbLmdb, self).tearDown()
702 class SearchTests(LdbBaseTest):
704 shutil.rmtree(self.testdir)
705 super(SearchTests, self).tearDown()
707 # Ensure the LDB is closed now, so we close the FD
711 super(SearchTests, self).setUp()
712 self.testdir = tempdir()
713 self.filename = os.path.join(self.testdir, "search_test.ldb")
714 options = ["modules:rdn_name"]
715 if hasattr(self, 'IDXCHECK'):
716 options.append("disable_full_db_scan_for_self_test:1")
717 self.l = ldb.Ldb(self.url(),
721 self.l.add(self.index)
722 except AttributeError:
725 self.l.add({"dn": "@ATTRIBUTES",
726 "DC": "CASE_INSENSITIVE"})
728 # Note that we can't use the name objectGUID here, as we
729 # want to stay clear of the objectGUID handler in LDB and
730 # instead use just the 16 bytes raw, which we just keep
731 # to printable chars here for ease of handling.
733 self.l.add({"dn": "DC=SAMBA,DC=ORG",
734 "name": b"samba.org",
735 "objectUUID": b"0123456789abcdef"})
736 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
739 "objectUUID": b"0123456789abcde1"})
740 self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG",
743 "objectUUID": b"0123456789abcde2"})
744 self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG",
747 "objectUUID": b"0123456789abcde3"})
748 self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG",
751 "objectUUID": b"0123456789abcde4"})
752 self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG",
755 "objectUUID": b"0123456789abcde5"})
756 self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG",
759 "objectUUID": b"0123456789abcde6"})
760 self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG",
763 "objectUUID": b"0123456789abcde7"})
764 self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG",
767 "objectUUID": b"0123456789abcde8"})
768 self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG",
771 "objectUUID": b"0123456789abcde9"})
772 self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG",
775 "objectUUID": b"0123456789abcde0"})
776 self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG",
779 "objectUUID": b"0123456789abcdea"})
780 self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG",
783 "objectUUID": b"0123456789abcdeb"})
784 self.l.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
787 "objectUUID": b"0123456789abcdec"})
788 self.l.add({"dn": "OU=OU12,DC=SAMBA,DC=ORG",
791 "objectUUID": b"0123456789abcded"})
792 self.l.add({"dn": "OU=OU13,DC=SAMBA,DC=ORG",
795 "objectUUID": b"0123456789abcdee"})
796 self.l.add({"dn": "OU=OU14,DC=SAMBA,DC=ORG",
799 "objectUUID": b"0123456789abcd01"})
800 self.l.add({"dn": "OU=OU15,DC=SAMBA,DC=ORG",
803 "objectUUID": b"0123456789abcd02"})
804 self.l.add({"dn": "OU=OU16,DC=SAMBA,DC=ORG",
807 "objectUUID": b"0123456789abcd03"})
808 self.l.add({"dn": "OU=OU17,DC=SAMBA,DC=ORG",
811 "objectUUID": b"0123456789abcd04"})
812 self.l.add({"dn": "OU=OU18,DC=SAMBA,DC=ORG",
815 "objectUUID": b"0123456789abcd05"})
816 self.l.add({"dn": "OU=OU19,DC=SAMBA,DC=ORG",
819 "objectUUID": b"0123456789abcd06"})
820 self.l.add({"dn": "OU=OU20,DC=SAMBA,DC=ORG",
823 "objectUUID": b"0123456789abcd07"})
824 self.l.add({"dn": "OU=OU21,DC=SAMBA,DC=ORG",
827 "objectUUID": b"0123456789abcd08"})
828 self.l.add({"dn": "OU=OU22,DC=SAMBA,DC=ORG",
831 "objectUUID": b"0123456789abcd09"})
834 """Testing a search"""
836 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
837 scope=ldb.SCOPE_BASE)
838 self.assertEqual(len(res11), 1)
840 def test_base_lower(self):
841 """Testing a search"""
843 res11 = self.l.search(base="OU=OU11,DC=samba,DC=org",
844 scope=ldb.SCOPE_BASE)
845 self.assertEqual(len(res11), 1)
847 def test_base_or(self):
848 """Testing a search"""
850 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
851 scope=ldb.SCOPE_BASE,
852 expression="(|(ou=ou11)(ou=ou12))")
853 self.assertEqual(len(res11), 1)
855 def test_base_or2(self):
856 """Testing a search"""
858 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
859 scope=ldb.SCOPE_BASE,
860 expression="(|(x=y)(y=b))")
861 self.assertEqual(len(res11), 1)
863 def test_base_and(self):
864 """Testing a search"""
866 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
867 scope=ldb.SCOPE_BASE,
868 expression="(&(ou=ou11)(ou=ou12))")
869 self.assertEqual(len(res11), 0)
871 def test_base_and2(self):
872 """Testing a search"""
874 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
875 scope=ldb.SCOPE_BASE,
876 expression="(&(x=y)(y=a))")
877 self.assertEqual(len(res11), 1)
879 def test_base_false(self):
880 """Testing a search"""
882 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
883 scope=ldb.SCOPE_BASE,
884 expression="(|(ou=ou13)(ou=ou12))")
885 self.assertEqual(len(res11), 0)
887 def test_check_base_false(self):
888 """Testing a search"""
889 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
890 scope=ldb.SCOPE_BASE,
891 expression="(|(ou=ou13)(ou=ou12))")
892 self.assertEqual(len(res11), 0)
894 def test_check_base_error(self):
895 """Testing a search"""
896 checkbaseonsearch = {"dn": "@OPTIONS",
897 "checkBaseOnSearch": b"TRUE"}
899 self.l.add(checkbaseonsearch)
900 except ldb.LdbError as err:
902 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
903 m = ldb.Message.from_dict(self.l,
908 res11 = self.l.search(base="OU=OU11x,DC=SAMBA,DC=ORG",
909 scope=ldb.SCOPE_BASE,
910 expression="(|(ou=ou13)(ou=ou12))")
911 self.fail("Should have failed on missing base")
912 except ldb.LdbError as err:
914 self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
916 def test_subtree_and(self):
917 """Testing a search"""
919 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
920 scope=ldb.SCOPE_SUBTREE,
921 expression="(&(ou=ou11)(ou=ou12))")
922 self.assertEqual(len(res11), 0)
924 def test_subtree_and2(self):
925 """Testing a search"""
927 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
928 scope=ldb.SCOPE_SUBTREE,
929 expression="(&(x=y)(|(y=b)(y=c)))")
930 self.assertEqual(len(res11), 1)
932 def test_subtree_and2_lower(self):
933 """Testing a search"""
935 res11 = self.l.search(base="DC=samba,DC=org",
936 scope=ldb.SCOPE_SUBTREE,
937 expression="(&(x=y)(|(y=b)(y=c)))")
938 self.assertEqual(len(res11), 1)
940 def test_subtree_or(self):
941 """Testing a search"""
943 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
944 scope=ldb.SCOPE_SUBTREE,
945 expression="(|(ou=ou11)(ou=ou12))")
946 self.assertEqual(len(res11), 2)
948 def test_subtree_or2(self):
949 """Testing a search"""
951 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
952 scope=ldb.SCOPE_SUBTREE,
953 expression="(|(x=y)(y=b))")
954 self.assertEqual(len(res11), 20)
956 def test_subtree_or3(self):
957 """Testing a search"""
959 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
960 scope=ldb.SCOPE_SUBTREE,
961 expression="(|(x=y)(y=b)(y=c))")
962 self.assertEqual(len(res11), 22)
964 def test_one_and(self):
965 """Testing a search"""
967 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
968 scope=ldb.SCOPE_ONELEVEL,
969 expression="(&(ou=ou11)(ou=ou12))")
970 self.assertEqual(len(res11), 0)
972 def test_one_and2(self):
973 """Testing a search"""
975 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
976 scope=ldb.SCOPE_ONELEVEL,
977 expression="(&(x=y)(y=b))")
978 self.assertEqual(len(res11), 1)
980 def test_one_or(self):
981 """Testing a search"""
983 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
984 scope=ldb.SCOPE_ONELEVEL,
985 expression="(|(ou=ou11)(ou=ou12))")
986 self.assertEqual(len(res11), 2)
988 def test_one_or2(self):
989 """Testing a search"""
991 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
992 scope=ldb.SCOPE_ONELEVEL,
993 expression="(|(x=y)(y=b))")
994 self.assertEqual(len(res11), 20)
996 def test_one_or2_lower(self):
997 """Testing a search"""
999 res11 = self.l.search(base="DC=samba,DC=org",
1000 scope=ldb.SCOPE_ONELEVEL,
1001 expression="(|(x=y)(y=b))")
1002 self.assertEqual(len(res11), 20)
1004 def test_one_unindexable(self):
1005 """Testing a search"""
1008 res11 = self.l.search(base="DC=samba,DC=org",
1009 scope=ldb.SCOPE_ONELEVEL,
1010 expression="(y=b*)")
1011 if hasattr(self, 'IDX') and \
1012 not hasattr(self, 'IDXONE') and \
1013 hasattr(self, 'IDXCHECK'):
1014 self.fail("Should have failed as un-indexed search")
1016 self.assertEqual(len(res11), 9)
1018 except ldb.LdbError as err:
1021 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1022 self.assertIn(estr, "ldb FULL SEARCH disabled")
1024 def test_one_unindexable_presence(self):
1025 """Testing a search"""
1028 res11 = self.l.search(base="DC=samba,DC=org",
1029 scope=ldb.SCOPE_ONELEVEL,
1031 if hasattr(self, 'IDX') and \
1032 not hasattr(self, 'IDXONE') and \
1033 hasattr(self, 'IDXCHECK'):
1034 self.fail("Should have failed as un-indexed search")
1036 self.assertEqual(len(res11), 24)
1038 except ldb.LdbError as err:
1041 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1042 self.assertIn(estr, "ldb FULL SEARCH disabled")
1044 def test_subtree_and_or(self):
1045 """Testing a search"""
1047 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1048 scope=ldb.SCOPE_SUBTREE,
1049 expression="(&(|(x=z)(y=b))(x=x)(y=c))")
1050 self.assertEqual(len(res11), 0)
1052 def test_subtree_and_or2(self):
1053 """Testing a search"""
1055 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1056 scope=ldb.SCOPE_SUBTREE,
1057 expression="(&(x=x)(y=c)(|(x=z)(y=b)))")
1058 self.assertEqual(len(res11), 0)
1060 def test_subtree_and_or3(self):
1061 """Testing a search"""
1063 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1064 scope=ldb.SCOPE_SUBTREE,
1065 expression="(&(|(ou=ou11)(ou=ou10))(|(x=y)(y=b)(y=c)))")
1066 self.assertEqual(len(res11), 2)
1068 def test_subtree_and_or4(self):
1069 """Testing a search"""
1071 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1072 scope=ldb.SCOPE_SUBTREE,
1073 expression="(&(|(x=y)(y=b)(y=c))(|(ou=ou11)(ou=ou10)))")
1074 self.assertEqual(len(res11), 2)
1076 def test_subtree_and_or5(self):
1077 """Testing a search"""
1079 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1080 scope=ldb.SCOPE_SUBTREE,
1081 expression="(&(|(x=y)(y=b)(y=c))(ou=ou11))")
1082 self.assertEqual(len(res11), 1)
1084 def test_subtree_or_and(self):
1085 """Testing a search"""
1087 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1088 scope=ldb.SCOPE_SUBTREE,
1089 expression="(|(x=x)(y=c)(&(x=z)(y=b)))")
1090 self.assertEqual(len(res11), 10)
1092 def test_subtree_large_and_unique(self):
1093 """Testing a search"""
1095 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1096 scope=ldb.SCOPE_SUBTREE,
1097 expression="(&(ou=ou10)(y=a))")
1098 self.assertEqual(len(res11), 1)
1100 def test_subtree_and_none(self):
1101 """Testing a search"""
1103 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1104 scope=ldb.SCOPE_SUBTREE,
1105 expression="(&(ou=ouX)(y=a))")
1106 self.assertEqual(len(res11), 0)
1108 def test_subtree_and_idx_record(self):
1109 """Testing a search against the index record"""
1111 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1112 scope=ldb.SCOPE_SUBTREE,
1113 expression="(@IDXDN=DC=SAMBA,DC=ORG)")
1114 self.assertEqual(len(res11), 0)
1116 def test_subtree_and_idxone_record(self):
1117 """Testing a search against the index record"""
1119 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1120 scope=ldb.SCOPE_SUBTREE,
1121 expression="(@IDXONE=DC=SAMBA,DC=ORG)")
1122 self.assertEqual(len(res11), 0)
1124 def test_subtree_unindexable(self):
1125 """Testing a search"""
1128 res11 = self.l.search(base="DC=samba,DC=org",
1129 scope=ldb.SCOPE_SUBTREE,
1130 expression="(y=b*)")
1131 if hasattr(self, 'IDX') and \
1132 hasattr(self, 'IDXCHECK'):
1133 self.fail("Should have failed as un-indexed search")
1135 self.assertEqual(len(res11), 9)
1137 except ldb.LdbError as err:
1140 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1141 self.assertIn(estr, "ldb FULL SEARCH disabled")
1143 def test_subtree_unindexable_presence(self):
1144 """Testing a search"""
1147 res11 = self.l.search(base="DC=samba,DC=org",
1148 scope=ldb.SCOPE_SUBTREE,
1150 if hasattr(self, 'IDX') and \
1151 hasattr(self, 'IDXCHECK'):
1152 self.fail("Should have failed as un-indexed search")
1154 self.assertEqual(len(res11), 24)
1156 except ldb.LdbError as err:
1159 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1160 self.assertIn(estr, "ldb FULL SEARCH disabled")
1162 def test_dn_filter_one(self):
1163 """Testing that a dn= filter succeeds
1164 (or fails with disallowDNFilter
1165 set and IDXGUID or (IDX and not IDXONE) mode)
1166 when the scope is SCOPE_ONELEVEL.
1168 This should be made more consistent, but for now lock in
1173 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1174 scope=ldb.SCOPE_ONELEVEL,
1175 expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1176 if hasattr(self, 'disallowDNFilter') and \
1177 hasattr(self, 'IDX') and \
1178 (hasattr(self, 'IDXGUID') or
1179 ((hasattr(self, 'IDXONE') == False and hasattr(self, 'IDX')))):
1180 self.assertEqual(len(res11), 0)
1182 self.assertEqual(len(res11), 1)
1184 def test_dn_filter_subtree(self):
1185 """Testing that a dn= filter succeeds
1186 (or fails with disallowDNFilter set)
1187 when the scope is SCOPE_SUBTREE"""
1189 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1190 scope=ldb.SCOPE_SUBTREE,
1191 expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1192 if hasattr(self, 'disallowDNFilter') \
1193 and hasattr(self, 'IDX'):
1194 self.assertEqual(len(res11), 0)
1196 self.assertEqual(len(res11), 1)
1198 def test_dn_filter_base(self):
1199 """Testing that (incorrectly) a dn= filter works
1200 when the scope is SCOPE_BASE"""
1202 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1203 scope=ldb.SCOPE_BASE,
1204 expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1206 # At some point we should fix this, but it isn't trivial
1207 self.assertEqual(len(res11), 1)
1209 def test_distinguishedName_filter_one(self):
1210 """Testing that a distinguishedName= filter succeeds
1211 when the scope is SCOPE_ONELEVEL.
1213 This should be made more consistent, but for now lock in
1218 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1219 scope=ldb.SCOPE_ONELEVEL,
1220 expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1221 self.assertEqual(len(res11), 1)
1223 def test_distinguishedName_filter_subtree(self):
1224 """Testing that a distinguishedName= filter succeeds
1225 when the scope is SCOPE_SUBTREE"""
1227 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1228 scope=ldb.SCOPE_SUBTREE,
1229 expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1230 self.assertEqual(len(res11), 1)
1232 def test_distinguishedName_filter_base(self):
1233 """Testing that (incorrectly) a distinguishedName= filter works
1234 when the scope is SCOPE_BASE"""
1236 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1237 scope=ldb.SCOPE_BASE,
1238 expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1240 # At some point we should fix this, but it isn't trivial
1241 self.assertEqual(len(res11), 1)
1243 def test_bad_dn_filter_base(self):
1244 """Testing that a dn= filter on an invalid DN works
1245 when the scope is SCOPE_BASE but
1246 returns zero results"""
1248 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1249 scope=ldb.SCOPE_BASE,
1250 expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1252 # At some point we should fix this, but it isn't trivial
1253 self.assertEqual(len(res11), 0)
1256 def test_bad_dn_filter_one(self):
1257 """Testing that a dn= filter succeeds but returns zero
1258 results when the DN is not valid on a SCOPE_ONELEVEL search
1262 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1263 scope=ldb.SCOPE_ONELEVEL,
1264 expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1265 self.assertEqual(len(res11), 0)
1267 def test_bad_dn_filter_subtree(self):
1268 """Testing that a dn= filter succeeds but returns zero
1269 results when the DN is not valid on a SCOPE_SUBTREE search
1273 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1274 scope=ldb.SCOPE_SUBTREE,
1275 expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1276 self.assertEqual(len(res11), 0)
1278 def test_bad_distinguishedName_filter_base(self):
1279 """Testing that a distinguishedName= filter on an invalid DN works
1280 when the scope is SCOPE_BASE but
1281 returns zero results"""
1283 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1284 scope=ldb.SCOPE_BASE,
1285 expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1287 # At some point we should fix this, but it isn't trivial
1288 self.assertEqual(len(res11), 0)
1291 def test_bad_distinguishedName_filter_one(self):
1292 """Testing that a distinguishedName= filter succeeds but returns zero
1293 results when the DN is not valid on a SCOPE_ONELEVEL search
1297 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1298 scope=ldb.SCOPE_ONELEVEL,
1299 expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1300 self.assertEqual(len(res11), 0)
1302 def test_bad_distinguishedName_filter_subtree(self):
1303 """Testing that a distinguishedName= filter succeeds but returns zero
1304 results when the DN is not valid on a SCOPE_SUBTREE search
1308 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1309 scope=ldb.SCOPE_SUBTREE,
1310 expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1311 self.assertEqual(len(res11), 0)
1313 def test_bad_dn_search_base(self):
1314 """Testing with a bad base DN (SCOPE_BASE)"""
1317 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DCXXX",
1318 scope=ldb.SCOPE_BASE)
1319 self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1320 except ldb.LdbError as err:
1322 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1325 def test_bad_dn_search_one(self):
1326 """Testing with a bad base DN (SCOPE_ONELEVEL)"""
1329 res11 = self.l.search(base="DC=SAMBA,DCXXXX",
1330 scope=ldb.SCOPE_ONELEVEL)
1331 self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1332 except ldb.LdbError as err:
1334 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1336 def test_bad_dn_search_subtree(self):
1337 """Testing with a bad base DN (SCOPE_SUBTREE)"""
1340 res11 = self.l.search(base="DC=SAMBA,DCXXXX",
1341 scope=ldb.SCOPE_SUBTREE)
1342 self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1343 except ldb.LdbError as err:
1345 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1349 # Run the search tests against an lmdb backend
1350 class SearchTestsLmdb(SearchTests):
1353 self.prefix = MDB_PREFIX
1354 self.index = MDB_INDEX_OBJ
1355 super(SearchTestsLmdb, self).setUp()
1358 super(SearchTestsLmdb, self).tearDown()
1361 class IndexedSearchTests(SearchTests):
1362 """Test searches using the index, to ensure the index doesn't
1366 super(IndexedSearchTests, self).setUp()
1367 self.l.add({"dn": "@INDEXLIST",
1368 "@IDXATTR": [b"x", b"y", b"ou"]})
1372 class IndexedCheckSearchTests(IndexedSearchTests):
1373 """Test searches using the index, to ensure the index doesn't
1374 break things (full scan disabled)"""
1377 self.IDXCHECK = True
1378 super(IndexedCheckSearchTests, self).setUp()
1381 class IndexedSearchDnFilterTests(SearchTests):
1382 """Test searches using the index, to ensure the index doesn't
1386 super(IndexedSearchDnFilterTests, self).setUp()
1387 self.l.add({"dn": "@OPTIONS",
1388 "disallowDNFilter": "TRUE"})
1389 self.disallowDNFilter = True
1391 self.l.add({"dn": "@INDEXLIST",
1392 "@IDXATTR": [b"x", b"y", b"ou"]})
1396 class IndexedAndOneLevelSearchTests(SearchTests):
1397 """Test searches using the index including @IDXONE, to ensure
1398 the index doesn't break things"""
1401 super(IndexedAndOneLevelSearchTests, self).setUp()
1402 self.l.add({"dn": "@INDEXLIST",
1403 "@IDXATTR": [b"x", b"y", b"ou"],
1409 class IndexedCheckedAndOneLevelSearchTests(IndexedAndOneLevelSearchTests):
1410 """Test searches using the index including @IDXONE, to ensure
1411 the index doesn't break things (full scan disabled)"""
1414 self.IDXCHECK = True
1415 super(IndexedCheckedAndOneLevelSearchTests, self).setUp()
1418 class IndexedAndOneLevelDNFilterSearchTests(SearchTests):
1419 """Test searches using the index including @IDXONE, to ensure
1420 the index doesn't break things"""
1423 super(IndexedAndOneLevelDNFilterSearchTests, self).setUp()
1424 self.l.add({"dn": "@OPTIONS",
1425 "disallowDNFilter": "TRUE",
1426 "checkBaseOnSearch": "TRUE"})
1427 self.disallowDNFilter = True
1428 self.checkBaseOnSearch = True
1430 self.l.add({"dn": "@INDEXLIST",
1431 "@IDXATTR": [b"x", b"y", b"ou"],
1437 class GUIDIndexedSearchTests(SearchTests):
1438 """Test searches using the index, to ensure the index doesn't
1442 self.index = {"dn": "@INDEXLIST",
1443 "@IDXATTR": [b"x", b"y", b"ou"],
1444 "@IDXGUID": [b"objectUUID"],
1445 "@IDX_DN_GUID": [b"GUID"]}
1446 super(GUIDIndexedSearchTests, self).setUp()
1452 class GUIDIndexedDNFilterSearchTests(SearchTests):
1453 """Test searches using the index, to ensure the index doesn't
1457 self.index = {"dn": "@INDEXLIST",
1458 "@IDXATTR": [b"x", b"y", b"ou"],
1459 "@IDXGUID": [b"objectUUID"],
1460 "@IDX_DN_GUID": [b"GUID"]}
1461 super(GUIDIndexedDNFilterSearchTests, self).setUp()
1462 self.l.add({"dn": "@OPTIONS",
1463 "disallowDNFilter": "TRUE",
1464 "checkBaseOnSearch": "TRUE"})
1465 self.disallowDNFilter = True
1466 self.checkBaseOnSearch = True
1471 class GUIDAndOneLevelIndexedSearchTests(SearchTests):
1472 """Test searches using the index including @IDXONE, to ensure
1473 the index doesn't break things"""
1476 self.index = {"dn": "@INDEXLIST",
1477 "@IDXATTR": [b"x", b"y", b"ou"],
1478 "@IDXGUID": [b"objectUUID"],
1479 "@IDX_DN_GUID": [b"GUID"]}
1480 super(GUIDAndOneLevelIndexedSearchTests, self).setUp()
1481 self.l.add({"dn": "@OPTIONS",
1482 "disallowDNFilter": "TRUE",
1483 "checkBaseOnSearch": "TRUE"})
1484 self.disallowDNFilter = True
1485 self.checkBaseOnSearch = True
1491 class GUIDIndexedSearchTestsLmdb(GUIDIndexedSearchTests):
1494 self.prefix = MDB_PREFIX
1495 super(GUIDIndexedSearchTestsLmdb, self).setUp()
1498 super(GUIDIndexedSearchTestsLmdb, self).tearDown()
1501 class GUIDIndexedDNFilterSearchTestsLmdb(GUIDIndexedDNFilterSearchTests):
1504 self.prefix = MDB_PREFIX
1505 super(GUIDIndexedDNFilterSearchTestsLmdb, self).setUp()
1508 super(GUIDIndexedDNFilterSearchTestsLmdb, self).tearDown()
1511 class GUIDAndOneLevelIndexedSearchTestsLmdb(GUIDAndOneLevelIndexedSearchTests):
1514 self.prefix = MDB_PREFIX
1515 super(GUIDAndOneLevelIndexedSearchTestsLmdb, self).setUp()
1518 super(GUIDAndOneLevelIndexedSearchTestsLmdb, self).tearDown()
1521 class AddModifyTests(LdbBaseTest):
1523 shutil.rmtree(self.testdir)
1524 super(AddModifyTests, self).tearDown()
1526 # Ensure the LDB is closed now, so we close the FD
1530 super(AddModifyTests, self).setUp()
1531 self.testdir = tempdir()
1532 self.filename = os.path.join(self.testdir, "add_test.ldb")
1533 self.l = ldb.Ldb(self.url(),
1535 options=["modules:rdn_name"])
1537 self.l.add(self.index)
1538 except AttributeError:
1541 self.l.add({"dn": "DC=SAMBA,DC=ORG",
1542 "name": b"samba.org",
1543 "objectUUID": b"0123456789abcdef"})
1544 self.l.add({"dn": "@ATTRIBUTES",
1545 "objectUUID": "UNIQUE_INDEX"})
1547 def test_add_dup(self):
1548 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1551 "objectUUID": b"0123456789abcde1"})
1553 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1556 "objectUUID": b"0123456789abcde2"})
1557 self.fail("Should have failed adding dupliate entry")
1558 except ldb.LdbError as err:
1560 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1562 def test_add_bad(self):
1564 self.l.add({"dn": "BAD,DC=SAMBA,DC=ORG",
1567 "objectUUID": b"0123456789abcde1"})
1568 self.fail("Should have failed adding entry with invalid DN")
1569 except ldb.LdbError as err:
1571 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1573 def test_add_del_add(self):
1574 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1577 "objectUUID": b"0123456789abcde1"})
1578 self.l.delete("OU=DUP,DC=SAMBA,DC=ORG")
1579 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1582 "objectUUID": b"0123456789abcde2"})
1584 def test_add_move_add(self):
1585 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1588 "objectUUID": b"0123456789abcde1"})
1589 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1590 "OU=DUP2,DC=SAMBA,DC=ORG")
1591 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1594 "objectUUID": b"0123456789abcde2"})
1596 def test_add_move_fail_move_move(self):
1597 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1600 "objectUUID": b"0123456789abcde1"})
1601 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1604 "objectUUID": b"0123456789abcde2"})
1606 res2 = self.l.search(base="DC=SAMBA,DC=ORG",
1607 scope=ldb.SCOPE_SUBTREE,
1608 expression="(objectUUID=0123456789abcde1)")
1609 self.assertEqual(len(res2), 1)
1610 self.assertEqual(str(res2[0].dn), "OU=DUP,DC=SAMBA,DC=ORG")
1612 res3 = self.l.search(base="DC=SAMBA,DC=ORG",
1613 scope=ldb.SCOPE_SUBTREE,
1614 expression="(objectUUID=0123456789abcde2)")
1615 self.assertEqual(len(res3), 1)
1616 self.assertEqual(str(res3[0].dn), "OU=DUP2,DC=SAMBA,DC=ORG")
1619 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1620 "OU=DUP2,DC=SAMBA,DC=ORG")
1621 self.fail("Should have failed on duplicate DN")
1622 except ldb.LdbError as err:
1624 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1626 self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1627 "OU=DUP3,DC=SAMBA,DC=ORG")
1629 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1630 "OU=DUP2,DC=SAMBA,DC=ORG")
1632 res2 = self.l.search(base="DC=SAMBA,DC=ORG",
1633 scope=ldb.SCOPE_SUBTREE,
1634 expression="(objectUUID=0123456789abcde1)")
1635 self.assertEqual(len(res2), 1)
1636 self.assertEqual(str(res2[0].dn), "OU=DUP2,DC=SAMBA,DC=ORG")
1638 res3 = self.l.search(base="DC=SAMBA,DC=ORG",
1639 scope=ldb.SCOPE_SUBTREE,
1640 expression="(objectUUID=0123456789abcde2)")
1641 self.assertEqual(len(res3), 1)
1642 self.assertEqual(str(res3[0].dn), "OU=DUP3,DC=SAMBA,DC=ORG")
1644 def test_move_missing(self):
1646 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1647 "OU=DUP2,DC=SAMBA,DC=ORG")
1648 self.fail("Should have failed on missing")
1649 except ldb.LdbError as err:
1651 self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1653 def test_move_missing2(self):
1654 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1657 "objectUUID": b"0123456789abcde2"})
1660 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1661 "OU=DUP2,DC=SAMBA,DC=ORG")
1662 self.fail("Should have failed on missing")
1663 except ldb.LdbError as err:
1665 self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1667 def test_move_bad(self):
1668 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1671 "objectUUID": b"0123456789abcde2"})
1674 self.l.rename("OUXDUP,DC=SAMBA,DC=ORG",
1675 "OU=DUP2,DC=SAMBA,DC=ORG")
1676 self.fail("Should have failed on invalid DN")
1677 except ldb.LdbError as err:
1679 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1681 def test_move_bad2(self):
1682 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1685 "objectUUID": b"0123456789abcde2"})
1688 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1689 "OUXDUP2,DC=SAMBA,DC=ORG")
1690 self.fail("Should have failed on missing")
1691 except ldb.LdbError as err:
1693 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1695 def test_move_fail_move_add(self):
1696 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1699 "objectUUID": b"0123456789abcde1"})
1700 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1703 "objectUUID": b"0123456789abcde2"})
1705 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1706 "OU=DUP2,DC=SAMBA,DC=ORG")
1707 self.fail("Should have failed on duplicate DN")
1708 except ldb.LdbError as err:
1710 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1712 self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1713 "OU=DUP3,DC=SAMBA,DC=ORG")
1715 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1718 "objectUUID": b"0123456789abcde3"})
1721 class AddModifyTestsLmdb(AddModifyTests):
1724 self.prefix = MDB_PREFIX
1725 self.index = MDB_INDEX_OBJ
1726 super(AddModifyTestsLmdb, self).setUp()
1729 super(AddModifyTestsLmdb, self).tearDown()
1732 class IndexedAddModifyTests(AddModifyTests):
1733 """Test searches using the index, to ensure the index doesn't
1737 if not hasattr(self, 'index'):
1738 self.index = {"dn": "@INDEXLIST",
1739 "@IDXATTR": [b"x", b"y", b"ou", b"objectUUID"],
1741 super(IndexedAddModifyTests, self).setUp()
1743 def test_duplicate_GUID(self):
1745 self.l.add({"dn": "OU=DUPGUID,DC=SAMBA,DC=ORG",
1748 "objectUUID": b"0123456789abcdef"})
1749 self.fail("Should have failed adding dupliate GUID")
1750 except ldb.LdbError as err:
1752 self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1754 def test_duplicate_name_dup_GUID(self):
1755 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1758 "objectUUID": b"a123456789abcdef"})
1760 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1763 "objectUUID": b"a123456789abcdef"})
1764 self.fail("Should have failed adding dupliate GUID")
1765 except ldb.LdbError as err:
1767 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1769 def test_duplicate_name_dup_GUID2(self):
1770 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1773 "objectUUID": b"abc3456789abcdef"})
1775 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1778 "objectUUID": b"aaa3456789abcdef"})
1779 self.fail("Should have failed adding dupliate DN")
1780 except ldb.LdbError as err:
1782 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1784 # Checking the GUID didn't stick in the index
1785 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1788 "objectUUID": b"aaa3456789abcdef"})
1790 def test_add_dup_guid_add(self):
1791 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1794 "objectUUID": b"0123456789abcde1"})
1796 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1799 "objectUUID": b"0123456789abcde1"})
1800 self.fail("Should have failed on duplicate GUID")
1802 except ldb.LdbError as err:
1804 self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1806 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1809 "objectUUID": b"0123456789abcde2"})
1812 class GUIDIndexedAddModifyTests(IndexedAddModifyTests):
1813 """Test searches using the index, to ensure the index doesn't
1817 self.index = {"dn": "@INDEXLIST",
1818 "@IDXATTR": [b"x", b"y", b"ou"],
1820 "@IDXGUID": [b"objectUUID"],
1821 "@IDX_DN_GUID": [b"GUID"]}
1822 super(GUIDIndexedAddModifyTests, self).setUp()
1825 class GUIDTransIndexedAddModifyTests(GUIDIndexedAddModifyTests):
1826 """Test GUID index behaviour insdie the transaction"""
1829 super(GUIDTransIndexedAddModifyTests, self).setUp()
1830 self.l.transaction_start()
1833 self.l.transaction_commit()
1834 super(GUIDTransIndexedAddModifyTests, self).tearDown()
1837 class TransIndexedAddModifyTests(IndexedAddModifyTests):
1838 """Test index behaviour insdie the transaction"""
1841 super(TransIndexedAddModifyTests, self).setUp()
1842 self.l.transaction_start()
1845 self.l.transaction_commit()
1846 super(TransIndexedAddModifyTests, self).tearDown()
1849 class GuidIndexedAddModifyTestsLmdb(GUIDIndexedAddModifyTests):
1852 self.prefix = MDB_PREFIX
1853 super(GuidIndexedAddModifyTestsLmdb, self).setUp()
1856 super(GuidIndexedAddModifyTestsLmdb, self).tearDown()
1859 class GuidTransIndexedAddModifyTestsLmdb(GUIDTransIndexedAddModifyTests):
1862 self.prefix = MDB_PREFIX
1863 super(GuidTransIndexedAddModifyTestsLmdb, self).setUp()
1866 super(GuidTransIndexedAddModifyTestsLmdb, self).tearDown()
1869 class BadIndexTests(LdbBaseTest):
1871 super(BadIndexTests, self).setUp()
1872 self.testdir = tempdir()
1873 self.filename = os.path.join(self.testdir, "test.ldb")
1874 self.ldb = ldb.Ldb(self.url(), flags=self.flags())
1875 if hasattr(self, 'IDXGUID'):
1876 self.ldb.add({"dn": "@INDEXLIST",
1877 "@IDXATTR": [b"x", b"y", b"ou"],
1878 "@IDXGUID": [b"objectUUID"],
1879 "@IDX_DN_GUID": [b"GUID"]})
1881 self.ldb.add({"dn": "@INDEXLIST",
1882 "@IDXATTR": [b"x", b"y", b"ou"]})
1884 super(BadIndexTests, self).setUp()
1886 def test_unique(self):
1887 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1888 "objectUUID": b"0123456789abcde1",
1890 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1891 "objectUUID": b"0123456789abcde2",
1893 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1894 "objectUUID": b"0123456789abcde3",
1897 res = self.ldb.search(expression="(y=1)",
1898 base="dc=samba,dc=org")
1899 self.assertEquals(len(res), 3)
1901 # Now set this to unique index, but forget to check the result
1903 self.ldb.add({"dn": "@ATTRIBUTES",
1904 "y": "UNIQUE_INDEX"})
1906 except ldb.LdbError:
1909 # We must still have a working index
1910 res = self.ldb.search(expression="(y=1)",
1911 base="dc=samba,dc=org")
1912 self.assertEquals(len(res), 3)
1914 def test_unique_transaction(self):
1915 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1916 "objectUUID": b"0123456789abcde1",
1918 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1919 "objectUUID": b"0123456789abcde2",
1921 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1922 "objectUUID": b"0123456789abcde3",
1925 res = self.ldb.search(expression="(y=1)",
1926 base="dc=samba,dc=org")
1927 self.assertEquals(len(res), 3)
1929 self.ldb.transaction_start()
1931 # Now set this to unique index, but forget to check the result
1933 self.ldb.add({"dn": "@ATTRIBUTES",
1934 "y": "UNIQUE_INDEX"})
1935 except ldb.LdbError:
1939 self.ldb.transaction_commit()
1942 except ldb.LdbError as err:
1944 self.assertEqual(enum, ldb.ERR_OPERATIONS_ERROR)
1946 # We must still have a working index
1947 res = self.ldb.search(expression="(y=1)",
1948 base="dc=samba,dc=org")
1950 self.assertEquals(len(res), 3)
1952 def test_casefold(self):
1953 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1954 "objectUUID": b"0123456789abcde1",
1956 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1957 "objectUUID": b"0123456789abcde2",
1959 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1960 "objectUUID": b"0123456789abcde3",
1963 res = self.ldb.search(expression="(y=a)",
1964 base="dc=samba,dc=org")
1965 self.assertEquals(len(res), 2)
1967 self.ldb.add({"dn": "@ATTRIBUTES",
1968 "y": "CASE_INSENSITIVE"})
1970 # We must still have a working index
1971 res = self.ldb.search(expression="(y=a)",
1972 base="dc=samba,dc=org")
1974 if hasattr(self, 'IDXGUID'):
1975 self.assertEquals(len(res), 3)
1977 # We should not return this entry twice, but sadly
1978 # we have not yet fixed
1979 # https://bugzilla.samba.org/show_bug.cgi?id=13361
1980 self.assertEquals(len(res), 4)
1982 def test_casefold_transaction(self):
1983 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1984 "objectUUID": b"0123456789abcde1",
1986 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1987 "objectUUID": b"0123456789abcde2",
1989 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1990 "objectUUID": b"0123456789abcde3",
1993 res = self.ldb.search(expression="(y=a)",
1994 base="dc=samba,dc=org")
1995 self.assertEquals(len(res), 2)
1997 self.ldb.transaction_start()
1999 self.ldb.add({"dn": "@ATTRIBUTES",
2000 "y": "CASE_INSENSITIVE"})
2002 self.ldb.transaction_commit()
2004 # We must still have a working index
2005 res = self.ldb.search(expression="(y=a)",
2006 base="dc=samba,dc=org")
2008 if hasattr(self, 'IDXGUID'):
2009 self.assertEquals(len(res), 3)
2011 # We should not return this entry twice, but sadly
2012 # we have not yet fixed
2013 # https://bugzilla.samba.org/show_bug.cgi?id=13361
2014 self.assertEquals(len(res), 4)
2017 super(BadIndexTests, self).tearDown()
2020 class GUIDBadIndexTests(BadIndexTests):
2021 """Test Bad index things with GUID index mode"""
2026 super(GUIDBadIndexTests, self).setUp()
2029 class DnTests(TestCase):
2032 super(DnTests, self).setUp()
2033 self.ldb = ldb.Ldb()
2036 super(DnTests, self).tearDown()
2039 def test_set_dn_invalid(self):
2044 self.assertRaises(TypeError, assign)
2047 x = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
2048 y = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
2049 self.assertEqual(x, y)
2050 y = ldb.Dn(self.ldb, "dc=foo11,bar=blie")
2051 self.assertNotEqual(x, y)
2054 x = ldb.Dn(self.ldb, "dc=foo12,bar=bloe")
2055 self.assertEqual(x.__str__(), "dc=foo12,bar=bloe")
2057 def test_repr(self):
2058 x = ldb.Dn(self.ldb, "dc=foo13,bla=blie")
2059 self.assertEqual(x.__repr__(), "Dn('dc=foo13,bla=blie')")
2061 def test_get_casefold(self):
2062 x = ldb.Dn(self.ldb, "dc=foo14,bar=bloe")
2063 self.assertEqual(x.get_casefold(), "DC=FOO14,BAR=bloe")
2065 def test_validate(self):
2066 x = ldb.Dn(self.ldb, "dc=foo15,bar=bloe")
2067 self.assertTrue(x.validate())
2069 def test_parent(self):
2070 x = ldb.Dn(self.ldb, "dc=foo16,bar=bloe")
2071 self.assertEqual("bar=bloe", x.parent().__str__())
2073 def test_parent_nonexistent(self):
2074 x = ldb.Dn(self.ldb, "@BLA")
2075 self.assertEqual(None, x.parent())
2077 def test_is_valid(self):
2078 x = ldb.Dn(self.ldb, "dc=foo18,dc=bloe")
2079 self.assertTrue(x.is_valid())
2080 x = ldb.Dn(self.ldb, "")
2081 self.assertTrue(x.is_valid())
2083 def test_is_special(self):
2084 x = ldb.Dn(self.ldb, "dc=foo19,bar=bloe")
2085 self.assertFalse(x.is_special())
2086 x = ldb.Dn(self.ldb, "@FOOBAR")
2087 self.assertTrue(x.is_special())
2089 def test_check_special(self):
2090 x = ldb.Dn(self.ldb, "dc=foo20,bar=bloe")
2091 self.assertFalse(x.check_special("FOOBAR"))
2092 x = ldb.Dn(self.ldb, "@FOOBAR")
2093 self.assertTrue(x.check_special("@FOOBAR"))
2096 x = ldb.Dn(self.ldb, "dc=foo21,bar=bloe")
2097 self.assertEqual(2, len(x))
2098 x = ldb.Dn(self.ldb, "dc=foo21")
2099 self.assertEqual(1, len(x))
2101 def test_add_child(self):
2102 x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
2103 self.assertTrue(x.add_child(ldb.Dn(self.ldb, "bla=bloe")))
2104 self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
2106 def test_add_base(self):
2107 x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
2108 base = ldb.Dn(self.ldb, "bla=bloe")
2109 self.assertTrue(x.add_base(base))
2110 self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
2112 def test_add_child_str(self):
2113 x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
2114 self.assertTrue(x.add_child("bla=bloe"))
2115 self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
2117 def test_add_base_str(self):
2118 x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
2120 self.assertTrue(x.add_base(base))
2121 self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
2124 x = ldb.Dn(self.ldb, "dc=foo24")
2125 y = ldb.Dn(self.ldb, "bar=bla")
2126 self.assertEqual("dc=foo24,bar=bla", str(x + y))
2128 def test_remove_base_components(self):
2129 x = ldb.Dn(self.ldb, "dc=foo24,dc=samba,dc=org")
2130 x.remove_base_components(len(x) - 1)
2131 self.assertEqual("dc=foo24", str(x))
2133 def test_parse_ldif(self):
2134 msgs = self.ldb.parse_ldif("dn: foo=bar\n")
2136 self.assertEqual("foo=bar", str(msg[1].dn))
2137 self.assertTrue(isinstance(msg[1], ldb.Message))
2138 ldif = self.ldb.write_ldif(msg[1], ldb.CHANGETYPE_NONE)
2139 self.assertEqual("dn: foo=bar\n\n", ldif)
2141 def test_parse_ldif_more(self):
2142 msgs = self.ldb.parse_ldif("dn: foo=bar\n\n\ndn: bar=bar")
2144 self.assertEqual("foo=bar", str(msg[1].dn))
2146 self.assertEqual("bar=bar", str(msg[1].dn))
2148 def test_canonical_string(self):
2149 x = ldb.Dn(self.ldb, "dc=foo25,bar=bloe")
2150 self.assertEqual("/bloe/foo25", x.canonical_str())
2152 def test_canonical_ex_string(self):
2153 x = ldb.Dn(self.ldb, "dc=foo26,bar=bloe")
2154 self.assertEqual("/bloe\nfoo26", x.canonical_ex_str())
2156 def test_ldb_is_child_of(self):
2157 """Testing ldb_dn_compare_dn"""
2158 dn1 = ldb.Dn(self.ldb, "dc=base")
2159 dn2 = ldb.Dn(self.ldb, "cn=foo,dc=base")
2160 dn3 = ldb.Dn(self.ldb, "cn=bar,dc=base")
2161 dn4 = ldb.Dn(self.ldb, "cn=baz,cn=bar,dc=base")
2163 self.assertTrue(dn1.is_child_of(dn1))
2164 self.assertTrue(dn2.is_child_of(dn1))
2165 self.assertTrue(dn4.is_child_of(dn1))
2166 self.assertTrue(dn4.is_child_of(dn3))
2167 self.assertTrue(dn4.is_child_of(dn4))
2168 self.assertFalse(dn3.is_child_of(dn2))
2169 self.assertFalse(dn1.is_child_of(dn4))
2171 def test_ldb_is_child_of_str(self):
2172 """Testing ldb_dn_compare_dn"""
2174 dn2_str = "cn=foo,dc=base"
2175 dn3_str = "cn=bar,dc=base"
2176 dn4_str = "cn=baz,cn=bar,dc=base"
2178 dn1 = ldb.Dn(self.ldb, dn1_str)
2179 dn2 = ldb.Dn(self.ldb, dn2_str)
2180 dn3 = ldb.Dn(self.ldb, dn3_str)
2181 dn4 = ldb.Dn(self.ldb, dn4_str)
2183 self.assertTrue(dn1.is_child_of(dn1_str))
2184 self.assertTrue(dn2.is_child_of(dn1_str))
2185 self.assertTrue(dn4.is_child_of(dn1_str))
2186 self.assertTrue(dn4.is_child_of(dn3_str))
2187 self.assertTrue(dn4.is_child_of(dn4_str))
2188 self.assertFalse(dn3.is_child_of(dn2_str))
2189 self.assertFalse(dn1.is_child_of(dn4_str))
2191 def test_get_component_name(self):
2192 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2193 self.assertEqual(dn.get_component_name(0), 'cn')
2194 self.assertEqual(dn.get_component_name(1), 'dc')
2195 self.assertEqual(dn.get_component_name(2), None)
2196 self.assertEqual(dn.get_component_name(-1), None)
2198 def test_get_component_value(self):
2199 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2200 self.assertEqual(dn.get_component_value(0), 'foo')
2201 self.assertEqual(dn.get_component_value(1), 'base')
2202 self.assertEqual(dn.get_component_name(2), None)
2203 self.assertEqual(dn.get_component_name(-1), None)
2205 def test_set_component(self):
2206 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2207 dn.set_component(0, 'cn', 'bar')
2208 self.assertEqual(str(dn), "cn=bar,dc=base")
2209 dn.set_component(1, 'o', 'asep')
2210 self.assertEqual(str(dn), "cn=bar,o=asep")
2211 self.assertRaises(TypeError, dn.set_component, 2, 'dc', 'base')
2212 self.assertEqual(str(dn), "cn=bar,o=asep")
2213 dn.set_component(1, 'o', 'a,b+c')
2214 self.assertEqual(str(dn), r"cn=bar,o=a\,b\+c")
2216 def test_set_component_bytes(self):
2217 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2218 dn.set_component(0, 'cn', b'bar')
2219 self.assertEqual(str(dn), "cn=bar,dc=base")
2220 dn.set_component(1, 'o', b'asep')
2221 self.assertEqual(str(dn), "cn=bar,o=asep")
2223 def test_set_component_none(self):
2224 dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
2225 self.assertRaises(TypeError, dn.set_component, 1, 'cn', None)
2227 def test_get_extended_component_null(self):
2228 dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
2229 self.assertEqual(dn.get_extended_component("TEST"), None)
2231 def test_get_extended_component(self):
2232 self.ldb._register_test_extensions()
2233 dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
2234 self.assertEqual(dn.get_extended_component("TEST"), b"foo")
2236 def test_set_extended_component(self):
2237 self.ldb._register_test_extensions()
2238 dn = ldb.Dn(self.ldb, "dc=base")
2239 dn.set_extended_component("TEST", "foo")
2240 self.assertEqual(dn.get_extended_component("TEST"), b"foo")
2241 dn.set_extended_component("TEST", b"bar")
2242 self.assertEqual(dn.get_extended_component("TEST"), b"bar")
2244 def test_extended_str(self):
2245 self.ldb._register_test_extensions()
2246 dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
2247 self.assertEqual(dn.extended_str(), "<TEST=foo>;cn=bar,dc=base")
2249 def test_get_rdn_name(self):
2250 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2251 self.assertEqual(dn.get_rdn_name(), 'cn')
2253 def test_get_rdn_value(self):
2254 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2255 self.assertEqual(dn.get_rdn_value(), 'foo')
2257 def test_get_casefold(self):
2258 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2259 self.assertEqual(dn.get_casefold(), 'CN=FOO,DC=BASE')
2261 def test_get_linearized(self):
2262 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2263 self.assertEqual(dn.get_linearized(), 'cn=foo,dc=base')
2265 def test_is_null(self):
2266 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2267 self.assertFalse(dn.is_null())
2269 dn = ldb.Dn(self.ldb, '')
2270 self.assertTrue(dn.is_null())
2273 class LdbMsgTests(TestCase):
2276 super(LdbMsgTests, self).setUp()
2277 self.msg = ldb.Message()
2279 def test_init_dn(self):
2280 self.msg = ldb.Message(ldb.Dn(ldb.Ldb(), "dc=foo27"))
2281 self.assertEqual("dc=foo27", str(self.msg.dn))
2283 def test_iter_items(self):
2284 self.assertEqual(0, len(self.msg.items()))
2285 self.msg.dn = ldb.Dn(ldb.Ldb(), "dc=foo28")
2286 self.assertEqual(1, len(self.msg.items()))
2288 def test_repr(self):
2289 self.msg.dn = ldb.Dn(ldb.Ldb(), "dc=foo29")
2290 self.msg["dc"] = b"foo"
2292 self.assertIn(repr(self.msg), [
2293 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])})",
2294 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')})",
2296 self.assertIn(repr(self.msg.text), [
2297 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])}).text",
2298 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')}).text",
2303 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])})")
2305 repr(self.msg.text),
2306 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])}).text")
2309 self.assertEqual(0, len(self.msg))
2311 def test_notpresent(self):
2312 self.assertRaises(KeyError, lambda: self.msg["foo"])
2318 self.msg.add(ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla"))
2320 def test_add_text(self):
2321 self.msg.add(ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla"))
2323 def test_elements_empty(self):
2324 self.assertEqual([], self.msg.elements())
2326 def test_elements(self):
2327 el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
2329 self.assertEqual([el], self.msg.elements())
2330 self.assertEqual([el.text], self.msg.text.elements())
2332 def test_add_value(self):
2333 self.assertEqual(0, len(self.msg))
2334 self.msg["foo"] = [b"foo"]
2335 self.assertEqual(1, len(self.msg))
2337 def test_add_value_text(self):
2338 self.assertEqual(0, len(self.msg))
2339 self.msg["foo"] = ["foo"]
2340 self.assertEqual(1, len(self.msg))
2342 def test_add_value_multiple(self):
2343 self.assertEqual(0, len(self.msg))
2344 self.msg["foo"] = [b"foo", b"bla"]
2345 self.assertEqual(1, len(self.msg))
2346 self.assertEqual([b"foo", b"bla"], list(self.msg["foo"]))
2348 def test_add_value_multiple_text(self):
2349 self.assertEqual(0, len(self.msg))
2350 self.msg["foo"] = ["foo", "bla"]
2351 self.assertEqual(1, len(self.msg))
2352 self.assertEqual(["foo", "bla"], list(self.msg.text["foo"]))
2354 def test_set_value(self):
2355 self.msg["foo"] = [b"fool"]
2356 self.assertEqual([b"fool"], list(self.msg["foo"]))
2357 self.msg["foo"] = [b"bar"]
2358 self.assertEqual([b"bar"], list(self.msg["foo"]))
2360 def test_set_value_text(self):
2361 self.msg["foo"] = ["fool"]
2362 self.assertEqual(["fool"], list(self.msg.text["foo"]))
2363 self.msg["foo"] = ["bar"]
2364 self.assertEqual(["bar"], list(self.msg.text["foo"]))
2366 def test_keys(self):
2367 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2368 self.msg["foo"] = [b"bla"]
2369 self.msg["bar"] = [b"bla"]
2370 self.assertEqual(["dn", "foo", "bar"], list(self.msg.keys()))
2372 def test_keys_text(self):
2373 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2374 self.msg["foo"] = ["bla"]
2375 self.msg["bar"] = ["bla"]
2376 self.assertEqual(["dn", "foo", "bar"], list(self.msg.text.keys()))
2379 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2380 self.assertEqual("@BASEINFO", self.msg.dn.__str__())
2382 def test_get_dn(self):
2383 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2384 self.assertEqual("@BASEINFO", self.msg.get("dn").__str__())
2386 def test_dn_text(self):
2387 self.msg.text.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2388 self.assertEqual("@BASEINFO", str(self.msg.dn))
2389 self.assertEqual("@BASEINFO", str(self.msg.text.dn))
2391 def test_get_dn_text(self):
2392 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2393 self.assertEqual("@BASEINFO", str(self.msg.get("dn")))
2394 self.assertEqual("@BASEINFO", str(self.msg.text.get("dn")))
2396 def test_get_invalid(self):
2397 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2398 self.assertRaises(TypeError, self.msg.get, 42)
2400 def test_get_other(self):
2401 self.msg["foo"] = [b"bar"]
2402 self.assertEqual(b"bar", self.msg.get("foo")[0])
2403 self.assertEqual(b"bar", self.msg.get("foo", idx=0))
2404 self.assertEqual(None, self.msg.get("foo", idx=1))
2405 self.assertEqual("", self.msg.get("foo", default='', idx=1))
2407 def test_get_other_text(self):
2408 self.msg["foo"] = ["bar"]
2409 self.assertEqual(["bar"], list(self.msg.text.get("foo")))
2410 self.assertEqual("bar", self.msg.text.get("foo")[0])
2411 self.assertEqual("bar", self.msg.text.get("foo", idx=0))
2412 self.assertEqual(None, self.msg.get("foo", idx=1))
2413 self.assertEqual("", self.msg.get("foo", default='', idx=1))
2415 def test_get_default(self):
2416 self.assertEqual(None, self.msg.get("tatayoyo", idx=0))
2417 self.assertEqual("anniecordie", self.msg.get("tatayoyo", "anniecordie"))
2419 def test_get_default_text(self):
2420 self.assertEqual(None, self.msg.text.get("tatayoyo", idx=0))
2421 self.assertEqual("anniecordie", self.msg.text.get("tatayoyo", "anniecordie"))
2423 def test_get_unknown(self):
2424 self.assertEqual(None, self.msg.get("lalalala"))
2426 def test_get_unknown_text(self):
2427 self.assertEqual(None, self.msg.text.get("lalalala"))
2429 def test_msg_diff(self):
2431 msgs = l.parse_ldif("dn: foo=bar\nfoo: bar\nbaz: do\n\ndn: foo=bar\nfoo: bar\nbaz: dont\n")
2432 msg1 = next(msgs)[1]
2433 msg2 = next(msgs)[1]
2434 msgdiff = l.msg_diff(msg1, msg2)
2435 self.assertEqual("foo=bar", msgdiff.get("dn").__str__())
2436 self.assertRaises(KeyError, lambda: msgdiff["foo"])
2437 self.assertEqual(1, len(msgdiff))
2439 def test_equal_empty(self):
2440 msg1 = ldb.Message()
2441 msg2 = ldb.Message()
2442 self.assertEqual(msg1, msg2)
2444 def test_equal_simplel(self):
2446 msg1 = ldb.Message()
2447 msg1.dn = ldb.Dn(db, "foo=bar")
2448 msg2 = ldb.Message()
2449 msg2.dn = ldb.Dn(db, "foo=bar")
2450 self.assertEqual(msg1, msg2)
2451 msg1['foo'] = b'bar'
2452 msg2['foo'] = b'bar'
2453 self.assertEqual(msg1, msg2)
2454 msg2['foo'] = b'blie'
2455 self.assertNotEqual(msg1, msg2)
2456 msg2['foo'] = b'blie'
2458 def test_from_dict(self):
2459 rec = {"dn": "dc=fromdict",
2460 "a1": [b"a1-val1", b"a1-val1"]}
2462 # check different types of input Flags
2463 for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
2464 m = ldb.Message.from_dict(l, rec, flags)
2465 self.assertEqual(rec["a1"], list(m["a1"]))
2466 self.assertEqual(flags, m["a1"].flags())
2467 # check input params
2468 self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
2469 self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
2470 self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
2471 # Message.from_dict expects dictionary with 'dn'
2472 err_rec = {"a1": [b"a1-val1", b"a1-val1"]}
2473 self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
2475 def test_from_dict_text(self):
2476 rec = {"dn": "dc=fromdict",
2477 "a1": ["a1-val1", "a1-val1"]}
2479 # check different types of input Flags
2480 for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
2481 m = ldb.Message.from_dict(l, rec, flags)
2482 self.assertEqual(rec["a1"], list(m.text["a1"]))
2483 self.assertEqual(flags, m.text["a1"].flags())
2484 # check input params
2485 self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
2486 self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
2487 self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
2488 # Message.from_dict expects dictionary with 'dn'
2489 err_rec = {"a1": ["a1-val1", "a1-val1"]}
2490 self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
2492 def test_copy_add_message_element(self):
2494 m["1"] = ldb.MessageElement([b"val 111"], ldb.FLAG_MOD_ADD, "1")
2495 m["2"] = ldb.MessageElement([b"val 222"], ldb.FLAG_MOD_ADD, "2")
2499 self.assertEqual(mto["1"], m["1"])
2500 self.assertEqual(mto["2"], m["2"])
2504 self.assertEqual(mto["1"], m["1"])
2505 self.assertEqual(mto["2"], m["2"])
2507 def test_copy_add_message_element_text(self):
2509 m["1"] = ldb.MessageElement(["val 111"], ldb.FLAG_MOD_ADD, "1")
2510 m["2"] = ldb.MessageElement(["val 222"], ldb.FLAG_MOD_ADD, "2")
2514 self.assertEqual(mto["1"], m.text["1"])
2515 self.assertEqual(mto["2"], m.text["2"])
2519 self.assertEqual(mto.text["1"], m.text["1"])
2520 self.assertEqual(mto.text["2"], m.text["2"])
2521 self.assertEqual(mto["1"], m["1"])
2522 self.assertEqual(mto["2"], m["2"])
2525 class MessageElementTests(TestCase):
2527 def test_cmp_element(self):
2528 x = ldb.MessageElement([b"foo"])
2529 y = ldb.MessageElement([b"foo"])
2530 z = ldb.MessageElement([b"bzr"])
2531 self.assertEqual(x, y)
2532 self.assertNotEqual(x, z)
2534 def test_cmp_element_text(self):
2535 x = ldb.MessageElement([b"foo"])
2536 y = ldb.MessageElement(["foo"])
2537 self.assertEqual(x, y)
2539 def test_create_iterable(self):
2540 x = ldb.MessageElement([b"foo"])
2541 self.assertEqual([b"foo"], list(x))
2542 self.assertEqual(["foo"], list(x.text))
2544 def test_repr(self):
2545 x = ldb.MessageElement([b"foo"])
2547 self.assertEqual("MessageElement([b'foo'])", repr(x))
2548 self.assertEqual("MessageElement([b'foo']).text", repr(x.text))
2550 self.assertEqual("MessageElement(['foo'])", repr(x))
2551 self.assertEqual("MessageElement(['foo']).text", repr(x.text))
2552 x = ldb.MessageElement([b"foo", b"bla"])
2553 self.assertEqual(2, len(x))
2555 self.assertEqual("MessageElement([b'foo',b'bla'])", repr(x))
2556 self.assertEqual("MessageElement([b'foo',b'bla']).text", repr(x.text))
2558 self.assertEqual("MessageElement(['foo','bla'])", repr(x))
2559 self.assertEqual("MessageElement(['foo','bla']).text", repr(x.text))
2561 def test_get_item(self):
2562 x = ldb.MessageElement([b"foo", b"bar"])
2563 self.assertEqual(b"foo", x[0])
2564 self.assertEqual(b"bar", x[1])
2565 self.assertEqual(b"bar", x[-1])
2566 self.assertRaises(IndexError, lambda: x[45])
2568 def test_get_item_text(self):
2569 x = ldb.MessageElement(["foo", "bar"])
2570 self.assertEqual("foo", x.text[0])
2571 self.assertEqual("bar", x.text[1])
2572 self.assertEqual("bar", x.text[-1])
2573 self.assertRaises(IndexError, lambda: x[45])
2576 x = ldb.MessageElement([b"foo", b"bar"])
2577 self.assertEqual(2, len(x))
2580 x = ldb.MessageElement([b"foo", b"bar"])
2581 y = ldb.MessageElement([b"foo", b"bar"])
2582 self.assertEqual(y, x)
2583 x = ldb.MessageElement([b"foo"])
2584 self.assertNotEqual(y, x)
2585 y = ldb.MessageElement([b"foo"])
2586 self.assertEqual(y, x)
2588 def test_extended(self):
2589 el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
2591 self.assertEqual("MessageElement([b'456'])", repr(el))
2592 self.assertEqual("MessageElement([b'456']).text", repr(el.text))
2594 self.assertEqual("MessageElement(['456'])", repr(el))
2595 self.assertEqual("MessageElement(['456']).text", repr(el.text))
2597 def test_bad_text(self):
2598 el = ldb.MessageElement(b'\xba\xdd')
2599 self.assertRaises(UnicodeDecodeError, el.text.__getitem__, 0)
2602 class ModuleTests(TestCase):
2605 super(ModuleTests, self).setUp()
2606 self.testdir = tempdir()
2607 self.filename = os.path.join(self.testdir, "test.ldb")
2608 self.ldb = ldb.Ldb(self.filename)
2611 shutil.rmtree(self.testdir)
2612 super(ModuleTests, self).setUp()
2614 def test_register_module(self):
2615 class ExampleModule:
2617 ldb.register_module(ExampleModule)
2619 def test_use_module(self):
2622 class ExampleModule:
2625 def __init__(self, ldb, next):
2629 def search(self, *args, **kwargs):
2630 return self.next.search(*args, **kwargs)
2632 def request(self, *args, **kwargs):
2635 ldb.register_module(ExampleModule)
2636 l = ldb.Ldb(self.filename)
2637 l.add({"dn": "@MODULES", "@LIST": "bla"})
2638 self.assertEqual([], ops)
2639 l = ldb.Ldb(self.filename)
2640 self.assertEqual(["init"], ops)
2643 class LdbResultTests(LdbBaseTest):
2646 super(LdbResultTests, self).setUp()
2647 self.testdir = tempdir()
2648 self.filename = os.path.join(self.testdir, "test.ldb")
2649 self.l = ldb.Ldb(self.url(), flags=self.flags())
2651 self.l.add(self.index)
2652 except AttributeError:
2654 self.l.add({"dn": "DC=SAMBA,DC=ORG", "name": b"samba.org",
2655 "objectUUID": b"0123456789abcde0"})
2656 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG", "name": b"Admins",
2657 "objectUUID": b"0123456789abcde1"})
2658 self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG", "name": b"Users",
2659 "objectUUID": b"0123456789abcde2"})
2660 self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG", "name": b"OU #1",
2661 "objectUUID": b"0123456789abcde3"})
2662 self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG", "name": b"OU #2",
2663 "objectUUID": b"0123456789abcde4"})
2664 self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG", "name": b"OU #3",
2665 "objectUUID": b"0123456789abcde5"})
2666 self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG", "name": b"OU #4",
2667 "objectUUID": b"0123456789abcde6"})
2668 self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG", "name": b"OU #5",
2669 "objectUUID": b"0123456789abcde7"})
2670 self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG", "name": b"OU #6",
2671 "objectUUID": b"0123456789abcde8"})
2672 self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG", "name": b"OU #7",
2673 "objectUUID": b"0123456789abcde9"})
2674 self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG", "name": b"OU #8",
2675 "objectUUID": b"0123456789abcdea"})
2676 self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG", "name": b"OU #9",
2677 "objectUUID": b"0123456789abcdeb"})
2678 self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG", "name": b"OU #10",
2679 "objectUUID": b"0123456789abcdec"})
2682 shutil.rmtree(self.testdir)
2683 super(LdbResultTests, self).tearDown()
2684 # Ensure the LDB is closed now, so we close the FD
2687 def test_return_type(self):
2688 res = self.l.search()
2689 self.assertEqual(str(res), "<ldb result>")
2691 def test_get_msgs(self):
2692 res = self.l.search()
2695 def test_get_controls(self):
2696 res = self.l.search()
2699 def test_get_referals(self):
2700 res = self.l.search()
2703 def test_iter_msgs(self):
2705 for l in self.l.search().msgs:
2706 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2708 self.assertTrue(found)
2710 def test_iter_msgs_count(self):
2711 self.assertTrue(self.l.search().count > 0)
2712 # 13 objects has been added to the DC=SAMBA, DC=ORG
2713 self.assertEqual(self.l.search(base="DC=SAMBA,DC=ORG").count, 13)
2715 def test_iter_controls(self):
2716 res = self.l.search().controls
2719 def test_create_control(self):
2720 self.assertRaises(ValueError, ldb.Control, self.l, "tatayoyo:0")
2721 c = ldb.Control(self.l, "relax:1")
2722 self.assertEqual(c.critical, True)
2723 self.assertEqual(c.oid, "1.3.6.1.4.1.4203.666.5.12")
2725 def test_iter_refs(self):
2726 res = self.l.search().referals
2729 def test_search_sequence_msgs(self):
2731 res = self.l.search().msgs
2733 for i in range(0, len(res)):
2735 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2737 self.assertTrue(found)
2739 def test_search_as_iter(self):
2741 res = self.l.search()
2744 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2746 self.assertTrue(found)
2748 def test_search_iter(self):
2750 res = self.l.search_iterator()
2753 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2755 self.assertTrue(found)
2757 # Show that search results can't see into a transaction
2759 def test_search_against_trans(self):
2762 (r1, w1) = os.pipe()
2764 (r2, w2) = os.pipe()
2766 # For the first element, fork a child that will
2770 # In the child, re-open
2774 child_ldb = ldb.Ldb(self.url(), flags=self.flags())
2775 # start a transaction
2776 child_ldb.transaction_start()
2779 child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
2780 "name": b"samba.org",
2781 "objectUUID": b"o123456789acbdef"})
2783 os.write(w1, b"added")
2785 # Now wait for the search to be done
2790 child_ldb.transaction_commit()
2791 except LdbError as err:
2792 # We print this here to see what went wrong in the child
2796 os.write(w1, b"transaction")
2799 self.assertEqual(os.read(r1, 5), b"added")
2801 # This should not turn up until the transaction is concluded
2802 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2803 scope=ldb.SCOPE_BASE)
2804 self.assertEqual(len(res11), 0)
2806 os.write(w2, b"search")
2808 # Now wait for the transaction to be done. This should
2809 # deadlock, but the search doesn't hold a read lock for the
2810 # iterator lifetime currently.
2811 self.assertEqual(os.read(r1, 11), b"transaction")
2813 # This should now turn up, as the transaction is over
2814 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2815 scope=ldb.SCOPE_BASE)
2816 self.assertEqual(len(res11), 1)
2818 self.assertFalse(found11)
2820 (got_pid, status) = os.waitpid(pid, 0)
2821 self.assertEqual(got_pid, pid)
2823 def test_search_iter_against_trans(self):
2827 # We need to hold this iterator open to hold the all-record
2829 res = self.l.search_iterator()
2831 (r1, w1) = os.pipe()
2833 (r2, w2) = os.pipe()
2835 # For the first element, with the sequence open (which
2836 # means with ldb locks held), fork a child that will
2840 # In the child, re-open
2845 child_ldb = ldb.Ldb(self.url(), flags=self.flags())
2846 # start a transaction
2847 child_ldb.transaction_start()
2850 child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
2851 "name": b"samba.org",
2852 "objectUUID": b"o123456789acbdef"})
2854 os.write(w1, b"added")
2856 # Now wait for the search to be done
2861 child_ldb.transaction_commit()
2862 except LdbError as err:
2863 # We print this here to see what went wrong in the child
2867 os.write(w1, b"transaction")
2870 self.assertEqual(os.read(r1, 5), b"added")
2872 # This should not turn up until the transaction is concluded
2873 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2874 scope=ldb.SCOPE_BASE)
2875 self.assertEqual(len(res11), 0)
2877 os.write(w2, b"search")
2879 # allow the transaction to start
2882 # This should not turn up until the search finishes and
2883 # removed the read lock, but for ldb_tdb that happened as soon
2884 # as we called the first res.next()
2885 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2886 scope=ldb.SCOPE_BASE)
2887 self.assertEqual(len(res11), 0)
2889 # These results are all collected at the first next(res) call
2891 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2893 if str(l.dn) == "OU=OU11,DC=SAMBA,DC=ORG":
2896 # Now wait for the transaction to be done.
2897 self.assertEqual(os.read(r1, 11), b"transaction")
2899 # This should now turn up, as the transaction is over and all
2900 # read locks are gone
2901 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2902 scope=ldb.SCOPE_BASE)
2903 self.assertEqual(len(res11), 1)
2905 self.assertTrue(found)
2906 self.assertFalse(found11)
2908 (got_pid, status) = os.waitpid(pid, 0)
2909 self.assertEqual(got_pid, pid)
2912 class LdbResultTestsLmdb(LdbResultTests):
2915 self.prefix = MDB_PREFIX
2916 self.index = MDB_INDEX_OBJ
2917 super(LdbResultTestsLmdb, self).setUp()
2920 super(LdbResultTestsLmdb, self).tearDown()
2923 class BadTypeTests(TestCase):
2924 def test_control(self):
2926 self.assertRaises(TypeError, ldb.Control, '<bad type>', 'relax:1')
2927 self.assertRaises(TypeError, ldb.Control, ldb, 1234)
2929 def test_modify(self):
2931 dn = ldb.Dn(l, 'a=b')
2933 self.assertRaises(TypeError, l.modify, '<bad type>')
2934 self.assertRaises(TypeError, l.modify, m, '<bad type>')
2938 dn = ldb.Dn(l, 'a=b')
2940 self.assertRaises(TypeError, l.add, '<bad type>')
2941 self.assertRaises(TypeError, l.add, m, '<bad type>')
2943 def test_delete(self):
2945 dn = ldb.Dn(l, 'a=b')
2946 self.assertRaises(TypeError, l.add, '<bad type>')
2947 self.assertRaises(TypeError, l.add, dn, '<bad type>')
2949 def test_rename(self):
2951 dn = ldb.Dn(l, 'a=b')
2952 self.assertRaises(TypeError, l.add, '<bad type>', dn)
2953 self.assertRaises(TypeError, l.add, dn, '<bad type>')
2954 self.assertRaises(TypeError, l.add, dn, dn, '<bad type>')
2956 def test_search(self):
2958 self.assertRaises(TypeError, l.search, base=1234)
2959 self.assertRaises(TypeError, l.search, scope='<bad type>')
2960 self.assertRaises(TypeError, l.search, expression=1234)
2961 self.assertRaises(TypeError, l.search, attrs='<bad type>')
2962 self.assertRaises(TypeError, l.search, controls='<bad type>')
2965 class VersionTests(TestCase):
2967 def test_version(self):
2968 self.assertTrue(isinstance(ldb.__version__, str))
2971 if __name__ == '__main__':
2973 unittest.TestProgram()