s4-samba3samtest: use system credentials for creating users
[ira/wip.git] / source4 / dsdb / samdb / ldb_modules / tests / samba3sam.py
1 #!/usr/bin/python
2
3 # Unix SMB/CIFS implementation.
4 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2005-2008
5 # Copyright (C) Martin Kuehl <mkhl@samba.org> 2006
6 #
7 # This is a Python port of the original in testprogs/ejs/samba3sam.js
8 #   
9 # This program is free software; you can redistribute it and/or modify
10 # it under the terms of the GNU General Public License as published by
11 # the Free Software Foundation; either version 3 of the License, or
12 # (at your option) any later version.
13 #   
14 # This program is distributed in the hope that it will be useful,
15 # but WITHOUT ANY WARRANTY; without even the implied warranty of
16 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17 # GNU General Public License for more details.
18 #   
19 # You should have received a copy of the GNU General Public License
20 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
21 #
22
23 """Tests for the samba3sam LDB module, which maps Samba3 LDAP to AD LDAP."""
24
25 import os
26 import ldb
27 from ldb import SCOPE_DEFAULT, SCOPE_BASE, SCOPE_SUBTREE
28 from samba import Ldb, substitute_var
29 from samba.tests import LdbTestCase, TestCaseInTempDir, cmdline_loadparm
30 import samba.dcerpc.security
31 import samba.ndr
32 from samba.auth import system_session
33
34 datadir = os.path.join(os.path.dirname(__file__), 
35                        "../../../../../testdata/samba3")
36
37 def read_datafile(filename):
38     return open(os.path.join(datadir, filename), 'r').read()
39
40 def ldb_debug(l, text):
41     print text
42
43
44 class MapBaseTestCase(TestCaseInTempDir):
45     """Base test case for mapping tests."""
46
47     def setup_modules(self, ldb, s3, s4):
48         ldb.add({"dn": "@MAP=samba3sam",
49                  "@FROM": s4.basedn,
50                  "@TO": "sambaDomainName=TESTS," + s3.basedn})
51
52         ldb.add({"dn": "@MODULES",
53                  "@LIST": "rootdse,paged_results,server_sort,asq,samldb,password_hash,operational,objectguid,rdn_name,samba3sam,partition"})
54
55         ldb.add({"dn": "@PARTITION",
56             "partition": ["%s" % (s4.basedn_casefold), 
57                           "%s" % (s3.basedn_casefold)],
58             "replicateEntries": ["@ATTRIBUTES", "@INDEXLIST"],
59             "modules": "*:"})
60
61     def setUp(self):
62         super(MapBaseTestCase, self).setUp()
63
64         def make_dn(basedn, rdn):
65             return "%s,sambaDomainName=TESTS,%s" % (rdn, basedn)
66
67         def make_s4dn(basedn, rdn):
68             return "%s,%s" % (rdn, basedn)
69
70         self.ldbfile = os.path.join(self.tempdir, "test.ldb")
71         self.ldburl = "tdb://" + self.ldbfile
72
73         tempdir = self.tempdir
74
75         class Target:
76             """Simple helper class that contains data for a specific SAM 
77             connection."""
78             def __init__(self, basedn, dn):
79                 self.db = Ldb(lp=cmdline_loadparm, session_info=system_session())
80                 self.basedn = basedn
81                 self.basedn_casefold = ldb.Dn(self.db, basedn).get_casefold()
82                 self.substvars = {"BASEDN": self.basedn}
83                 self.file = os.path.join(tempdir, "%s.ldb" % self.basedn_casefold)
84                 self.url = "tdb://" + self.file
85                 self._dn = dn
86
87             def dn(self, rdn):
88                 return self._dn(self.basedn, rdn)
89
90             def connect(self):
91                 return self.db.connect(self.url)
92
93             def setup_data(self, path):
94                 self.add_ldif(read_datafile(path))
95
96             def subst(self, text):
97                 return substitute_var(text, self.substvars)
98
99             def add_ldif(self, ldif):
100                 self.db.add_ldif(self.subst(ldif))
101
102             def modify_ldif(self, ldif):
103                 self.db.modify_ldif(self.subst(ldif))
104
105         self.samba4 = Target("dc=vernstok,dc=nl", make_s4dn)
106         self.samba3 = Target("cn=Samba3Sam", make_dn)
107
108         self.samba3.connect()
109         self.samba4.connect()
110
111     def tearDown(self):
112         os.unlink(self.ldbfile)
113         os.unlink(self.samba3.file)
114         os.unlink(self.samba4.file)
115         super(MapBaseTestCase, self).tearDown()
116
117     def assertSidEquals(self, text, ndr_sid):
118         sid_obj1 = samba.ndr.ndr_unpack(samba.dcerpc.security.dom_sid,
119                                         str(ndr_sid[0]))
120         sid_obj2 = samba.dcerpc.security.dom_sid(text)
121         self.assertEquals(sid_obj1, sid_obj2)
122
123
124 class Samba3SamTestCase(MapBaseTestCase):
125
126     def setUp(self):
127         super(Samba3SamTestCase, self).setUp()
128         ldb = Ldb(self.ldburl, lp=cmdline_loadparm, session_info=system_session())
129         self.samba3.setup_data("samba3.ldif")
130         ldif = read_datafile("provision_samba3sam.ldif")
131         ldb.add_ldif(self.samba4.subst(ldif))
132         self.setup_modules(ldb, self.samba3, self.samba4)
133         del ldb
134         self.ldb = Ldb(self.ldburl, lp=cmdline_loadparm, session_info=system_session())
135
136     def test_search_non_mapped(self):
137         """Looking up by non-mapped attribute"""
138         msg = self.ldb.search(expression="(cn=Administrator)")
139         self.assertEquals(len(msg), 1)
140         self.assertEquals(msg[0]["cn"], "Administrator")
141
142     def test_search_non_mapped(self):
143         """Looking up by mapped attribute"""
144         msg = self.ldb.search(expression="(name=Backup Operators)")
145         self.assertEquals(len(msg), 1)
146         self.assertEquals(str(msg[0]["name"]), "Backup Operators")
147
148     def test_old_name_of_renamed(self):
149         """Looking up by old name of renamed attribute"""
150         msg = self.ldb.search(expression="(displayName=Backup Operators)")
151         self.assertEquals(len(msg), 0)
152
153     def test_mapped_containing_sid(self):
154         """Looking up mapped entry containing SID"""
155         msg = self.ldb.search(expression="(cn=Replicator)")
156         self.assertEquals(len(msg), 1)
157         self.assertEquals(str(msg[0].dn), 
158                           "cn=Replicator,ou=Groups,dc=vernstok,dc=nl")
159         self.assertTrue("objectSid" in msg[0]) 
160         self.assertSidEquals("S-1-5-21-4231626423-2410014848-2360679739-552",
161                              msg[0]["objectSid"])
162         oc = set(msg[0]["objectClass"])
163         self.assertEquals(oc, set(["group"]))
164
165     def test_search_by_objclass(self):
166         """Looking up by objectClass"""
167         msg = self.ldb.search(expression="(|(objectClass=user)(cn=Administrator))")
168         self.assertEquals(set([str(m.dn) for m in msg]), 
169                 set(["unixName=Administrator,ou=Users,dc=vernstok,dc=nl", 
170                      "unixName=nobody,ou=Users,dc=vernstok,dc=nl"]))
171
172     def test_s3sam_modify(self):
173         # Adding a record that will be fallbacked
174         self.ldb.add({"dn": "cn=Foo", 
175             "foo": "bar", 
176             "blah": "Blie", 
177             "cn": "Foo", 
178             "showInAdvancedViewOnly": "TRUE"}
179             )
180
181         # Checking for existence of record (local)
182         # TODO: This record must be searched in the local database, which is 
183         # currently only supported for base searches
184         # msg = ldb.search(expression="(cn=Foo)", ['foo','blah','cn','showInAdvancedViewOnly')]
185         # TODO: Actually, this version should work as well but doesn't...
186         # 
187         #    
188         msg = self.ldb.search(expression="(cn=Foo)", base="cn=Foo", 
189                 scope=SCOPE_BASE, 
190                 attrs=['foo','blah','cn','showInAdvancedViewOnly'])
191         self.assertEquals(len(msg), 1)
192         self.assertEquals(str(msg[0]["showInAdvancedViewOnly"]), "TRUE")
193         self.assertEquals(str(msg[0]["foo"]), "bar")
194         self.assertEquals(str(msg[0]["blah"]), "Blie")
195
196         # Adding record that will be mapped
197         self.ldb.add({"dn": "cn=Niemand,cn=Users,dc=vernstok,dc=nl",
198                  "objectClass": "user",
199                  "unixName": "bin",
200                  "sambaUnicodePwd": "geheim",
201                  "cn": "Niemand"})
202
203         # Checking for existence of record (remote)
204         msg = self.ldb.search(expression="(unixName=bin)", 
205                               attrs=['unixName','cn','dn', 'sambaUnicodePwd'])
206         self.assertEquals(len(msg), 1)
207         self.assertEquals(str(msg[0]["cn"]), "Niemand")
208         self.assertEquals(str(msg[0]["sambaUnicodePwd"]), "geheim")
209
210         # Checking for existence of record (local && remote)
211         msg = self.ldb.search(expression="(&(unixName=bin)(sambaUnicodePwd=geheim))", 
212                          attrs=['unixName','cn','dn', 'sambaUnicodePwd'])
213         self.assertEquals(len(msg), 1)           # TODO: should check with more records
214         self.assertEquals(str(msg[0]["cn"]), "Niemand")
215         self.assertEquals(str(msg[0]["unixName"]), "bin")
216         self.assertEquals(str(msg[0]["sambaUnicodePwd"]), "geheim")
217
218         # Checking for existence of record (local || remote)
219         msg = self.ldb.search(expression="(|(unixName=bin)(sambaUnicodePwd=geheim))", 
220                          attrs=['unixName','cn','dn', 'sambaUnicodePwd'])
221         #print "got %d replies" % len(msg)
222         self.assertEquals(len(msg), 1)        # TODO: should check with more records
223         self.assertEquals(str(msg[0]["cn"]), "Niemand")
224         self.assertEquals(str(msg[0]["unixName"]), "bin")
225         self.assertEquals(str(msg[0]["sambaUnicodePwd"]), "geheim")
226
227         # Checking for data in destination database
228         msg = self.samba3.db.search(expression="(cn=Niemand)")
229         self.assertTrue(len(msg) >= 1)
230         self.assertEquals(str(msg[0]["sambaSID"]), 
231                 "S-1-5-21-4231626423-2410014848-2360679739-2001")
232         self.assertEquals(str(msg[0]["displayName"]), "Niemand")
233
234         # Adding attribute...
235         self.ldb.modify_ldif("""
236 dn: cn=Niemand,cn=Users,dc=vernstok,dc=nl
237 changetype: modify
238 add: description
239 description: Blah
240 """)
241
242         # Checking whether changes are still there...
243         msg = self.ldb.search(expression="(cn=Niemand)")
244         self.assertTrue(len(msg) >= 1)
245         self.assertEquals(str(msg[0]["cn"]), "Niemand")
246         self.assertEquals(str(msg[0]["description"]), "Blah")
247
248         # Modifying attribute...
249         self.ldb.modify_ldif("""
250 dn: cn=Niemand,cn=Users,dc=vernstok,dc=nl
251 changetype: modify
252 replace: description
253 description: Blie
254 """)
255
256         # Checking whether changes are still there...
257         msg = self.ldb.search(expression="(cn=Niemand)")
258         self.assertTrue(len(msg) >= 1)
259         self.assertEquals(str(msg[0]["description"]), "Blie")
260
261         # Deleting attribute...
262         self.ldb.modify_ldif("""
263 dn: cn=Niemand,cn=Users,dc=vernstok,dc=nl
264 changetype: modify
265 delete: description
266 """)
267
268         # Checking whether changes are no longer there...
269         msg = self.ldb.search(expression="(cn=Niemand)")
270         self.assertTrue(len(msg) >= 1)
271         self.assertTrue(not "description" in msg[0])
272
273         # Renaming record...
274         self.ldb.rename("cn=Niemand,cn=Users,dc=vernstok,dc=nl", 
275                         "cn=Niemand2,cn=Users,dc=vernstok,dc=nl")
276
277         # Checking whether DN has changed...
278         msg = self.ldb.search(expression="(cn=Niemand2)")
279         self.assertEquals(len(msg), 1)
280         self.assertEquals(str(msg[0].dn), 
281                           "cn=Niemand2,cn=Users,dc=vernstok,dc=nl")
282
283         # Deleting record...
284         self.ldb.delete("cn=Niemand2,cn=Users,dc=vernstok,dc=nl")
285
286         # Checking whether record is gone...
287         msg = self.ldb.search(expression="(cn=Niemand2)")
288         self.assertEquals(len(msg), 0)
289
290
291 class MapTestCase(MapBaseTestCase):
292
293     def setUp(self):
294         super(MapTestCase, self).setUp()
295         ldb = Ldb(self.ldburl, lp=cmdline_loadparm, session_info=system_session())
296         ldif = read_datafile("provision_samba3sam.ldif")
297         ldb.add_ldif(self.samba4.subst(ldif))
298         self.setup_modules(ldb, self.samba3, self.samba4)
299         del ldb
300         self.ldb = Ldb(self.ldburl, lp=cmdline_loadparm, session_info=system_session())
301
302     def test_map_search(self):
303         """Running search tests on mapped data."""
304         self.samba3.db.add({
305             "dn": "sambaDomainName=TESTS," + self.samba3.basedn,
306             "objectclass": ["sambaDomain", "top"],
307             "sambaSID": "S-1-5-21-4231626423-2410014848-2360679739",
308             "sambaNextRid": "2000",
309             "sambaDomainName": "TESTS"
310             })
311
312         # Add a set of split records
313         self.ldb.add_ldif("""
314 dn: """+ self.samba4.dn("cn=Domain Users") + """
315 objectClass: group
316 cn: Domain Users
317 objectSid: S-1-5-21-4231626423-2410014848-2360679739-513
318 """)
319
320         # Add a set of split records
321         self.ldb.add_ldif("""
322 dn: """+ self.samba4.dn("cn=X") + """
323 objectClass: user
324 cn: X
325 codePage: x
326 revision: x
327 dnsHostName: x
328 nextRid: y
329 lastLogon: x
330 description: x
331 objectSid: S-1-5-21-4231626423-2410014848-2360679739-552
332 """)
333
334         self.ldb.add({
335             "dn": self.samba4.dn("cn=Y"),
336             "objectClass": "top",
337             "cn": "Y",
338             "codePage": "x",
339             "revision": "x",
340             "dnsHostName": "y",
341             "nextRid": "y",
342             "lastLogon": "y",
343             "description": "x"})
344
345         self.ldb.add({
346             "dn": self.samba4.dn("cn=Z"),
347             "objectClass": "top",
348             "cn": "Z",
349             "codePage": "x",
350             "revision": "y",
351             "dnsHostName": "z",
352             "nextRid": "y",
353             "lastLogon": "z",
354             "description": "y"})
355
356         # Add a set of remote records
357
358         self.samba3.db.add({
359             "dn": self.samba3.dn("cn=A"),
360             "objectClass": "posixAccount",
361             "cn": "A",
362             "sambaNextRid": "x",
363             "sambaBadPasswordCount": "x", 
364             "sambaLogonTime": "x",
365             "description": "x",
366             "sambaSID": "S-1-5-21-4231626423-2410014848-2360679739-552",
367             "sambaPrimaryGroupSID": "S-1-5-21-4231626423-2410014848-2360679739-512"})
368
369         self.samba3.db.add({
370             "dn": self.samba3.dn("cn=B"),
371             "objectClass": "top",
372             "cn": "B",
373             "sambaNextRid": "x",
374             "sambaBadPasswordCount": "x",
375             "sambaLogonTime": "y",
376             "description": "x"})
377
378         self.samba3.db.add({
379             "dn": self.samba3.dn("cn=C"),
380             "objectClass": "top",
381             "cn": "C",
382             "sambaNextRid": "x",
383             "sambaBadPasswordCount": "y",
384             "sambaLogonTime": "z",
385             "description": "y"})
386
387         # Testing search by DN
388
389         # Search remote record by local DN
390         dn = self.samba4.dn("cn=A")
391         res = self.ldb.search(dn, scope=SCOPE_BASE, 
392                 attrs=["dnsHostName", "lastLogon"])
393         self.assertEquals(len(res), 1)
394         self.assertEquals(str(res[0].dn), dn)
395         self.assertTrue(not "dnsHostName" in res[0])
396         self.assertEquals(str(res[0]["lastLogon"]), "x")
397
398         # Search remote record by remote DN
399         dn = self.samba3.dn("cn=A")
400         res = self.samba3.db.search(dn, scope=SCOPE_BASE, 
401                 attrs=["dnsHostName", "lastLogon", "sambaLogonTime"])
402         self.assertEquals(len(res), 1)
403         self.assertEquals(str(res[0].dn), dn)
404         self.assertTrue(not "dnsHostName" in res[0])
405         self.assertTrue(not "lastLogon" in res[0])
406         self.assertEquals(str(res[0]["sambaLogonTime"]), "x")
407
408         # Search split record by local DN
409         dn = self.samba4.dn("cn=X")
410         res = self.ldb.search(dn, scope=SCOPE_BASE, 
411                 attrs=["dnsHostName", "lastLogon"])
412         self.assertEquals(len(res), 1)
413         self.assertEquals(str(res[0].dn), dn)
414         self.assertEquals(str(res[0]["dnsHostName"]), "x")
415         self.assertEquals(str(res[0]["lastLogon"]), "x")
416
417         # Search split record by remote DN
418         dn = self.samba3.dn("cn=X")
419         res = self.samba3.db.search(dn, scope=SCOPE_BASE, 
420                 attrs=["dnsHostName", "lastLogon", "sambaLogonTime"])
421         self.assertEquals(len(res), 1)
422         self.assertEquals(str(res[0].dn), dn)
423         self.assertTrue(not "dnsHostName" in res[0])
424         self.assertTrue(not "lastLogon" in res[0])
425         self.assertEquals(str(res[0]["sambaLogonTime"]), "x")
426
427         # Testing search by attribute
428
429         # Search by ignored attribute
430         res = self.ldb.search(expression="(revision=x)", scope=SCOPE_DEFAULT, 
431                 attrs=["dnsHostName", "lastLogon"])
432         self.assertEquals(len(res), 2)
433         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
434         self.assertEquals(str(res[0]["dnsHostName"]), "y")
435         self.assertEquals(str(res[0]["lastLogon"]), "y")
436         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
437         self.assertEquals(str(res[1]["dnsHostName"]), "x")
438         self.assertEquals(str(res[1]["lastLogon"]), "x")
439
440         # Search by kept attribute
441         res = self.ldb.search(expression="(description=y)", 
442                 scope=SCOPE_DEFAULT, attrs=["dnsHostName", "lastLogon"])
443         self.assertEquals(len(res), 2)
444         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Z"))
445         self.assertEquals(str(res[0]["dnsHostName"]), "z")
446         self.assertEquals(str(res[0]["lastLogon"]), "z")
447         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=C"))
448         self.assertTrue(not "dnsHostName" in res[1])
449         self.assertEquals(str(res[1]["lastLogon"]), "z")
450
451         # Search by renamed attribute
452         res = self.ldb.search(expression="(badPwdCount=x)", scope=SCOPE_DEFAULT,
453                               attrs=["dnsHostName", "lastLogon"])
454         self.assertEquals(len(res), 2)
455         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
456         self.assertTrue(not "dnsHostName" in res[0])
457         self.assertEquals(str(res[0]["lastLogon"]), "y")
458         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
459         self.assertTrue(not "dnsHostName" in res[1])
460         self.assertEquals(str(res[1]["lastLogon"]), "x")
461
462         # Search by converted attribute
463         # TODO:
464         #   Using the SID directly in the parse tree leads to conversion
465         #   errors, letting the search fail with no results.
466         #res = self.ldb.search("(objectSid=S-1-5-21-4231626423-2410014848-2360679739-552)", scope=SCOPE_DEFAULT, attrs)
467         res = self.ldb.search(expression="(objectSid=*)", base=None, scope=SCOPE_DEFAULT, attrs=["dnsHostName", "lastLogon", "objectSid"])
468         self.assertEquals(len(res), 4)
469         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=X"))
470         self.assertEquals(str(res[0]["dnsHostName"]), "x")
471         self.assertEquals(str(res[0]["lastLogon"]), "x")
472         self.assertSidEquals("S-1-5-21-4231626423-2410014848-2360679739-552", 
473                              res[0]["objectSid"])
474         self.assertTrue("objectSid" in res[0])
475         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
476         self.assertTrue(not "dnsHostName" in res[1])
477         self.assertEquals(str(res[1]["lastLogon"]), "x")
478         self.assertSidEquals("S-1-5-21-4231626423-2410014848-2360679739-552",
479                              res[1]["objectSid"])
480         self.assertTrue("objectSid" in res[1])
481
482         # Search by generated attribute 
483         # In most cases, this even works when the mapping is missing
484         # a `convert_operator' by enumerating the remote db.
485         res = self.ldb.search(expression="(primaryGroupID=512)", 
486                            attrs=["dnsHostName", "lastLogon", "primaryGroupID"])
487         self.assertEquals(len(res), 1)
488         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=A"))
489         self.assertTrue(not "dnsHostName" in res[0])
490         self.assertEquals(str(res[0]["lastLogon"]), "x")
491         self.assertEquals(str(res[0]["primaryGroupID"]), "512")
492
493         # Note that Xs "objectSid" seems to be fine in the previous search for
494         # "objectSid"...
495         #res = ldb.search(expression="(primaryGroupID=*)", NULL, ldb. SCOPE_DEFAULT, attrs)
496         #print len(res) + " results found"
497         #for i in range(len(res)):
498         #    for (obj in res[i]) {
499         #        print obj + ": " + res[i][obj]
500         #    }
501         #    print "---"
502         #    
503
504         # Search by remote name of renamed attribute */
505         res = self.ldb.search(expression="(sambaBadPasswordCount=*)", 
506                               attrs=["dnsHostName", "lastLogon"])
507         self.assertEquals(len(res), 0)
508
509         # Search by objectClass
510         attrs = ["dnsHostName", "lastLogon", "objectClass"]
511         res = self.ldb.search(expression="(objectClass=user)", attrs=attrs)
512         self.assertEquals(len(res), 2)
513         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=X"))
514         self.assertEquals(str(res[0]["dnsHostName"]), "x")
515         self.assertEquals(str(res[0]["lastLogon"]), "x")
516         self.assertEquals(str(res[0]["objectClass"][0]), "user")
517         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
518         self.assertTrue(not "dnsHostName" in res[1])
519         self.assertEquals(str(res[1]["lastLogon"]), "x")
520         self.assertEquals(str(res[1]["objectClass"][0]), "user")
521
522         # Prove that the objectClass is actually used for the search
523         res = self.ldb.search(expression="(|(objectClass=user)(badPwdCount=x))",
524                               attrs=attrs)
525         self.assertEquals(len(res), 3)
526         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
527         self.assertTrue(not "dnsHostName" in res[0])
528         self.assertEquals(str(res[0]["lastLogon"]), "y")
529         self.assertEquals(set(res[0]["objectClass"]), set(["top"]))
530         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
531         self.assertEquals(str(res[1]["dnsHostName"]), "x")
532         self.assertEquals(str(res[1]["lastLogon"]), "x")
533         self.assertEquals(str(res[1]["objectClass"][0]), "user")
534         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=A"))
535         self.assertTrue(not "dnsHostName" in res[2])
536         self.assertEquals(str(res[2]["lastLogon"]), "x")
537         self.assertEquals(res[2]["objectClass"][0], "user")
538
539         # Testing search by parse tree
540
541         # Search by conjunction of local attributes
542         res = self.ldb.search(expression="(&(codePage=x)(revision=x))", 
543                               attrs=["dnsHostName", "lastLogon"])
544         self.assertEquals(len(res), 2)
545         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
546         self.assertEquals(str(res[0]["dnsHostName"]), "y")
547         self.assertEquals(str(res[0]["lastLogon"]), "y")
548         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
549         self.assertEquals(str(res[1]["dnsHostName"]), "x")
550         self.assertEquals(str(res[1]["lastLogon"]), "x")
551
552         # Search by conjunction of remote attributes
553         res = self.ldb.search(expression="(&(lastLogon=x)(description=x))", 
554                               attrs=["dnsHostName", "lastLogon"])
555         self.assertEquals(len(res), 2)
556         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=X"))
557         self.assertEquals(str(res[0]["dnsHostName"]), "x")
558         self.assertEquals(str(res[0]["lastLogon"]), "x")
559         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
560         self.assertTrue(not "dnsHostName" in res[1])
561         self.assertEquals(str(res[1]["lastLogon"]), "x")
562         
563         # Search by conjunction of local and remote attribute 
564         res = self.ldb.search(expression="(&(codePage=x)(description=x))", 
565                               attrs=["dnsHostName", "lastLogon"])
566         self.assertEquals(len(res), 2)
567         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
568         self.assertEquals(str(res[0]["dnsHostName"]), "y")
569         self.assertEquals(str(res[0]["lastLogon"]), "y")
570         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
571         self.assertEquals(str(res[1]["dnsHostName"]), "x")
572         self.assertEquals(str(res[1]["lastLogon"]), "x")
573
574         # Search by conjunction of local and remote attribute w/o match
575         attrs = ["dnsHostName", "lastLogon"]
576         res = self.ldb.search(expression="(&(codePage=x)(nextRid=x))", 
577                               attrs=attrs)
578         self.assertEquals(len(res), 0)
579         res = self.ldb.search(expression="(&(revision=x)(lastLogon=z))", 
580                               attrs=attrs)
581         self.assertEquals(len(res), 0)
582
583         # Search by disjunction of local attributes
584         res = self.ldb.search(expression="(|(revision=x)(dnsHostName=x))", 
585                               attrs=["dnsHostName", "lastLogon"])
586         self.assertEquals(len(res), 2)
587         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
588         self.assertEquals(str(res[0]["dnsHostName"]), "y")
589         self.assertEquals(str(res[0]["lastLogon"]), "y")
590         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
591         self.assertEquals(str(res[1]["dnsHostName"]), "x")
592         self.assertEquals(str(res[1]["lastLogon"]), "x")
593
594         # Search by disjunction of remote attributes
595         res = self.ldb.search(expression="(|(badPwdCount=x)(lastLogon=x))", 
596                               attrs=["dnsHostName", "lastLogon"])
597         self.assertEquals(len(res), 3)
598         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
599         self.assertFalse("dnsHostName" in res[0])
600         self.assertEquals(str(res[0]["lastLogon"]), "y")
601         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
602         self.assertEquals(str(res[1]["dnsHostName"]), "x")
603         self.assertEquals(str(res[1]["lastLogon"]), "x")
604         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=A"))
605         self.assertFalse("dnsHostName" in res[2])
606         self.assertEquals(str(res[2]["lastLogon"]), "x")
607
608         # Search by disjunction of local and remote attribute
609         res = self.ldb.search(expression="(|(revision=x)(lastLogon=y))", 
610                               attrs=["dnsHostName", "lastLogon"])
611         self.assertEquals(len(res), 3)
612         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
613         self.assertEquals(str(res[0]["dnsHostName"]), "y")
614         self.assertEquals(str(res[0]["lastLogon"]), "y")
615         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=B"))
616         self.assertFalse("dnsHostName" in res[1])
617         self.assertEquals(str(res[1]["lastLogon"]), "y")
618         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=X"))
619         self.assertEquals(str(res[2]["dnsHostName"]), "x")
620         self.assertEquals(str(res[2]["lastLogon"]), "x")
621
622         # Search by disjunction of local and remote attribute w/o match
623         res = self.ldb.search(expression="(|(codePage=y)(nextRid=z))", 
624                               attrs=["dnsHostName", "lastLogon"])
625         self.assertEquals(len(res), 0)
626
627         # Search by negated local attribute
628         res = self.ldb.search(expression="(!(revision=x))", 
629                               attrs=["dnsHostName", "lastLogon"])
630         self.assertEquals(len(res), 6)
631         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
632         self.assertTrue(not "dnsHostName" in res[0])
633         self.assertEquals(str(res[0]["lastLogon"]), "y")
634         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
635         self.assertTrue(not "dnsHostName" in res[1])
636         self.assertEquals(str(res[1]["lastLogon"]), "x")
637         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
638         self.assertEquals(str(res[2]["dnsHostName"]), "z")
639         self.assertEquals(str(res[2]["lastLogon"]), "z")
640         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
641         self.assertTrue(not "dnsHostName" in res[3])
642         self.assertEquals(str(res[3]["lastLogon"]), "z")
643
644         # Search by negated remote attribute
645         res = self.ldb.search(expression="(!(description=x))", 
646                               attrs=["dnsHostName", "lastLogon"])
647         self.assertEquals(len(res), 4)
648         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Z"))
649         self.assertEquals(str(res[0]["dnsHostName"]), "z")
650         self.assertEquals(str(res[0]["lastLogon"]), "z")
651         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=C"))
652         self.assertTrue(not "dnsHostName" in res[1])
653         self.assertEquals(str(res[1]["lastLogon"]), "z")
654
655         # Search by negated conjunction of local attributes
656         res = self.ldb.search(expression="(!(&(codePage=x)(revision=x)))", 
657                               attrs=["dnsHostName", "lastLogon"])
658         self.assertEquals(len(res), 6)
659         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
660         self.assertTrue(not "dnsHostName" in res[0])
661         self.assertEquals(str(res[0]["lastLogon"]), "y")
662         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
663         self.assertTrue(not "dnsHostName" in res[1])
664         self.assertEquals(str(res[1]["lastLogon"]), "x")
665         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
666         self.assertEquals(str(res[2]["dnsHostName"]), "z")
667         self.assertEquals(str(res[2]["lastLogon"]), "z")
668         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
669         self.assertTrue(not "dnsHostName" in res[3])
670         self.assertEquals(str(res[3]["lastLogon"]), "z")
671
672         # Search by negated conjunction of remote attributes
673         res = self.ldb.search(expression="(!(&(lastLogon=x)(description=x)))", 
674                               attrs=["dnsHostName", "lastLogon"])
675         self.assertEquals(len(res), 6)
676         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
677         self.assertEquals(str(res[0]["dnsHostName"]), "y")
678         self.assertEquals(str(res[0]["lastLogon"]), "y")
679         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=B"))
680         self.assertTrue(not "dnsHostName" in res[1])
681         self.assertEquals(str(res[1]["lastLogon"]), "y")
682         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
683         self.assertEquals(str(res[2]["dnsHostName"]), "z")
684         self.assertEquals(str(res[2]["lastLogon"]), "z")
685         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
686         self.assertTrue(not "dnsHostName" in res[3])
687         self.assertEquals(str(res[3]["lastLogon"]), "z")
688
689         # Search by negated conjunction of local and remote attribute
690         res = self.ldb.search(expression="(!(&(codePage=x)(description=x)))", 
691                               attrs=["dnsHostName", "lastLogon"])
692         self.assertEquals(len(res), 6)
693         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
694         self.assertTrue(not "dnsHostName" in res[0])
695         self.assertEquals(str(res[0]["lastLogon"]), "y")
696         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
697         self.assertTrue(not "dnsHostName" in res[1])
698         self.assertEquals(str(res[1]["lastLogon"]), "x")
699         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
700         self.assertEquals(str(res[2]["dnsHostName"]), "z")
701         self.assertEquals(str(res[2]["lastLogon"]), "z")
702         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
703         self.assertTrue(not "dnsHostName" in res[3])
704         self.assertEquals(str(res[3]["lastLogon"]), "z")
705
706         # Search by negated disjunction of local attributes
707         res = self.ldb.search(expression="(!(|(revision=x)(dnsHostName=x)))", 
708                               attrs=["dnsHostName", "lastLogon"])
709         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
710         self.assertTrue(not "dnsHostName" in res[0])
711         self.assertEquals(str(res[0]["lastLogon"]), "y")
712         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
713         self.assertTrue(not "dnsHostName" in res[1])
714         self.assertEquals(str(res[1]["lastLogon"]), "x")
715         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
716         self.assertEquals(str(res[2]["dnsHostName"]), "z")
717         self.assertEquals(str(res[2]["lastLogon"]), "z")
718         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
719         self.assertTrue(not "dnsHostName" in res[3])
720         self.assertEquals(str(res[3]["lastLogon"]), "z")
721
722         # Search by negated disjunction of remote attributes
723         res = self.ldb.search(expression="(!(|(badPwdCount=x)(lastLogon=x)))", 
724                               attrs=["dnsHostName", "lastLogon"])
725         self.assertEquals(len(res), 5)
726         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
727         self.assertEquals(str(res[0]["dnsHostName"]), "y")
728         self.assertEquals(str(res[0]["lastLogon"]), "y")
729         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=Z"))
730         self.assertEquals(str(res[1]["dnsHostName"]), "z")
731         self.assertEquals(str(res[1]["lastLogon"]), "z")
732         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=C"))
733         self.assertTrue(not "dnsHostName" in res[2])
734         self.assertEquals(str(res[2]["lastLogon"]), "z")
735
736         # Search by negated disjunction of local and remote attribute
737         res = self.ldb.search(expression="(!(|(revision=x)(lastLogon=y)))", 
738                               attrs=["dnsHostName", "lastLogon"])
739         self.assertEquals(len(res), 5)
740         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=A"))
741         self.assertTrue(not "dnsHostName" in res[0])
742         self.assertEquals(str(res[0]["lastLogon"]), "x")
743         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=Z"))
744         self.assertEquals(str(res[1]["dnsHostName"]), "z")
745         self.assertEquals(str(res[1]["lastLogon"]), "z")
746         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=C"))
747         self.assertTrue(not "dnsHostName" in res[2])
748         self.assertEquals(str(res[2]["lastLogon"]), "z")
749
750         # Search by complex parse tree
751         res = self.ldb.search(expression="(|(&(revision=x)(dnsHostName=x))(!(&(description=x)(nextRid=y)))(badPwdCount=y))", attrs=["dnsHostName", "lastLogon"])
752         self.assertEquals(len(res), 7)
753         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
754         self.assertTrue(not "dnsHostName" in res[0])
755         self.assertEquals(str(res[0]["lastLogon"]), "y")
756         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
757         self.assertEquals(str(res[1]["dnsHostName"]), "x")
758         self.assertEquals(str(res[1]["lastLogon"]), "x")
759         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=A"))
760         self.assertTrue(not "dnsHostName" in res[2])
761         self.assertEquals(str(res[2]["lastLogon"]), "x")
762         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=Z"))
763         self.assertEquals(str(res[3]["dnsHostName"]), "z")
764         self.assertEquals(str(res[3]["lastLogon"]), "z")
765         self.assertEquals(str(res[4].dn), self.samba4.dn("cn=C"))
766         self.assertTrue(not "dnsHostName" in res[4])
767         self.assertEquals(str(res[4]["lastLogon"]), "z")
768
769         # Clean up
770         dns = [self.samba4.dn("cn=%s" % n) for n in ["A","B","C","X","Y","Z"]]
771         for dn in dns:
772             self.ldb.delete(dn)
773
774     def test_map_modify_local(self):
775         """Modification of local records."""
776         # Add local record
777         dn = "cn=test,dc=idealx,dc=org"
778         self.ldb.add({"dn": dn, 
779                  "cn": "test",
780                  "foo": "bar",
781                  "revision": "1",
782                  "description": "test"})
783         # Check it's there
784         attrs = ["foo", "revision", "description"]
785         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
786         self.assertEquals(len(res), 1)
787         self.assertEquals(str(res[0].dn), dn)
788         self.assertEquals(str(res[0]["foo"]), "bar")
789         self.assertEquals(str(res[0]["revision"]), "1")
790         self.assertEquals(str(res[0]["description"]), "test")
791         # Check it's not in the local db
792         res = self.samba4.db.search(expression="(cn=test)", 
793                                     scope=SCOPE_DEFAULT, attrs=attrs)
794         self.assertEquals(len(res), 0)
795         # Check it's not in the remote db
796         res = self.samba3.db.search(expression="(cn=test)", 
797                                     scope=SCOPE_DEFAULT, attrs=attrs)
798         self.assertEquals(len(res), 0)
799
800         # Modify local record
801         ldif = """
802 dn: """ + dn + """
803 replace: foo
804 foo: baz
805 replace: description
806 description: foo
807 """
808         self.ldb.modify_ldif(ldif)
809         # Check in local db
810         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
811         self.assertEquals(len(res), 1)
812         self.assertEquals(str(res[0].dn), dn)
813         self.assertEquals(str(res[0]["foo"]), "baz")
814         self.assertEquals(str(res[0]["revision"]), "1")
815         self.assertEquals(str(res[0]["description"]), "foo")
816
817         # Rename local record
818         dn2 = "cn=toast,dc=idealx,dc=org"
819         self.ldb.rename(dn, dn2)
820         # Check in local db
821         res = self.ldb.search(dn2, scope=SCOPE_BASE, attrs=attrs)
822         self.assertEquals(len(res), 1)
823         self.assertEquals(str(res[0].dn), dn2)
824         self.assertEquals(str(res[0]["foo"]), "baz")
825         self.assertEquals(str(res[0]["revision"]), "1")
826         self.assertEquals(str(res[0]["description"]), "foo")
827
828         # Delete local record
829         self.ldb.delete(dn2)
830         # Check it's gone
831         res = self.ldb.search(dn2, scope=SCOPE_BASE)
832         self.assertEquals(len(res), 0)
833
834     def test_map_modify_remote_remote(self):
835         """Modification of remote data of remote records"""
836         # Add remote record
837         dn = self.samba4.dn("cn=test")
838         dn2 = self.samba3.dn("cn=test")
839         self.samba3.db.add({"dn": dn2, 
840                    "cn": "test",
841                    "description": "foo",
842                    "sambaBadPasswordCount": "3",
843                    "sambaNextRid": "1001"})
844         # Check it's there
845         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, 
846                 attrs=["description", "sambaBadPasswordCount", "sambaNextRid"])
847         self.assertEquals(len(res), 1)
848         self.assertEquals(str(res[0].dn), dn2)
849         self.assertEquals(str(res[0]["description"]), "foo")
850         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "3")
851         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
852         # Check in mapped db
853         attrs = ["description", "badPwdCount", "nextRid"]
854         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs, expression="")
855         self.assertEquals(len(res), 1)
856         self.assertEquals(str(res[0].dn), dn)
857         self.assertEquals(str(res[0]["description"]), "foo")
858         self.assertEquals(str(res[0]["badPwdCount"]), "3")
859         self.assertEquals(str(res[0]["nextRid"]), "1001")
860         # Check in local db
861         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
862         self.assertEquals(len(res), 0)
863
864         # Modify remote data of remote record
865         ldif = """
866 dn: """ + dn + """
867 replace: description
868 description: test
869 replace: badPwdCount
870 badPwdCount: 4
871 """
872         self.ldb.modify_ldif(ldif)
873         # Check in mapped db
874         res = self.ldb.search(dn, scope=SCOPE_BASE, 
875                 attrs=["description", "badPwdCount", "nextRid"])
876         self.assertEquals(len(res), 1)
877         self.assertEquals(str(res[0].dn), dn)
878         self.assertEquals(str(res[0]["description"]), "test")
879         self.assertEquals(str(res[0]["badPwdCount"]), "4")
880         self.assertEquals(str(res[0]["nextRid"]), "1001")
881         # Check in remote db
882         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, 
883                 attrs=["description", "sambaBadPasswordCount", "sambaNextRid"])
884         self.assertEquals(len(res), 1)
885         self.assertEquals(str(res[0].dn), dn2)
886         self.assertEquals(str(res[0]["description"]), "test")
887         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
888         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
889
890         # Rename remote record
891         dn2 = self.samba4.dn("cn=toast")
892         self.ldb.rename(dn, dn2)
893         # Check in mapped db
894         dn = dn2
895         res = self.ldb.search(dn, scope=SCOPE_BASE, 
896                 attrs=["description", "badPwdCount", "nextRid"])
897         self.assertEquals(len(res), 1)
898         self.assertEquals(str(res[0].dn), dn)
899         self.assertEquals(str(res[0]["description"]), "test")
900         self.assertEquals(str(res[0]["badPwdCount"]), "4")
901         self.assertEquals(str(res[0]["nextRid"]), "1001")
902         # Check in remote db 
903         dn2 = self.samba3.dn("cn=toast")
904         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, 
905                 attrs=["description", "sambaBadPasswordCount", "sambaNextRid"])
906         self.assertEquals(len(res), 1)
907         self.assertEquals(str(res[0].dn), dn2)
908         self.assertEquals(str(res[0]["description"]), "test")
909         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
910         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
911
912         # Delete remote record
913         self.ldb.delete(dn)
914         # Check in mapped db that it's removed
915         res = self.ldb.search(dn, scope=SCOPE_BASE)
916         self.assertEquals(len(res), 0)
917         # Check in remote db
918         res = self.samba3.db.search(dn2, scope=SCOPE_BASE)
919         self.assertEquals(len(res), 0)
920
921     def test_map_modify_remote_local(self):
922         """Modification of local data of remote records"""
923         # Add remote record (same as before)
924         dn = self.samba4.dn("cn=test")
925         dn2 = self.samba3.dn("cn=test")
926         self.samba3.db.add({"dn": dn2, 
927                    "cn": "test",
928                    "description": "foo",
929                    "sambaBadPasswordCount": "3",
930                    "sambaNextRid": "1001"})
931
932         # Modify local data of remote record
933         ldif = """
934 dn: """ + dn + """
935 add: revision
936 revision: 1
937 replace: description
938 description: test
939
940 """
941         self.ldb.modify_ldif(ldif)
942         # Check in mapped db
943         attrs = ["revision", "description"]
944         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
945         self.assertEquals(len(res), 1)
946         self.assertEquals(str(res[0].dn), dn)
947         self.assertEquals(str(res[0]["description"]), "test")
948         self.assertEquals(str(res[0]["revision"]), "1")
949         # Check in remote db
950         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, attrs=attrs)
951         self.assertEquals(len(res), 1)
952         self.assertEquals(str(res[0].dn), dn2)
953         self.assertEquals(str(res[0]["description"]), "test")
954         self.assertTrue(not "revision" in res[0])
955         # Check in local db
956         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
957         self.assertEquals(len(res), 1)
958         self.assertEquals(str(res[0].dn), dn)
959         self.assertTrue(not "description" in res[0])
960         self.assertEquals(str(res[0]["revision"]), "1")
961
962         # Delete (newly) split record
963         self.ldb.delete(dn)
964
965     def test_map_modify_split(self):
966         """Testing modification of split records"""
967         # Add split record
968         dn = self.samba4.dn("cn=test")
969         dn2 = self.samba3.dn("cn=test")
970         self.ldb.add({
971             "dn": dn,
972             "cn": "test",
973             "description": "foo",
974             "badPwdCount": "3",
975             "nextRid": "1001",
976             "revision": "1"})
977         # Check it's there
978         attrs = ["description", "badPwdCount", "nextRid", "revision"]
979         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
980         self.assertEquals(len(res), 1)
981         self.assertEquals(str(res[0].dn), dn)
982         self.assertEquals(str(res[0]["description"]), "foo")
983         self.assertEquals(str(res[0]["badPwdCount"]), "3")
984         self.assertEquals(str(res[0]["nextRid"]), "1001")
985         self.assertEquals(str(res[0]["revision"]), "1")
986         # Check in local db
987         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
988         self.assertEquals(len(res), 1)
989         self.assertEquals(str(res[0].dn), dn)
990         self.assertTrue(not "description" in res[0])
991         self.assertTrue(not "badPwdCount" in res[0])
992         self.assertTrue(not "nextRid" in res[0])
993         self.assertEquals(str(res[0]["revision"]), "1")
994         # Check in remote db
995         attrs = ["description", "sambaBadPasswordCount", "sambaNextRid", 
996                  "revision"]
997         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, attrs=attrs)
998         self.assertEquals(len(res), 1)
999         self.assertEquals(str(res[0].dn), dn2)
1000         self.assertEquals(str(res[0]["description"]), "foo")
1001         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "3")
1002         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
1003         self.assertTrue(not "revision" in res[0])
1004
1005         # Modify of split record
1006         ldif = """
1007 dn: """ + dn + """
1008 replace: description
1009 description: test
1010 replace: badPwdCount
1011 badPwdCount: 4
1012 replace: revision
1013 revision: 2
1014 """
1015         self.ldb.modify_ldif(ldif)
1016         # Check in mapped db
1017         attrs = ["description", "badPwdCount", "nextRid", "revision"]
1018         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
1019         self.assertEquals(len(res), 1)
1020         self.assertEquals(str(res[0].dn), dn)
1021         self.assertEquals(str(res[0]["description"]), "test")
1022         self.assertEquals(str(res[0]["badPwdCount"]), "4")
1023         self.assertEquals(str(res[0]["nextRid"]), "1001")
1024         self.assertEquals(str(res[0]["revision"]), "2")
1025         # Check in local db
1026         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
1027         self.assertEquals(len(res), 1)
1028         self.assertEquals(str(res[0].dn), dn)
1029         self.assertTrue(not "description" in res[0])
1030         self.assertTrue(not "badPwdCount" in res[0])
1031         self.assertTrue(not "nextRid" in res[0])
1032         self.assertEquals(str(res[0]["revision"]), "2")
1033         # Check in remote db
1034         attrs = ["description", "sambaBadPasswordCount", "sambaNextRid", 
1035                  "revision"]
1036         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, attrs=attrs)
1037         self.assertEquals(len(res), 1)
1038         self.assertEquals(str(res[0].dn), dn2)
1039         self.assertEquals(str(res[0]["description"]), "test")
1040         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
1041         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
1042         self.assertTrue(not "revision" in res[0])
1043
1044         # Rename split record
1045         dn2 = self.samba4.dn("cn=toast")
1046         self.ldb.rename(dn, dn2)
1047         # Check in mapped db
1048         dn = dn2
1049         attrs = ["description", "badPwdCount", "nextRid", "revision"]
1050         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
1051         self.assertEquals(len(res), 1)
1052         self.assertEquals(str(res[0].dn), dn)
1053         self.assertEquals(str(res[0]["description"]), "test")
1054         self.assertEquals(str(res[0]["badPwdCount"]), "4")
1055         self.assertEquals(str(res[0]["nextRid"]), "1001")
1056         self.assertEquals(str(res[0]["revision"]), "2")
1057         # Check in local db
1058         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
1059         self.assertEquals(len(res), 1)
1060         self.assertEquals(str(res[0].dn), dn)
1061         self.assertTrue(not "description" in res[0])
1062         self.assertTrue(not "badPwdCount" in res[0])
1063         self.assertTrue(not "nextRid" in res[0])
1064         self.assertEquals(str(res[0]["revision"]), "2")
1065         # Check in remote db
1066         dn2 = self.samba3.dn("cn=toast")
1067         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, 
1068           attrs=["description", "sambaBadPasswordCount", "sambaNextRid", 
1069                  "revision"])
1070         self.assertEquals(len(res), 1)
1071         self.assertEquals(str(res[0].dn), dn2)
1072         self.assertEquals(str(res[0]["description"]), "test")
1073         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
1074         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
1075         self.assertTrue(not "revision" in res[0])
1076
1077         # Delete split record
1078         self.ldb.delete(dn)
1079         # Check in mapped db
1080         res = self.ldb.search(dn, scope=SCOPE_BASE)
1081         self.assertEquals(len(res), 0)
1082         # Check in local db
1083         res = self.samba4.db.search(dn, scope=SCOPE_BASE)
1084         self.assertEquals(len(res), 0)
1085         # Check in remote db
1086         res = self.samba3.db.search(dn2, scope=SCOPE_BASE)
1087         self.assertEquals(len(res), 0)