self.testdir = tempdir()
self.filename = os.path.join(self.testdir, "test.ldb")
self.ldb = ldb.Ldb(self.url(), flags=self.flags())
+ try:
+ self.ldb.add(self.index)
+ except AttributeError:
+ pass
def tearDown(self):
shutil.rmtree(self.testdir)
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=foo1")
m["b"] = [b"a"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
self.assertRaises(ldb.LdbError, lambda: l.delete(m.dn, ["search_options:1:2"]))
l.delete(m.dn)
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=foo3")
m["b"] = ["a"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
try:
self.assertTrue(ldb.Dn(l, "dc=foo3") in l)
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=foo4")
m["bla"] = b"bla"
+ m["objectUUID"] = b"0123456789abcdef"
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
m1 = ldb.Message()
m1.dn = ldb.Dn(l, "dc=foo4")
m1["bla"] = b"bla"
+ m1["objectUUID"] = b"0123456789abcdef"
l.add(m1)
try:
s = l.search_iterator()
m2 = ldb.Message()
m2.dn = ldb.Dn(l, "dc=foo5")
m2["bla"] = b"bla"
+ m2["objectUUID"] = b"0123456789abcdee"
l.add(m2)
s = l.search_iterator()
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=foo4")
m["bla"] = "bla"
+ m["objectUUID"] = b"0123456789abcdef"
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
def test_add_dict(self):
l = ldb.Ldb(self.url(), flags=self.flags())
m = {"dn": ldb.Dn(l, "dc=foo5"),
- "bla": b"bla"}
+ "bla": b"bla",
+ "objectUUID": b"0123456789abcdef"}
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
def test_add_dict_text(self):
l = ldb.Ldb(self.url(), flags=self.flags())
m = {"dn": ldb.Dn(l, "dc=foo5"),
- "bla": "bla"}
+ "bla": "bla",
+ "objectUUID": b"0123456789abcdef"}
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
def test_add_dict_string_dn(self):
l = ldb.Ldb(self.url(), flags=self.flags())
- m = {"dn": "dc=foo6", "bla": b"bla"}
+ m = {"dn": "dc=foo6", "bla": b"bla",
+ "objectUUID": b"0123456789abcdef"}
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
def test_add_dict_bytes_dn(self):
l = ldb.Ldb(self.url(), flags=self.flags())
- m = {"dn": b"dc=foo6", "bla": b"bla"}
+ m = {"dn": b"dc=foo6", "bla": b"bla",
+ "objectUUID": b"0123456789abcdef"}
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=foo7")
m["bla"] = b"bla"
+ m["objectUUID"] = b"0123456789abcdef"
self.assertEqual(len(l.search()), 0)
l.add(m)
try:
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=foo8")
m["bla"] = b"bla"
+ m["objectUUID"] = b"0123456789abcdef"
self.assertEqual(len(l.search()), 0)
l.add(m)
self.assertEqual(len(l.search()), 1)
self.assertEqual(0, len(l.search()))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=empty")
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
rm = l.search()
self.assertEqual(1, len(rm))
- self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
+ self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
+ set(rm[0].keys()))
rm = l.search(m.dn)
self.assertEqual(1, len(rm))
- self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
+ self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
+ set(rm[0].keys()))
rm = l.search(m.dn, attrs=["blah"])
self.assertEqual(1, len(rm))
self.assertEqual(0, len(rm[0]))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=modifydelete")
m["bla"] = [b"1234"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
rm = l.search(m.dn)[0]
self.assertEqual([b"1234"], list(rm["bla"]))
l.modify(m)
rm = l.search(m.dn)
self.assertEqual(1, len(rm))
- self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
+ self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
+ set(rm[0].keys()))
rm = l.search(m.dn, attrs=["bla"])
self.assertEqual(1, len(rm))
self.assertEqual(0, len(rm[0]))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=modifydelete")
m.text["bla"] = ["1234"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
rm = l.search(m.dn)[0]
self.assertEqual(["1234"], list(rm.text["bla"]))
l.modify(m)
rm = l.search(m.dn)
self.assertEqual(1, len(rm))
- self.assertEqual(set(["dn", "distinguishedName"]), set(rm[0].keys()))
+ self.assertEqual(set(["dn", "distinguishedName", "objectUUID"]),
+ set(rm[0].keys()))
rm = l.search(m.dn, attrs=["bla"])
self.assertEqual(1, len(rm))
self.assertEqual(0, len(rm[0]))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=add")
m["bla"] = [b"1234"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
try:
m = ldb.Message()
self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
l.modify(m)
rm = l.search(m.dn)[0]
- self.assertEqual(2, len(rm))
+ self.assertEqual(3, len(rm))
self.assertEqual([b"1234", b"456"], list(rm["bla"]))
finally:
l.delete(ldb.Dn(l, "dc=add"))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=add")
m.text["bla"] = ["1234"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
try:
m = ldb.Message()
self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
l.modify(m)
rm = l.search(m.dn)[0]
- self.assertEqual(2, len(rm))
+ self.assertEqual(3, len(rm))
self.assertEqual(["1234", "456"], list(rm.text["bla"]))
finally:
l.delete(ldb.Dn(l, "dc=add"))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=modify2")
m["bla"] = [b"1234", b"456"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
try:
m = ldb.Message()
self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
l.modify(m)
rm = l.search(m.dn)[0]
- self.assertEqual(2, len(rm))
+ self.assertEqual(3, len(rm))
self.assertEqual([b"789"], list(rm["bla"]))
rm = l.search(m.dn, attrs=["bla"])[0]
self.assertEqual(1, len(rm))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=modify2")
m.text["bla"] = ["1234", "456"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
try:
m = ldb.Message()
self.assertEqual(ldb.FLAG_MOD_REPLACE, m["bla"].flags())
l.modify(m)
rm = l.search(m.dn)[0]
- self.assertEqual(2, len(rm))
+ self.assertEqual(3, len(rm))
self.assertEqual(["789"], list(rm.text["bla"]))
rm = l.search(m.dn, attrs=["bla"])[0]
self.assertEqual(1, len(rm))
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=add")
m["bla"] = [b"1234"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
try:
m = ldb.Message()
self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
l.modify(m)
rm = l.search(m.dn)[0]
- self.assertEqual(2, len(rm))
+ self.assertEqual(3, len(rm))
self.assertEqual([b"1234", b"456"], list(rm["bla"]))
# Now create another modify, but switch the flags before we do it
m = ldb.Message()
m.dn = ldb.Dn(l, "dc=add")
m.text["bla"] = ["1234"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
try:
m = ldb.Message()
self.assertEqual(ldb.FLAG_MOD_ADD, m["bla"].flags())
l.modify(m)
rm = l.search(m.dn)[0]
- self.assertEqual(2, len(rm))
+ self.assertEqual(3, len(rm))
self.assertEqual(["1234", "456"], list(rm.text["bla"]))
# Now create another modify, but switch the flags before we do it
l.transaction_start()
m = ldb.Message(ldb.Dn(l, "dc=foo9"))
m["foo"] = [b"bar"]
+ m["objectUUID"] = b"0123456789abcdef"
l.add(m)
l.transaction_commit()
l.delete(m.dn)
l.transaction_start()
m = ldb.Message(ldb.Dn(l, "dc=foo10"))
m["foo"] = [b"bar"]
+ m["objectUUID"] = b"0123456789abcdee"
l.add(m)
l.transaction_cancel()
self.assertEqual(0, len(l.search(ldb.Dn(l, "dc=foo10"))))
"cN" : b"LDAPtestUSER",
"givenname" : b"ldap",
"displayname" : b"foo\0bar",
+ "objectUUID" : b"0123456789abcdef"
})
res = l.search(expression="(dn=dc=somedn)")
self.assertEqual(b"foo\0bar", res[0]["displayname"][0])
self.l = ldb.Ldb(self.url(),
flags=self.flags(),
options=["modules:rdn_name"])
+ try:
+ self.l.add(self.index)
+ except AttributeError:
+ pass
self.l.add({"dn": "@ATTRIBUTES",
"DC": "CASE_INSENSITIVE"})
self.l.add({"dn": "DC=SAMBA,DC=ORG",
"name": b"samba.org",
- "objectUUID": b"0123456789abcddf"})
+ "objectUUID": b"0123456789abcdef"})
self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG",
"name": b"Admins",
"x": "z", "y": "a",
"""Test searches using the index, to ensure the index doesn't
break things"""
def setUp(self):
+ self.index = {"dn": "@INDEXLIST",
+ "@IDXATTR": [b"x", b"y", b"ou"],
+ "@IDXGUID": [b"objectUUID"],
+ "@IDX_DN_GUID": [b"GUID"]}
super(GUIDIndexedSearchTests, self).setUp()
- self.l.add({"dn": "@INDEXLIST",
- "@IDXATTR": [b"x", b"y", b"ou"],
- "@IDXGUID": [b"objectUUID"],
- "@IDX_DN_GUID": [b"GUID"]})
self.IDXGUID = True
self.IDXONE = True
"""Test searches using the index, to ensure the index doesn't
break things"""
def setUp(self):
+ self.index = {"dn": "@INDEXLIST",
+ "@IDXATTR": [b"x", b"y", b"ou"],
+ "@IDXGUID": [b"objectUUID"],
+ "@IDX_DN_GUID": [b"GUID"]}
super(GUIDIndexedDNFilterSearchTests, self).setUp()
self.l.add({"dn": "@OPTIONS",
"disallowDNFilter": "TRUE"})
self.disallowDNFilter = True
-
- self.l.add({"dn": "@INDEXLIST",
- "@IDXATTR": [b"x", b"y", b"ou"],
- "@IDXGUID": [b"objectUUID"],
- "@IDX_DN_GUID": [b"GUID"]})
self.IDX = True
self.IDXGUID = True
"""Test searches using the index including @IDXONE, to ensure
the index doesn't break things"""
def setUp(self):
+ self.index = {"dn": "@INDEXLIST",
+ "@IDXATTR": [b"x", b"y", b"ou"],
+ "@IDXGUID": [b"objectUUID"],
+ "@IDX_DN_GUID": [b"GUID"]}
super(GUIDAndOneLevelIndexedSearchTests, self).setUp()
self.l.add({"dn": "@OPTIONS",
"disallowDNFilter": "TRUE"})
self.disallowDNFilter = True
-
- self.l.add({"dn": "@INDEXLIST",
- "@IDXATTR": [b"x", b"y", b"ou"],
- "@IDXONE": [b"1"],
- "@IDXGUID": [b"objectUUID"],
- "@IDX_DN_GUID": [b"GUID"]})
self.IDX = True
self.IDXGUID = True
self.IDXONE = True
self.l = ldb.Ldb(self.url(),
flags=self.flags(),
options=["modules:rdn_name"])
+ try:
+ self.l.add(self.index)
+ except AttributeError:
+ pass
+
self.l.add({"dn": "DC=SAMBA,DC=ORG",
"name": b"samba.org",
"objectUUID": b"0123456789abcdef"})
"""Test searches using the index, to ensure the index doesn't
break things"""
def setUp(self):
+ if not hasattr(self, 'index'):
+ self.index = {"dn": "@INDEXLIST",
+ "@IDXATTR": [b"x", b"y", b"ou", b"objectUUID"],
+ "@IDXONE": [b"1"]}
super(IndexedAddModifyTests, self).setUp()
- self.l.add({"dn": "@INDEXLIST",
- "@IDXATTR": [b"x", b"y", b"ou", b"objectUUID"],
- "@IDXONE": [b"1"]})
def test_duplicate_GUID(self):
try:
"""Test searches using the index, to ensure the index doesn't
break things"""
def setUp(self):
+ self.index = {"dn": "@INDEXLIST",
+ "@IDXATTR": [b"x", b"y", b"ou"],
+ "@IDXONE": [b"1"],
+ "@IDXGUID": [b"objectUUID"],
+ "@IDX_DN_GUID": [b"GUID"]}
super(GUIDIndexedAddModifyTests, self).setUp()
- indexlist = {"dn": "@INDEXLIST",
- "@IDXATTR": [b"x", b"y", b"ou"],
- "@IDXONE": [b"1"],
- "@IDXGUID": [b"objectUUID"],
- "@IDX_DN_GUID": [b"GUID"]}
- m = ldb.Message.from_dict(self.l, indexlist, ldb.FLAG_MOD_REPLACE)
- self.l.modify(m)
class GUIDTransIndexedAddModifyTests(GUIDIndexedAddModifyTests):
self.testdir = tempdir()
self.filename = os.path.join(self.testdir, "test.ldb")
self.l = ldb.Ldb(self.url(), flags=self.flags())
- self.l.add({"dn": "DC=SAMBA,DC=ORG", "name": b"samba.org"})
- self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG", "name": b"Admins"})
- self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG", "name": b"Users"})
- self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG", "name": b"OU #1"})
- self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG", "name": b"OU #2"})
- self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG", "name": b"OU #3"})
- self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG", "name": b"OU #4"})
- self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG", "name": b"OU #5"})
- self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG", "name": b"OU #6"})
- self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG", "name": b"OU #7"})
- self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG", "name": b"OU #8"})
- self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG", "name": b"OU #9"})
- self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG", "name": b"OU #10"})
+ try:
+ self.l.add(self.index)
+ except AttributeError:
+ pass
+ self.l.add({"dn": "DC=SAMBA,DC=ORG", "name": b"samba.org",
+ "objectUUID": b"0123456789abcde0"})
+ self.l.add({"dn": "OU=ADMIN,DC=SAMBA,DC=ORG", "name": b"Admins",
+ "objectUUID": b"0123456789abcde1"})
+ self.l.add({"dn": "OU=USERS,DC=SAMBA,DC=ORG", "name": b"Users",
+ "objectUUID": b"0123456789abcde2"})
+ self.l.add({"dn": "OU=OU1,DC=SAMBA,DC=ORG", "name": b"OU #1",
+ "objectUUID": b"0123456789abcde3"})
+ self.l.add({"dn": "OU=OU2,DC=SAMBA,DC=ORG", "name": b"OU #2",
+ "objectUUID": b"0123456789abcde4"})
+ self.l.add({"dn": "OU=OU3,DC=SAMBA,DC=ORG", "name": b"OU #3",
+ "objectUUID": b"0123456789abcde5"})
+ self.l.add({"dn": "OU=OU4,DC=SAMBA,DC=ORG", "name": b"OU #4",
+ "objectUUID": b"0123456789abcde6"})
+ self.l.add({"dn": "OU=OU5,DC=SAMBA,DC=ORG", "name": b"OU #5",
+ "objectUUID": b"0123456789abcde7"})
+ self.l.add({"dn": "OU=OU6,DC=SAMBA,DC=ORG", "name": b"OU #6",
+ "objectUUID": b"0123456789abcde8"})
+ self.l.add({"dn": "OU=OU7,DC=SAMBA,DC=ORG", "name": b"OU #7",
+ "objectUUID": b"0123456789abcde9"})
+ self.l.add({"dn": "OU=OU8,DC=SAMBA,DC=ORG", "name": b"OU #8",
+ "objectUUID": b"0123456789abcdea"})
+ self.l.add({"dn": "OU=OU9,DC=SAMBA,DC=ORG", "name": b"OU #9",
+ "objectUUID": b"0123456789abcdeb"})
+ self.l.add({"dn": "OU=OU10,DC=SAMBA,DC=ORG", "name": b"OU #10",
+ "objectUUID": b"0123456789abcdec"})
def tearDown(self):
shutil.rmtree(self.testdir)
# write to it
child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
- "name": b"samba.org"})
+ "name": b"samba.org",
+ "objectUUID": b"o123456789acbdef"})
os.write(w1, b"added")
# write to it
child_ldb.add({"dn": "OU=OU11,DC=SAMBA,DC=ORG",
- "name": b"samba.org"})
+ "name": b"samba.org",
+ "objectUUID": b"o123456789acbdef"})
os.write(w1, b"added")