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