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