PEP8: fix E203: whitespace before ':'
[nivanova/samba-autobuild/.git] / lib / ldb / tests / python / api.py
1 #!/usr/bin/env python
2 # Simple tests for the ldb python bindings.
3 # Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
4
5 import os
6 from unittest import TestCase
7 import sys
8 import gc
9 import time
10 import ldb
11 import shutil
12
13 PY3 = sys.version_info > (3, 0)
14
15 TDB_PREFIX = "tdb://"
16 MDB_PREFIX = "mdb://"
17
18 MDB_INDEX_OBJ = {
19     "dn": "@INDEXLIST",
20     "@IDXONE": [b"1"],
21     "@IDXGUID": [b"objectUUID"],
22     "@IDX_DN_GUID": [b"GUID"]
23 }
24
25 def tempdir():
26     import tempfile
27     try:
28         dir_prefix = os.path.join(os.environ["SELFTEST_PREFIX"], "tmp")
29     except KeyError:
30         dir_prefix = None
31     return tempfile.mkdtemp(dir=dir_prefix)
32
33
34 class NoContextTests(TestCase):
35
36     def test_valid_attr_name(self):
37         self.assertTrue(ldb.valid_attr_name("foo"))
38         self.assertFalse(ldb.valid_attr_name("24foo"))
39
40     def test_timestring(self):
41         self.assertEqual("19700101000000.0Z", ldb.timestring(0))
42         self.assertEqual("20071119191012.0Z", ldb.timestring(1195499412))
43
44     def test_string_to_time(self):
45         self.assertEqual(0, ldb.string_to_time("19700101000000.0Z"))
46         self.assertEqual(1195499412, ldb.string_to_time("20071119191012.0Z"))
47
48     def test_binary_encode(self):
49         encoded = ldb.binary_encode(b'test\\x')
50         decoded = ldb.binary_decode(encoded)
51         self.assertEqual(decoded, b'test\\x')
52
53         encoded2 = ldb.binary_encode('test\\x')
54         self.assertEqual(encoded2, encoded)
55
56
57 class LdbBaseTest(TestCase):
58     def setUp(self):
59         super(LdbBaseTest, self).setUp()
60         try:
61             if self.prefix is None:
62                 self.prefix = TDB_PREFIX
63         except AttributeError:
64             self.prefix = TDB_PREFIX
65
66     def tearDown(self):
67         super(LdbBaseTest, self).tearDown()
68
69     def url(self):
70         return self.prefix + self.filename
71
72     def flags(self):
73         if self.prefix == MDB_PREFIX:
74             return ldb.FLG_NOSYNC
75         else:
76             return 0
77
78
79 class SimpleLdb(LdbBaseTest):
80
81     def setUp(self):
82         super(SimpleLdb, self).setUp()
83         self.testdir = tempdir()
84         self.filename = os.path.join(self.testdir, "test.ldb")
85         self.ldb = ldb.Ldb(self.url(), flags=self.flags())
86         try:
87             self.ldb.add(self.index)
88         except AttributeError:
89             pass
90
91     def tearDown(self):
92         shutil.rmtree(self.testdir)
93         super(SimpleLdb, self).tearDown()
94         # Ensure the LDB is closed now, so we close the FD
95         del(self.ldb)
96
97     def test_connect(self):
98         ldb.Ldb(self.url(), flags=self.flags())
99
100     def test_connect_none(self):
101         ldb.Ldb()
102
103     def test_connect_later(self):
104         x = ldb.Ldb()
105         x.connect(self.url(), flags=self.flags())
106
107     def test_repr(self):
108         x = ldb.Ldb()
109         self.assertTrue(repr(x).startswith("<ldb connection"))
110
111     def test_set_create_perms(self):
112         x = ldb.Ldb()
113         x.set_create_perms(0o600)
114
115     def test_modules_none(self):
116         x = ldb.Ldb()
117         self.assertEqual([], x.modules())
118
119     def test_modules_tdb(self):
120         x = ldb.Ldb(self.url(), flags=self.flags())
121         self.assertEqual("[<ldb module 'tdb'>]", repr(x.modules()))
122
123     def test_firstmodule_none(self):
124         x = ldb.Ldb()
125         self.assertEqual(x.firstmodule, None)
126
127     def test_firstmodule_tdb(self):
128         x = ldb.Ldb(self.url(), flags=self.flags())
129         mod = x.firstmodule
130         self.assertEqual(repr(mod), "<ldb module 'tdb'>")
131
132     def test_search(self):
133         l = ldb.Ldb(self.url(), flags=self.flags())
134         self.assertEqual(len(l.search()), 0)
135
136     def test_search_controls(self):
137         l = ldb.Ldb(self.url(), flags=self.flags())
138         self.assertEqual(len(l.search(controls=["paged_results:0:5"])), 0)
139
140     def test_search_attrs(self):
141         l = ldb.Ldb(self.url(), flags=self.flags())
142         self.assertEqual(len(l.search(ldb.Dn(l, ""), ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
143
144     def test_search_string_dn(self):
145         l = ldb.Ldb(self.url(), flags=self.flags())
146         self.assertEqual(len(l.search("", ldb.SCOPE_SUBTREE, "(dc=*)", ["dc"])), 0)
147
148     def test_search_attr_string(self):
149         l = ldb.Ldb(self.url(), flags=self.flags())
150         self.assertRaises(TypeError, l.search, attrs="dc")
151         self.assertRaises(TypeError, l.search, attrs=b"dc")
152
153     def test_opaque(self):
154         l = ldb.Ldb(self.url(), flags=self.flags())
155         l.set_opaque("my_opaque", l)
156         self.assertTrue(l.get_opaque("my_opaque") is not None)
157         self.assertEqual(None, l.get_opaque("unknown"))
158
159     def test_search_scope_base_empty_db(self):
160         l = ldb.Ldb(self.url(), flags=self.flags())
161         self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
162                                       ldb.SCOPE_BASE)), 0)
163
164     def test_search_scope_onelevel_empty_db(self):
165         l = ldb.Ldb(self.url(), flags=self.flags())
166         self.assertEqual(len(l.search(ldb.Dn(l, "dc=foo1"),
167                                       ldb.SCOPE_ONELEVEL)), 0)
168
169     def test_delete(self):
170         l = ldb.Ldb(self.url(), flags=self.flags())
171         self.assertRaises(ldb.LdbError, lambda: l.delete(ldb.Dn(l, "dc=foo2")))
172
173     def test_delete_w_unhandled_ctrl(self):
174         l = ldb.Ldb(self.url(), flags=self.flags())
175         m = ldb.Message()
176         m.dn = ldb.Dn(l, "dc=foo1")
177         m["b"] = [b"a"]
178         m["objectUUID"] = b"0123456789abcdef"
179         l.add(m)
180         self.assertRaises(ldb.LdbError, lambda: l.delete(m.dn, ["search_options:1:2"]))
181         l.delete(m.dn)
182
183     def test_contains(self):
184         name = self.url()
185         l = ldb.Ldb(name, flags=self.flags())
186         self.assertFalse(ldb.Dn(l, "dc=foo3") in l)
187         l = ldb.Ldb(name, flags=self.flags())
188         m = ldb.Message()
189         m.dn = ldb.Dn(l, "dc=foo3")
190         m["b"] = ["a"]
191         m["objectUUID"] = b"0123456789abcdef"
192         l.add(m)
193         try:
194             self.assertTrue(ldb.Dn(l, "dc=foo3") in l)
195             self.assertFalse(ldb.Dn(l, "dc=foo4") in l)
196         finally:
197             l.delete(m.dn)
198
199     def test_get_config_basedn(self):
200         l = ldb.Ldb(self.url(), flags=self.flags())
201         self.assertEqual(None, l.get_config_basedn())
202
203     def test_get_root_basedn(self):
204         l = ldb.Ldb(self.url(), flags=self.flags())
205         self.assertEqual(None, l.get_root_basedn())
206
207     def test_get_schema_basedn(self):
208         l = ldb.Ldb(self.url(), flags=self.flags())
209         self.assertEqual(None, l.get_schema_basedn())
210
211     def test_get_default_basedn(self):
212         l = ldb.Ldb(self.url(), flags=self.flags())
213         self.assertEqual(None, l.get_default_basedn())
214
215     def test_add(self):
216         l = ldb.Ldb(self.url(), flags=self.flags())
217         m = ldb.Message()
218         m.dn = ldb.Dn(l, "dc=foo4")
219         m["bla"] = b"bla"
220         m["objectUUID"] = b"0123456789abcdef"
221         self.assertEqual(len(l.search()), 0)
222         l.add(m)
223         try:
224             self.assertEqual(len(l.search()), 1)
225         finally:
226             l.delete(ldb.Dn(l, "dc=foo4"))
227
228     def test_search_iterator(self):
229         l = ldb.Ldb(self.url(), flags=self.flags())
230         s = l.search_iterator()
231         s.abandon()
232         try:
233             for me in s:
234                 self.fail()
235             self.fail()
236         except RuntimeError as re:
237             pass
238         try:
239             s.abandon()
240             self.fail()
241         except RuntimeError as re:
242             pass
243         try:
244             s.result()
245             self.fail()
246         except RuntimeError as re:
247             pass
248
249         s = l.search_iterator()
250         count = 0
251         for me in s:
252             self.assertTrue(isinstance(me, ldb.Message))
253             count += 1
254         r = s.result()
255         self.assertEqual(len(r), 0)
256         self.assertEqual(count, 0)
257
258         m1 = ldb.Message()
259         m1.dn = ldb.Dn(l, "dc=foo4")
260         m1["bla"] = b"bla"
261         m1["objectUUID"] = b"0123456789abcdef"
262         l.add(m1)
263         try:
264             s = l.search_iterator()
265             msgs = []
266             for me in s:
267                 self.assertTrue(isinstance(me, ldb.Message))
268                 count += 1
269                 msgs.append(me)
270             r = s.result()
271             self.assertEqual(len(r), 0)
272             self.assertEqual(len(msgs), 1)
273             self.assertEqual(msgs[0].dn, m1.dn)
274
275             m2 = ldb.Message()
276             m2.dn = ldb.Dn(l, "dc=foo5")
277             m2["bla"] = b"bla"
278             m2["objectUUID"] = b"0123456789abcdee"
279             l.add(m2)
280
281             s = l.search_iterator()
282             msgs = []
283             for me in s:
284                 self.assertTrue(isinstance(me, ldb.Message))
285                 count += 1
286                 msgs.append(me)
287             r = s.result()
288             self.assertEqual(len(r), 0)
289             self.assertEqual(len(msgs), 2)
290             if msgs[0].dn == m1.dn:
291                 self.assertEqual(msgs[0].dn, m1.dn)
292                 self.assertEqual(msgs[1].dn, m2.dn)
293             else:
294                 self.assertEqual(msgs[0].dn, m2.dn)
295                 self.assertEqual(msgs[1].dn, m1.dn)
296
297             s = l.search_iterator()
298             msgs = []
299             for me in s:
300                 self.assertTrue(isinstance(me, ldb.Message))
301                 count += 1
302                 msgs.append(me)
303                 break
304             try:
305                 s.result()
306                 self.fail()
307             except RuntimeError as re:
308                 pass
309             for me in s:
310                 self.assertTrue(isinstance(me, ldb.Message))
311                 count += 1
312                 msgs.append(me)
313                 break
314             for me in s:
315                 self.fail()
316
317             r = s.result()
318             self.assertEqual(len(r), 0)
319             self.assertEqual(len(msgs), 2)
320             if msgs[0].dn == m1.dn:
321                 self.assertEqual(msgs[0].dn, m1.dn)
322                 self.assertEqual(msgs[1].dn, m2.dn)
323             else:
324                 self.assertEqual(msgs[0].dn, m2.dn)
325                 self.assertEqual(msgs[1].dn, m1.dn)
326         finally:
327             l.delete(ldb.Dn(l, "dc=foo4"))
328             l.delete(ldb.Dn(l, "dc=foo5"))
329
330     def test_add_text(self):
331         l = ldb.Ldb(self.url(), flags=self.flags())
332         m = ldb.Message()
333         m.dn = ldb.Dn(l, "dc=foo4")
334         m["bla"] = "bla"
335         m["objectUUID"] = b"0123456789abcdef"
336         self.assertEqual(len(l.search()), 0)
337         l.add(m)
338         try:
339             self.assertEqual(len(l.search()), 1)
340         finally:
341             l.delete(ldb.Dn(l, "dc=foo4"))
342
343     def test_add_w_unhandled_ctrl(self):
344         l = ldb.Ldb(self.url(), flags=self.flags())
345         m = ldb.Message()
346         m.dn = ldb.Dn(l, "dc=foo4")
347         m["bla"] = b"bla"
348         self.assertEqual(len(l.search()), 0)
349         self.assertRaises(ldb.LdbError, lambda: l.add(m,["search_options:1:2"]))
350
351     def test_add_dict(self):
352         l = ldb.Ldb(self.url(), flags=self.flags())
353         m = {"dn": ldb.Dn(l, "dc=foo5"),
354              "bla": b"bla",
355              "objectUUID": b"0123456789abcdef"}
356         self.assertEqual(len(l.search()), 0)
357         l.add(m)
358         try:
359             self.assertEqual(len(l.search()), 1)
360         finally:
361             l.delete(ldb.Dn(l, "dc=foo5"))
362
363     def test_add_dict_text(self):
364         l = ldb.Ldb(self.url(), flags=self.flags())
365         m = {"dn": ldb.Dn(l, "dc=foo5"),
366              "bla": "bla",
367              "objectUUID": b"0123456789abcdef"}
368         self.assertEqual(len(l.search()), 0)
369         l.add(m)
370         try:
371             self.assertEqual(len(l.search()), 1)
372         finally:
373             l.delete(ldb.Dn(l, "dc=foo5"))
374
375     def test_add_dict_string_dn(self):
376         l = ldb.Ldb(self.url(), flags=self.flags())
377         m = {"dn": "dc=foo6", "bla": b"bla",
378              "objectUUID": b"0123456789abcdef"}
379         self.assertEqual(len(l.search()), 0)
380         l.add(m)
381         try:
382             self.assertEqual(len(l.search()), 1)
383         finally:
384             l.delete(ldb.Dn(l, "dc=foo6"))
385
386     def test_add_dict_bytes_dn(self):
387         l = ldb.Ldb(self.url(), flags=self.flags())
388         m = {"dn": b"dc=foo6", "bla": b"bla",
389              "objectUUID": b"0123456789abcdef"}
390         self.assertEqual(len(l.search()), 0)
391         l.add(m)
392         try:
393             self.assertEqual(len(l.search()), 1)
394         finally:
395             l.delete(ldb.Dn(l, "dc=foo6"))
396
397     def test_rename(self):
398         l = ldb.Ldb(self.url(), flags=self.flags())
399         m = ldb.Message()
400         m.dn = ldb.Dn(l, "dc=foo7")
401         m["bla"] = b"bla"
402         m["objectUUID"] = b"0123456789abcdef"
403         self.assertEqual(len(l.search()), 0)
404         l.add(m)
405         try:
406             l.rename(ldb.Dn(l, "dc=foo7"), ldb.Dn(l, "dc=bar"))
407             self.assertEqual(len(l.search()), 1)
408         finally:
409             l.delete(ldb.Dn(l, "dc=bar"))
410
411     def test_rename_string_dns(self):
412         l = ldb.Ldb(self.url(), flags=self.flags())
413         m = ldb.Message()
414         m.dn = ldb.Dn(l, "dc=foo8")
415         m["bla"] = b"bla"
416         m["objectUUID"] = b"0123456789abcdef"
417         self.assertEqual(len(l.search()), 0)
418         l.add(m)
419         self.assertEqual(len(l.search()), 1)
420         try:
421             l.rename("dc=foo8", "dc=bar")
422             self.assertEqual(len(l.search()), 1)
423         finally:
424             l.delete(ldb.Dn(l, "dc=bar"))
425
426     def test_rename_bad_string_dns(self):
427         l = ldb.Ldb(self.url(), flags=self.flags())
428         m = ldb.Message()
429         m.dn = ldb.Dn(l, "dc=foo8")
430         m["bla"] = b"bla"
431         m["objectUUID"] = b"0123456789abcdef"
432         self.assertEqual(len(l.search()), 0)
433         l.add(m)
434         self.assertEqual(len(l.search()), 1)
435         self.assertRaises(ldb.LdbError,lambda: l.rename("dcXfoo8", "dc=bar"))
436         self.assertRaises(ldb.LdbError,lambda: l.rename("dc=foo8", "dcXbar"))
437         l.delete(ldb.Dn(l, "dc=foo8"))
438
439     def test_empty_dn(self):
440         l = ldb.Ldb(self.url(), flags=self.flags())
441         self.assertEqual(0, len(l.search()))
442         m = ldb.Message()
443         m.dn = ldb.Dn(l, "dc=empty")
444         m["objectUUID"] = b"0123456789abcdef"
445         l.add(m)
446         rm = l.search()
447         self.assertEqual(1, len(rm))
448         self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
449                          set(rm[0].keys()))
450
451         rm = l.search(m.dn)
452         self.assertEqual(1, len(rm))
453         self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
454                          set(rm[0].keys()))
455         rm = l.search(m.dn, attrs=["blah"])
456         self.assertEqual(1, len(rm))
457         self.assertEqual(0, len(rm[0]))
458
459     def test_modify_delete(self):
460         l = ldb.Ldb(self.url(), flags=self.flags())
461         m = ldb.Message()
462         m.dn = ldb.Dn(l, "dc=modifydelete")
463         m["bla"] = [b"1234"]
464         m["objectUUID"] = b"0123456789abcdef"
465         l.add(m)
466         rm = l.search(m.dn)[0]
467         self.assertEqual([b"1234"], list(rm["bla"]))
468         try:
469             m = ldb.Message()
470             m.dn = ldb.Dn(l, "dc=modifydelete")
471             m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
472             self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
473             l.modify(m)
474             rm = l.search(m.dn)
475             self.assertEqual(1, len(rm))
476             self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
477                              set(rm[0].keys()))
478             rm = l.search(m.dn, attrs=["bla"])
479             self.assertEqual(1, len(rm))
480             self.assertEqual(0, len(rm[0]))
481         finally:
482             l.delete(ldb.Dn(l, "dc=modifydelete"))
483
484     def test_modify_delete_text(self):
485         l = ldb.Ldb(self.url(), flags=self.flags())
486         m = ldb.Message()
487         m.dn = ldb.Dn(l, "dc=modifydelete")
488         m.text["bla"] = ["1234"]
489         m["objectUUID"] = b"0123456789abcdef"
490         l.add(m)
491         rm = l.search(m.dn)[0]
492         self.assertEqual(["1234"], list(rm.text["bla"]))
493         try:
494             m = ldb.Message()
495             m.dn = ldb.Dn(l, "dc=modifydelete")
496             m["bla"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "bla")
497             self.assertEqual(ldb.FLAG_MOD_DELETE, m["bla"].flags())
498             l.modify(m)
499             rm = l.search(m.dn)
500             self.assertEqual(1, len(rm))
501             self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
502                              set(rm[0].keys()))
503             rm = l.search(m.dn, attrs=["bla"])
504             self.assertEqual(1, len(rm))
505             self.assertEqual(0, len(rm[0]))
506         finally:
507             l.delete(ldb.Dn(l, "dc=modifydelete"))
508
509     def test_modify_add(self):
510         l = ldb.Ldb(self.url(), flags=self.flags())
511         m = ldb.Message()
512         m.dn = ldb.Dn(l, "dc=add")
513         m["bla"] = [b"1234"]
514         m["objectUUID"] = b"0123456789abcdef"
515         l.add(m)
516         try:
517             m = ldb.Message()
518             m.dn = ldb.Dn(l, "dc=add")
519             m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
520             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
521             l.modify(m)
522             rm = l.search(m.dn)[0]
523             self.assertEqual(3, len(rm))
524             self.assertEqual([b"1234", b"456"], list(rm["bla"]))
525         finally:
526             l.delete(ldb.Dn(l, "dc=add"))
527
528     def test_modify_add_text(self):
529         l = ldb.Ldb(self.url(), flags=self.flags())
530         m = ldb.Message()
531         m.dn = ldb.Dn(l, "dc=add")
532         m.text["bla"] = ["1234"]
533         m["objectUUID"] = b"0123456789abcdef"
534         l.add(m)
535         try:
536             m = ldb.Message()
537             m.dn = ldb.Dn(l, "dc=add")
538             m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
539             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
540             l.modify(m)
541             rm = l.search(m.dn)[0]
542             self.assertEqual(3, len(rm))
543             self.assertEqual(["1234", "456"], list(rm.text["bla"]))
544         finally:
545             l.delete(ldb.Dn(l, "dc=add"))
546
547     def test_modify_replace(self):
548         l = ldb.Ldb(self.url(), flags=self.flags())
549         m = ldb.Message()
550         m.dn = ldb.Dn(l, "dc=modify2")
551         m["bla"] = [b"1234", b"456"]
552         m["objectUUID"] = b"0123456789abcdef"
553         l.add(m)
554         try:
555             m = ldb.Message()
556             m.dn = ldb.Dn(l, "dc=modify2")
557             m["bla"] = ldb.MessageElement([b"789"], ldb.FLAG_MOD_REPLACE, "bla")
558             self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
559             l.modify(m)
560             rm = l.search(m.dn)[0]
561             self.assertEqual(3, len(rm))
562             self.assertEqual([b"789"], list(rm["bla"]))
563             rm = l.search(m.dn, attrs=["bla"])[0]
564             self.assertEqual(1, len(rm))
565         finally:
566             l.delete(ldb.Dn(l, "dc=modify2"))
567
568     def test_modify_replace_text(self):
569         l = ldb.Ldb(self.url(), flags=self.flags())
570         m = ldb.Message()
571         m.dn = ldb.Dn(l, "dc=modify2")
572         m.text["bla"] = ["1234", "456"]
573         m["objectUUID"] = b"0123456789abcdef"
574         l.add(m)
575         try:
576             m = ldb.Message()
577             m.dn = ldb.Dn(l, "dc=modify2")
578             m["bla"] = ldb.MessageElement(["789"], ldb.FLAG_MOD_REPLACE, "bla")
579             self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
580             l.modify(m)
581             rm = l.search(m.dn)[0]
582             self.assertEqual(3, len(rm))
583             self.assertEqual(["789"], list(rm.text["bla"]))
584             rm = l.search(m.dn, attrs=["bla"])[0]
585             self.assertEqual(1, len(rm))
586         finally:
587             l.delete(ldb.Dn(l, "dc=modify2"))
588
589     def test_modify_flags_change(self):
590         l = ldb.Ldb(self.url(), flags=self.flags())
591         m = ldb.Message()
592         m.dn = ldb.Dn(l, "dc=add")
593         m["bla"] = [b"1234"]
594         m["objectUUID"] = b"0123456789abcdef"
595         l.add(m)
596         try:
597             m = ldb.Message()
598             m.dn = ldb.Dn(l, "dc=add")
599             m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
600             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
601             l.modify(m)
602             rm = l.search(m.dn)[0]
603             self.assertEqual(3, len(rm))
604             self.assertEqual([b"1234", b"456"], list(rm["bla"]))
605
606             # Now create another modify, but switch the flags before we do it
607             m["bla"] = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
608             m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
609             l.modify(m)
610             rm = l.search(m.dn, attrs=["bla"])[0]
611             self.assertEqual(1, len(rm))
612             self.assertEqual([b"1234"], list(rm["bla"]))
613         finally:
614             l.delete(ldb.Dn(l, "dc=add"))
615
616     def test_modify_flags_change_text(self):
617         l = ldb.Ldb(self.url(), flags=self.flags())
618         m = ldb.Message()
619         m.dn = ldb.Dn(l, "dc=add")
620         m.text["bla"] = ["1234"]
621         m["objectUUID"] = b"0123456789abcdef"
622         l.add(m)
623         try:
624             m = ldb.Message()
625             m.dn = ldb.Dn(l, "dc=add")
626             m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
627             self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
628             l.modify(m)
629             rm = l.search(m.dn)[0]
630             self.assertEqual(3, len(rm))
631             self.assertEqual(["1234", "456"], list(rm.text["bla"]))
632
633             # Now create another modify, but switch the flags before we do it
634             m["bla"] = ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla")
635             m["bla"].set_flags(ldb.FLAG_MOD_DELETE)
636             l.modify(m)
637             rm = l.search(m.dn, attrs=["bla"])[0]
638             self.assertEqual(1, len(rm))
639             self.assertEqual(["1234"], list(rm.text["bla"]))
640         finally:
641             l.delete(ldb.Dn(l, "dc=add"))
642
643     def test_transaction_commit(self):
644         l = ldb.Ldb(self.url(), flags=self.flags())
645         l.transaction_start()
646         m = ldb.Message(ldb.Dn(l, "dc=foo9"))
647         m["foo"] = [b"bar"]
648         m["objectUUID"] = b"0123456789abcdef"
649         l.add(m)
650         l.transaction_commit()
651         l.delete(m.dn)
652
653     def test_transaction_cancel(self):
654         l = ldb.Ldb(self.url(), flags=self.flags())
655         l.transaction_start()
656         m = ldb.Message(ldb.Dn(l, "dc=foo10"))
657         m["foo"] = [b"bar"]
658         m["objectUUID"] = b"0123456789abcdee"
659         l.add(m)
660         l.transaction_cancel()
661         self.assertEqual(0, len(l.search(ldb.Dn(l, "dc=foo10"))))
662
663     def test_set_debug(self):
664         def my_report_fn(level, text):
665             pass
666         l = ldb.Ldb(self.url(), flags=self.flags())
667         l.set_debug(my_report_fn)
668
669     def test_zero_byte_string(self):
670         """Testing we do not get trapped in the \0 byte in a property string."""
671         l = ldb.Ldb(self.url(), flags=self.flags())
672         l.add({
673             "dn": b"dc=somedn",
674             "objectclass": b"user",
675             "cN": b"LDAPtestUSER",
676             "givenname": b"ldap",
677             "displayname": b"foo\0bar",
678             "objectUUID": b"0123456789abcdef"
679         })
680         res = l.search(expression="(dn=dc=somedn)")
681         self.assertEqual(b"foo\0bar", res[0]["displayname"][0])
682
683     def test_no_crash_broken_expr(self):
684         l = ldb.Ldb(self.url(), flags=self.flags())
685         self.assertRaises(ldb.LdbError,lambda: l.search("", ldb.SCOPE_SUBTREE, "&(dc=*)(dn=*)", ["dc"]))
686
687 # Run the SimpleLdb tests against an lmdb backend
688 class SimpleLdbLmdb(SimpleLdb):
689
690     def setUp(self):
691         self.prefix = MDB_PREFIX
692         self.index = MDB_INDEX_OBJ
693         super(SimpleLdbLmdb, self).setUp()
694
695     def tearDown(self):
696         super(SimpleLdbLmdb, self).tearDown()
697
698 class SearchTests(LdbBaseTest):
699     def tearDown(self):
700         shutil.rmtree(self.testdir)
701         super(SearchTests, self).tearDown()
702
703         # Ensure the LDB is closed now, so we close the FD
704         del(self.l)
705
706
707     def setUp(self):
708         super(SearchTests, self).setUp()
709         self.testdir = tempdir()
710         self.filename = os.path.join(self.testdir, "search_test.ldb")
711         options = ["modules:rdn_name"]
712         if hasattr(self, 'IDXCHECK'):
713             options.append("disable_full_db_scan_for_self_test:1")
714         self.l = ldb.Ldb(self.url(),
715                          flags=self.flags(),
716                          options=options)
717         try:
718             self.l.add(self.index)
719         except AttributeError:
720             pass
721
722         self.l.add({"dn": "@ATTRIBUTES",
723                     "DC": "CASE_INSENSITIVE"})
724
725         # Note that we can't use the name objectGUID here, as we
726         # want to stay clear of the objectGUID handler in LDB and
727         # instead use just the 16 bytes raw, which we just keep
728         # to printable chars here for ease of handling.
729
730         self.l.add({"dn": "DC=SAMBA,DC=ORG",
731                     "name": b"samba.org",
732                     "objectUUID": b"0123456789abcdef"})
733         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
734                     "name": b"Admins",
735                     "x": "z", "y": "a",
736                     "objectUUID": b"0123456789abcde1"})
737         self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG",
738                     "name": b"Users",
739                     "x": "z", "y": "a",
740                     "objectUUID": b"0123456789abcde2"})
741         self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG",
742                     "name": b"OU #1",
743                     "x": "y", "y": "a",
744                     "objectUUID": b"0123456789abcde3"})
745         self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG",
746                     "name": b"OU #2",
747                     "x": "y", "y": "a",
748                     "objectUUID": b"0123456789abcde4"})
749         self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG",
750                     "name": b"OU #3",
751                     "x": "y", "y": "a",
752                     "objectUUID": b"0123456789abcde5"})
753         self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG",
754                     "name": b"OU #4",
755                     "x": "y", "y": "a",
756                     "objectUUID": b"0123456789abcde6"})
757         self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG",
758                     "name": b"OU #5",
759                     "x": "y", "y": "a",
760                     "objectUUID": b"0123456789abcde7"})
761         self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG",
762                     "name": b"OU #6",
763                     "x": "y", "y": "a",
764                     "objectUUID": b"0123456789abcde8"})
765         self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG",
766                     "name": b"OU #7",
767                     "x": "y", "y": "a",
768                     "objectUUID": b"0123456789abcde9"})
769         self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG",
770                     "name": b"OU #8",
771                     "x": "y", "y": "a",
772                     "objectUUID": b"0123456789abcde0"})
773         self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG",
774                     "name": b"OU #9",
775                     "x": "y", "y": "a",
776                     "objectUUID": b"0123456789abcdea"})
777         self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG",
778                     "name": b"OU #10",
779                     "x": "y", "y": "a",
780                     "objectUUID": b"0123456789abcdeb"})
781         self.l.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
782                     "name": b"OU #10",
783                     "x": "y", "y": "a",
784                     "objectUUID": b"0123456789abcdec"})
785         self.l.add({"dn": "OU=OU12,DC=SAMBA,DC=ORG",
786                     "name": b"OU #10",
787                     "x": "y", "y": "b",
788                     "objectUUID": b"0123456789abcded"})
789         self.l.add({"dn": "OU=OU13,DC=SAMBA,DC=ORG",
790                     "name": b"OU #10",
791                     "x": "x", "y": "b",
792                     "objectUUID": b"0123456789abcdee"})
793         self.l.add({"dn": "OU=OU14,DC=SAMBA,DC=ORG",
794                     "name": b"OU #10",
795                     "x": "x", "y": "b",
796                     "objectUUID": b"0123456789abcd01"})
797         self.l.add({"dn": "OU=OU15,DC=SAMBA,DC=ORG",
798                     "name": b"OU #10",
799                     "x": "x", "y": "b",
800                     "objectUUID": b"0123456789abcd02"})
801         self.l.add({"dn": "OU=OU16,DC=SAMBA,DC=ORG",
802                     "name": b"OU #10",
803                     "x": "x", "y": "b",
804                     "objectUUID": b"0123456789abcd03"})
805         self.l.add({"dn": "OU=OU17,DC=SAMBA,DC=ORG",
806                     "name": b"OU #10",
807                     "x": "x", "y": "b",
808                     "objectUUID": b"0123456789abcd04"})
809         self.l.add({"dn": "OU=OU18,DC=SAMBA,DC=ORG",
810                     "name": b"OU #10",
811                     "x": "x", "y": "b",
812                     "objectUUID": b"0123456789abcd05"})
813         self.l.add({"dn": "OU=OU19,DC=SAMBA,DC=ORG",
814                     "name": b"OU #10",
815                     "x": "x", "y": "b",
816                     "objectUUID": b"0123456789abcd06"})
817         self.l.add({"dn": "OU=OU20,DC=SAMBA,DC=ORG",
818                     "name": b"OU #10",
819                     "x": "x", "y": "b",
820                     "objectUUID": b"0123456789abcd07"})
821         self.l.add({"dn": "OU=OU21,DC=SAMBA,DC=ORG",
822                     "name": b"OU #10",
823                     "x": "x", "y": "c",
824                     "objectUUID": b"0123456789abcd08"})
825         self.l.add({"dn": "OU=OU22,DC=SAMBA,DC=ORG",
826                     "name": b"OU #10",
827                     "x": "x", "y": "c",
828                     "objectUUID": b"0123456789abcd09"})
829
830     def test_base(self):
831         """Testing a search"""
832
833         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
834                               scope=ldb.SCOPE_BASE)
835         self.assertEqual(len(res11), 1)
836
837     def test_base_lower(self):
838         """Testing a search"""
839
840         res11 = self.l.search(base="OU=OU11,DC=samba,DC=org",
841                               scope=ldb.SCOPE_BASE)
842         self.assertEqual(len(res11), 1)
843
844     def test_base_or(self):
845         """Testing a search"""
846
847         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
848                               scope=ldb.SCOPE_BASE,
849                               expression="(|(ou=ou11)(ou=ou12))")
850         self.assertEqual(len(res11), 1)
851
852     def test_base_or2(self):
853         """Testing a search"""
854
855         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
856                               scope=ldb.SCOPE_BASE,
857                               expression="(|(x=y)(y=b))")
858         self.assertEqual(len(res11), 1)
859
860     def test_base_and(self):
861         """Testing a search"""
862
863         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
864                               scope=ldb.SCOPE_BASE,
865                               expression="(&(ou=ou11)(ou=ou12))")
866         self.assertEqual(len(res11), 0)
867
868     def test_base_and2(self):
869         """Testing a search"""
870
871         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
872                               scope=ldb.SCOPE_BASE,
873                               expression="(&(x=y)(y=a))")
874         self.assertEqual(len(res11), 1)
875
876     def test_base_false(self):
877         """Testing a search"""
878
879         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
880                               scope=ldb.SCOPE_BASE,
881                               expression="(|(ou=ou13)(ou=ou12))")
882         self.assertEqual(len(res11), 0)
883
884     def test_check_base_false(self):
885         """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)
890
891     def test_check_base_error(self):
892         """Testing a search"""
893         checkbaseonsearch = {"dn": "@OPTIONS",
894                              "checkBaseOnSearch": b"TRUE"}
895         try:
896             self.l.add(checkbaseonsearch)
897         except ldb.LdbError as err:
898             enum = err.args[0]
899             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
900             m = ldb.Message.from_dict(self.l,
901                                       checkbaseonsearch)
902             self.l.modify(m)
903
904         try:
905             res11 = self.l.search(base="OU=OU11x,DC=SAMBA,DC=ORG",
906                                   scope=ldb.SCOPE_BASE,
907                                   expression="(|(ou=ou13)(ou=ou12))")
908             self.fail("Should have failed on missing base")
909         except ldb.LdbError as err:
910             enum = err.args[0]
911             self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
912
913     def test_subtree_and(self):
914         """Testing a search"""
915
916         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
917                               scope=ldb.SCOPE_SUBTREE,
918                               expression="(&(ou=ou11)(ou=ou12))")
919         self.assertEqual(len(res11), 0)
920
921     def test_subtree_and2(self):
922         """Testing a search"""
923
924         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
925                               scope=ldb.SCOPE_SUBTREE,
926                               expression="(&(x=y)(|(y=b)(y=c)))")
927         self.assertEqual(len(res11), 1)
928
929     def test_subtree_and2_lower(self):
930         """Testing a search"""
931
932         res11 = self.l.search(base="DC=samba,DC=org",
933                               scope=ldb.SCOPE_SUBTREE,
934                               expression="(&(x=y)(|(y=b)(y=c)))")
935         self.assertEqual(len(res11), 1)
936
937     def test_subtree_or(self):
938         """Testing a search"""
939
940         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
941                               scope=ldb.SCOPE_SUBTREE,
942                               expression="(|(ou=ou11)(ou=ou12))")
943         self.assertEqual(len(res11), 2)
944
945     def test_subtree_or2(self):
946         """Testing a search"""
947
948         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
949                               scope=ldb.SCOPE_SUBTREE,
950                               expression="(|(x=y)(y=b))")
951         self.assertEqual(len(res11), 20)
952
953     def test_subtree_or3(self):
954         """Testing a search"""
955
956         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
957                               scope=ldb.SCOPE_SUBTREE,
958                               expression="(|(x=y)(y=b)(y=c))")
959         self.assertEqual(len(res11), 22)
960
961     def test_one_and(self):
962         """Testing a search"""
963
964         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
965                               scope=ldb.SCOPE_ONELEVEL,
966                               expression="(&(ou=ou11)(ou=ou12))")
967         self.assertEqual(len(res11), 0)
968
969     def test_one_and2(self):
970         """Testing a search"""
971
972         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
973                               scope=ldb.SCOPE_ONELEVEL,
974                               expression="(&(x=y)(y=b))")
975         self.assertEqual(len(res11), 1)
976
977     def test_one_or(self):
978         """Testing a search"""
979
980         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
981                               scope=ldb.SCOPE_ONELEVEL,
982                               expression="(|(ou=ou11)(ou=ou12))")
983         self.assertEqual(len(res11), 2)
984
985     def test_one_or2(self):
986         """Testing a search"""
987
988         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
989                               scope=ldb.SCOPE_ONELEVEL,
990                               expression="(|(x=y)(y=b))")
991         self.assertEqual(len(res11), 20)
992
993     def test_one_or2_lower(self):
994         """Testing a search"""
995
996         res11 = self.l.search(base="DC=samba,DC=org",
997                               scope=ldb.SCOPE_ONELEVEL,
998                               expression="(|(x=y)(y=b))")
999         self.assertEqual(len(res11), 20)
1000
1001     def test_one_unindexable(self):
1002         """Testing a search"""
1003
1004         try:
1005             res11 = self.l.search(base="DC=samba,DC=org",
1006                                   scope=ldb.SCOPE_ONELEVEL,
1007                                   expression="(y=b*)")
1008             if hasattr(self, 'IDX') and \
1009                not hasattr(self, 'IDXONE') and \
1010                hasattr(self, 'IDXCHECK'):
1011                 self.fail("Should have failed as un-indexed search")
1012
1013             self.assertEqual(len(res11), 9)
1014
1015         except ldb.LdbError as err:
1016             enum = err.args[0]
1017             estr = err.args[1]
1018             self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1019             self.assertIn(estr, "ldb FULL SEARCH disabled")
1020
1021     def test_one_unindexable_presence(self):
1022         """Testing a search"""
1023
1024         try:
1025             res11 = self.l.search(base="DC=samba,DC=org",
1026                                   scope=ldb.SCOPE_ONELEVEL,
1027                                   expression="(y=*)")
1028             if hasattr(self, 'IDX') and \
1029                not hasattr(self, 'IDXONE') and \
1030                hasattr(self, 'IDXCHECK'):
1031                 self.fail("Should have failed as un-indexed search")
1032
1033             self.assertEqual(len(res11), 24)
1034
1035         except ldb.LdbError as err:
1036             enum = err.args[0]
1037             estr = err.args[1]
1038             self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1039             self.assertIn(estr, "ldb FULL SEARCH disabled")
1040
1041
1042     def test_subtree_and_or(self):
1043         """Testing a search"""
1044
1045         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1046                               scope=ldb.SCOPE_SUBTREE,
1047                               expression="(&(|(x=z)(y=b))(x=x)(y=c))")
1048         self.assertEqual(len(res11), 0)
1049
1050     def test_subtree_and_or2(self):
1051         """Testing a search"""
1052
1053         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1054                               scope=ldb.SCOPE_SUBTREE,
1055                               expression="(&(x=x)(y=c)(|(x=z)(y=b)))")
1056         self.assertEqual(len(res11), 0)
1057
1058     def test_subtree_and_or3(self):
1059         """Testing a search"""
1060
1061         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1062                               scope=ldb.SCOPE_SUBTREE,
1063                               expression="(&(|(ou=ou11)(ou=ou10))(|(x=y)(y=b)(y=c)))")
1064         self.assertEqual(len(res11), 2)
1065
1066     def test_subtree_and_or4(self):
1067         """Testing a search"""
1068
1069         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1070                               scope=ldb.SCOPE_SUBTREE,
1071                               expression="(&(|(x=y)(y=b)(y=c))(|(ou=ou11)(ou=ou10)))")
1072         self.assertEqual(len(res11), 2)
1073
1074     def test_subtree_and_or5(self):
1075         """Testing a search"""
1076
1077         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1078                               scope=ldb.SCOPE_SUBTREE,
1079                               expression="(&(|(x=y)(y=b)(y=c))(ou=ou11))")
1080         self.assertEqual(len(res11), 1)
1081
1082     def test_subtree_or_and(self):
1083         """Testing a search"""
1084
1085         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1086                               scope=ldb.SCOPE_SUBTREE,
1087                               expression="(|(x=x)(y=c)(&(x=z)(y=b)))")
1088         self.assertEqual(len(res11), 10)
1089
1090     def test_subtree_large_and_unique(self):
1091         """Testing a search"""
1092
1093         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1094                               scope=ldb.SCOPE_SUBTREE,
1095                               expression="(&(ou=ou10)(y=a))")
1096         self.assertEqual(len(res11), 1)
1097
1098     def test_subtree_and_none(self):
1099         """Testing a search"""
1100
1101         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1102                               scope=ldb.SCOPE_SUBTREE,
1103                               expression="(&(ou=ouX)(y=a))")
1104         self.assertEqual(len(res11), 0)
1105
1106     def test_subtree_and_idx_record(self):
1107         """Testing a search against the index record"""
1108
1109         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1110                               scope=ldb.SCOPE_SUBTREE,
1111                               expression="(@IDXDN=DC=SAMBA,DC=ORG)")
1112         self.assertEqual(len(res11), 0)
1113
1114     def test_subtree_and_idxone_record(self):
1115         """Testing a search against the index record"""
1116
1117         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1118                               scope=ldb.SCOPE_SUBTREE,
1119                               expression="(@IDXONE=DC=SAMBA,DC=ORG)")
1120         self.assertEqual(len(res11), 0)
1121
1122     def test_subtree_unindexable(self):
1123         """Testing a search"""
1124
1125         try:
1126             res11 = self.l.search(base="DC=samba,DC=org",
1127                                   scope=ldb.SCOPE_SUBTREE,
1128                                   expression="(y=b*)")
1129             if hasattr(self, 'IDX') and \
1130                hasattr(self, 'IDXCHECK'):
1131                 self.fail("Should have failed as un-indexed search")
1132
1133             self.assertEqual(len(res11), 9)
1134
1135         except ldb.LdbError as err:
1136             enum = err.args[0]
1137             estr = err.args[1]
1138             self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1139             self.assertIn(estr, "ldb FULL SEARCH disabled")
1140
1141     def test_subtree_unindexable_presence(self):
1142         """Testing a search"""
1143
1144         try:
1145             res11 = self.l.search(base="DC=samba,DC=org",
1146                                   scope=ldb.SCOPE_SUBTREE,
1147                                   expression="(y=*)")
1148             if hasattr(self, 'IDX') and \
1149                hasattr(self, 'IDXCHECK'):
1150                 self.fail("Should have failed as un-indexed search")
1151
1152             self.assertEqual(len(res11), 24)
1153
1154         except ldb.LdbError as err:
1155             enum = err.args[0]
1156             estr = err.args[1]
1157             self.assertEqual(enum, ldb.ERR_INAPPROPRIATE_MATCHING)
1158             self.assertIn(estr, "ldb FULL SEARCH disabled")
1159
1160
1161     def test_dn_filter_one(self):
1162         """Testing that a dn= filter succeeds
1163         (or fails with disallowDNFilter
1164         set and IDXGUID or (IDX and not IDXONE) mode)
1165         when the scope is SCOPE_ONELEVEL.
1166
1167         This should be made more consistent, but for now lock in
1168         the behaviour
1169
1170         """
1171
1172         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1173                               scope=ldb.SCOPE_ONELEVEL,
1174                               expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1175         if hasattr(self, 'disallowDNFilter') and \
1176            hasattr(self, 'IDX') and \
1177            (hasattr(self, 'IDXGUID') or \
1178             ((hasattr(self, 'IDXONE') == False and hasattr(self, 'IDX')))):
1179             self.assertEqual(len(res11), 0)
1180         else:
1181             self.assertEqual(len(res11), 1)
1182
1183     def test_dn_filter_subtree(self):
1184         """Testing that a dn= filter succeeds
1185         (or fails with disallowDNFilter set)
1186         when the scope is SCOPE_SUBTREE"""
1187
1188         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1189                               scope=ldb.SCOPE_SUBTREE,
1190                               expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1191         if hasattr(self, 'disallowDNFilter') \
1192            and hasattr(self, 'IDX'):
1193             self.assertEqual(len(res11), 0)
1194         else:
1195             self.assertEqual(len(res11), 1)
1196
1197     def test_dn_filter_base(self):
1198         """Testing that (incorrectly) a dn= filter works
1199         when the scope is SCOPE_BASE"""
1200
1201         res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1202                               scope=ldb.SCOPE_BASE,
1203                               expression="(dn=OU=OU1,DC=SAMBA,DC=ORG)")
1204
1205         # At some point we should fix this, but it isn't trivial
1206         self.assertEqual(len(res11), 1)
1207
1208     def test_distinguishedName_filter_one(self):
1209         """Testing that a distinguishedName= filter succeeds
1210         when the scope is SCOPE_ONELEVEL.
1211
1212         This should be made more consistent, but for now lock in
1213         the behaviour
1214
1215         """
1216
1217         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1218                               scope=ldb.SCOPE_ONELEVEL,
1219                               expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1220         self.assertEqual(len(res11), 1)
1221
1222     def test_distinguishedName_filter_subtree(self):
1223         """Testing that a distinguishedName= filter succeeds
1224         when the scope is SCOPE_SUBTREE"""
1225
1226         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1227                               scope=ldb.SCOPE_SUBTREE,
1228                               expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1229         self.assertEqual(len(res11), 1)
1230
1231     def test_distinguishedName_filter_base(self):
1232         """Testing that (incorrectly) a distinguishedName= filter works
1233         when the scope is SCOPE_BASE"""
1234
1235         res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1236                               scope=ldb.SCOPE_BASE,
1237                               expression="(distinguishedName=OU=OU1,DC=SAMBA,DC=ORG)")
1238
1239         # At some point we should fix this, but it isn't trivial
1240         self.assertEqual(len(res11), 1)
1241
1242     def test_bad_dn_filter_base(self):
1243         """Testing that a dn= filter on an invalid DN works
1244         when the scope is SCOPE_BASE but
1245         returns zero results"""
1246
1247         res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1248                               scope=ldb.SCOPE_BASE,
1249                               expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1250
1251         # At some point we should fix this, but it isn't trivial
1252         self.assertEqual(len(res11), 0)
1253
1254
1255     def test_bad_dn_filter_one(self):
1256         """Testing that a dn= filter succeeds but returns zero
1257         results when the DN is not valid on a SCOPE_ONELEVEL search
1258
1259         """
1260
1261         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1262                               scope=ldb.SCOPE_ONELEVEL,
1263                               expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1264         self.assertEqual(len(res11), 0)
1265
1266     def test_bad_dn_filter_subtree(self):
1267         """Testing that a dn= filter succeeds but returns zero
1268         results when the DN is not valid on a SCOPE_SUBTREE search
1269
1270         """
1271
1272         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1273                               scope=ldb.SCOPE_SUBTREE,
1274                               expression="(dn=OU=OU1,DC=SAMBA,DCXXXX)")
1275         self.assertEqual(len(res11), 0)
1276
1277     def test_bad_distinguishedName_filter_base(self):
1278         """Testing that a distinguishedName= filter on an invalid DN works
1279         when the scope is SCOPE_BASE but
1280         returns zero results"""
1281
1282         res11 = self.l.search(base="OU=OU1,DC=SAMBA,DC=ORG",
1283                               scope=ldb.SCOPE_BASE,
1284                               expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1285
1286         # At some point we should fix this, but it isn't trivial
1287         self.assertEqual(len(res11), 0)
1288
1289
1290     def test_bad_distinguishedName_filter_one(self):
1291         """Testing that a distinguishedName= filter succeeds but returns zero
1292         results when the DN is not valid on a SCOPE_ONELEVEL search
1293
1294         """
1295
1296         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1297                               scope=ldb.SCOPE_ONELEVEL,
1298                               expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1299         self.assertEqual(len(res11), 0)
1300
1301     def test_bad_distinguishedName_filter_subtree(self):
1302         """Testing that a distinguishedName= filter succeeds but returns zero
1303         results when the DN is not valid on a SCOPE_SUBTREE search
1304
1305         """
1306
1307         res11 = self.l.search(base="DC=SAMBA,DC=ORG",
1308                               scope=ldb.SCOPE_SUBTREE,
1309                               expression="(distinguishedName=OU=OU1,DC=SAMBA,DCXXXX)")
1310         self.assertEqual(len(res11), 0)
1311
1312     def test_bad_dn_search_base(self):
1313         """Testing with a bad base DN (SCOPE_BASE)"""
1314
1315         try:
1316             res11 = self.l.search(base="OU=OU1,DC=SAMBA,DCXXX",
1317                                   scope=ldb.SCOPE_BASE)
1318             self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1319         except ldb.LdbError as err:
1320             enum = err.args[0]
1321             self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1322
1323
1324     def test_bad_dn_search_one(self):
1325         """Testing with a bad base DN (SCOPE_ONELEVEL)"""
1326
1327         try:
1328             res11 = self.l.search(base="DC=SAMBA,DCXXXX",
1329                               scope=ldb.SCOPE_ONELEVEL)
1330             self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1331         except ldb.LdbError as err:
1332             enum = err.args[0]
1333             self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1334
1335     def test_bad_dn_search_subtree(self):
1336         """Testing with a bad base DN (SCOPE_SUBTREE)"""
1337
1338         try:
1339             res11 = self.l.search(base="DC=SAMBA,DCXXXX",
1340                                   scope=ldb.SCOPE_SUBTREE)
1341             self.fail("Should have failed with ERR_INVALID_DN_SYNTAX")
1342         except ldb.LdbError as err:
1343             enum = err.args[0]
1344             self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1345
1346
1347
1348 # Run the search tests against an lmdb backend
1349 class SearchTestsLmdb(SearchTests):
1350
1351     def setUp(self):
1352         self.prefix = MDB_PREFIX
1353         self.index = MDB_INDEX_OBJ
1354         super(SearchTestsLmdb, self).setUp()
1355
1356     def tearDown(self):
1357         super(SearchTestsLmdb, self).tearDown()
1358
1359
1360 class IndexedSearchTests(SearchTests):
1361     """Test searches using the index, to ensure the index doesn't
1362        break things"""
1363     def setUp(self):
1364         super(IndexedSearchTests, self).setUp()
1365         self.l.add({"dn": "@INDEXLIST",
1366                     "@IDXATTR": [b"x", b"y", b"ou"]})
1367         self.IDX = True
1368
1369 class IndexedCheckSearchTests(IndexedSearchTests):
1370     """Test searches using the index, to ensure the index doesn't
1371        break things (full scan disabled)"""
1372     def setUp(self):
1373         self.IDXCHECK = True
1374         super(IndexedCheckSearchTests, self).setUp()
1375
1376 class IndexedSearchDnFilterTests(SearchTests):
1377     """Test searches using the index, to ensure the index doesn't
1378        break things"""
1379     def setUp(self):
1380         super(IndexedSearchDnFilterTests, self).setUp()
1381         self.l.add({"dn": "@OPTIONS",
1382                     "disallowDNFilter": "TRUE"})
1383         self.disallowDNFilter = True
1384
1385         self.l.add({"dn": "@INDEXLIST",
1386                     "@IDXATTR": [b"x", b"y", b"ou"]})
1387         self.IDX = True
1388
1389 class IndexedAndOneLevelSearchTests(SearchTests):
1390     """Test searches using the index including @IDXONE, to ensure
1391        the index doesn't break things"""
1392     def setUp(self):
1393         super(IndexedAndOneLevelSearchTests, self).setUp()
1394         self.l.add({"dn": "@INDEXLIST",
1395                     "@IDXATTR": [b"x", b"y", b"ou"],
1396                     "@IDXONE": [b"1"]})
1397         self.IDX = True
1398         self.IDXONE = True
1399
1400 class IndexedCheckedAndOneLevelSearchTests(IndexedAndOneLevelSearchTests):
1401     """Test searches using the index including @IDXONE, to ensure
1402        the index doesn't break things (full scan disabled)"""
1403     def setUp(self):
1404         self.IDXCHECK = True
1405         super(IndexedCheckedAndOneLevelSearchTests, self).setUp()
1406
1407 class IndexedAndOneLevelDNFilterSearchTests(SearchTests):
1408     """Test searches using the index including @IDXONE, to ensure
1409        the index doesn't break things"""
1410     def setUp(self):
1411         super(IndexedAndOneLevelDNFilterSearchTests, self).setUp()
1412         self.l.add({"dn": "@OPTIONS",
1413                     "disallowDNFilter": "TRUE",
1414                     "checkBaseOnSearch": "TRUE"})
1415         self.disallowDNFilter = True
1416         self.checkBaseOnSearch = True
1417
1418         self.l.add({"dn": "@INDEXLIST",
1419                     "@IDXATTR": [b"x", b"y", b"ou"],
1420                     "@IDXONE": [b"1"]})
1421         self.IDX = True
1422         self.IDXONE = True
1423
1424 class GUIDIndexedSearchTests(SearchTests):
1425     """Test searches using the index, to ensure the index doesn't
1426        break things"""
1427     def setUp(self):
1428         self.index = {"dn": "@INDEXLIST",
1429                       "@IDXATTR": [b"x", b"y", b"ou"],
1430                       "@IDXGUID": [b"objectUUID"],
1431                       "@IDX_DN_GUID": [b"GUID"]}
1432         super(GUIDIndexedSearchTests, self).setUp()
1433
1434         self.IDXGUID = True
1435         self.IDXONE = True
1436
1437
1438 class GUIDIndexedDNFilterSearchTests(SearchTests):
1439     """Test searches using the index, to ensure the index doesn't
1440        break things"""
1441     def setUp(self):
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(GUIDIndexedDNFilterSearchTests, self).setUp()
1447         self.l.add({"dn": "@OPTIONS",
1448                     "disallowDNFilter": "TRUE",
1449                     "checkBaseOnSearch": "TRUE"})
1450         self.disallowDNFilter = True
1451         self.checkBaseOnSearch = True
1452         self.IDX = True
1453         self.IDXGUID = True
1454
1455 class GUIDAndOneLevelIndexedSearchTests(SearchTests):
1456     """Test searches using the index including @IDXONE, to ensure
1457        the index doesn't break things"""
1458     def setUp(self):
1459         self.index = {"dn": "@INDEXLIST",
1460                       "@IDXATTR": [b"x", b"y", b"ou"],
1461                       "@IDXGUID": [b"objectUUID"],
1462                       "@IDX_DN_GUID": [b"GUID"]}
1463         super(GUIDAndOneLevelIndexedSearchTests, self).setUp()
1464         self.l.add({"dn": "@OPTIONS",
1465                     "disallowDNFilter": "TRUE",
1466                     "checkBaseOnSearch": "TRUE"})
1467         self.disallowDNFilter = True
1468         self.checkBaseOnSearch = True
1469         self.IDX = True
1470         self.IDXGUID = True
1471         self.IDXONE = True
1472
1473 class GUIDIndexedSearchTestsLmdb(GUIDIndexedSearchTests):
1474
1475     def setUp(self):
1476         self.prefix = MDB_PREFIX
1477         super(GUIDIndexedSearchTestsLmdb, self).setUp()
1478
1479     def tearDown(self):
1480         super(GUIDIndexedSearchTestsLmdb, self).tearDown()
1481
1482
1483 class GUIDIndexedDNFilterSearchTestsLmdb(GUIDIndexedDNFilterSearchTests):
1484
1485     def setUp(self):
1486         self.prefix = MDB_PREFIX
1487         super(GUIDIndexedDNFilterSearchTestsLmdb, self).setUp()
1488
1489     def tearDown(self):
1490         super(GUIDIndexedDNFilterSearchTestsLmdb, self).tearDown()
1491
1492
1493 class GUIDAndOneLevelIndexedSearchTestsLmdb(GUIDAndOneLevelIndexedSearchTests):
1494
1495     def setUp(self):
1496         self.prefix = MDB_PREFIX
1497         super(GUIDAndOneLevelIndexedSearchTestsLmdb, self).setUp()
1498
1499     def tearDown(self):
1500         super(GUIDAndOneLevelIndexedSearchTestsLmdb, self).tearDown()
1501
1502
1503 class AddModifyTests(LdbBaseTest):
1504     def tearDown(self):
1505         shutil.rmtree(self.testdir)
1506         super(AddModifyTests, self).tearDown()
1507
1508         # Ensure the LDB is closed now, so we close the FD
1509         del(self.l)
1510
1511     def setUp(self):
1512         super(AddModifyTests, self).setUp()
1513         self.testdir = tempdir()
1514         self.filename = os.path.join(self.testdir, "add_test.ldb")
1515         self.l = ldb.Ldb(self.url(),
1516                          flags=self.flags(),
1517                          options=["modules:rdn_name"])
1518         try:
1519             self.l.add(self.index)
1520         except AttributeError:
1521             pass
1522
1523         self.l.add({"dn": "DC=SAMBA,DC=ORG",
1524                     "name": b"samba.org",
1525                     "objectUUID": b"0123456789abcdef"})
1526         self.l.add({"dn": "@ATTRIBUTES",
1527                     "objectUUID": "UNIQUE_INDEX"})
1528
1529     def test_add_dup(self):
1530         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1531                     "name": b"Admins",
1532                     "x": "z", "y": "a",
1533                     "objectUUID": b"0123456789abcde1"})
1534         try:
1535             self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1536                         "name": b"Admins",
1537                         "x": "z", "y": "a",
1538                         "objectUUID": b"0123456789abcde2"})
1539             self.fail("Should have failed adding dupliate entry")
1540         except ldb.LdbError as err:
1541             enum = err.args[0]
1542             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1543
1544     def test_add_bad(self):
1545         try:
1546             self.l.add({"dn": "BAD,DC=SAMBA,DC=ORG",
1547                         "name": b"Admins",
1548                         "x": "z", "y": "a",
1549                         "objectUUID": b"0123456789abcde1"})
1550             self.fail("Should have failed adding entry with invalid DN")
1551         except ldb.LdbError as err:
1552             enum = err.args[0]
1553             self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1554
1555     def test_add_del_add(self):
1556         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1557                     "name": b"Admins",
1558                     "x": "z", "y": "a",
1559                     "objectUUID": b"0123456789abcde1"})
1560         self.l.delete("OU=DUP,DC=SAMBA,DC=ORG")
1561         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1562                     "name": b"Admins",
1563                     "x": "z", "y": "a",
1564                     "objectUUID": b"0123456789abcde2"})
1565
1566     def test_add_move_add(self):
1567         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1568                     "name": b"Admins",
1569                     "x": "z", "y": "a",
1570                     "objectUUID": b"0123456789abcde1"})
1571         self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1572                       "OU=DUP2,DC=SAMBA,DC=ORG")
1573         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1574                     "name": b"Admins",
1575                     "x": "z", "y": "a",
1576                     "objectUUID": b"0123456789abcde2"})
1577
1578     def test_add_move_fail_move_move(self):
1579         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1580                     "name": b"Admins",
1581                     "x": "z", "y": "a",
1582                     "objectUUID": b"0123456789abcde1"})
1583         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1584                     "name": b"Admins",
1585                     "x": "z", "y": "a",
1586                     "objectUUID": b"0123456789abcde2"})
1587
1588         res2 = self.l.search(base="DC=SAMBA,DC=ORG",
1589                              scope=ldb.SCOPE_SUBTREE,
1590                              expression="(objectUUID=0123456789abcde1)")
1591         self.assertEqual(len(res2), 1)
1592         self.assertEqual(str(res2[0].dn), "OU=DUP,DC=SAMBA,DC=ORG")
1593
1594         res3 = self.l.search(base="DC=SAMBA,DC=ORG",
1595                              scope=ldb.SCOPE_SUBTREE,
1596                              expression="(objectUUID=0123456789abcde2)")
1597         self.assertEqual(len(res3), 1)
1598         self.assertEqual(str(res3[0].dn), "OU=DUP2,DC=SAMBA,DC=ORG")
1599
1600         try:
1601             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1602                           "OU=DUP2,DC=SAMBA,DC=ORG")
1603             self.fail("Should have failed on duplicate DN")
1604         except ldb.LdbError as err:
1605             enum = err.args[0]
1606             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1607
1608         self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1609                       "OU=DUP3,DC=SAMBA,DC=ORG")
1610
1611         self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1612                       "OU=DUP2,DC=SAMBA,DC=ORG")
1613
1614         res2 = self.l.search(base="DC=SAMBA,DC=ORG",
1615                              scope=ldb.SCOPE_SUBTREE,
1616                              expression="(objectUUID=0123456789abcde1)")
1617         self.assertEqual(len(res2), 1)
1618         self.assertEqual(str(res2[0].dn), "OU=DUP2,DC=SAMBA,DC=ORG")
1619
1620         res3 = self.l.search(base="DC=SAMBA,DC=ORG",
1621                              scope=ldb.SCOPE_SUBTREE,
1622                              expression="(objectUUID=0123456789abcde2)")
1623         self.assertEqual(len(res3), 1)
1624         self.assertEqual(str(res3[0].dn), "OU=DUP3,DC=SAMBA,DC=ORG")
1625
1626     def test_move_missing(self):
1627         try:
1628             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1629                           "OU=DUP2,DC=SAMBA,DC=ORG")
1630             self.fail("Should have failed on missing")
1631         except ldb.LdbError as err:
1632             enum = err.args[0]
1633             self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1634
1635     def test_move_missing2(self):
1636         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1637                     "name": b"Admins",
1638                     "x": "z", "y": "a",
1639                     "objectUUID": b"0123456789abcde2"})
1640
1641         try:
1642             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1643                           "OU=DUP2,DC=SAMBA,DC=ORG")
1644             self.fail("Should have failed on missing")
1645         except ldb.LdbError as err:
1646             enum = err.args[0]
1647             self.assertEqual(enum, ldb.ERR_NO_SUCH_OBJECT)
1648
1649     def test_move_bad(self):
1650         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1651                     "name": b"Admins",
1652                     "x": "z", "y": "a",
1653                     "objectUUID": b"0123456789abcde2"})
1654
1655         try:
1656             self.l.rename("OUXDUP,DC=SAMBA,DC=ORG",
1657                           "OU=DUP2,DC=SAMBA,DC=ORG")
1658             self.fail("Should have failed on invalid DN")
1659         except ldb.LdbError as err:
1660             enum = err.args[0]
1661             self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1662
1663     def test_move_bad2(self):
1664         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1665                     "name": b"Admins",
1666                     "x": "z", "y": "a",
1667                     "objectUUID": b"0123456789abcde2"})
1668
1669         try:
1670             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1671                           "OUXDUP2,DC=SAMBA,DC=ORG")
1672             self.fail("Should have failed on missing")
1673         except ldb.LdbError as err:
1674             enum = err.args[0]
1675             self.assertEqual(enum, ldb.ERR_INVALID_DN_SYNTAX)
1676
1677     def test_move_fail_move_add(self):
1678         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1679                     "name": b"Admins",
1680                     "x": "z", "y": "a",
1681                     "objectUUID": b"0123456789abcde1"})
1682         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1683                     "name": b"Admins",
1684                     "x": "z", "y": "a",
1685                     "objectUUID": b"0123456789abcde2"})
1686         try:
1687             self.l.rename("OU=DUP,DC=SAMBA,DC=ORG",
1688                           "OU=DUP2,DC=SAMBA,DC=ORG")
1689             self.fail("Should have failed on duplicate DN")
1690         except ldb.LdbError as err:
1691             enum = err.args[0]
1692             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1693
1694         self.l.rename("OU=DUP2,DC=SAMBA,DC=ORG",
1695                       "OU=DUP3,DC=SAMBA,DC=ORG")
1696
1697         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1698                     "name": b"Admins",
1699                     "x": "z", "y": "a",
1700                     "objectUUID": b"0123456789abcde3"})
1701
1702
1703 class AddModifyTestsLmdb(AddModifyTests):
1704
1705     def setUp(self):
1706         self.prefix = MDB_PREFIX
1707         self.index = MDB_INDEX_OBJ
1708         super(AddModifyTestsLmdb, self).setUp()
1709
1710     def tearDown(self):
1711         super(AddModifyTestsLmdb, self).tearDown()
1712
1713 class IndexedAddModifyTests(AddModifyTests):
1714     """Test searches using the index, to ensure the index doesn't
1715        break things"""
1716     def setUp(self):
1717         if not hasattr(self, 'index'):
1718             self.index = {"dn": "@INDEXLIST",
1719                           "@IDXATTR": [b"x", b"y", b"ou", b"objectUUID"],
1720                           "@IDXONE": [b"1"]}
1721         super(IndexedAddModifyTests, self).setUp()
1722
1723     def test_duplicate_GUID(self):
1724         try:
1725             self.l.add({"dn": "OU=DUPGUID,DC=SAMBA,DC=ORG",
1726                         "name": b"Admins",
1727                         "x": "z", "y": "a",
1728                         "objectUUID": b"0123456789abcdef"})
1729             self.fail("Should have failed adding dupliate GUID")
1730         except ldb.LdbError as err:
1731             enum = err.args[0]
1732             self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1733
1734     def test_duplicate_name_dup_GUID(self):
1735         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1736                     "name": b"Admins",
1737                     "x": "z", "y": "a",
1738                     "objectUUID": b"a123456789abcdef"})
1739         try:
1740             self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1741                         "name": b"Admins",
1742                         "x": "z", "y": "a",
1743                         "objectUUID": b"a123456789abcdef"})
1744             self.fail("Should have failed adding dupliate GUID")
1745         except ldb.LdbError as err:
1746             enum = err.args[0]
1747             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1748
1749     def test_duplicate_name_dup_GUID2(self):
1750         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1751                     "name": b"Admins",
1752                     "x": "z", "y": "a",
1753                     "objectUUID": b"abc3456789abcdef"})
1754         try:
1755             self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
1756                         "name": b"Admins",
1757                         "x": "z", "y": "a",
1758                         "objectUUID": b"aaa3456789abcdef"})
1759             self.fail("Should have failed adding dupliate DN")
1760         except ldb.LdbError as err:
1761             enum = err.args[0]
1762             self.assertEqual(enum, ldb.ERR_ENTRY_ALREADY_EXISTS)
1763
1764         # Checking the GUID didn't stick in the index
1765         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1766                     "name": b"Admins",
1767                     "x": "z", "y": "a",
1768                     "objectUUID": b"aaa3456789abcdef"})
1769
1770     def test_add_dup_guid_add(self):
1771         self.l.add({"dn": "OU=DUP,DC=SAMBA,DC=ORG",
1772                     "name": b"Admins",
1773                     "x": "z", "y": "a",
1774                     "objectUUID": b"0123456789abcde1"})
1775         try:
1776             self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1777                         "name": b"Admins",
1778                         "x": "z", "y": "a",
1779                         "objectUUID": b"0123456789abcde1"})
1780             self.fail("Should have failed on duplicate GUID")
1781
1782         except ldb.LdbError as err:
1783             enum = err.args[0]
1784             self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
1785
1786         self.l.add({"dn": "OU=DUP2,DC=SAMBA,DC=ORG",
1787                     "name": b"Admins",
1788                     "x": "z", "y": "a",
1789                     "objectUUID": b"0123456789abcde2"})
1790
1791 class GUIDIndexedAddModifyTests(IndexedAddModifyTests):
1792     """Test searches using the index, to ensure the index doesn't
1793        break things"""
1794     def setUp(self):
1795         self.index = {"dn": "@INDEXLIST",
1796                       "@IDXATTR": [b"x", b"y", b"ou"],
1797                       "@IDXONE": [b"1"],
1798                       "@IDXGUID": [b"objectUUID"],
1799                       "@IDX_DN_GUID": [b"GUID"]}
1800         super(GUIDIndexedAddModifyTests, self).setUp()
1801
1802
1803 class GUIDTransIndexedAddModifyTests(GUIDIndexedAddModifyTests):
1804     """Test GUID index behaviour insdie the transaction"""
1805     def setUp(self):
1806         super(GUIDTransIndexedAddModifyTests, self).setUp()
1807         self.l.transaction_start()
1808
1809     def tearDown(self):
1810         self.l.transaction_commit()
1811         super(GUIDTransIndexedAddModifyTests, self).tearDown()
1812
1813 class TransIndexedAddModifyTests(IndexedAddModifyTests):
1814     """Test index behaviour insdie the transaction"""
1815     def setUp(self):
1816         super(TransIndexedAddModifyTests, self).setUp()
1817         self.l.transaction_start()
1818
1819     def tearDown(self):
1820         self.l.transaction_commit()
1821         super(TransIndexedAddModifyTests, self).tearDown()
1822
1823 class GuidIndexedAddModifyTestsLmdb(GUIDIndexedAddModifyTests):
1824
1825     def setUp(self):
1826         self.prefix = MDB_PREFIX
1827         super(GuidIndexedAddModifyTestsLmdb, self).setUp()
1828
1829     def tearDown(self):
1830         super(GuidIndexedAddModifyTestsLmdb, self).tearDown()
1831
1832 class GuidTransIndexedAddModifyTestsLmdb(GUIDTransIndexedAddModifyTests):
1833
1834     def setUp(self):
1835         self.prefix = MDB_PREFIX
1836         super(GuidTransIndexedAddModifyTestsLmdb, self).setUp()
1837
1838     def tearDown(self):
1839         super(GuidTransIndexedAddModifyTestsLmdb, self).tearDown()
1840
1841 class BadIndexTests(LdbBaseTest):
1842     def setUp(self):
1843         super(BadIndexTests, self).setUp()
1844         self.testdir = tempdir()
1845         self.filename = os.path.join(self.testdir, "test.ldb")
1846         self.ldb = ldb.Ldb(self.url(), flags=self.flags())
1847         if hasattr(self, 'IDXGUID'):
1848             self.ldb.add({"dn": "@INDEXLIST",
1849                           "@IDXATTR": [b"x", b"y", b"ou"],
1850                           "@IDXGUID": [b"objectUUID"],
1851                           "@IDX_DN_GUID": [b"GUID"]})
1852         else:
1853             self.ldb.add({"dn": "@INDEXLIST",
1854                           "@IDXATTR": [b"x", b"y", b"ou"]})
1855
1856         super(BadIndexTests, self).setUp()
1857
1858     def test_unique(self):
1859         self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1860                       "objectUUID": b"0123456789abcde1",
1861                       "y": "1"})
1862         self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1863                       "objectUUID": b"0123456789abcde2",
1864                       "y": "1"})
1865         self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1866                       "objectUUID": b"0123456789abcde3",
1867                       "y": "1"})
1868
1869         res = self.ldb.search(expression="(y=1)",
1870                               base="dc=samba,dc=org")
1871         self.assertEquals(len(res), 3)
1872
1873         # Now set this to unique index, but forget to check the result
1874         try:
1875             self.ldb.add({"dn": "@ATTRIBUTES",
1876                           "y": "UNIQUE_INDEX"})
1877             self.fail()
1878         except ldb.LdbError:
1879             pass
1880
1881         # We must still have a working index
1882         res = self.ldb.search(expression="(y=1)",
1883                               base="dc=samba,dc=org")
1884         self.assertEquals(len(res), 3)
1885
1886     def test_unique_transaction(self):
1887         self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1888                       "objectUUID": b"0123456789abcde1",
1889                       "y": "1"})
1890         self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1891                       "objectUUID": b"0123456789abcde2",
1892                       "y": "1"})
1893         self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1894                       "objectUUID": b"0123456789abcde3",
1895                       "y": "1"})
1896
1897         res = self.ldb.search(expression="(y=1)",
1898                               base="dc=samba,dc=org")
1899         self.assertEquals(len(res), 3)
1900
1901         self.ldb.transaction_start()
1902
1903         # Now set this to unique index, but forget to check the result
1904         try:
1905             self.ldb.add({"dn": "@ATTRIBUTES",
1906                           "y": "UNIQUE_INDEX"})
1907         except ldb.LdbError:
1908             pass
1909
1910         try:
1911             self.ldb.transaction_commit()
1912             self.fail()
1913
1914         except ldb.LdbError as err:
1915             enum = err.args[0]
1916             self.assertEqual(enum, ldb.ERR_OPERATIONS_ERROR)
1917
1918         # We must still have a working index
1919         res = self.ldb.search(expression="(y=1)",
1920                               base="dc=samba,dc=org")
1921
1922         self.assertEquals(len(res), 3)
1923
1924     def test_casefold(self):
1925         self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1926                       "objectUUID": b"0123456789abcde1",
1927                       "y": "a"})
1928         self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1929                       "objectUUID": b"0123456789abcde2",
1930                       "y": "A"})
1931         self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1932                       "objectUUID": b"0123456789abcde3",
1933                       "y": ["a", "A"]})
1934
1935         res = self.ldb.search(expression="(y=a)",
1936                               base="dc=samba,dc=org")
1937         self.assertEquals(len(res), 2)
1938
1939         self.ldb.add({"dn": "@ATTRIBUTES",
1940                       "y": "CASE_INSENSITIVE"})
1941
1942         # We must still have a working index
1943         res = self.ldb.search(expression="(y=a)",
1944                               base="dc=samba,dc=org")
1945
1946         if hasattr(self, 'IDXGUID'):
1947             self.assertEquals(len(res), 3)
1948         else:
1949             # We should not return this entry twice, but sadly
1950             # we have not yet fixed
1951             # https://bugzilla.samba.org/show_bug.cgi?id=13361
1952             self.assertEquals(len(res), 4)
1953
1954     def test_casefold_transaction(self):
1955         self.ldb.add({"dn": "x=x,dc=samba,dc=org",
1956                       "objectUUID": b"0123456789abcde1",
1957                       "y": "a"})
1958         self.ldb.add({"dn": "x=y,dc=samba,dc=org",
1959                       "objectUUID": b"0123456789abcde2",
1960                       "y": "A"})
1961         self.ldb.add({"dn": "x=z,dc=samba,dc=org",
1962                       "objectUUID": b"0123456789abcde3",
1963                       "y": ["a", "A"]})
1964
1965         res = self.ldb.search(expression="(y=a)",
1966                               base="dc=samba,dc=org")
1967         self.assertEquals(len(res), 2)
1968
1969         self.ldb.transaction_start()
1970
1971         self.ldb.add({"dn": "@ATTRIBUTES",
1972                       "y": "CASE_INSENSITIVE"})
1973
1974         self.ldb.transaction_commit()
1975
1976         # We must still have a working index
1977         res = self.ldb.search(expression="(y=a)",
1978                               base="dc=samba,dc=org")
1979
1980         if hasattr(self, 'IDXGUID'):
1981             self.assertEquals(len(res), 3)
1982         else:
1983             # We should not return this entry twice, but sadly
1984             # we have not yet fixed
1985             # https://bugzilla.samba.org/show_bug.cgi?id=13361
1986             self.assertEquals(len(res), 4)
1987
1988
1989     def tearDown(self):
1990         super(BadIndexTests, self).tearDown()
1991
1992
1993 class GUIDBadIndexTests(BadIndexTests):
1994     """Test Bad index things with GUID index mode"""
1995     def setUp(self):
1996         self.IDXGUID = True
1997
1998         super(GUIDBadIndexTests, self).setUp()
1999
2000 class DnTests(TestCase):
2001
2002     def setUp(self):
2003         super(DnTests, self).setUp()
2004         self.ldb = ldb.Ldb()
2005
2006     def tearDown(self):
2007         super(DnTests, self).tearDown()
2008         del(self.ldb)
2009
2010     def test_set_dn_invalid(self):
2011         x = ldb.Message()
2012         def assign():
2013             x.dn = "astring"
2014         self.assertRaises(TypeError, assign)
2015
2016     def test_eq(self):
2017         x = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
2018         y = ldb.Dn(self.ldb, "dc=foo11,bar=bloe")
2019         self.assertEqual(x, y)
2020         y = ldb.Dn(self.ldb, "dc=foo11,bar=blie")
2021         self.assertNotEqual(x, y)
2022
2023     def test_str(self):
2024         x = ldb.Dn(self.ldb, "dc=foo12,bar=bloe")
2025         self.assertEqual(x.__str__(), "dc=foo12,bar=bloe")
2026
2027     def test_repr(self):
2028         x = ldb.Dn(self.ldb, "dc=foo13,bla=blie")
2029         self.assertEqual(x.__repr__(), "Dn('dc=foo13,bla=blie')")
2030
2031     def test_get_casefold(self):
2032         x = ldb.Dn(self.ldb, "dc=foo14,bar=bloe")
2033         self.assertEqual(x.get_casefold(), "DC=FOO14,BAR=bloe")
2034
2035     def test_validate(self):
2036         x = ldb.Dn(self.ldb, "dc=foo15,bar=bloe")
2037         self.assertTrue(x.validate())
2038
2039     def test_parent(self):
2040         x = ldb.Dn(self.ldb, "dc=foo16,bar=bloe")
2041         self.assertEqual("bar=bloe", x.parent().__str__())
2042
2043     def test_parent_nonexistent(self):
2044         x = ldb.Dn(self.ldb, "@BLA")
2045         self.assertEqual(None, x.parent())
2046
2047     def test_is_valid(self):
2048         x = ldb.Dn(self.ldb, "dc=foo18,dc=bloe")
2049         self.assertTrue(x.is_valid())
2050         x = ldb.Dn(self.ldb, "")
2051         self.assertTrue(x.is_valid())
2052
2053     def test_is_special(self):
2054         x = ldb.Dn(self.ldb, "dc=foo19,bar=bloe")
2055         self.assertFalse(x.is_special())
2056         x = ldb.Dn(self.ldb, "@FOOBAR")
2057         self.assertTrue(x.is_special())
2058
2059     def test_check_special(self):
2060         x = ldb.Dn(self.ldb, "dc=foo20,bar=bloe")
2061         self.assertFalse(x.check_special("FOOBAR"))
2062         x = ldb.Dn(self.ldb, "@FOOBAR")
2063         self.assertTrue(x.check_special("@FOOBAR"))
2064
2065     def test_len(self):
2066         x = ldb.Dn(self.ldb, "dc=foo21,bar=bloe")
2067         self.assertEqual(2, len(x))
2068         x = ldb.Dn(self.ldb, "dc=foo21")
2069         self.assertEqual(1, len(x))
2070
2071     def test_add_child(self):
2072         x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
2073         self.assertTrue(x.add_child(ldb.Dn(self.ldb, "bla=bloe")))
2074         self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
2075
2076     def test_add_base(self):
2077         x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
2078         base = ldb.Dn(self.ldb, "bla=bloe")
2079         self.assertTrue(x.add_base(base))
2080         self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
2081
2082     def test_add_child_str(self):
2083         x = ldb.Dn(self.ldb, "dc=foo22,bar=bloe")
2084         self.assertTrue(x.add_child("bla=bloe"))
2085         self.assertEqual("bla=bloe,dc=foo22,bar=bloe", x.__str__())
2086
2087     def test_add_base_str(self):
2088         x = ldb.Dn(self.ldb, "dc=foo23,bar=bloe")
2089         base = "bla=bloe"
2090         self.assertTrue(x.add_base(base))
2091         self.assertEqual("dc=foo23,bar=bloe,bla=bloe", x.__str__())
2092
2093     def test_add(self):
2094         x = ldb.Dn(self.ldb, "dc=foo24")
2095         y = ldb.Dn(self.ldb, "bar=bla")
2096         self.assertEqual("dc=foo24,bar=bla", str(x + y))
2097
2098     def test_remove_base_components(self):
2099         x = ldb.Dn(self.ldb, "dc=foo24,dc=samba,dc=org")
2100         x.remove_base_components(len(x)-1)
2101         self.assertEqual("dc=foo24", str(x))
2102
2103     def test_parse_ldif(self):
2104         msgs = self.ldb.parse_ldif("dn: foo=bar\n")
2105         msg = next(msgs)
2106         self.assertEqual("foo=bar", str(msg[1].dn))
2107         self.assertTrue(isinstance(msg[1], ldb.Message))
2108         ldif = self.ldb.write_ldif(msg[1], ldb.CHANGETYPE_NONE)
2109         self.assertEqual("dn: foo=bar\n\n", ldif)
2110
2111     def test_parse_ldif_more(self):
2112         msgs = self.ldb.parse_ldif("dn: foo=bar\n\n\ndn: bar=bar")
2113         msg = next(msgs)
2114         self.assertEqual("foo=bar", str(msg[1].dn))
2115         msg = next(msgs)
2116         self.assertEqual("bar=bar", str(msg[1].dn))
2117
2118     def test_canonical_string(self):
2119         x = ldb.Dn(self.ldb, "dc=foo25,bar=bloe")
2120         self.assertEqual("/bloe/foo25", x.canonical_str())
2121
2122     def test_canonical_ex_string(self):
2123         x = ldb.Dn(self.ldb, "dc=foo26,bar=bloe")
2124         self.assertEqual("/bloe\nfoo26", x.canonical_ex_str())
2125
2126     def test_ldb_is_child_of(self):
2127         """Testing ldb_dn_compare_dn"""
2128         dn1 = ldb.Dn(self.ldb, "dc=base")
2129         dn2 = ldb.Dn(self.ldb, "cn=foo,dc=base")
2130         dn3 = ldb.Dn(self.ldb, "cn=bar,dc=base")
2131         dn4 = ldb.Dn(self.ldb, "cn=baz,cn=bar,dc=base")
2132
2133         self.assertTrue(dn1.is_child_of(dn1))
2134         self.assertTrue(dn2.is_child_of(dn1))
2135         self.assertTrue(dn4.is_child_of(dn1))
2136         self.assertTrue(dn4.is_child_of(dn3))
2137         self.assertTrue(dn4.is_child_of(dn4))
2138         self.assertFalse(dn3.is_child_of(dn2))
2139         self.assertFalse(dn1.is_child_of(dn4))
2140
2141     def test_ldb_is_child_of_str(self):
2142         """Testing ldb_dn_compare_dn"""
2143         dn1_str = "dc=base"
2144         dn2_str = "cn=foo,dc=base"
2145         dn3_str = "cn=bar,dc=base"
2146         dn4_str = "cn=baz,cn=bar,dc=base"
2147
2148         dn1 = ldb.Dn(self.ldb, dn1_str)
2149         dn2 = ldb.Dn(self.ldb, dn2_str)
2150         dn3 = ldb.Dn(self.ldb, dn3_str)
2151         dn4 = ldb.Dn(self.ldb, dn4_str)
2152
2153         self.assertTrue(dn1.is_child_of(dn1_str))
2154         self.assertTrue(dn2.is_child_of(dn1_str))
2155         self.assertTrue(dn4.is_child_of(dn1_str))
2156         self.assertTrue(dn4.is_child_of(dn3_str))
2157         self.assertTrue(dn4.is_child_of(dn4_str))
2158         self.assertFalse(dn3.is_child_of(dn2_str))
2159         self.assertFalse(dn1.is_child_of(dn4_str))
2160
2161     def test_get_component_name(self):
2162         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2163         self.assertEqual(dn.get_component_name(0), 'cn')
2164         self.assertEqual(dn.get_component_name(1), 'dc')
2165         self.assertEqual(dn.get_component_name(2), None)
2166         self.assertEqual(dn.get_component_name(-1), None)
2167
2168     def test_get_component_value(self):
2169         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2170         self.assertEqual(dn.get_component_value(0), 'foo')
2171         self.assertEqual(dn.get_component_value(1), 'base')
2172         self.assertEqual(dn.get_component_name(2), None)
2173         self.assertEqual(dn.get_component_name(-1), None)
2174
2175     def test_set_component(self):
2176         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2177         dn.set_component(0, 'cn', 'bar')
2178         self.assertEqual(str(dn), "cn=bar,dc=base")
2179         dn.set_component(1, 'o', 'asep')
2180         self.assertEqual(str(dn), "cn=bar,o=asep")
2181         self.assertRaises(TypeError, dn.set_component, 2, 'dc', 'base')
2182         self.assertEqual(str(dn), "cn=bar,o=asep")
2183         dn.set_component(1, 'o', 'a,b+c')
2184         self.assertEqual(str(dn), r"cn=bar,o=a\,b\+c")
2185
2186     def test_set_component_bytes(self):
2187         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2188         dn.set_component(0, 'cn', b'bar')
2189         self.assertEqual(str(dn), "cn=bar,dc=base")
2190         dn.set_component(1, 'o', b'asep')
2191         self.assertEqual(str(dn), "cn=bar,o=asep")
2192
2193     def test_set_component_none(self):
2194         dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
2195         self.assertRaises(TypeError, dn.set_component, 1, 'cn', None)
2196
2197     def test_get_extended_component_null(self):
2198         dn = ldb.Dn(self.ldb, "cn=foo,cn=bar,dc=base")
2199         self.assertEqual(dn.get_extended_component("TEST"), None)
2200
2201     def test_get_extended_component(self):
2202         self.ldb._register_test_extensions()
2203         dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
2204         self.assertEqual(dn.get_extended_component("TEST"), b"foo")
2205
2206     def test_set_extended_component(self):
2207         self.ldb._register_test_extensions()
2208         dn = ldb.Dn(self.ldb, "dc=base")
2209         dn.set_extended_component("TEST", "foo")
2210         self.assertEqual(dn.get_extended_component("TEST"), b"foo")
2211         dn.set_extended_component("TEST", b"bar")
2212         self.assertEqual(dn.get_extended_component("TEST"), b"bar")
2213
2214     def test_extended_str(self):
2215         self.ldb._register_test_extensions()
2216         dn = ldb.Dn(self.ldb, "<TEST=foo>;cn=bar,dc=base")
2217         self.assertEqual(dn.extended_str(), "<TEST=foo>;cn=bar,dc=base")
2218
2219     def test_get_rdn_name(self):
2220         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2221         self.assertEqual(dn.get_rdn_name(), 'cn')
2222
2223     def test_get_rdn_value(self):
2224         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2225         self.assertEqual(dn.get_rdn_value(), 'foo')
2226
2227     def test_get_casefold(self):
2228         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2229         self.assertEqual(dn.get_casefold(), 'CN=FOO,DC=BASE')
2230
2231     def test_get_linearized(self):
2232         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2233         self.assertEqual(dn.get_linearized(), 'cn=foo,dc=base')
2234
2235     def test_is_null(self):
2236         dn = ldb.Dn(self.ldb, "cn=foo,dc=base")
2237         self.assertFalse(dn.is_null())
2238
2239         dn = ldb.Dn(self.ldb, '')
2240         self.assertTrue(dn.is_null())
2241
2242 class LdbMsgTests(TestCase):
2243
2244     def setUp(self):
2245         super(LdbMsgTests, self).setUp()
2246         self.msg = ldb.Message()
2247
2248     def test_init_dn(self):
2249         self.msg = ldb.Message(ldb.Dn(ldb.Ldb(), "dc=foo27"))
2250         self.assertEqual("dc=foo27", str(self.msg.dn))
2251
2252     def test_iter_items(self):
2253         self.assertEqual(0, len(self.msg.items()))
2254         self.msg.dn = ldb.Dn(ldb.Ldb(), "dc=foo28")
2255         self.assertEqual(1, len(self.msg.items()))
2256
2257     def test_repr(self):
2258         self.msg.dn = ldb.Dn(ldb.Ldb(), "dc=foo29")
2259         self.msg["dc"] = b"foo"
2260         if PY3:
2261             self.assertIn(repr(self.msg), [
2262                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])})",
2263                 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')})",
2264             ])
2265             self.assertIn(repr(self.msg.text), [
2266                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement([b'foo'])}).text",
2267                 "Message({'dc': MessageElement([b'foo']), 'dn': Dn('dc=foo29')}).text",
2268             ])
2269         else:
2270             self.assertEqual(
2271                 repr(self.msg),
2272                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])})")
2273             self.assertEqual(
2274                 repr(self.msg.text),
2275                 "Message({'dn': Dn('dc=foo29'), 'dc': MessageElement(['foo'])}).text")
2276
2277     def test_len(self):
2278         self.assertEqual(0, len(self.msg))
2279
2280     def test_notpresent(self):
2281         self.assertRaises(KeyError, lambda: self.msg["foo"])
2282
2283     def test_del(self):
2284         del self.msg["foo"]
2285
2286     def test_add(self):
2287         self.msg.add(ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla"))
2288
2289     def test_add_text(self):
2290         self.msg.add(ldb.MessageElement(["456"], ldb.FLAG_MOD_ADD, "bla"))
2291
2292     def test_elements_empty(self):
2293         self.assertEqual([], self.msg.elements())
2294
2295     def test_elements(self):
2296         el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
2297         self.msg.add(el)
2298         self.assertEqual([el], self.msg.elements())
2299         self.assertEqual([el.text], self.msg.text.elements())
2300
2301     def test_add_value(self):
2302         self.assertEqual(0, len(self.msg))
2303         self.msg["foo"] = [b"foo"]
2304         self.assertEqual(1, len(self.msg))
2305
2306     def test_add_value_text(self):
2307         self.assertEqual(0, len(self.msg))
2308         self.msg["foo"] = ["foo"]
2309         self.assertEqual(1, len(self.msg))
2310
2311     def test_add_value_multiple(self):
2312         self.assertEqual(0, len(self.msg))
2313         self.msg["foo"] = [b"foo", b"bla"]
2314         self.assertEqual(1, len(self.msg))
2315         self.assertEqual([b"foo", b"bla"], list(self.msg["foo"]))
2316
2317     def test_add_value_multiple_text(self):
2318         self.assertEqual(0, len(self.msg))
2319         self.msg["foo"] = ["foo", "bla"]
2320         self.assertEqual(1, len(self.msg))
2321         self.assertEqual(["foo", "bla"], list(self.msg.text["foo"]))
2322
2323     def test_set_value(self):
2324         self.msg["foo"] = [b"fool"]
2325         self.assertEqual([b"fool"], list(self.msg["foo"]))
2326         self.msg["foo"] = [b"bar"]
2327         self.assertEqual([b"bar"], list(self.msg["foo"]))
2328
2329     def test_set_value_text(self):
2330         self.msg["foo"] = ["fool"]
2331         self.assertEqual(["fool"], list(self.msg.text["foo"]))
2332         self.msg["foo"] = ["bar"]
2333         self.assertEqual(["bar"], list(self.msg.text["foo"]))
2334
2335     def test_keys(self):
2336         self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2337         self.msg["foo"] = [b"bla"]
2338         self.msg["bar"] = [b"bla"]
2339         self.assertEqual(["dn", "foo", "bar"], list(self.msg.keys()))
2340
2341     def test_keys_text(self):
2342         self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2343         self.msg["foo"] = ["bla"]
2344         self.msg["bar"] = ["bla"]
2345         self.assertEqual(["dn", "foo", "bar"], list(self.msg.text.keys()))
2346
2347     def test_dn(self):
2348         self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2349         self.assertEqual("@BASEINFO", self.msg.dn.__str__())
2350
2351     def test_get_dn(self):
2352         self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2353         self.assertEqual("@BASEINFO", self.msg.get("dn").__str__())
2354
2355     def test_dn_text(self):
2356         self.msg.text.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2357         self.assertEqual("@BASEINFO", str(self.msg.dn))
2358         self.assertEqual("@BASEINFO", str(self.msg.text.dn))
2359
2360     def test_get_dn_text(self):
2361         self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2362         self.assertEqual("@BASEINFO", str(self.msg.get("dn")))
2363         self.assertEqual("@BASEINFO", str(self.msg.text.get("dn")))
2364
2365     def test_get_invalid(self):
2366         self.msg.dn = ldb.Dn(ldb.Ldb(), "@BASEINFO")
2367         self.assertRaises(TypeError, self.msg.get, 42)
2368
2369     def test_get_other(self):
2370         self.msg["foo"] = [b"bar"]
2371         self.assertEqual(b"bar", self.msg.get("foo")[0])
2372         self.assertEqual(b"bar", self.msg.get("foo", idx=0))
2373         self.assertEqual(None, self.msg.get("foo", idx=1))
2374         self.assertEqual("", self.msg.get("foo", default='', idx=1))
2375
2376     def test_get_other_text(self):
2377         self.msg["foo"] = ["bar"]
2378         self.assertEqual(["bar"], list(self.msg.text.get("foo")))
2379         self.assertEqual("bar", self.msg.text.get("foo")[0])
2380         self.assertEqual("bar", self.msg.text.get("foo", idx=0))
2381         self.assertEqual(None, self.msg.get("foo", idx=1))
2382         self.assertEqual("", self.msg.get("foo", default='', idx=1))
2383
2384     def test_get_default(self):
2385         self.assertEqual(None, self.msg.get("tatayoyo", idx=0))
2386         self.assertEqual("anniecordie", self.msg.get("tatayoyo", "anniecordie"))
2387
2388     def test_get_default_text(self):
2389         self.assertEqual(None, self.msg.text.get("tatayoyo", idx=0))
2390         self.assertEqual("anniecordie", self.msg.text.get("tatayoyo", "anniecordie"))
2391
2392     def test_get_unknown(self):
2393         self.assertEqual(None, self.msg.get("lalalala"))
2394
2395     def test_get_unknown_text(self):
2396         self.assertEqual(None, self.msg.text.get("lalalala"))
2397
2398     def test_msg_diff(self):
2399         l = ldb.Ldb()
2400         msgs = l.parse_ldif("dn: foo=bar\nfoo: bar\nbaz: do\n\ndn: foo=bar\nfoo: bar\nbaz: dont\n")
2401         msg1 = next(msgs)[1]
2402         msg2 = next(msgs)[1]
2403         msgdiff = l.msg_diff(msg1, msg2)
2404         self.assertEqual("foo=bar", msgdiff.get("dn").__str__())
2405         self.assertRaises(KeyError, lambda: msgdiff["foo"])
2406         self.assertEqual(1, len(msgdiff))
2407
2408     def test_equal_empty(self):
2409         msg1 = ldb.Message()
2410         msg2 = ldb.Message()
2411         self.assertEqual(msg1, msg2)
2412
2413     def test_equal_simplel(self):
2414         db = ldb.Ldb()
2415         msg1 = ldb.Message()
2416         msg1.dn = ldb.Dn(db, "foo=bar")
2417         msg2 = ldb.Message()
2418         msg2.dn = ldb.Dn(db, "foo=bar")
2419         self.assertEqual(msg1, msg2)
2420         msg1['foo'] = b'bar'
2421         msg2['foo'] = b'bar'
2422         self.assertEqual(msg1, msg2)
2423         msg2['foo'] = b'blie'
2424         self.assertNotEqual(msg1, msg2)
2425         msg2['foo'] = b'blie'
2426
2427     def test_from_dict(self):
2428         rec = {"dn": "dc=fromdict",
2429                "a1": [b"a1-val1", b"a1-val1"]}
2430         l = ldb.Ldb()
2431         # check different types of input Flags
2432         for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
2433             m = ldb.Message.from_dict(l, rec, flags)
2434             self.assertEqual(rec["a1"], list(m["a1"]))
2435             self.assertEqual(flags, m["a1"].flags())
2436         # check input params
2437         self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
2438         self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
2439         self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
2440         # Message.from_dict expects dictionary with 'dn'
2441         err_rec = {"a1": [b"a1-val1", b"a1-val1"]}
2442         self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
2443
2444     def test_from_dict_text(self):
2445         rec = {"dn": "dc=fromdict",
2446                "a1": ["a1-val1", "a1-val1"]}
2447         l = ldb.Ldb()
2448         # check different types of input Flags
2449         for flags in [ldb.FLAG_MOD_ADD, ldb.FLAG_MOD_REPLACE, ldb.FLAG_MOD_DELETE]:
2450             m = ldb.Message.from_dict(l, rec, flags)
2451             self.assertEqual(rec["a1"], list(m.text["a1"]))
2452             self.assertEqual(flags, m.text["a1"].flags())
2453         # check input params
2454         self.assertRaises(TypeError, ldb.Message.from_dict, dict(), rec, ldb.FLAG_MOD_REPLACE)
2455         self.assertRaises(TypeError, ldb.Message.from_dict, l, list(), ldb.FLAG_MOD_REPLACE)
2456         self.assertRaises(ValueError, ldb.Message.from_dict, l, rec, 0)
2457         # Message.from_dict expects dictionary with 'dn'
2458         err_rec = {"a1": ["a1-val1", "a1-val1"]}
2459         self.assertRaises(TypeError, ldb.Message.from_dict, l, err_rec, ldb.FLAG_MOD_REPLACE)
2460
2461     def test_copy_add_message_element(self):
2462         m = ldb.Message()
2463         m["1"] = ldb.MessageElement([b"val 111"], ldb.FLAG_MOD_ADD, "1")
2464         m["2"] = ldb.MessageElement([b"val 222"], ldb.FLAG_MOD_ADD, "2")
2465         mto = ldb.Message()
2466         mto["1"] = m["1"]
2467         mto["2"] = m["2"]
2468         self.assertEqual(mto["1"], m["1"])
2469         self.assertEqual(mto["2"], m["2"])
2470         mto = ldb.Message()
2471         mto.add(m["1"])
2472         mto.add(m["2"])
2473         self.assertEqual(mto["1"], m["1"])
2474         self.assertEqual(mto["2"], m["2"])
2475
2476     def test_copy_add_message_element_text(self):
2477         m = ldb.Message()
2478         m["1"] = ldb.MessageElement(["val 111"], ldb.FLAG_MOD_ADD, "1")
2479         m["2"] = ldb.MessageElement(["val 222"], ldb.FLAG_MOD_ADD, "2")
2480         mto = ldb.Message()
2481         mto["1"] = m["1"]
2482         mto["2"] = m["2"]
2483         self.assertEqual(mto["1"], m.text["1"])
2484         self.assertEqual(mto["2"], m.text["2"])
2485         mto = ldb.Message()
2486         mto.add(m["1"])
2487         mto.add(m["2"])
2488         self.assertEqual(mto.text["1"], m.text["1"])
2489         self.assertEqual(mto.text["2"], m.text["2"])
2490         self.assertEqual(mto["1"], m["1"])
2491         self.assertEqual(mto["2"], m["2"])
2492
2493
2494 class MessageElementTests(TestCase):
2495
2496     def test_cmp_element(self):
2497         x = ldb.MessageElement([b"foo"])
2498         y = ldb.MessageElement([b"foo"])
2499         z = ldb.MessageElement([b"bzr"])
2500         self.assertEqual(x, y)
2501         self.assertNotEqual(x, z)
2502
2503     def test_cmp_element_text(self):
2504         x = ldb.MessageElement([b"foo"])
2505         y = ldb.MessageElement(["foo"])
2506         self.assertEqual(x, y)
2507
2508     def test_create_iterable(self):
2509         x = ldb.MessageElement([b"foo"])
2510         self.assertEqual([b"foo"], list(x))
2511         self.assertEqual(["foo"], list(x.text))
2512
2513     def test_repr(self):
2514         x = ldb.MessageElement([b"foo"])
2515         if PY3:
2516             self.assertEqual("MessageElement([b'foo'])", repr(x))
2517             self.assertEqual("MessageElement([b'foo']).text", repr(x.text))
2518         else:
2519             self.assertEqual("MessageElement(['foo'])", repr(x))
2520             self.assertEqual("MessageElement(['foo']).text", repr(x.text))
2521         x = ldb.MessageElement([b"foo", b"bla"])
2522         self.assertEqual(2, len(x))
2523         if PY3:
2524             self.assertEqual("MessageElement([b'foo',b'bla'])", repr(x))
2525             self.assertEqual("MessageElement([b'foo',b'bla']).text", repr(x.text))
2526         else:
2527             self.assertEqual("MessageElement(['foo','bla'])", repr(x))
2528             self.assertEqual("MessageElement(['foo','bla']).text", repr(x.text))
2529
2530     def test_get_item(self):
2531         x = ldb.MessageElement([b"foo", b"bar"])
2532         self.assertEqual(b"foo", x[0])
2533         self.assertEqual(b"bar", x[1])
2534         self.assertEqual(b"bar", x[-1])
2535         self.assertRaises(IndexError, lambda: x[45])
2536
2537     def test_get_item_text(self):
2538         x = ldb.MessageElement(["foo", "bar"])
2539         self.assertEqual("foo", x.text[0])
2540         self.assertEqual("bar", x.text[1])
2541         self.assertEqual("bar", x.text[-1])
2542         self.assertRaises(IndexError, lambda: x[45])
2543
2544     def test_len(self):
2545         x = ldb.MessageElement([b"foo", b"bar"])
2546         self.assertEqual(2, len(x))
2547
2548     def test_eq(self):
2549         x = ldb.MessageElement([b"foo", b"bar"])
2550         y = ldb.MessageElement([b"foo", b"bar"])
2551         self.assertEqual(y, x)
2552         x = ldb.MessageElement([b"foo"])
2553         self.assertNotEqual(y, x)
2554         y = ldb.MessageElement([b"foo"])
2555         self.assertEqual(y, x)
2556
2557     def test_extended(self):
2558         el = ldb.MessageElement([b"456"], ldb.FLAG_MOD_ADD, "bla")
2559         if PY3:
2560             self.assertEqual("MessageElement([b'456'])", repr(el))
2561             self.assertEqual("MessageElement([b'456']).text", repr(el.text))
2562         else:
2563             self.assertEqual("MessageElement(['456'])", repr(el))
2564             self.assertEqual("MessageElement(['456']).text", repr(el.text))
2565
2566     def test_bad_text(self):
2567         el = ldb.MessageElement(b'\xba\xdd')
2568         self.assertRaises(UnicodeDecodeError, el.text.__getitem__, 0)
2569
2570
2571 class ModuleTests(TestCase):
2572
2573     def setUp(self):
2574         super(ModuleTests, self).setUp()
2575         self.testdir = tempdir()
2576         self.filename = os.path.join(self.testdir, "test.ldb")
2577         self.ldb = ldb.Ldb(self.filename)
2578
2579     def tearDown(self):
2580         shutil.rmtree(self.testdir)
2581         super(ModuleTests, self).setUp()
2582
2583     def test_register_module(self):
2584         class ExampleModule:
2585             name = "example"
2586         ldb.register_module(ExampleModule)
2587
2588     def test_use_module(self):
2589         ops = []
2590         class ExampleModule:
2591             name = "bla"
2592
2593             def __init__(self, ldb, next):
2594                 ops.append("init")
2595                 self.next = next
2596
2597             def search(self, *args, **kwargs):
2598                 return self.next.search(*args, **kwargs)
2599
2600             def request(self, *args, **kwargs):
2601                 pass
2602
2603         ldb.register_module(ExampleModule)
2604         l = ldb.Ldb(self.filename)
2605         l.add({"dn": "@MODULES", "@LIST": "bla"})
2606         self.assertEqual([], ops)
2607         l = ldb.Ldb(self.filename)
2608         self.assertEqual(["init"], ops)
2609
2610 class LdbResultTests(LdbBaseTest):
2611
2612     def setUp(self):
2613         super(LdbResultTests, self).setUp()
2614         self.testdir = tempdir()
2615         self.filename = os.path.join(self.testdir, "test.ldb")
2616         self.l = ldb.Ldb(self.url(), flags=self.flags())
2617         try:
2618             self.l.add(self.index)
2619         except AttributeError:
2620             pass
2621         self.l.add({"dn": "DC=SAMBA,DC=ORG", "name": b"samba.org",
2622                     "objectUUID": b"0123456789abcde0"})
2623         self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG", "name": b"Admins",
2624                     "objectUUID": b"0123456789abcde1"})
2625         self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG", "name": b"Users",
2626                     "objectUUID": b"0123456789abcde2"})
2627         self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG", "name": b"OU #1",
2628                     "objectUUID": b"0123456789abcde3"})
2629         self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG", "name": b"OU #2",
2630                     "objectUUID": b"0123456789abcde4"})
2631         self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG", "name": b"OU #3",
2632                     "objectUUID": b"0123456789abcde5"})
2633         self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG", "name": b"OU #4",
2634                     "objectUUID": b"0123456789abcde6"})
2635         self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG", "name": b"OU #5",
2636                     "objectUUID": b"0123456789abcde7"})
2637         self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG", "name": b"OU #6",
2638                     "objectUUID": b"0123456789abcde8"})
2639         self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG", "name": b"OU #7",
2640                     "objectUUID": b"0123456789abcde9"})
2641         self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG", "name": b"OU #8",
2642                     "objectUUID": b"0123456789abcdea"})
2643         self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG", "name": b"OU #9",
2644                     "objectUUID": b"0123456789abcdeb"})
2645         self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG", "name": b"OU #10",
2646                     "objectUUID": b"0123456789abcdec"})
2647
2648     def tearDown(self):
2649         shutil.rmtree(self.testdir)
2650         super(LdbResultTests, self).tearDown()
2651         # Ensure the LDB is closed now, so we close the FD
2652         del(self.l)
2653
2654     def test_return_type(self):
2655         res = self.l.search()
2656         self.assertEqual(str(res), "<ldb result>")
2657
2658     def test_get_msgs(self):
2659         res = self.l.search()
2660         list = res.msgs
2661
2662     def test_get_controls(self):
2663         res = self.l.search()
2664         list = res.controls
2665
2666     def test_get_referals(self):
2667         res = self.l.search()
2668         list = res.referals
2669
2670     def test_iter_msgs(self):
2671         found = False
2672         for l in self.l.search().msgs:
2673             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2674                 found = True
2675         self.assertTrue(found)
2676
2677     def test_iter_msgs_count(self):
2678         self.assertTrue(self.l.search().count > 0)
2679         # 13 objects has been added to the DC=SAMBA, DC=ORG
2680         self.assertEqual(self.l.search(base="DC=SAMBA,DC=ORG").count, 13)
2681
2682     def test_iter_controls(self):
2683         res = self.l.search().controls
2684         it = iter(res)
2685
2686     def test_create_control(self):
2687         self.assertRaises(ValueError, ldb.Control, self.l, "tatayoyo:0")
2688         c = ldb.Control(self.l, "relax:1")
2689         self.assertEqual(c.critical, True)
2690         self.assertEqual(c.oid, "1.3.6.1.4.1.4203.666.5.12")
2691
2692     def test_iter_refs(self):
2693         res = self.l.search().referals
2694         it = iter(res)
2695
2696     def test_search_sequence_msgs(self):
2697         found = False
2698         res = self.l.search().msgs
2699
2700         for i in range(0, len(res)):
2701             l = res[i]
2702             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2703                 found = True
2704         self.assertTrue(found)
2705
2706     def test_search_as_iter(self):
2707         found = False
2708         res = self.l.search()
2709
2710         for l in res:
2711             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2712                 found = True
2713         self.assertTrue(found)
2714
2715     def test_search_iter(self):
2716         found = False
2717         res = self.l.search_iterator()
2718
2719         for l in res:
2720             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2721                 found = True
2722         self.assertTrue(found)
2723
2724
2725     # Show that search results can't see into a transaction
2726     def test_search_against_trans(self):
2727         found11 = False
2728
2729         (r1, w1) = os.pipe()
2730
2731         (r2, w2) = os.pipe()
2732
2733         # For the first element, fork a child that will
2734         # write to the DB
2735         pid = os.fork()
2736         if pid == 0:
2737             # In the child, re-open
2738             del(self.l)
2739             gc.collect()
2740
2741             child_ldb = ldb.Ldb(self.url(), flags=self.flags())
2742             # start a transaction
2743             child_ldb.transaction_start()
2744
2745             # write to it
2746             child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
2747                            "name": b"samba.org",
2748                            "objectUUID": b"o123456789acbdef"})
2749
2750             os.write(w1, b"added")
2751
2752             # Now wait for the search to be done
2753             os.read(r2, 6)
2754
2755             # and commit
2756             try:
2757                 child_ldb.transaction_commit()
2758             except LdbError as err:
2759                 # We print this here to see what went wrong in the child
2760                 print(err)
2761                 os._exit(1)
2762
2763             os.write(w1, b"transaction")
2764             os._exit(0)
2765
2766         self.assertEqual(os.read(r1, 5), b"added")
2767
2768         # This should not turn up until the transaction is concluded
2769         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2770                               scope=ldb.SCOPE_BASE)
2771         self.assertEqual(len(res11), 0)
2772
2773         os.write(w2, b"search")
2774
2775         # Now wait for the transaction to be done.  This should
2776         # deadlock, but the search doesn't hold a read lock for the
2777         # iterator lifetime currently.
2778         self.assertEqual(os.read(r1, 11), b"transaction")
2779
2780         # This should now turn up, as the transaction is over
2781         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2782                               scope=ldb.SCOPE_BASE)
2783         self.assertEqual(len(res11), 1)
2784
2785         self.assertFalse(found11)
2786
2787         (got_pid, status) = os.waitpid(pid, 0)
2788         self.assertEqual(got_pid, pid)
2789
2790
2791     def test_search_iter_against_trans(self):
2792         found = False
2793         found11 = False
2794
2795         # We need to hold this iterator open to hold the all-record
2796         # lock
2797         res = self.l.search_iterator()
2798
2799         (r1, w1) = os.pipe()
2800
2801         (r2, w2) = os.pipe()
2802
2803         # For the first element, with the sequence open (which
2804         # means with ldb locks held), fork a child that will
2805         # write to the DB
2806         pid = os.fork()
2807         if pid == 0:
2808             # In the child, re-open
2809             del(res)
2810             del(self.l)
2811             gc.collect()
2812
2813             child_ldb = ldb.Ldb(self.url(), flags=self.flags())
2814             # start a transaction
2815             child_ldb.transaction_start()
2816
2817             # write to it
2818             child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
2819                            "name": b"samba.org",
2820                            "objectUUID": b"o123456789acbdef"})
2821
2822             os.write(w1, b"added")
2823
2824             # Now wait for the search to be done
2825             os.read(r2, 6)
2826
2827             # and commit
2828             try:
2829                 child_ldb.transaction_commit()
2830             except LdbError as err:
2831                 # We print this here to see what went wrong in the child
2832                 print(err)
2833                 os._exit(1)
2834
2835             os.write(w1, b"transaction")
2836             os._exit(0)
2837
2838         self.assertEqual(os.read(r1, 5), b"added")
2839
2840         # This should not turn up until the transaction is concluded
2841         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2842                               scope=ldb.SCOPE_BASE)
2843         self.assertEqual(len(res11), 0)
2844
2845         os.write(w2, b"search")
2846
2847         # allow the transaction to start
2848         time.sleep(1)
2849
2850         # This should not turn up until the search finishes and
2851         # removed the read lock, but for ldb_tdb that happened as soon
2852         # as we called the first res.next()
2853         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2854                               scope=ldb.SCOPE_BASE)
2855         self.assertEqual(len(res11), 0)
2856
2857         # These results are all collected at the first next(res) call
2858         for l in res:
2859             if str(l.dn) == "OU=OU10,DC=SAMBA,DC=ORG":
2860                 found = True
2861             if str(l.dn) == "OU=OU11,DC=SAMBA,DC=ORG":
2862                 found11 = True
2863
2864         # Now wait for the transaction to be done.
2865         self.assertEqual(os.read(r1, 11), b"transaction")
2866
2867         # This should now turn up, as the transaction is over and all
2868         # read locks are gone
2869         res11 = self.l.search(base="OU=OU11,DC=SAMBA,DC=ORG",
2870                               scope=ldb.SCOPE_BASE)
2871         self.assertEqual(len(res11), 1)
2872
2873         self.assertTrue(found)
2874         self.assertFalse(found11)
2875
2876         (got_pid, status) = os.waitpid(pid, 0)
2877         self.assertEqual(got_pid, pid)
2878
2879
2880 class LdbResultTestsLmdb(LdbResultTests):
2881
2882     def setUp(self):
2883         self.prefix = MDB_PREFIX
2884         self.index = MDB_INDEX_OBJ
2885         super(LdbResultTestsLmdb, self).setUp()
2886
2887     def tearDown(self):
2888         super(LdbResultTestsLmdb, self).tearDown()
2889
2890
2891 class BadTypeTests(TestCase):
2892     def test_control(self):
2893         l = ldb.Ldb()
2894         self.assertRaises(TypeError, ldb.Control, '<bad type>', 'relax:1')
2895         self.assertRaises(TypeError, ldb.Control, ldb, 1234)
2896
2897     def test_modify(self):
2898         l = ldb.Ldb()
2899         dn = ldb.Dn(l, 'a=b')
2900         m = ldb.Message(dn)
2901         self.assertRaises(TypeError, l.modify, '<bad type>')
2902         self.assertRaises(TypeError, l.modify, m, '<bad type>')
2903
2904     def test_add(self):
2905         l = ldb.Ldb()
2906         dn = ldb.Dn(l, 'a=b')
2907         m = ldb.Message(dn)
2908         self.assertRaises(TypeError, l.add, '<bad type>')
2909         self.assertRaises(TypeError, l.add, m, '<bad type>')
2910
2911     def test_delete(self):
2912         l = ldb.Ldb()
2913         dn = ldb.Dn(l, 'a=b')
2914         self.assertRaises(TypeError, l.add, '<bad type>')
2915         self.assertRaises(TypeError, l.add, dn, '<bad type>')
2916
2917     def test_rename(self):
2918         l = ldb.Ldb()
2919         dn = ldb.Dn(l, 'a=b')
2920         self.assertRaises(TypeError, l.add, '<bad type>', dn)
2921         self.assertRaises(TypeError, l.add, dn, '<bad type>')
2922         self.assertRaises(TypeError, l.add, dn, dn, '<bad type>')
2923
2924     def test_search(self):
2925         l = ldb.Ldb()
2926         self.assertRaises(TypeError, l.search, base=1234)
2927         self.assertRaises(TypeError, l.search, scope='<bad type>')
2928         self.assertRaises(TypeError, l.search, expression=1234)
2929         self.assertRaises(TypeError, l.search, attrs='<bad type>')
2930         self.assertRaises(TypeError, l.search, controls='<bad type>')
2931
2932
2933 class VersionTests(TestCase):
2934
2935     def test_version(self):
2936         self.assertTrue(isinstance(ldb.__version__, str))
2937
2938
2939 if __name__ == '__main__':
2940     import unittest
2941     unittest.TestProgram()