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_utf8_ldb_Dn(self):
142 l = ldb.Ldb(self.url(), flags=self.flags())
143 dn = ldb.Dn(l, (b'a=' + b'\xc4\x85\xc4\x87\xc4\x99\xc5\x82\xc5\x84\xc3\xb3\xc5\x9b\xc5\xba\xc5\xbc').decode('utf8'))
145 def test_search_attrs(self):
146 l = ldb.Ldb(self.url(), flags=self.flags())
147 self.assertEqual(len(l.search(ldb.Dn(l, ""), ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
149 def test_search_string_dn(self):
150 l = ldb.Ldb(self.url(), flags=self.flags())
151 self.assertEqual(len(l.search("", ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
153 def test_search_attr_string(self):
154 l = ldb.Ldb(self.url(), flags=self.flags())
155 self.assertRaises(TypeError, l.search, attrs="dc")
156 self.assertRaises(TypeError, l.search, attrs=b"dc")
158 def test_opaque(self):
159 l = ldb.Ldb(self.url(), flags=self.flags())
160 l.set_opaque("my_opaque", l)
161 self.assertTrue(l.get_opaque("my_opaque") is not None)
162 self.assertEqual(None, l.get_opaque("unknown"))
164 def test_search_scope_base_empty_db(self):
165 l = ldb.Ldb(self.url(), flags=self.flags())
166 self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
169 def test_search_scope_onelevel_empty_db(self):
170 l = ldb.Ldb(self.url(), flags=self.flags())
171 self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
172 ldb.SCOPE_ONELEVEL)), 0)
174 def test_delete(self):
175 l = ldb.Ldb(self.url(), flags=self.flags())
176 self.assertRaises(ldb.LdbError, lambda: l.delete(ldb.Dn(l, "dc=foo2")))
178 def test_delete_w_unhandled_ctrl(self):
179 l = ldb.Ldb(self.url(), flags=self.flags())
181 m.dn = ldb.Dn(l, "dc=foo1")
183 m["objectUUID"] = b"0123456789abcdef"
185 self.assertRaises(ldb.LdbError, lambda: l.delete(m.dn, ["search_options:1:2"]))
188 def test_contains(self):
190 l = ldb.Ldb(name, flags=self.flags())
191 self.assertFalse(ldb.Dn(l, "dc=foo3") in l)
192 l = ldb.Ldb(name, flags=self.flags())
194 m.dn = ldb.Dn(l, "dc=foo3")
196 m["objectUUID"] = b"0123456789abcdef"
199 self.assertTrue(ldb.Dn(l, "dc=foo3") in l)
200 self.assertFalse(ldb.Dn(l, "dc=foo4") in l)
204 def test_get_config_basedn(self):
205 l = ldb.Ldb(self.url(), flags=self.flags())
206 self.assertEqual(None, l.get_config_basedn())
208 def test_get_root_basedn(self):
209 l = ldb.Ldb(self.url(), flags=self.flags())
210 self.assertEqual(None, l.get_root_basedn())
212 def test_get_schema_basedn(self):
213 l = ldb.Ldb(self.url(), flags=self.flags())
214 self.assertEqual(None, l.get_schema_basedn())
216 def test_get_default_basedn(self):
217 l = ldb.Ldb(self.url(), flags=self.flags())
218 self.assertEqual(None, l.get_default_basedn())
221 l = ldb.Ldb(self.url(), flags=self.flags())
223 m.dn = ldb.Dn(l, "dc=foo4")
225 m["objectUUID"] = b"0123456789abcdef"
226 self.assertEqual(len(l.search()), 0)
229 self.assertEqual(len(l.search()), 1)
231 l.delete(ldb.Dn(l, "dc=foo4"))
233 def test_search_iterator(self):
234 l = ldb.Ldb(self.url(), flags=self.flags())
235 s = l.search_iterator()
241 except RuntimeError as re:
246 except RuntimeError as re:
251 except RuntimeError as re:
254 s = l.search_iterator()
257 self.assertTrue(isinstance(me, ldb.Message))
260 self.assertEqual(len(r), 0)
261 self.assertEqual(count, 0)
264 m1.dn = ldb.Dn(l, "dc=foo4")
266 m1["objectUUID"] = b"0123456789abcdef"
269 s = l.search_iterator()
272 self.assertTrue(isinstance(me, ldb.Message))
276 self.assertEqual(len(r), 0)
277 self.assertEqual(len(msgs), 1)
278 self.assertEqual(msgs[0].dn, m1.dn)
281 m2.dn = ldb.Dn(l, "dc=foo5")
283 m2["objectUUID"] = b"0123456789abcdee"
286 s = l.search_iterator()
289 self.assertTrue(isinstance(me, ldb.Message))
293 self.assertEqual(len(r), 0)
294 self.assertEqual(len(msgs), 2)
295 if msgs[0].dn == m1.dn:
296 self.assertEqual(msgs[0].dn, m1.dn)
297 self.assertEqual(msgs[1].dn, m2.dn)
299 self.assertEqual(msgs[0].dn, m2.dn)
300 self.assertEqual(msgs[1].dn, m1.dn)
302 s = l.search_iterator()
305 self.assertTrue(isinstance(me, ldb.Message))
312 except RuntimeError as re:
315 self.assertTrue(isinstance(me, ldb.Message))
323 self.assertEqual(len(r), 0)
324 self.assertEqual(len(msgs), 2)
325 if msgs[0].dn == m1.dn:
326 self.assertEqual(msgs[0].dn, m1.dn)
327 self.assertEqual(msgs[1].dn, m2.dn)
329 self.assertEqual(msgs[0].dn, m2.dn)
330 self.assertEqual(msgs[1].dn, m1.dn)
332 l.delete(ldb.Dn(l, "dc=foo4"))
333 l.delete(ldb.Dn(l, "dc=foo5"))
335 def test_add_text(self):
336 l = ldb.Ldb(self.url(), flags=self.flags())
338 m.dn = ldb.Dn(l, "dc=foo4")
340 m["objectUUID"] = b"0123456789abcdef"
341 self.assertEqual(len(l.search()), 0)
344 self.assertEqual(len(l.search()), 1)
346 l.delete(ldb.Dn(l, "dc=foo4"))
348 def test_add_w_unhandled_ctrl(self):
349 l = ldb.Ldb(self.url(), flags=self.flags())
351 m.dn = ldb.Dn(l, "dc=foo4")
353 self.assertEqual(len(l.search()), 0)
354 self.assertRaises(ldb.LdbError, lambda: l.add(m, ["search_options:1:2"]))
356 def test_add_dict(self):
357 l = ldb.Ldb(self.url(), flags=self.flags())
358 m = {"dn": ldb.Dn(l, "dc=foo5"),
360 "objectUUID": b"0123456789abcdef"}
361 self.assertEqual(len(l.search()), 0)
364 self.assertEqual(len(l.search()), 1)
366 l.delete(ldb.Dn(l, "dc=foo5"))
368 def test_add_dict_text(self):
369 l = ldb.Ldb(self.url(), flags=self.flags())
370 m = {"dn": ldb.Dn(l, "dc=foo5"),
372 "objectUUID": b"0123456789abcdef"}
373 self.assertEqual(len(l.search()), 0)
376 self.assertEqual(len(l.search()), 1)
378 l.delete(ldb.Dn(l, "dc=foo5"))
380 def test_add_dict_string_dn(self):
381 l = ldb.Ldb(self.url(), flags=self.flags())
382 m = {"dn": "dc=foo6", "bla": b"bla",
383 "objectUUID": b"0123456789abcdef"}
384 self.assertEqual(len(l.search()), 0)
387 self.assertEqual(len(l.search()), 1)
389 l.delete(ldb.Dn(l, "dc=foo6"))
391 def test_add_dict_bytes_dn(self):
392 l = ldb.Ldb(self.url(), flags=self.flags())
393 m = {"dn": b"dc=foo6", "bla": b"bla",
394 "objectUUID": b"0123456789abcdef"}
395 self.assertEqual(len(l.search()), 0)
398 self.assertEqual(len(l.search()), 1)
400 l.delete(ldb.Dn(l, "dc=foo6"))
402 def test_rename(self):
403 l = ldb.Ldb(self.url(), flags=self.flags())
405 m.dn = ldb.Dn(l, "dc=foo7")
407 m["objectUUID"] = b"0123456789abcdef"
408 self.assertEqual(len(l.search()), 0)
411 l.rename(ldb.Dn(l, "dc=foo7"), ldb.Dn(l, "dc=bar"))
412 self.assertEqual(len(l.search()), 1)
414 l.delete(ldb.Dn(l, "dc=bar"))
416 def test_rename_string_dns(self):
417 l = ldb.Ldb(self.url(), flags=self.flags())
419 m.dn = ldb.Dn(l, "dc=foo8")
421 m["objectUUID"] = b"0123456789abcdef"
422 self.assertEqual(len(l.search()), 0)
424 self.assertEqual(len(l.search()), 1)
426 l.rename("dc=foo8", "dc=bar")
427 self.assertEqual(len(l.search()), 1)
429 l.delete(ldb.Dn(l, "dc=bar"))
431 def test_rename_bad_string_dns(self):
432 l = ldb.Ldb(self.url(), flags=self.flags())
434 m.dn = ldb.Dn(l, "dc=foo8")
436 m["objectUUID"] = b"0123456789abcdef"
437 self.assertEqual(len(l.search()), 0)
439 self.assertEqual(len(l.search()), 1)
440 self.assertRaises(ldb.LdbError,lambda: l.rename("dcXfoo8", "dc=bar"))
441 self.assertRaises(ldb.LdbError,lambda: l.rename("dc=foo8", "dcXbar"))
442 l.delete(ldb.Dn(l, "dc=foo8"))
444 def test_empty_dn(self):
445 l = ldb.Ldb(self.url(), flags=self.flags())
446 self.assertEqual(0, len(l.search()))
448 m.dn = ldb.Dn(l, "dc=empty")
449 m["objectUUID"] = b"0123456789abcdef"
452 self.assertEqual(1, len(rm))
453 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
457 self.assertEqual(1, len(rm))
458 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
460 rm = l.search(m.dn, attrs=["blah"])
461 self.assertEqual(1, len(rm))
462 self.assertEqual(0, len(rm[0]))
464 def test_modify_delete(self):
465 l = ldb.Ldb(self.url(), flags=self.flags())
467 m.dn = ldb.Dn(l, "dc=modifydelete")
469 m["objectUUID"] = b"0123456789abcdef"
471 rm = l.search(m.dn)[0]
472 self.assertEqual([b"1234"], list(rm["bla"]))
475 m.dn = ldb.Dn(l, "dc=modifydelete")
476 m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
477 self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
480 self.assertEqual(1, len(rm))
481 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
483 rm = l.search(m.dn, attrs=["bla"])
484 self.assertEqual(1, len(rm))
485 self.assertEqual(0, len(rm[0]))
487 l.delete(ldb.Dn(l, "dc=modifydelete"))
489 def test_modify_delete_text(self):
490 l = ldb.Ldb(self.url(), flags=self.flags())
492 m.dn = ldb.Dn(l, "dc=modifydelete")
493 m.text["bla"] = ["1234"]
494 m["objectUUID"] = b"0123456789abcdef"
496 rm = l.search(m.dn)[0]
497 self.assertEqual(["1234"], list(rm.text["bla"]))
500 m.dn = ldb.Dn(l, "dc=modifydelete")
501 m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
502 self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
505 self.assertEqual(1, len(rm))
506 self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
508 rm = l.search(m.dn, attrs=["bla"])
509 self.assertEqual(1, len(rm))
510 self.assertEqual(0, len(rm[0]))
512 l.delete(ldb.Dn(l, "dc=modifydelete"))
514 def test_modify_add(self):
515 l = ldb.Ldb(self.url(), flags=self.flags())
517 m.dn = ldb.Dn(l, "dc=add")
519 m["objectUUID"] = b"0123456789abcdef"
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())
527 rm = l.search(m.dn)[0]
528 self.assertEqual(3, len(rm))
529 self.assertEqual([b"1234", b"456"], list(rm["bla"]))
531 l.delete(ldb.Dn(l, "dc=add"))
533 def test_modify_add_text(self):
534 l = ldb.Ldb(self.url(), flags=self.flags())
536 m.dn = ldb.Dn(l, "dc=add")
537 m.text["bla"] = ["1234"]
538 m["objectUUID"] = b"0123456789abcdef"
542 m.dn = ldb.Dn(l, "dc=add")
543 m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
544 self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
546 rm = l.search(m.dn)[0]
547 self.assertEqual(3, len(rm))
548 self.assertEqual(["1234", "456"], list(rm.text["bla"]))
550 l.delete(ldb.Dn(l, "dc=add"))
552 def test_modify_replace(self):
553 l = ldb.Ldb(self.url(), flags=self.flags())
555 m.dn = ldb.Dn(l, "dc=modify2")
556 m["bla"] = [b"1234", b"456"]
557 m["objectUUID"] = b"0123456789abcdef"
561 m.dn = ldb.Dn(l, "dc=modify2")
562 m["bla"] = ldb.MessageElement([b"789"], ldb.FLAG_MOD_REPLACE, "bla")
563 self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
565 rm = l.search(m.dn)[0]
566 self.assertEqual(3, len(rm))
567 self.assertEqual([b"789"], list(rm["bla"]))
568 rm = l.search(m.dn, attrs=["bla"])[0]
569 self.assertEqual(1, len(rm))
571 l.delete(ldb.Dn(l, "dc=modify2"))
573 def test_modify_replace_text(self):
574 l = ldb.Ldb(self.url(), flags=self.flags())
576 m.dn = ldb.Dn(l, "dc=modify2")
577 m.text["bla"] = ["1234", "456"]
578 m["objectUUID"] = b"0123456789abcdef"
582 m.dn = ldb.Dn(l, "dc=modify2")
583 m["bla"] = ldb.MessageElement(["789"], ldb.FLAG_MOD_REPLACE, "bla")
584 self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
586 rm = l.search(m.dn)[0]
587 self.assertEqual(3, len(rm))
588 self.assertEqual(["789"], list(rm.text["bla"]))
589 rm = l.search(m.dn, attrs=["bla"])[0]
590 self.assertEqual(1, len(rm))
592 l.delete(ldb.Dn(l, "dc=modify2"))
594 def test_modify_flags_change(self):
595 l = ldb.Ldb(self.url(), flags=self.flags())
597 m.dn = ldb.Dn(l, "dc=add")
599 m["objectUUID"] = b"0123456789abcdef"
603 m.dn = ldb.Dn(l, "dc=add")
604 m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
605 self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
607 rm = l.search(m.dn)[0]
608 self.assertEqual(3, len(rm))
609 self.assertEqual([b"1234", b"456"], list(rm["bla"]))
611 # Now create another modify, but switch the flags before we do it
612 m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
613 m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
615 rm = l.search(m.dn, attrs=["bla"])[0]
616 self.assertEqual(1, len(rm))
617 self.assertEqual([b"1234"], list(rm["bla"]))
619 l.delete(ldb.Dn(l, "dc=add"))
621 def test_modify_flags_change_text(self):
622 l = ldb.Ldb(self.url(), flags=self.flags())
624 m.dn = ldb.Dn(l, "dc=add")
625 m.text["bla"] = ["1234"]
626 m["objectUUID"] = b"0123456789abcdef"
630 m.dn = ldb.Dn(l, "dc=add")
631 m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
632 self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
634 rm = l.search(m.dn)[0]
635 self.assertEqual(3, len(rm))
636 self.assertEqual(["1234", "456"], list(rm.text["bla"]))
638 # Now create another modify, but switch the flags before we do it
639 m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
640 m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
642 rm = l.search(m.dn, attrs=["bla"])[0]
643 self.assertEqual(1, len(rm))
644 self.assertEqual(["1234"], list(rm.text["bla"]))
646 l.delete(ldb.Dn(l, "dc=add"))
648 def test_transaction_commit(self):
649 l = ldb.Ldb(self.url(), flags=self.flags())
650 l.transaction_start()
651 m = ldb.Message(ldb.Dn(l, "dc=foo9"))
653 m["objectUUID"] = b"0123456789abcdef"
655 l.transaction_commit()
658 def test_transaction_cancel(self):
659 l = ldb.Ldb(self.url(), flags=self.flags())
660 l.transaction_start()
661 m = ldb.Message(ldb.Dn(l, "dc=foo10"))
663 m["objectUUID"] = b"0123456789abcdee"
665 l.transaction_cancel()
666 self.assertEqual(0, len(l.search(ldb.Dn(l, "dc=foo10"))))
668 def test_set_debug(self):
669 def my_report_fn(level, text):
671 l = ldb.Ldb(self.url(), flags=self.flags())
672 l.set_debug(my_report_fn)
674 def test_zero_byte_string(self):
675 """Testing we do not get trapped in the \0 byte in a property string."""
676 l = ldb.Ldb(self.url(), flags=self.flags())
679 "objectclass": b"user",
680 "cN": b"LDAPtestUSER",
681 "givenname": b"ldap",
682 "displayname": b"foo\0bar",
683 "objectUUID": b"0123456789abcdef"
685 res = l.search(expression="(dn=dc=somedn)")
686 self.assertEqual(b"foo\0bar", res[0]["displayname"][0])
688 def test_no_crash_broken_expr(self):
689 l = ldb.Ldb(self.url(), flags=self.flags())
690 self.assertRaises(ldb.LdbError, lambda: l.search("", ldb.SCOPE_SUBTREE, "&(dc=*)(dn=*)", ["dc"]))
692 # Run the SimpleLdb tests against an lmdb backend
695 class SimpleLdbLmdb(SimpleLdb):
698 self.prefix = MDB_PREFIX
699 self.index = MDB_INDEX_OBJ
700 super(SimpleLdbLmdb, self).setUp()
703 super(SimpleLdbLmdb, self).tearDown()
706 class SearchTests(LdbBaseTest):
708 shutil.rmtree(self.testdir)
709 super(SearchTests, self).tearDown()
711 # Ensure the LDB is closed now, so we close the FD
715 super(SearchTests, self).setUp()
716 self.testdir = tempdir()
717 self.filename = os.path.join(self.testdir, "search_test.ldb")
718 options = ["modules:rdn_name"]
719 if hasattr(self, 'IDXCHECK'):
720 options.append("disable_full_db_scan_for_self_test:1")
721 self.l = ldb.Ldb(self.url(),
725 self.l.add(self.index)
726 except AttributeError:
729 self.l.add({"dn": "@ATTRIBUTES",
730 "DC": "CASE_INSENSITIVE"})
732 # Note that we can't use the name objectGUID here, as we
733 # want to stay clear of the objectGUID handler in LDB and
734 # instead use just the 16 bytes raw, which we just keep
735 # to printable chars here for ease of handling.
737 self.l.add({"dn": "DC=SAMBA,DC=ORG",
738 "name": b"samba.org",
739 "objectUUID": b"0123456789abcdef"})
740 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
743 "objectUUID": b"0123456789abcde1"})
744 self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG",
747 "objectUUID": b"0123456789abcde2"})
748 self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG",
751 "objectUUID": b"0123456789abcde3"})
752 self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG",
755 "objectUUID": b"0123456789abcde4"})
756 self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG",
759 "objectUUID": b"0123456789abcde5"})
760 self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG",
763 "objectUUID": b"0123456789abcde6"})
764 self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG",
767 "objectUUID": b"0123456789abcde7"})
768 self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG",
771 "objectUUID": b"0123456789abcde8"})
772 self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG",
775 "objectUUID": b"0123456789abcde9"})
776 self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG",
779 "objectUUID": b"0123456789abcde0"})
780 self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG",
783 "objectUUID": b"0123456789abcdea"})
784 self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG",
787 "objectUUID": b"0123456789abcdeb"})
788 self.l.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
791 "objectUUID": b"0123456789abcdec"})
792 self.l.add({"dn": "OU=OU12,DC=SAMBA,DC=ORG",
795 "objectUUID": b"0123456789abcded"})
796 self.l.add({"dn": "OU=OU13,DC=SAMBA,DC=ORG",
799 "objectUUID": b"0123456789abcdee"})
800 self.l.add({"dn": "OU=OU14,DC=SAMBA,DC=ORG",
803 "objectUUID": b"0123456789abcd01"})
804 self.l.add({"dn": "OU=OU15,DC=SAMBA,DC=ORG",
807 "objectUUID": b"0123456789abcd02"})
808 self.l.add({"dn": "OU=OU16,DC=SAMBA,DC=ORG",
811 "objectUUID": b"0123456789abcd03"})
812 self.l.add({"dn": "OU=OU17,DC=SAMBA,DC=ORG",
815 "objectUUID": b"0123456789abcd04"})
816 self.l.add({"dn": "OU=OU18,DC=SAMBA,DC=ORG",
819 "objectUUID": b"0123456789abcd05"})
820 self.l.add({"dn": "OU=OU19,DC=SAMBA,DC=ORG",
823 "objectUUID": b"0123456789abcd06"})
824 self.l.add({"dn": "OU=OU20,DC=SAMBA,DC=ORG",
827 "objectUUID": b"0123456789abcd07"})
828 self.l.add({"dn": "OU=OU21,DC=SAMBA,DC=ORG",
831 "objectUUID": b"0123456789abcd08"})
832 self.l.add({"dn": "OU=OU22,DC=SAMBA,DC=ORG",
835 "objectUUID": b"0123456789abcd09"})
838 """Testing a search"""
840 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
841 scope=ldb.SCOPE_BASE)
842 self.assertEqual(len(res11), 1)
844 def test_base_lower(self):
845 """Testing a search"""
847 res11 = self.l.search(base="OU=OU11,DC=samba,DC=org",
848 scope=ldb.SCOPE_BASE)
849 self.assertEqual(len(res11), 1)
851 def test_base_or(self):
852 """Testing a search"""
854 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
855 scope=ldb.SCOPE_BASE,
856 expression="(|(ou=ou11)(ou=ou12))")
857 self.assertEqual(len(res11), 1)
859 def test_base_or2(self):
860 """Testing a search"""
862 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
863 scope=ldb.SCOPE_BASE,
864 expression="(|(x=y)(y=b))")
865 self.assertEqual(len(res11), 1)
867 def test_base_and(self):
868 """Testing a search"""
870 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
871 scope=ldb.SCOPE_BASE,
872 expression="(&(ou=ou11)(ou=ou12))")
873 self.assertEqual(len(res11), 0)
875 def test_base_and2(self):
876 """Testing a search"""
878 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
879 scope=ldb.SCOPE_BASE,
880 expression="(&(x=y)(y=a))")
881 self.assertEqual(len(res11), 1)
883 def test_base_false(self):
884 """Testing a search"""
886 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
887 scope=ldb.SCOPE_BASE,
888 expression="(|(ou=ou13)(ou=ou12))")
889 self.assertEqual(len(res11), 0)
891 def test_check_base_false(self):
892 """Testing a search"""
893 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
894 scope=ldb.SCOPE_BASE,
895 expression="(|(ou=ou13)(ou=ou12))")
896 self.assertEqual(len(res11), 0)
898 def test_check_base_error(self):
899 """Testing a search"""
900 checkbaseonsearch = {"dn": "@OPTIONS",
901 "checkBaseOnSearch": b"TRUE"}
903 self.l.add(checkbaseonsearch)
904 except ldb.LdbError as err:
906 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
907 m = ldb.Message.from_dict(self.l,
912 res11 = self.l.search(base="OU=OU11x,DC=SAMBA,DC=ORG",
913 scope=ldb.SCOPE_BASE,
914 expression="(|(ou=ou13)(ou=ou12))")
915 self.fail("Should have failed on missing base")
916 except ldb.LdbError as err:
918 self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
920 def test_subtree_and(self):
921 """Testing a search"""
923 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
924 scope=ldb.SCOPE_SUBTREE,
925 expression="(&(ou=ou11)(ou=ou12))")
926 self.assertEqual(len(res11), 0)
928 def test_subtree_and2(self):
929 """Testing a search"""
931 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
932 scope=ldb.SCOPE_SUBTREE,
933 expression="(&(x=y)(|(y=b)(y=c)))")
934 self.assertEqual(len(res11), 1)
936 def test_subtree_and2_lower(self):
937 """Testing a search"""
939 res11 = self.l.search(base="DC=samba,DC=org",
940 scope=ldb.SCOPE_SUBTREE,
941 expression="(&(x=y)(|(y=b)(y=c)))")
942 self.assertEqual(len(res11), 1)
944 def test_subtree_or(self):
945 """Testing a search"""
947 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
948 scope=ldb.SCOPE_SUBTREE,
949 expression="(|(ou=ou11)(ou=ou12))")
950 self.assertEqual(len(res11), 2)
952 def test_subtree_or2(self):
953 """Testing a search"""
955 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
956 scope=ldb.SCOPE_SUBTREE,
957 expression="(|(x=y)(y=b))")
958 self.assertEqual(len(res11), 20)
960 def test_subtree_or3(self):
961 """Testing a search"""
963 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
964 scope=ldb.SCOPE_SUBTREE,
965 expression="(|(x=y)(y=b)(y=c))")
966 self.assertEqual(len(res11), 22)
968 def test_one_and(self):
969 """Testing a search"""
971 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
972 scope=ldb.SCOPE_ONELEVEL,
973 expression="(&(ou=ou11)(ou=ou12))")
974 self.assertEqual(len(res11), 0)
976 def test_one_and2(self):
977 """Testing a search"""
979 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
980 scope=ldb.SCOPE_ONELEVEL,
981 expression="(&(x=y)(y=b))")
982 self.assertEqual(len(res11), 1)
984 def test_one_or(self):
985 """Testing a search"""
987 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
988 scope=ldb.SCOPE_ONELEVEL,
989 expression="(|(ou=ou11)(ou=ou12))")
990 self.assertEqual(len(res11), 2)
992 def test_one_or2(self):
993 """Testing a search"""
995 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
996 scope=ldb.SCOPE_ONELEVEL,
997 expression="(|(x=y)(y=b))")
998 self.assertEqual(len(res11), 20)
1000 def test_one_or2_lower(self):
1001 """Testing a search"""
1003 res11 = self.l.search(base="DC=samba,DC=org",
1004 scope=ldb.SCOPE_ONELEVEL,
1005 expression="(|(x=y)(y=b))")
1006 self.assertEqual(len(res11), 20)
1008 def test_one_unindexable(self):
1009 """Testing a search"""
1012 res11 = self.l.search(base="DC=samba,DC=org",
1013 scope=ldb.SCOPE_ONELEVEL,
1014 expression="(y=b*)")
1015 if hasattr(self, 'IDX') and \
1016 not hasattr(self, 'IDXONE') and \
1017 hasattr(self, 'IDXCHECK'):
1018 self.fail("Should have failed as un-indexed search")
1020 self.assertEqual(len(res11), 9)
1022 except ldb.LdbError as err:
1025 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1026 self.assertIn(estr, "ldb FULL SEARCH disabled")
1028 def test_one_unindexable_presence(self):
1029 """Testing a search"""
1032 res11 = self.l.search(base="DC=samba,DC=org",
1033 scope=ldb.SCOPE_ONELEVEL,
1035 if hasattr(self, 'IDX') and \
1036 not hasattr(self, 'IDXONE') and \
1037 hasattr(self, 'IDXCHECK'):
1038 self.fail("Should have failed as un-indexed search")
1040 self.assertEqual(len(res11), 24)
1042 except ldb.LdbError as err:
1045 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1046 self.assertIn(estr, "ldb FULL SEARCH disabled")
1048 def test_subtree_and_or(self):
1049 """Testing a search"""
1051 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1052 scope=ldb.SCOPE_SUBTREE,
1053 expression="(&(|(x=z)(y=b))(x=x)(y=c))")
1054 self.assertEqual(len(res11), 0)
1056 def test_subtree_and_or2(self):
1057 """Testing a search"""
1059 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1060 scope=ldb.SCOPE_SUBTREE,
1061 expression="(&(x=x)(y=c)(|(x=z)(y=b)))")
1062 self.assertEqual(len(res11), 0)
1064 def test_subtree_and_or3(self):
1065 """Testing a search"""
1067 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1068 scope=ldb.SCOPE_SUBTREE,
1069 expression="(&(|(ou=ou11)(ou=ou10))(|(x=y)(y=b)(y=c)))")
1070 self.assertEqual(len(res11), 2)
1072 def test_subtree_and_or4(self):
1073 """Testing a search"""
1075 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1076 scope=ldb.SCOPE_SUBTREE,
1077 expression="(&(|(x=y)(y=b)(y=c))(|(ou=ou11)(ou=ou10)))")
1078 self.assertEqual(len(res11), 2)
1080 def test_subtree_and_or5(self):
1081 """Testing a search"""
1083 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1084 scope=ldb.SCOPE_SUBTREE,
1085 expression="(&(|(x=y)(y=b)(y=c))(ou=ou11))")
1086 self.assertEqual(len(res11), 1)
1088 def test_subtree_or_and(self):
1089 """Testing a search"""
1091 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1092 scope=ldb.SCOPE_SUBTREE,
1093 expression="(|(x=x)(y=c)(&(x=z)(y=b)))")
1094 self.assertEqual(len(res11), 10)
1096 def test_subtree_large_and_unique(self):
1097 """Testing a search"""
1099 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1100 scope=ldb.SCOPE_SUBTREE,
1101 expression="(&(ou=ou10)(y=a))")
1102 self.assertEqual(len(res11), 1)
1104 def test_subtree_and_none(self):
1105 """Testing a search"""
1107 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1108 scope=ldb.SCOPE_SUBTREE,
1109 expression="(&(ou=ouX)(y=a))")
1110 self.assertEqual(len(res11), 0)
1112 def test_subtree_and_idx_record(self):
1113 """Testing a search against the index record"""
1115 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1116 scope=ldb.SCOPE_SUBTREE,
1117 expression="(@IDXDN=DC=SAMBA,DC=ORG)")
1118 self.assertEqual(len(res11), 0)
1120 def test_subtree_and_idxone_record(self):
1121 """Testing a search against the index record"""
1123 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1124 scope=ldb.SCOPE_SUBTREE,
1125 expression="(@IDXONE=DC=SAMBA,DC=ORG)")
1126 self.assertEqual(len(res11), 0)
1128 def test_subtree_unindexable(self):
1129 """Testing a search"""
1132 res11 = self.l.search(base="DC=samba,DC=org",
1133 scope=ldb.SCOPE_SUBTREE,
1134 expression="(y=b*)")
1135 if hasattr(self, 'IDX') and \
1136 hasattr(self, 'IDXCHECK'):
1137 self.fail("Should have failed as un-indexed search")
1139 self.assertEqual(len(res11), 9)
1141 except ldb.LdbError as err:
1144 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1145 self.assertIn(estr, "ldb FULL SEARCH disabled")
1147 def test_subtree_unindexable_presence(self):
1148 """Testing a search"""
1151 res11 = self.l.search(base="DC=samba,DC=org",
1152 scope=ldb.SCOPE_SUBTREE,
1154 if hasattr(self, 'IDX') and \
1155 hasattr(self, 'IDXCHECK'):
1156 self.fail("Should have failed as un-indexed search")
1158 self.assertEqual(len(res11), 24)
1160 except ldb.LdbError as err:
1163 self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1164 self.assertIn(estr, "ldb FULL SEARCH disabled")
1166 def test_dn_filter_one(self):
1167 """Testing that a dn= filter succeeds
1168 (or fails with disallowDNFilter
1169 set and IDXGUID or (IDX and not IDXONE) mode)
1170 when the scope is SCOPE_ONELEVEL.
1172 This should be made more consistent, but for now lock in
1177 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1178 scope=ldb.SCOPE_ONELEVEL,
1179 expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1180 if hasattr(self, 'disallowDNFilter') and \
1181 hasattr(self, 'IDX') and \
1182 (hasattr(self, 'IDXGUID') or
1183 ((hasattr(self, 'IDXONE') == False and hasattr(self, 'IDX')))):
1184 self.assertEqual(len(res11), 0)
1186 self.assertEqual(len(res11), 1)
1188 def test_dn_filter_subtree(self):
1189 """Testing that a dn= filter succeeds
1190 (or fails with disallowDNFilter set)
1191 when the scope is SCOPE_SUBTREE"""
1193 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1194 scope=ldb.SCOPE_SUBTREE,
1195 expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1196 if hasattr(self, 'disallowDNFilter') \
1197 and hasattr(self, 'IDX'):
1198 self.assertEqual(len(res11), 0)
1200 self.assertEqual(len(res11), 1)
1202 def test_dn_filter_base(self):
1203 """Testing that (incorrectly) a dn= filter works
1204 when the scope is SCOPE_BASE"""
1206 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1207 scope=ldb.SCOPE_BASE,
1208 expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1210 # At some point we should fix this, but it isn't trivial
1211 self.assertEqual(len(res11), 1)
1213 def test_distinguishedName_filter_one(self):
1214 """Testing that a distinguishedName= filter succeeds
1215 when the scope is SCOPE_ONELEVEL.
1217 This should be made more consistent, but for now lock in
1222 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1223 scope=ldb.SCOPE_ONELEVEL,
1224 expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1225 self.assertEqual(len(res11), 1)
1227 def test_distinguishedName_filter_subtree(self):
1228 """Testing that a distinguishedName= filter succeeds
1229 when the scope is SCOPE_SUBTREE"""
1231 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1232 scope=ldb.SCOPE_SUBTREE,
1233 expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1234 self.assertEqual(len(res11), 1)
1236 def test_distinguishedName_filter_base(self):
1237 """Testing that (incorrectly) a distinguishedName= filter works
1238 when the scope is SCOPE_BASE"""
1240 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1241 scope=ldb.SCOPE_BASE,
1242 expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1244 # At some point we should fix this, but it isn't trivial
1245 self.assertEqual(len(res11), 1)
1247 def test_bad_dn_filter_base(self):
1248 """Testing that a dn= filter on an invalid DN works
1249 when the scope is SCOPE_BASE but
1250 returns zero results"""
1252 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1253 scope=ldb.SCOPE_BASE,
1254 expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1256 # At some point we should fix this, but it isn't trivial
1257 self.assertEqual(len(res11), 0)
1260 def test_bad_dn_filter_one(self):
1261 """Testing that a dn= filter succeeds but returns zero
1262 results when the DN is not valid on a SCOPE_ONELEVEL search
1266 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1267 scope=ldb.SCOPE_ONELEVEL,
1268 expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1269 self.assertEqual(len(res11), 0)
1271 def test_bad_dn_filter_subtree(self):
1272 """Testing that a dn= filter succeeds but returns zero
1273 results when the DN is not valid on a SCOPE_SUBTREE search
1277 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1278 scope=ldb.SCOPE_SUBTREE,
1279 expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1280 self.assertEqual(len(res11), 0)
1282 def test_bad_distinguishedName_filter_base(self):
1283 """Testing that a distinguishedName= filter on an invalid DN works
1284 when the scope is SCOPE_BASE but
1285 returns zero results"""
1287 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1288 scope=ldb.SCOPE_BASE,
1289 expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1291 # At some point we should fix this, but it isn't trivial
1292 self.assertEqual(len(res11), 0)
1295 def test_bad_distinguishedName_filter_one(self):
1296 """Testing that a distinguishedName= filter succeeds but returns zero
1297 results when the DN is not valid on a SCOPE_ONELEVEL search
1301 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1302 scope=ldb.SCOPE_ONELEVEL,
1303 expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1304 self.assertEqual(len(res11), 0)
1306 def test_bad_distinguishedName_filter_subtree(self):
1307 """Testing that a distinguishedName= filter succeeds but returns zero
1308 results when the DN is not valid on a SCOPE_SUBTREE search
1312 res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1313 scope=ldb.SCOPE_SUBTREE,
1314 expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1315 self.assertEqual(len(res11), 0)
1317 def test_bad_dn_search_base(self):
1318 """Testing with a bad base DN (SCOPE_BASE)"""
1321 res11 = self.l.search(base="OU=OU1,DC=SAMBA,DCXXX",
1322 scope=ldb.SCOPE_BASE)
1323 self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1324 except ldb.LdbError as err:
1326 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1329 def test_bad_dn_search_one(self):
1330 """Testing with a bad base DN (SCOPE_ONELEVEL)"""
1333 res11 = self.l.search(base="DC=SAMBA,DCXXXX",
1334 scope=ldb.SCOPE_ONELEVEL)
1335 self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1336 except ldb.LdbError as err:
1338 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1340 def test_bad_dn_search_subtree(self):
1341 """Testing with a bad base DN (SCOPE_SUBTREE)"""
1344 res11 = self.l.search(base="DC=SAMBA,DCXXXX",
1345 scope=ldb.SCOPE_SUBTREE)
1346 self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1347 except ldb.LdbError as err:
1349 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1353 # Run the search tests against an lmdb backend
1354 class SearchTestsLmdb(SearchTests):
1357 self.prefix = MDB_PREFIX
1358 self.index = MDB_INDEX_OBJ
1359 super(SearchTestsLmdb, self).setUp()
1362 super(SearchTestsLmdb, self).tearDown()
1365 class IndexedSearchTests(SearchTests):
1366 """Test searches using the index, to ensure the index doesn't
1370 super(IndexedSearchTests, self).setUp()
1371 self.l.add({"dn": "@INDEXLIST",
1372 "@IDXATTR": [b"x", b"y", b"ou"]})
1376 class IndexedCheckSearchTests(IndexedSearchTests):
1377 """Test searches using the index, to ensure the index doesn't
1378 break things (full scan disabled)"""
1381 self.IDXCHECK = True
1382 super(IndexedCheckSearchTests, self).setUp()
1385 class IndexedSearchDnFilterTests(SearchTests):
1386 """Test searches using the index, to ensure the index doesn't
1390 super(IndexedSearchDnFilterTests, self).setUp()
1391 self.l.add({"dn": "@OPTIONS",
1392 "disallowDNFilter": "TRUE"})
1393 self.disallowDNFilter = True
1395 self.l.add({"dn": "@INDEXLIST",
1396 "@IDXATTR": [b"x", b"y", b"ou"]})
1400 class IndexedAndOneLevelSearchTests(SearchTests):
1401 """Test searches using the index including @IDXONE, to ensure
1402 the index doesn't break things"""
1405 super(IndexedAndOneLevelSearchTests, self).setUp()
1406 self.l.add({"dn": "@INDEXLIST",
1407 "@IDXATTR": [b"x", b"y", b"ou"],
1413 class IndexedCheckedAndOneLevelSearchTests(IndexedAndOneLevelSearchTests):
1414 """Test searches using the index including @IDXONE, to ensure
1415 the index doesn't break things (full scan disabled)"""
1418 self.IDXCHECK = True
1419 super(IndexedCheckedAndOneLevelSearchTests, self).setUp()
1422 class IndexedAndOneLevelDNFilterSearchTests(SearchTests):
1423 """Test searches using the index including @IDXONE, to ensure
1424 the index doesn't break things"""
1427 super(IndexedAndOneLevelDNFilterSearchTests, self).setUp()
1428 self.l.add({"dn": "@OPTIONS",
1429 "disallowDNFilter": "TRUE",
1430 "checkBaseOnSearch": "TRUE"})
1431 self.disallowDNFilter = True
1432 self.checkBaseOnSearch = True
1434 self.l.add({"dn": "@INDEXLIST",
1435 "@IDXATTR": [b"x", b"y", b"ou"],
1441 class GUIDIndexedSearchTests(SearchTests):
1442 """Test searches using the index, to ensure the index doesn't
1446 self.index = {"dn": "@INDEXLIST",
1447 "@IDXATTR": [b"x", b"y", b"ou"],
1448 "@IDXGUID": [b"objectUUID"],
1449 "@IDX_DN_GUID": [b"GUID"]}
1450 super(GUIDIndexedSearchTests, self).setUp()
1456 class GUIDIndexedDNFilterSearchTests(SearchTests):
1457 """Test searches using the index, to ensure the index doesn't
1461 self.index = {"dn": "@INDEXLIST",
1462 "@IDXATTR": [b"x", b"y", b"ou"],
1463 "@IDXGUID": [b"objectUUID"],
1464 "@IDX_DN_GUID": [b"GUID"]}
1465 super(GUIDIndexedDNFilterSearchTests, self).setUp()
1466 self.l.add({"dn": "@OPTIONS",
1467 "disallowDNFilter": "TRUE",
1468 "checkBaseOnSearch": "TRUE"})
1469 self.disallowDNFilter = True
1470 self.checkBaseOnSearch = True
1475 class GUIDAndOneLevelIndexedSearchTests(SearchTests):
1476 """Test searches using the index including @IDXONE, to ensure
1477 the index doesn't break things"""
1480 self.index = {"dn": "@INDEXLIST",
1481 "@IDXATTR": [b"x", b"y", b"ou"],
1482 "@IDXGUID": [b"objectUUID"],
1483 "@IDX_DN_GUID": [b"GUID"]}
1484 super(GUIDAndOneLevelIndexedSearchTests, self).setUp()
1485 self.l.add({"dn": "@OPTIONS",
1486 "disallowDNFilter": "TRUE",
1487 "checkBaseOnSearch": "TRUE"})
1488 self.disallowDNFilter = True
1489 self.checkBaseOnSearch = True
1495 class GUIDIndexedSearchTestsLmdb(GUIDIndexedSearchTests):
1498 self.prefix = MDB_PREFIX
1499 super(GUIDIndexedSearchTestsLmdb, self).setUp()
1502 super(GUIDIndexedSearchTestsLmdb, self).tearDown()
1505 class GUIDIndexedDNFilterSearchTestsLmdb(GUIDIndexedDNFilterSearchTests):
1508 self.prefix = MDB_PREFIX
1509 super(GUIDIndexedDNFilterSearchTestsLmdb, self).setUp()
1512 super(GUIDIndexedDNFilterSearchTestsLmdb, self).tearDown()
1515 class GUIDAndOneLevelIndexedSearchTestsLmdb(GUIDAndOneLevelIndexedSearchTests):
1518 self.prefix = MDB_PREFIX
1519 super(GUIDAndOneLevelIndexedSearchTestsLmdb, self).setUp()
1522 super(GUIDAndOneLevelIndexedSearchTestsLmdb, self).tearDown()
1525 class AddModifyTests(LdbBaseTest):
1527 shutil.rmtree(self.testdir)
1528 super(AddModifyTests, self).tearDown()
1530 # Ensure the LDB is closed now, so we close the FD
1534 super(AddModifyTests, self).setUp()
1535 self.testdir = tempdir()
1536 self.filename = os.path.join(self.testdir, "add_test.ldb")
1537 self.l = ldb.Ldb(self.url(),
1539 options=["modules:rdn_name"])
1541 self.l.add(self.index)
1542 except AttributeError:
1545 self.l.add({"dn": "DC=SAMBA,DC=ORG",
1546 "name": b"samba.org",
1547 "objectUUID": b"0123456789abcdef"})
1548 self.l.add({"dn": "@ATTRIBUTES",
1549 "objectUUID": "UNIQUE_INDEX"})
1551 def test_add_dup(self):
1552 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1555 "objectUUID": b"0123456789abcde1"})
1557 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1560 "objectUUID": b"0123456789abcde2"})
1561 self.fail("Should have failed adding dupliate entry")
1562 except ldb.LdbError as err:
1564 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1566 def test_add_bad(self):
1568 self.l.add({"dn": "BAD,DC=SAMBA,DC=ORG",
1571 "objectUUID": b"0123456789abcde1"})
1572 self.fail("Should have failed adding entry with invalid DN")
1573 except ldb.LdbError as err:
1575 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1577 def test_add_del_add(self):
1578 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1581 "objectUUID": b"0123456789abcde1"})
1582 self.l.delete("OU=DUP,DC=SAMBA,DC=ORG")
1583 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1586 "objectUUID": b"0123456789abcde2"})
1588 def test_add_move_add(self):
1589 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1592 "objectUUID": b"0123456789abcde1"})
1593 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1594 "OU=DUP2,DC=SAMBA,DC=ORG")
1595 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1598 "objectUUID": b"0123456789abcde2"})
1600 def test_add_move_fail_move_move(self):
1601 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1604 "objectUUID": b"0123456789abcde1"})
1605 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1608 "objectUUID": b"0123456789abcde2"})
1610 res2 = self.l.search(base="DC=SAMBA,DC=ORG",
1611 scope=ldb.SCOPE_SUBTREE,
1612 expression="(objectUUID=0123456789abcde1)")
1613 self.assertEqual(len(res2), 1)
1614 self.assertEqual(str(res2[0].dn), "OU=DUP,DC=SAMBA,DC=ORG")
1616 res3 = self.l.search(base="DC=SAMBA,DC=ORG",
1617 scope=ldb.SCOPE_SUBTREE,
1618 expression="(objectUUID=0123456789abcde2)")
1619 self.assertEqual(len(res3), 1)
1620 self.assertEqual(str(res3[0].dn), "OU=DUP2,DC=SAMBA,DC=ORG")
1623 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1624 "OU=DUP2,DC=SAMBA,DC=ORG")
1625 self.fail("Should have failed on duplicate DN")
1626 except ldb.LdbError as err:
1628 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1630 self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1631 "OU=DUP3,DC=SAMBA,DC=ORG")
1633 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1634 "OU=DUP2,DC=SAMBA,DC=ORG")
1636 res2 = self.l.search(base="DC=SAMBA,DC=ORG",
1637 scope=ldb.SCOPE_SUBTREE,
1638 expression="(objectUUID=0123456789abcde1)")
1639 self.assertEqual(len(res2), 1)
1640 self.assertEqual(str(res2[0].dn), "OU=DUP2,DC=SAMBA,DC=ORG")
1642 res3 = self.l.search(base="DC=SAMBA,DC=ORG",
1643 scope=ldb.SCOPE_SUBTREE,
1644 expression="(objectUUID=0123456789abcde2)")
1645 self.assertEqual(len(res3), 1)
1646 self.assertEqual(str(res3[0].dn), "OU=DUP3,DC=SAMBA,DC=ORG")
1648 def test_move_missing(self):
1650 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1651 "OU=DUP2,DC=SAMBA,DC=ORG")
1652 self.fail("Should have failed on missing")
1653 except ldb.LdbError as err:
1655 self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1657 def test_move_missing2(self):
1658 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1661 "objectUUID": b"0123456789abcde2"})
1664 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1665 "OU=DUP2,DC=SAMBA,DC=ORG")
1666 self.fail("Should have failed on missing")
1667 except ldb.LdbError as err:
1669 self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1671 def test_move_bad(self):
1672 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1675 "objectUUID": b"0123456789abcde2"})
1678 self.l.rename("OUXDUP,DC=SAMBA,DC=ORG",
1679 "OU=DUP2,DC=SAMBA,DC=ORG")
1680 self.fail("Should have failed on invalid DN")
1681 except ldb.LdbError as err:
1683 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1685 def test_move_bad2(self):
1686 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1689 "objectUUID": b"0123456789abcde2"})
1692 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1693 "OUXDUP2,DC=SAMBA,DC=ORG")
1694 self.fail("Should have failed on missing")
1695 except ldb.LdbError as err:
1697 self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1699 def test_move_fail_move_add(self):
1700 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1703 "objectUUID": b"0123456789abcde1"})
1704 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1707 "objectUUID": b"0123456789abcde2"})
1709 self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1710 "OU=DUP2,DC=SAMBA,DC=ORG")
1711 self.fail("Should have failed on duplicate DN")
1712 except ldb.LdbError as err:
1714 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1716 self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1717 "OU=DUP3,DC=SAMBA,DC=ORG")
1719 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1722 "objectUUID": b"0123456789abcde3"})
1725 class AddModifyTestsLmdb(AddModifyTests):
1728 self.prefix = MDB_PREFIX
1729 self.index = MDB_INDEX_OBJ
1730 super(AddModifyTestsLmdb, self).setUp()
1733 super(AddModifyTestsLmdb, self).tearDown()
1736 class IndexedAddModifyTests(AddModifyTests):
1737 """Test searches using the index, to ensure the index doesn't
1741 if not hasattr(self, 'index'):
1742 self.index = {"dn": "@INDEXLIST",
1743 "@IDXATTR": [b"x", b"y", b"ou", b"objectUUID"],
1745 super(IndexedAddModifyTests, self).setUp()
1747 def test_duplicate_GUID(self):
1749 self.l.add({"dn": "OU=DUPGUID,DC=SAMBA,DC=ORG",
1752 "objectUUID": b"0123456789abcdef"})
1753 self.fail("Should have failed adding dupliate GUID")
1754 except ldb.LdbError as err:
1756 self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1758 def test_duplicate_name_dup_GUID(self):
1759 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1762 "objectUUID": b"a123456789abcdef"})
1764 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1767 "objectUUID": b"a123456789abcdef"})
1768 self.fail("Should have failed adding dupliate GUID")
1769 except ldb.LdbError as err:
1771 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1773 def test_duplicate_name_dup_GUID2(self):
1774 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1777 "objectUUID": b"abc3456789abcdef"})
1779 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1782 "objectUUID": b"aaa3456789abcdef"})
1783 self.fail("Should have failed adding dupliate DN")
1784 except ldb.LdbError as err:
1786 self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1788 # Checking the GUID didn't stick in the index
1789 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1792 "objectUUID": b"aaa3456789abcdef"})
1794 def test_add_dup_guid_add(self):
1795 self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1798 "objectUUID": b"0123456789abcde1"})
1800 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1803 "objectUUID": b"0123456789abcde1"})
1804 self.fail("Should have failed on duplicate GUID")
1806 except ldb.LdbError as err:
1808 self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1810 self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1813 "objectUUID": b"0123456789abcde2"})
1816 class GUIDIndexedAddModifyTests(IndexedAddModifyTests):
1817 """Test searches using the index, to ensure the index doesn't
1821 self.index = {"dn": "@INDEXLIST",
1822 "@IDXATTR": [b"x", b"y", b"ou"],
1824 "@IDXGUID": [b"objectUUID"],
1825 "@IDX_DN_GUID": [b"GUID"]}
1826 super(GUIDIndexedAddModifyTests, self).setUp()
1829 class GUIDTransIndexedAddModifyTests(GUIDIndexedAddModifyTests):
1830 """Test GUID index behaviour insdie the transaction"""
1833 super(GUIDTransIndexedAddModifyTests, self).setUp()
1834 self.l.transaction_start()
1837 self.l.transaction_commit()
1838 super(GUIDTransIndexedAddModifyTests, self).tearDown()
1841 class TransIndexedAddModifyTests(IndexedAddModifyTests):
1842 """Test index behaviour insdie the transaction"""
1845 super(TransIndexedAddModifyTests, self).setUp()
1846 self.l.transaction_start()
1849 self.l.transaction_commit()
1850 super(TransIndexedAddModifyTests, self).tearDown()
1853 class GuidIndexedAddModifyTestsLmdb(GUIDIndexedAddModifyTests):
1856 self.prefix = MDB_PREFIX
1857 super(GuidIndexedAddModifyTestsLmdb, self).setUp()
1860 super(GuidIndexedAddModifyTestsLmdb, self).tearDown()
1863 class GuidTransIndexedAddModifyTestsLmdb(GUIDTransIndexedAddModifyTests):
1866 self.prefix = MDB_PREFIX
1867 super(GuidTransIndexedAddModifyTestsLmdb, self).setUp()
1870 super(GuidTransIndexedAddModifyTestsLmdb, self).tearDown()
1873 class BadIndexTests(LdbBaseTest):
1875 super(BadIndexTests, self).setUp()
1876 self.testdir = tempdir()
1877 self.filename = os.path.join(self.testdir, "test.ldb")
1878 self.ldb = ldb.Ldb(self.url(), flags=self.flags())
1879 if hasattr(self, 'IDXGUID'):
1880 self.ldb.add({"dn": "@INDEXLIST",
1881 "@IDXATTR": [b"x", b"y", b"ou"],
1882 "@IDXGUID": [b"objectUUID"],
1883 "@IDX_DN_GUID": [b"GUID"]})
1885 self.ldb.add({"dn": "@INDEXLIST",
1886 "@IDXATTR": [b"x", b"y", b"ou"]})
1888 super(BadIndexTests, self).setUp()
1890 def test_unique(self):
1891 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1892 "objectUUID": b"0123456789abcde1",
1894 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1895 "objectUUID": b"0123456789abcde2",
1897 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1898 "objectUUID": b"0123456789abcde3",
1901 res = self.ldb.search(expression="(y=1)",
1902 base="dc=samba,dc=org")
1903 self.assertEquals(len(res), 3)
1905 # Now set this to unique index, but forget to check the result
1907 self.ldb.add({"dn": "@ATTRIBUTES",
1908 "y": "UNIQUE_INDEX"})
1910 except ldb.LdbError:
1913 # We must still have a working index
1914 res = self.ldb.search(expression="(y=1)",
1915 base="dc=samba,dc=org")
1916 self.assertEquals(len(res), 3)
1918 def test_unique_transaction(self):
1919 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1920 "objectUUID": b"0123456789abcde1",
1922 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1923 "objectUUID": b"0123456789abcde2",
1925 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1926 "objectUUID": b"0123456789abcde3",
1929 res = self.ldb.search(expression="(y=1)",
1930 base="dc=samba,dc=org")
1931 self.assertEquals(len(res), 3)
1933 self.ldb.transaction_start()
1935 # Now set this to unique index, but forget to check the result
1937 self.ldb.add({"dn": "@ATTRIBUTES",
1938 "y": "UNIQUE_INDEX"})
1939 except ldb.LdbError:
1943 self.ldb.transaction_commit()
1946 except ldb.LdbError as err:
1948 self.assertEqual(enum, ldb.ERR_OPERATIONS_ERROR)
1950 # We must still have a working index
1951 res = self.ldb.search(expression="(y=1)",
1952 base="dc=samba,dc=org")
1954 self.assertEquals(len(res), 3)
1956 def test_casefold(self):
1957 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1958 "objectUUID": b"0123456789abcde1",
1960 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1961 "objectUUID": b"0123456789abcde2",
1963 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1964 "objectUUID": b"0123456789abcde3",
1967 res = self.ldb.search(expression="(y=a)",
1968 base="dc=samba,dc=org")
1969 self.assertEquals(len(res), 2)
1971 self.ldb.add({"dn": "@ATTRIBUTES",
1972 "y": "CASE_INSENSITIVE"})
1974 # We must still have a working index
1975 res = self.ldb.search(expression="(y=a)",
1976 base="dc=samba,dc=org")
1978 if hasattr(self, 'IDXGUID'):
1979 self.assertEquals(len(res), 3)
1981 # We should not return this entry twice, but sadly
1982 # we have not yet fixed
1983 # https://bugzilla.samba.org/show_bug.cgi?id=13361
1984 self.assertEquals(len(res), 4)
1986 def test_casefold_transaction(self):
1987 self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1988 "objectUUID": b"0123456789abcde1",
1990 self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1991 "objectUUID": b"0123456789abcde2",
1993 self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1994 "objectUUID": b"0123456789abcde3",
1997 res = self.ldb.search(expression="(y=a)",
1998 base="dc=samba,dc=org")
1999 self.assertEquals(len(res), 2)
2001 self.ldb.transaction_start()
2003 self.ldb.add({"dn": "@ATTRIBUTES",
2004 "y": "CASE_INSENSITIVE"})
2006 self.ldb.transaction_commit()
2008 # We must still have a working index
2009 res = self.ldb.search(expression="(y=a)",
2010 base="dc=samba,dc=org")
2012 if hasattr(self, 'IDXGUID'):
2013 self.assertEquals(len(res), 3)
2015 # We should not return this entry twice, but sadly
2016 # we have not yet fixed
2017 # https://bugzilla.samba.org/show_bug.cgi?id=13361
2018 self.assertEquals(len(res), 4)
2021 super(BadIndexTests, self).tearDown()
2024 class GUIDBadIndexTests(BadIndexTests):
2025 """Test Bad index things with GUID index mode"""
2030 super(GUIDBadIndexTests, self).setUp()
2033 class DnTests(TestCase):
2036 super(DnTests, self).setUp()
2037 self.ldb = ldb.Ldb()
2040 super(DnTests, self).tearDown()
2043 def test_set_dn_invalid(self):
2048 self.assertRaises(TypeError, assign)
2051 x = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
2052 y = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
2053 self.assertEqual(x, y)
2054 y = ldb.Dn(self.ldb, "dc=foo11,bar=blie")
2055 self.assertNotEqual(x, y)
2058 x = ldb.Dn(self.ldb, "dc=foo12,bar=bloe")
2059 self.assertEqual(x.__str__(), "dc=foo12,bar=bloe")
2061 def test_repr(self):
2062 x = ldb.Dn(self.ldb, "dc=foo13,bla=blie")
2063 self.assertEqual(x.__repr__(), "Dn('dc=foo13,bla=blie')")
2065 def test_get_casefold(self):
2066 x = ldb.Dn(self.ldb, "dc=foo14,bar=bloe")
2067 self.assertEqual(x.get_casefold(), "DC=FOO14,BAR=bloe")
2069 def test_validate(self):
2070 x = ldb.Dn(self.ldb, "dc=foo15,bar=bloe")
2071 self.assertTrue(x.validate())
2073 def test_parent(self):
2074 x = ldb.Dn(self.ldb, "dc=foo16,bar=bloe")
2075 self.assertEqual("bar=bloe", x.parent().__str__())
2077 def test_parent_nonexistent(self):
2078 x = ldb.Dn(self.ldb, "@BLA")
2079 self.assertEqual(None, x.parent())
2081 def test_is_valid(self):
2082 x = ldb.Dn(self.ldb, "dc=foo18,dc=bloe")
2083 self.assertTrue(x.is_valid())
2084 x = ldb.Dn(self.ldb, "")
2085 self.assertTrue(x.is_valid())
2087 def test_is_special(self):
2088 x = ldb.Dn(self.ldb, "dc=foo19,bar=bloe")
2089 self.assertFalse(x.is_special())
2090 x = ldb.Dn(self.ldb, "@FOOBAR")
2091 self.assertTrue(x.is_special())
2093 def test_check_special(self):
2094 x = ldb.Dn(self.ldb, "dc=foo20,bar=bloe")
2095 self.assertFalse(x.check_special("FOOBAR"))
2096 x = ldb.Dn(self.ldb, "@FOOBAR")
2097 self.assertTrue(x.check_special("@FOOBAR"))
2100 x = ldb.Dn(self.ldb, "dc=foo21,bar=bloe")
2101 self.assertEqual(2, len(x))
2102 x = ldb.Dn(self.ldb, "dc=foo21")
2103 self.assertEqual(1, len(x))
2105 def test_add_child(self):
2106 x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
2107 self.assertTrue(x.add_child(ldb.Dn(self.ldb, "bla=bloe")))
2108 self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
2110 def test_add_base(self):
2111 x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
2112 base = ldb.Dn(self.ldb, "bla=bloe")
2113 self.assertTrue(x.add_base(base))
2114 self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
2116 def test_add_child_str(self):
2117 x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
2118 self.assertTrue(x.add_child("bla=bloe"))
2119 self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
2121 def test_add_base_str(self):
2122 x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
2124 self.assertTrue(x.add_base(base))
2125 self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
2128 x = ldb.Dn(self.ldb, "dc=foo24")
2129 y = ldb.Dn(self.ldb, "bar=bla")
2130 self.assertEqual("dc=foo24,bar=bla", str(x + y))
2132 def test_remove_base_components(self):
2133 x = ldb.Dn(self.ldb, "dc=foo24,dc=samba,dc=org")
2134 x.remove_base_components(len(x) - 1)
2135 self.assertEqual("dc=foo24", str(x))
2137 def test_parse_ldif(self):
2138 msgs = self.ldb.parse_ldif("dn: foo=bar\n")
2140 self.assertEqual("foo=bar", str(msg[1].dn))
2141 self.assertTrue(isinstance(msg[1], ldb.Message))
2142 ldif = self.ldb.write_ldif(msg[1], ldb.CHANGETYPE_NONE)
2143 self.assertEqual("dn: foo=bar\n\n", ldif)
2145 def test_parse_ldif_more(self):
2146 msgs = self.ldb.parse_ldif("dn: foo=bar\n\n\ndn: bar=bar")
2148 self.assertEqual("foo=bar", str(msg[1].dn))
2150 self.assertEqual("bar=bar", str(msg[1].dn))
2152 def test_canonical_string(self):
2153 x = ldb.Dn(self.ldb, "dc=foo25,bar=bloe")
2154 self.assertEqual("/bloe/foo25", x.canonical_str())
2156 def test_canonical_ex_string(self):
2157 x = ldb.Dn(self.ldb, "dc=foo26,bar=bloe")
2158 self.assertEqual("/bloe\nfoo26", x.canonical_ex_str())
2160 def test_ldb_is_child_of(self):
2161 """Testing ldb_dn_compare_dn"""
2162 dn1 = ldb.Dn(self.ldb, "dc=base")
2163 dn2 = ldb.Dn(self.ldb, "cn=foo,dc=base")
2164 dn3 = ldb.Dn(self.ldb, "cn=bar,dc=base")
2165 dn4 = ldb.Dn(self.ldb, "cn=baz,cn=bar,dc=base")
2167 self.assertTrue(dn1.is_child_of(dn1))
2168 self.assertTrue(dn2.is_child_of(dn1))
2169 self.assertTrue(dn4.is_child_of(dn1))
2170 self.assertTrue(dn4.is_child_of(dn3))
2171 self.assertTrue(dn4.is_child_of(dn4))
2172 self.assertFalse(dn3.is_child_of(dn2))
2173 self.assertFalse(dn1.is_child_of(dn4))
2175 def test_ldb_is_child_of_str(self):
2176 """Testing ldb_dn_compare_dn"""
2178 dn2_str = "cn=foo,dc=base"
2179 dn3_str = "cn=bar,dc=base"
2180 dn4_str = "cn=baz,cn=bar,dc=base"
2182 dn1 = ldb.Dn(self.ldb, dn1_str)
2183 dn2 = ldb.Dn(self.ldb, dn2_str)
2184 dn3 = ldb.Dn(self.ldb, dn3_str)
2185 dn4 = ldb.Dn(self.ldb, dn4_str)
2187 self.assertTrue(dn1.is_child_of(dn1_str))
2188 self.assertTrue(dn2.is_child_of(dn1_str))
2189 self.assertTrue(dn4.is_child_of(dn1_str))
2190 self.assertTrue(dn4.is_child_of(dn3_str))
2191 self.assertTrue(dn4.is_child_of(dn4_str))
2192 self.assertFalse(dn3.is_child_of(dn2_str))
2193 self.assertFalse(dn1.is_child_of(dn4_str))
2195 def test_get_component_name(self):
2196 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2197 self.assertEqual(dn.get_component_name(0), 'cn')
2198 self.assertEqual(dn.get_component_name(1), 'dc')
2199 self.assertEqual(dn.get_component_name(2), None)
2200 self.assertEqual(dn.get_component_name(-1), None)
2202 def test_get_component_value(self):
2203 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2204 self.assertEqual(dn.get_component_value(0), 'foo')
2205 self.assertEqual(dn.get_component_value(1), 'base')
2206 self.assertEqual(dn.get_component_name(2), None)
2207 self.assertEqual(dn.get_component_name(-1), None)
2209 def test_set_component(self):
2210 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2211 dn.set_component(0, 'cn', 'bar')
2212 self.assertEqual(str(dn), "cn=bar,dc=base")
2213 dn.set_component(1, 'o', 'asep')
2214 self.assertEqual(str(dn), "cn=bar,o=asep")
2215 self.assertRaises(TypeError, dn.set_component, 2, 'dc', 'base')
2216 self.assertEqual(str(dn), "cn=bar,o=asep")
2217 dn.set_component(1, 'o', 'a,b+c')
2218 self.assertEqual(str(dn), r"cn=bar,o=a\,b\+c")
2220 def test_set_component_bytes(self):
2221 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2222 dn.set_component(0, 'cn', b'bar')
2223 self.assertEqual(str(dn), "cn=bar,dc=base")
2224 dn.set_component(1, 'o', b'asep')
2225 self.assertEqual(str(dn), "cn=bar,o=asep")
2227 def test_set_component_none(self):
2228 dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
2229 self.assertRaises(TypeError, dn.set_component, 1, 'cn', None)
2231 def test_get_extended_component_null(self):
2232 dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
2233 self.assertEqual(dn.get_extended_component("TEST"), None)
2235 def test_get_extended_component(self):
2236 self.ldb._register_test_extensions()
2237 dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
2238 self.assertEqual(dn.get_extended_component("TEST"), b"foo")
2240 def test_set_extended_component(self):
2241 self.ldb._register_test_extensions()
2242 dn = ldb.Dn(self.ldb, "dc=base")
2243 dn.set_extended_component("TEST", "foo")
2244 self.assertEqual(dn.get_extended_component("TEST"), b"foo")
2245 dn.set_extended_component("TEST", b"bar")
2246 self.assertEqual(dn.get_extended_component("TEST"), b"bar")
2248 def test_extended_str(self):
2249 self.ldb._register_test_extensions()
2250 dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
2251 self.assertEqual(dn.extended_str(), "<TEST=foo>;cn=bar,dc=base")
2253 def test_get_rdn_name(self):
2254 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2255 self.assertEqual(dn.get_rdn_name(), 'cn')
2257 def test_get_rdn_value(self):
2258 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2259 self.assertEqual(dn.get_rdn_value(), 'foo')
2261 def test_get_casefold(self):
2262 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2263 self.assertEqual(dn.get_casefold(), 'CN=FOO,DC=BASE')
2265 def test_get_linearized(self):
2266 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2267 self.assertEqual(dn.get_linearized(), 'cn=foo,dc=base')
2269 def test_is_null(self):
2270 dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2271 self.assertFalse(dn.is_null())
2273 dn = ldb.Dn(self.ldb, '')
2274 self.assertTrue(dn.is_null())
2277 class LdbMsgTests(TestCase):
2280 super(LdbMsgTests, self).setUp()
2281 self.msg = ldb.Message()
2283 def test_init_dn(self):
2284 self.msg = ldb.Message(ldb.Dn(ldb.Ldb(), "dc=foo27"))
2285 self.assertEqual("dc=foo27", str(self.msg.dn))
2287 def test_iter_items(self):
2288 self.assertEqual(0, len(self.msg.items()))
2289 self.msg.dn = ldb.Dn(ldb.Ldb(), "dc=foo28")
2290 self.assertEqual(1, len(self.msg.items()))
2292 def test_repr(self):
2293 self.msg.dn = ldb.Dn(ldb.Ldb(), "dc=foo29")
2294 self.msg["dc"] = b"foo"
2296 self.assertIn(repr(self.msg), [
2297 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])})",
2298 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')})",
2300 self.assertIn(repr(self.msg.text), [
2301 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])}).text",
2302 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')}).text",
2307 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])})")
2309 repr(self.msg.text),
2310 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])}).text")
2313 self.assertEqual(0, len(self.msg))
2315 def test_notpresent(self):
2316 self.assertRaises(KeyError, lambda: self.msg["foo"])
2322 self.msg.add(ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla"))
2324 def test_add_text(self):
2325 self.msg.add(ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla"))
2327 def test_elements_empty(self):
2328 self.assertEqual([], self.msg.elements())
2330 def test_elements(self):
2331 el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
2333 self.assertEqual([el], self.msg.elements())
2334 self.assertEqual([el.text], self.msg.text.elements())
2336 def test_add_value(self):
2337 self.assertEqual(0, len(self.msg))
2338 self.msg["foo"] = [b"foo"]
2339 self.assertEqual(1, len(self.msg))
2341 def test_add_value_text(self):
2342 self.assertEqual(0, len(self.msg))
2343 self.msg["foo"] = ["foo"]
2344 self.assertEqual(1, len(self.msg))
2346 def test_add_value_multiple(self):
2347 self.assertEqual(0, len(self.msg))
2348 self.msg["foo"] = [b"foo", b"bla"]
2349 self.assertEqual(1, len(self.msg))
2350 self.assertEqual([b"foo", b"bla"], list(self.msg["foo"]))
2352 def test_add_value_multiple_text(self):
2353 self.assertEqual(0, len(self.msg))
2354 self.msg["foo"] = ["foo", "bla"]
2355 self.assertEqual(1, len(self.msg))
2356 self.assertEqual(["foo", "bla"], list(self.msg.text["foo"]))
2358 def test_set_value(self):
2359 self.msg["foo"] = [b"fool"]
2360 self.assertEqual([b"fool"], list(self.msg["foo"]))
2361 self.msg["foo"] = [b"bar"]
2362 self.assertEqual([b"bar"], list(self.msg["foo"]))
2364 def test_set_value_text(self):
2365 self.msg["foo"] = ["fool"]
2366 self.assertEqual(["fool"], list(self.msg.text["foo"]))
2367 self.msg["foo"] = ["bar"]
2368 self.assertEqual(["bar"], list(self.msg.text["foo"]))
2370 def test_keys(self):
2371 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2372 self.msg["foo"] = [b"bla"]
2373 self.msg["bar"] = [b"bla"]
2374 self.assertEqual(["dn", "foo", "bar"], list(self.msg.keys()))
2376 def test_keys_text(self):
2377 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2378 self.msg["foo"] = ["bla"]
2379 self.msg["bar"] = ["bla"]
2380 self.assertEqual(["dn", "foo", "bar"], list(self.msg.text.keys()))
2383 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2384 self.assertEqual("@BASEINFO", self.msg.dn.__str__())
2386 def test_get_dn(self):
2387 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2388 self.assertEqual("@BASEINFO", self.msg.get("dn").__str__())
2390 def test_dn_text(self):
2391 self.msg.text.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2392 self.assertEqual("@BASEINFO", str(self.msg.dn))
2393 self.assertEqual("@BASEINFO", str(self.msg.text.dn))
2395 def test_get_dn_text(self):
2396 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2397 self.assertEqual("@BASEINFO", str(self.msg.get("dn")))
2398 self.assertEqual("@BASEINFO", str(self.msg.text.get("dn")))
2400 def test_get_invalid(self):
2401 self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2402 self.assertRaises(TypeError, self.msg.get, 42)
2404 def test_get_other(self):
2405 self.msg["foo"] = [b"bar"]
2406 self.assertEqual(b"bar", self.msg.get("foo")[0])
2407 self.assertEqual(b"bar", self.msg.get("foo", idx=0))
2408 self.assertEqual(None, self.msg.get("foo", idx=1))
2409 self.assertEqual("", self.msg.get("foo", default='', idx=1))
2411 def test_get_other_text(self):
2412 self.msg["foo"] = ["bar"]
2413 self.assertEqual(["bar"], list(self.msg.text.get("foo")))
2414 self.assertEqual("bar", self.msg.text.get("foo")[0])
2415 self.assertEqual("bar", self.msg.text.get("foo", idx=0))
2416 self.assertEqual(None, self.msg.get("foo", idx=1))
2417 self.assertEqual("", self.msg.get("foo", default='', idx=1))
2419 def test_get_default(self):
2420 self.assertEqual(None, self.msg.get("tatayoyo", idx=0))
2421 self.assertEqual("anniecordie", self.msg.get("tatayoyo", "anniecordie"))
2423 def test_get_default_text(self):
2424 self.assertEqual(None, self.msg.text.get("tatayoyo", idx=0))
2425 self.assertEqual("anniecordie", self.msg.text.get("tatayoyo", "anniecordie"))
2427 def test_get_unknown(self):
2428 self.assertEqual(None, self.msg.get("lalalala"))
2430 def test_get_unknown_text(self):
2431 self.assertEqual(None, self.msg.text.get("lalalala"))
2433 def test_msg_diff(self):
2435 msgs = l.parse_ldif("dn: foo=bar\nfoo: bar\nbaz: do\n\ndn: foo=bar\nfoo: bar\nbaz: dont\n")
2436 msg1 = next(msgs)[1]
2437 msg2 = next(msgs)[1]
2438 msgdiff = l.msg_diff(msg1, msg2)
2439 self.assertEqual("foo=bar", msgdiff.get("dn").__str__())
2440 self.assertRaises(KeyError, lambda: msgdiff["foo"])
2441 self.assertEqual(1, len(msgdiff))
2443 def test_equal_empty(self):
2444 msg1 = ldb.Message()
2445 msg2 = ldb.Message()
2446 self.assertEqual(msg1, msg2)
2448 def test_equal_simplel(self):
2450 msg1 = ldb.Message()
2451 msg1.dn = ldb.Dn(db, "foo=bar")
2452 msg2 = ldb.Message()
2453 msg2.dn = ldb.Dn(db, "foo=bar")
2454 self.assertEqual(msg1, msg2)
2455 msg1['foo'] = b'bar'
2456 msg2['foo'] = b'bar'
2457 self.assertEqual(msg1, msg2)
2458 msg2['foo'] = b'blie'
2459 self.assertNotEqual(msg1, msg2)
2460 msg2['foo'] = b'blie'
2462 def test_from_dict(self):
2463 rec = {"dn": "dc=fromdict",
2464 "a1": [b"a1-val1", b"a1-val1"]}
2466 # check different types of input Flags
2467 for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
2468 m = ldb.Message.from_dict(l, rec, flags)
2469 self.assertEqual(rec["a1"], list(m["a1"]))
2470 self.assertEqual(flags, m["a1"].flags())
2471 # check input params
2472 self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
2473 self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
2474 self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
2475 # Message.from_dict expects dictionary with 'dn'
2476 err_rec = {"a1": [b"a1-val1", b"a1-val1"]}
2477 self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
2479 def test_from_dict_text(self):
2480 rec = {"dn": "dc=fromdict",
2481 "a1": ["a1-val1", "a1-val1"]}
2483 # check different types of input Flags
2484 for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
2485 m = ldb.Message.from_dict(l, rec, flags)
2486 self.assertEqual(rec["a1"], list(m.text["a1"]))
2487 self.assertEqual(flags, m.text["a1"].flags())
2488 # check input params
2489 self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
2490 self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
2491 self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
2492 # Message.from_dict expects dictionary with 'dn'
2493 err_rec = {"a1": ["a1-val1", "a1-val1"]}
2494 self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
2496 def test_copy_add_message_element(self):
2498 m["1"] = ldb.MessageElement([b"val 111"], ldb.FLAG_MOD_ADD, "1")
2499 m["2"] = ldb.MessageElement([b"val 222"], ldb.FLAG_MOD_ADD, "2")
2503 self.assertEqual(mto["1"], m["1"])
2504 self.assertEqual(mto["2"], m["2"])
2508 self.assertEqual(mto["1"], m["1"])
2509 self.assertEqual(mto["2"], m["2"])
2511 def test_copy_add_message_element_text(self):
2513 m["1"] = ldb.MessageElement(["val 111"], ldb.FLAG_MOD_ADD, "1")
2514 m["2"] = ldb.MessageElement(["val 222"], ldb.FLAG_MOD_ADD, "2")
2518 self.assertEqual(mto["1"], m.text["1"])
2519 self.assertEqual(mto["2"], m.text["2"])
2523 self.assertEqual(mto.text["1"], m.text["1"])
2524 self.assertEqual(mto.text["2"], m.text["2"])
2525 self.assertEqual(mto["1"], m["1"])
2526 self.assertEqual(mto["2"], m["2"])
2529 class MessageElementTests(TestCase):
2531 def test_cmp_element(self):
2532 x = ldb.MessageElement([b"foo"])
2533 y = ldb.MessageElement([b"foo"])
2534 z = ldb.MessageElement([b"bzr"])
2535 self.assertEqual(x, y)
2536 self.assertNotEqual(x, z)
2538 def test_cmp_element_text(self):
2539 x = ldb.MessageElement([b"foo"])
2540 y = ldb.MessageElement(["foo"])
2541 self.assertEqual(x, y)
2543 def test_create_iterable(self):
2544 x = ldb.MessageElement([b"foo"])
2545 self.assertEqual([b"foo"], list(x))
2546 self.assertEqual(["foo"], list(x.text))
2548 def test_repr(self):
2549 x = ldb.MessageElement([b"foo"])
2551 self.assertEqual("MessageElement([b'foo'])", repr(x))
2552 self.assertEqual("MessageElement([b'foo']).text", repr(x.text))
2554 self.assertEqual("MessageElement(['foo'])", repr(x))
2555 self.assertEqual("MessageElement(['foo']).text", repr(x.text))
2556 x = ldb.MessageElement([b"foo", b"bla"])
2557 self.assertEqual(2, len(x))
2559 self.assertEqual("MessageElement([b'foo',b'bla'])", repr(x))
2560 self.assertEqual("MessageElement([b'foo',b'bla']).text", repr(x.text))
2562 self.assertEqual("MessageElement(['foo','bla'])", repr(x))
2563 self.assertEqual("MessageElement(['foo','bla']).text", repr(x.text))
2565 def test_get_item(self):
2566 x = ldb.MessageElement([b"foo", b"bar"])
2567 self.assertEqual(b"foo", x[0])
2568 self.assertEqual(b"bar", x[1])
2569 self.assertEqual(b"bar", x[-1])
2570 self.assertRaises(IndexError, lambda: x[45])
2572 def test_get_item_text(self):
2573 x = ldb.MessageElement(["foo", "bar"])
2574 self.assertEqual("foo", x.text[0])
2575 self.assertEqual("bar", x.text[1])
2576 self.assertEqual("bar", x.text[-1])
2577 self.assertRaises(IndexError, lambda: x[45])
2580 x = ldb.MessageElement([b"foo", b"bar"])
2581 self.assertEqual(2, len(x))
2584 x = ldb.MessageElement([b"foo", b"bar"])
2585 y = ldb.MessageElement([b"foo", b"bar"])
2586 self.assertEqual(y, x)
2587 x = ldb.MessageElement([b"foo"])
2588 self.assertNotEqual(y, x)
2589 y = ldb.MessageElement([b"foo"])
2590 self.assertEqual(y, x)
2592 def test_extended(self):
2593 el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
2595 self.assertEqual("MessageElement([b'456'])", repr(el))
2596 self.assertEqual("MessageElement([b'456']).text", repr(el.text))
2598 self.assertEqual("MessageElement(['456'])", repr(el))
2599 self.assertEqual("MessageElement(['456']).text", repr(el.text))
2601 def test_bad_text(self):
2602 el = ldb.MessageElement(b'\xba\xdd')
2603 self.assertRaises(UnicodeDecodeError, el.text.__getitem__, 0)
2606 class ModuleTests(TestCase):
2609 super(ModuleTests, self).setUp()
2610 self.testdir = tempdir()
2611 self.filename = os.path.join(self.testdir, "test.ldb")
2612 self.ldb = ldb.Ldb(self.filename)
2615 shutil.rmtree(self.testdir)
2616 super(ModuleTests, self).setUp()
2618 def test_register_module(self):
2619 class ExampleModule:
2621 ldb.register_module(ExampleModule)
2623 def test_use_module(self):
2626 class ExampleModule:
2629 def __init__(self, ldb, next):
2633 def search(self, *args, **kwargs):
2634 return self.next.search(*args, **kwargs)
2636 def request(self, *args, **kwargs):
2639 ldb.register_module(ExampleModule)
2640 l = ldb.Ldb(self.filename)
2641 l.add({"dn": "@MODULES", "@LIST": "bla"})
2642 self.assertEqual([], ops)
2643 l = ldb.Ldb(self.filename)
2644 self.assertEqual(["init"], ops)
2647 class LdbResultTests(LdbBaseTest):
2650 super(LdbResultTests, self).setUp()
2651 self.testdir = tempdir()
2652 self.filename = os.path.join(self.testdir, "test.ldb")
2653 self.l = ldb.Ldb(self.url(), flags=self.flags())
2655 self.l.add(self.index)
2656 except AttributeError:
2658 self.l.add({"dn": "DC=SAMBA,DC=ORG", "name": b"samba.org",
2659 "objectUUID": b"0123456789abcde0"})
2660 self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG", "name": b"Admins",
2661 "objectUUID": b"0123456789abcde1"})
2662 self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG", "name": b"Users",
2663 "objectUUID": b"0123456789abcde2"})
2664 self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG", "name": b"OU #1",
2665 "objectUUID": b"0123456789abcde3"})
2666 self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG", "name": b"OU #2",
2667 "objectUUID": b"0123456789abcde4"})
2668 self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG", "name": b"OU #3",
2669 "objectUUID": b"0123456789abcde5"})
2670 self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG", "name": b"OU #4",
2671 "objectUUID": b"0123456789abcde6"})
2672 self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG", "name": b"OU #5",
2673 "objectUUID": b"0123456789abcde7"})
2674 self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG", "name": b"OU #6",
2675 "objectUUID": b"0123456789abcde8"})
2676 self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG", "name": b"OU #7",
2677 "objectUUID": b"0123456789abcde9"})
2678 self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG", "name": b"OU #8",
2679 "objectUUID": b"0123456789abcdea"})
2680 self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG", "name": b"OU #9",
2681 "objectUUID": b"0123456789abcdeb"})
2682 self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG", "name": b"OU #10",
2683 "objectUUID": b"0123456789abcdec"})
2686 shutil.rmtree(self.testdir)
2687 super(LdbResultTests, self).tearDown()
2688 # Ensure the LDB is closed now, so we close the FD
2691 def test_return_type(self):
2692 res = self.l.search()
2693 self.assertEqual(str(res), "<ldb result>")
2695 def test_get_msgs(self):
2696 res = self.l.search()
2699 def test_get_controls(self):
2700 res = self.l.search()
2703 def test_get_referals(self):
2704 res = self.l.search()
2707 def test_iter_msgs(self):
2709 for l in self.l.search().msgs:
2710 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2712 self.assertTrue(found)
2714 def test_iter_msgs_count(self):
2715 self.assertTrue(self.l.search().count > 0)
2716 # 13 objects has been added to the DC=SAMBA, DC=ORG
2717 self.assertEqual(self.l.search(base="DC=SAMBA,DC=ORG").count, 13)
2719 def test_iter_controls(self):
2720 res = self.l.search().controls
2723 def test_create_control(self):
2724 self.assertRaises(ValueError, ldb.Control, self.l, "tatayoyo:0")
2725 c = ldb.Control(self.l, "relax:1")
2726 self.assertEqual(c.critical, True)
2727 self.assertEqual(c.oid, "1.3.6.1.4.1.4203.666.5.12")
2729 def test_iter_refs(self):
2730 res = self.l.search().referals
2733 def test_search_sequence_msgs(self):
2735 res = self.l.search().msgs
2737 for i in range(0, len(res)):
2739 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2741 self.assertTrue(found)
2743 def test_search_as_iter(self):
2745 res = self.l.search()
2748 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2750 self.assertTrue(found)
2752 def test_search_iter(self):
2754 res = self.l.search_iterator()
2757 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2759 self.assertTrue(found)
2761 # Show that search results can't see into a transaction
2763 def test_search_against_trans(self):
2766 (r1, w1) = os.pipe()
2768 (r2, w2) = os.pipe()
2770 # For the first element, fork a child that will
2774 # In the child, re-open
2778 child_ldb = ldb.Ldb(self.url(), flags=self.flags())
2779 # start a transaction
2780 child_ldb.transaction_start()
2783 child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
2784 "name": b"samba.org",
2785 "objectUUID": b"o123456789acbdef"})
2787 os.write(w1, b"added")
2789 # Now wait for the search to be done
2794 child_ldb.transaction_commit()
2795 except LdbError as err:
2796 # We print this here to see what went wrong in the child
2800 os.write(w1, b"transaction")
2803 self.assertEqual(os.read(r1, 5), b"added")
2805 # This should not turn up until the transaction is concluded
2806 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2807 scope=ldb.SCOPE_BASE)
2808 self.assertEqual(len(res11), 0)
2810 os.write(w2, b"search")
2812 # Now wait for the transaction to be done. This should
2813 # deadlock, but the search doesn't hold a read lock for the
2814 # iterator lifetime currently.
2815 self.assertEqual(os.read(r1, 11), b"transaction")
2817 # This should now turn up, as the transaction is over
2818 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2819 scope=ldb.SCOPE_BASE)
2820 self.assertEqual(len(res11), 1)
2822 self.assertFalse(found11)
2824 (got_pid, status) = os.waitpid(pid, 0)
2825 self.assertEqual(got_pid, pid)
2827 def test_search_iter_against_trans(self):
2831 # We need to hold this iterator open to hold the all-record
2833 res = self.l.search_iterator()
2835 (r1, w1) = os.pipe()
2837 (r2, w2) = os.pipe()
2839 # For the first element, with the sequence open (which
2840 # means with ldb locks held), fork a child that will
2844 # In the child, re-open
2849 child_ldb = ldb.Ldb(self.url(), flags=self.flags())
2850 # start a transaction
2851 child_ldb.transaction_start()
2854 child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
2855 "name": b"samba.org",
2856 "objectUUID": b"o123456789acbdef"})
2858 os.write(w1, b"added")
2860 # Now wait for the search to be done
2865 child_ldb.transaction_commit()
2866 except LdbError as err:
2867 # We print this here to see what went wrong in the child
2871 os.write(w1, b"transaction")
2874 self.assertEqual(os.read(r1, 5), b"added")
2876 # This should not turn up until the transaction is concluded
2877 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2878 scope=ldb.SCOPE_BASE)
2879 self.assertEqual(len(res11), 0)
2881 os.write(w2, b"search")
2883 # allow the transaction to start
2886 # This should not turn up until the search finishes and
2887 # removed the read lock, but for ldb_tdb that happened as soon
2888 # as we called the first res.next()
2889 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2890 scope=ldb.SCOPE_BASE)
2891 self.assertEqual(len(res11), 0)
2893 # These results are all collected at the first next(res) call
2895 if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2897 if str(l.dn) == "OU=OU11,DC=SAMBA,DC=ORG":
2900 # Now wait for the transaction to be done.
2901 self.assertEqual(os.read(r1, 11), b"transaction")
2903 # This should now turn up, as the transaction is over and all
2904 # read locks are gone
2905 res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2906 scope=ldb.SCOPE_BASE)
2907 self.assertEqual(len(res11), 1)
2909 self.assertTrue(found)
2910 self.assertFalse(found11)
2912 (got_pid, status) = os.waitpid(pid, 0)
2913 self.assertEqual(got_pid, pid)
2916 class LdbResultTestsLmdb(LdbResultTests):
2919 self.prefix = MDB_PREFIX
2920 self.index = MDB_INDEX_OBJ
2921 super(LdbResultTestsLmdb, self).setUp()
2924 super(LdbResultTestsLmdb, self).tearDown()
2927 class BadTypeTests(TestCase):
2928 def test_control(self):
2930 self.assertRaises(TypeError, ldb.Control, '<bad type>', 'relax:1')
2931 self.assertRaises(TypeError, ldb.Control, ldb, 1234)
2933 def test_modify(self):
2935 dn = ldb.Dn(l, 'a=b')
2937 self.assertRaises(TypeError, l.modify, '<bad type>')
2938 self.assertRaises(TypeError, l.modify, m, '<bad type>')
2942 dn = ldb.Dn(l, 'a=b')
2944 self.assertRaises(TypeError, l.add, '<bad type>')
2945 self.assertRaises(TypeError, l.add, m, '<bad type>')
2947 def test_delete(self):
2949 dn = ldb.Dn(l, 'a=b')
2950 self.assertRaises(TypeError, l.add, '<bad type>')
2951 self.assertRaises(TypeError, l.add, dn, '<bad type>')
2953 def test_rename(self):
2955 dn = ldb.Dn(l, 'a=b')
2956 self.assertRaises(TypeError, l.add, '<bad type>', dn)
2957 self.assertRaises(TypeError, l.add, dn, '<bad type>')
2958 self.assertRaises(TypeError, l.add, dn, dn, '<bad type>')
2960 def test_search(self):
2962 self.assertRaises(TypeError, l.search, base=1234)
2963 self.assertRaises(TypeError, l.search, scope='<bad type>')
2964 self.assertRaises(TypeError, l.search, expression=1234)
2965 self.assertRaises(TypeError, l.search, attrs='<bad type>')
2966 self.assertRaises(TypeError, l.search, controls='<bad type>')
2969 class VersionTests(TestCase):
2971 def test_version(self):
2972 self.assertTrue(isinstance(ldb.__version__, str))
2975 if __name__ == '__main__':
2977 unittest.TestProgram()