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=X") + """
 317 objectClass: user
 318 cn: X
 319 codePage: x
 320 revision: x
 321 dnsHostName: x
 322 nextRid: y
 323 lastLogon: x
 324 description: x
 325 objectSid: S-1-5-21-4231626423-2410014848-2360679739-552
 326 primaryGroupID: 1-5-21-4231626423-2410014848-2360679739-512
 327
 328 """)
 329
 330         self.ldb.add({
 331             "dn": self.samba4.dn("cn=Y"),
 332             "objectClass": "top",
 333             "cn": "Y",
 334             "codePage": "x",
 335             "revision": "x",
 336             "dnsHostName": "y",
 337             "nextRid": "y",
 338             "lastLogon": "y",
 339             "description": "x"})
 340
 341         self.ldb.add({
 342             "dn": self.samba4.dn("cn=Z"),
 343             "objectClass": "top",
 344             "cn": "Z",
 345             "codePage": "x",
 346             "revision": "y",
 347             "dnsHostName": "z",
 348             "nextRid": "y",
 349             "lastLogon": "z",
 350             "description": "y"})
 351
 352         # Add a set of remote records
 353
 354         self.samba3.db.add({
 355             "dn": self.samba3.dn("cn=A"),
 356             "objectClass": "posixAccount",
 357             "cn": "A",
 358             "sambaNextRid": "x",
 359             "sambaBadPasswordCount": "x",
 360             "sambaLogonTime": "x",
 361             "description": "x",
 362             "sambaSID": "S-1-5-21-4231626423-2410014848-2360679739-552",
 363             "sambaPrimaryGroupSID": "S-1-5-21-4231626423-2410014848-2360679739-512"})
 364
 365         self.samba3.db.add({
 366             "dn": self.samba3.dn("cn=B"),
 367             "objectClass": "top",
 368             "cn": "B",
 369             "sambaNextRid": "x",
 370             "sambaBadPasswordCount": "x",
 371             "sambaLogonTime": "y",
 372             "description": "x"})
 373
 374         self.samba3.db.add({
 375             "dn": self.samba3.dn("cn=C"),
 376             "objectClass": "top",
 377             "cn": "C",
 378             "sambaNextRid": "x",
 379             "sambaBadPasswordCount": "y",
 380             "sambaLogonTime": "z",
 381             "description": "y"})
 382
 383         # Testing search by DN
 384
 385         # Search remote record by local DN
 386         dn = self.samba4.dn("cn=A")
 387         res = self.ldb.search(dn, scope=SCOPE_BASE,
 388                 attrs=["dnsHostName", "lastLogon"])
 389         self.assertEquals(len(res), 1)
 390         self.assertEquals(str(res[0].dn), dn)
 391         self.assertTrue(not "dnsHostName" in res[0])
 392         self.assertEquals(str(res[0]["lastLogon"]), "x")
 393
 394         # Search remote record by remote DN
 395         dn = self.samba3.dn("cn=A")
 396         res = self.samba3.db.search(dn, scope=SCOPE_BASE,
 397                 attrs=["dnsHostName", "lastLogon", "sambaLogonTime"])
 398         self.assertEquals(len(res), 1)
 399         self.assertEquals(str(res[0].dn), dn)
 400         self.assertTrue(not "dnsHostName" in res[0])
 401         self.assertTrue(not "lastLogon" in res[0])
 402         self.assertEquals(str(res[0]["sambaLogonTime"]), "x")
 403
 404         # Search split record by local DN
 405         dn = self.samba4.dn("cn=X")
 406         res = self.ldb.search(dn, scope=SCOPE_BASE,
 407                 attrs=["dnsHostName", "lastLogon"])
 408         self.assertEquals(len(res), 1)
 409         self.assertEquals(str(res[0].dn), dn)
 410         self.assertEquals(str(res[0]["dnsHostName"]), "x")
 411         self.assertEquals(str(res[0]["lastLogon"]), "x")
 412
 413         # Search split record by remote DN
 414         dn = self.samba3.dn("cn=X")
 415         res = self.samba3.db.search(dn, scope=SCOPE_BASE,
 416                 attrs=["dnsHostName", "lastLogon", "sambaLogonTime"])
 417         self.assertEquals(len(res), 1)
 418         self.assertEquals(str(res[0].dn), dn)
 419         self.assertTrue(not "dnsHostName" in res[0])
 420         self.assertTrue(not "lastLogon" in res[0])
 421         self.assertEquals(str(res[0]["sambaLogonTime"]), "x")
 422
 423         # Testing search by attribute
 424
 425         # Search by ignored attribute
 426         res = self.ldb.search(expression="(revision=x)", scope=SCOPE_DEFAULT,
 427                 attrs=["dnsHostName", "lastLogon"])
 428         self.assertEquals(len(res), 2)
 429         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
 430         self.assertEquals(str(res[0]["dnsHostName"]), "y")
 431         self.assertEquals(str(res[0]["lastLogon"]), "y")
 432         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
 433         self.assertEquals(str(res[1]["dnsHostName"]), "x")
 434         self.assertEquals(str(res[1]["lastLogon"]), "x")
 435
 436         # Search by kept attribute
 437         res = self.ldb.search(expression="(description=y)",
 438                 scope=SCOPE_DEFAULT, attrs=["dnsHostName", "lastLogon"])
 439         self.assertEquals(len(res), 2)
 440         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Z"))
 441         self.assertEquals(str(res[0]["dnsHostName"]), "z")
 442         self.assertEquals(str(res[0]["lastLogon"]), "z")
 443         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=C"))
 444         self.assertTrue(not "dnsHostName" in res[1])
 445         self.assertEquals(str(res[1]["lastLogon"]), "z")
 446
 447         # Search by renamed attribute
 448         res = self.ldb.search(expression="(badPwdCount=x)", scope=SCOPE_DEFAULT,
 449                               attrs=["dnsHostName", "lastLogon"])
 450         self.assertEquals(len(res), 2)
 451         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 452         self.assertTrue(not "dnsHostName" in res[0])
 453         self.assertEquals(str(res[0]["lastLogon"]), "y")
 454         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 455         self.assertTrue(not "dnsHostName" in res[1])
 456         self.assertEquals(str(res[1]["lastLogon"]), "x")
 457
 458         # Search by converted attribute
 459         # TODO:
 460         #   Using the SID directly in the parse tree leads to conversion
 461         #   errors, letting the search fail with no results.
 462         #res = self.ldb.search("(objectSid=S-1-5-21-4231626423-2410014848-2360679739-552)", scope=SCOPE_DEFAULT, attrs)
 463         res = self.ldb.search(expression="(objectSid=*)", base=None, scope=SCOPE_DEFAULT, attrs=["dnsHostName", "lastLogon", "objectSid"])
 464         self.assertEquals(len(res), 3)
 465         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=X"))
 466         self.assertEquals(str(res[0]["dnsHostName"]), "x")
 467         self.assertEquals(str(res[0]["lastLogon"]), "x")
 468         self.assertSidEquals("S-1-5-21-4231626423-2410014848-2360679739-552",
 469                              res[0]["objectSid"])
 470         self.assertTrue("objectSid" in res[0])
 471         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 472         self.assertTrue(not "dnsHostName" in res[1])
 473         self.assertEquals(str(res[1]["lastLogon"]), "x")
 474         self.assertSidEquals("S-1-5-21-4231626423-2410014848-2360679739-552",
 475                              res[1]["objectSid"])
 476         self.assertTrue("objectSid" in res[1])
 477
 478         # Search by generated attribute
 479         # In most cases, this even works when the mapping is missing
 480         # a `convert_operator' by enumerating the remote db.
 481         res = self.ldb.search(expression="(primaryGroupID=512)",
 482                            attrs=["dnsHostName", "lastLogon", "primaryGroupID"])
 483         self.assertEquals(len(res), 1)
 484         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=A"))
 485         self.assertTrue(not "dnsHostName" in res[0])
 486         self.assertEquals(str(res[0]["lastLogon"]), "x")
 487         self.assertEquals(str(res[0]["primaryGroupID"]), "512")
 488
 489         # TODO: There should actually be two results, A and X.  The
 490         # primaryGroupID of X seems to get corrupted somewhere, and the
 491         # objectSid isn't available during the generation of remote (!) data,
 492         # which can be observed with the following search.  Also note that Xs
 493         # objectSid seems to be fine in the previous search for objectSid... */
 494         #res = ldb.search(expression="(primaryGroupID=*)", NULL, ldb. SCOPE_DEFAULT, attrs)
 495         #print len(res) + " results found"
 496         #for i in range(len(res)):
 497         #    for (obj in res[i]) {
 498         #        print obj + ": " + res[i][obj]
 499         #    }
 500         #    print "---"
 501         #
 502
 503         # Search by remote name of renamed attribute */
 504         res = self.ldb.search(expression="(sambaBadPasswordCount=*)",
 505                               attrs=["dnsHostName", "lastLogon"])
 506         self.assertEquals(len(res), 0)
 507
 508         # Search by objectClass
 509         attrs = ["dnsHostName", "lastLogon", "objectClass"]
 510         res = self.ldb.search(expression="(objectClass=user)", attrs=attrs)
 511         self.assertEquals(len(res), 2)
 512         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=X"))
 513         self.assertEquals(str(res[0]["dnsHostName"]), "x")
 514         self.assertEquals(str(res[0]["lastLogon"]), "x")
 515         self.assertEquals(str(res[0]["objectClass"][0]), "user")
 516         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 517         self.assertTrue(not "dnsHostName" in res[1])
 518         self.assertEquals(str(res[1]["lastLogon"]), "x")
 519         self.assertEquals(str(res[1]["objectClass"][0]), "user")
 520
 521         # Prove that the objectClass is actually used for the search
 522         res = self.ldb.search(expression="(|(objectClass=user)(badPwdCount=x))",
 523                               attrs=attrs)
 524         self.assertEquals(len(res), 3)
 525         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 526         self.assertTrue(not "dnsHostName" in res[0])
 527         self.assertEquals(str(res[0]["lastLogon"]), "y")
 528         self.assertEquals(set(res[0]["objectClass"]), set(["top"]))
 529         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
 530         self.assertEquals(str(res[1]["dnsHostName"]), "x")
 531         self.assertEquals(str(res[1]["lastLogon"]), "x")
 532         self.assertEquals(str(res[1]["objectClass"][0]), "user")
 533         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=A"))
 534         self.assertTrue(not "dnsHostName" in res[2])
 535         self.assertEquals(str(res[2]["lastLogon"]), "x")
 536         self.assertEquals(res[2]["objectClass"][0], "user")
 537
 538         # Testing search by parse tree
 539
 540         # Search by conjunction of local attributes
 541         res = self.ldb.search(expression="(&(codePage=x)(revision=x))",
 542                               attrs=["dnsHostName", "lastLogon"])
 543         self.assertEquals(len(res), 2)
 544         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
 545         self.assertEquals(str(res[0]["dnsHostName"]), "y")
 546         self.assertEquals(str(res[0]["lastLogon"]), "y")
 547         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
 548         self.assertEquals(str(res[1]["dnsHostName"]), "x")
 549         self.assertEquals(str(res[1]["lastLogon"]), "x")
 550
 551         # Search by conjunction of remote attributes
 552         res = self.ldb.search(expression="(&(lastLogon=x)(description=x))",
 553                               attrs=["dnsHostName", "lastLogon"])
 554         self.assertEquals(len(res), 2)
 555         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=X"))
 556         self.assertEquals(str(res[0]["dnsHostName"]), "x")
 557         self.assertEquals(str(res[0]["lastLogon"]), "x")
 558         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 559         self.assertTrue(not "dnsHostName" in res[1])
 560         self.assertEquals(str(res[1]["lastLogon"]), "x")
 561
 562         # Search by conjunction of local and remote attribute
 563         res = self.ldb.search(expression="(&(codePage=x)(description=x))",
 564                               attrs=["dnsHostName", "lastLogon"])
 565         self.assertEquals(len(res), 2)
 566         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
 567         self.assertEquals(str(res[0]["dnsHostName"]), "y")
 568         self.assertEquals(str(res[0]["lastLogon"]), "y")
 569         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
 570         self.assertEquals(str(res[1]["dnsHostName"]), "x")
 571         self.assertEquals(str(res[1]["lastLogon"]), "x")
 572
 573         # Search by conjunction of local and remote attribute w/o match
 574         attrs = ["dnsHostName", "lastLogon"]
 575         res = self.ldb.search(expression="(&(codePage=x)(nextRid=x))",
 576                               attrs=attrs)
 577         self.assertEquals(len(res), 0)
 578         res = self.ldb.search(expression="(&(revision=x)(lastLogon=z))",
 579                               attrs=attrs)
 580         self.assertEquals(len(res), 0)
 581
 582         # Search by disjunction of local attributes
 583         res = self.ldb.search(expression="(|(revision=x)(dnsHostName=x))",
 584                               attrs=["dnsHostName", "lastLogon"])
 585         self.assertEquals(len(res), 2)
 586         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
 587         self.assertEquals(str(res[0]["dnsHostName"]), "y")
 588         self.assertEquals(str(res[0]["lastLogon"]), "y")
 589         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
 590         self.assertEquals(str(res[1]["dnsHostName"]), "x")
 591         self.assertEquals(str(res[1]["lastLogon"]), "x")
 592
 593         # Search by disjunction of remote attributes
 594         res = self.ldb.search(expression="(|(badPwdCount=x)(lastLogon=x))",
 595                               attrs=["dnsHostName", "lastLogon"])
 596         self.assertEquals(len(res), 3)
 597         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 598         self.assertFalse("dnsHostName" in res[0])
 599         self.assertEquals(str(res[0]["lastLogon"]), "y")
 600         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
 601         self.assertEquals(str(res[1]["dnsHostName"]), "x")
 602         self.assertEquals(str(res[1]["lastLogon"]), "x")
 603         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=A"))
 604         self.assertFalse("dnsHostName" in res[2])
 605         self.assertEquals(str(res[2]["lastLogon"]), "x")
 606
 607         # Search by disjunction of local and remote attribute
 608         res = self.ldb.search(expression="(|(revision=x)(lastLogon=y))",
 609                               attrs=["dnsHostName", "lastLogon"])
 610         self.assertEquals(len(res), 3)
 611         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
 612         self.assertEquals(str(res[0]["dnsHostName"]), "y")
 613         self.assertEquals(str(res[0]["lastLogon"]), "y")
 614         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=B"))
 615         self.assertFalse("dnsHostName" in res[1])
 616         self.assertEquals(str(res[1]["lastLogon"]), "y")
 617         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=X"))
 618         self.assertEquals(str(res[2]["dnsHostName"]), "x")
 619         self.assertEquals(str(res[2]["lastLogon"]), "x")
 620
 621         # Search by disjunction of local and remote attribute w/o match
 622         res = self.ldb.search(expression="(|(codePage=y)(nextRid=z))",
 623                               attrs=["dnsHostName", "lastLogon"])
 624         self.assertEquals(len(res), 0)
 625
 626         # Search by negated local attribute
 627         res = self.ldb.search(expression="(!(revision=x))",
 628                               attrs=["dnsHostName", "lastLogon"])
 629         self.assertEquals(len(res), 5)
 630         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 631         self.assertTrue(not "dnsHostName" in res[0])
 632         self.assertEquals(str(res[0]["lastLogon"]), "y")
 633         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 634         self.assertTrue(not "dnsHostName" in res[1])
 635         self.assertEquals(str(res[1]["lastLogon"]), "x")
 636         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
 637         self.assertEquals(str(res[2]["dnsHostName"]), "z")
 638         self.assertEquals(str(res[2]["lastLogon"]), "z")
 639         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
 640         self.assertTrue(not "dnsHostName" in res[3])
 641         self.assertEquals(str(res[3]["lastLogon"]), "z")
 642
 643         # Search by negated remote attribute
 644         res = self.ldb.search(expression="(!(description=x))",
 645                               attrs=["dnsHostName", "lastLogon"])
 646         self.assertEquals(len(res), 3)
 647         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Z"))
 648         self.assertEquals(str(res[0]["dnsHostName"]), "z")
 649         self.assertEquals(str(res[0]["lastLogon"]), "z")
 650         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=C"))
 651         self.assertTrue(not "dnsHostName" in res[1])
 652         self.assertEquals(str(res[1]["lastLogon"]), "z")
 653
 654         # Search by negated conjunction of local attributes
 655         res = self.ldb.search(expression="(!(&(codePage=x)(revision=x)))",
 656                               attrs=["dnsHostName", "lastLogon"])
 657         self.assertEquals(len(res), 5)
 658         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 659         self.assertTrue(not "dnsHostName" in res[0])
 660         self.assertEquals(str(res[0]["lastLogon"]), "y")
 661         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 662         self.assertTrue(not "dnsHostName" in res[1])
 663         self.assertEquals(str(res[1]["lastLogon"]), "x")
 664         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
 665         self.assertEquals(str(res[2]["dnsHostName"]), "z")
 666         self.assertEquals(str(res[2]["lastLogon"]), "z")
 667         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
 668         self.assertTrue(not "dnsHostName" in res[3])
 669         self.assertEquals(str(res[3]["lastLogon"]), "z")
 670
 671         # Search by negated conjunction of remote attributes
 672         res = self.ldb.search(expression="(!(&(lastLogon=x)(description=x)))",
 673                               attrs=["dnsHostName", "lastLogon"])
 674         self.assertEquals(len(res), 5)
 675         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
 676         self.assertEquals(str(res[0]["dnsHostName"]), "y")
 677         self.assertEquals(str(res[0]["lastLogon"]), "y")
 678         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=B"))
 679         self.assertTrue(not "dnsHostName" in res[1])
 680         self.assertEquals(str(res[1]["lastLogon"]), "y")
 681         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
 682         self.assertEquals(str(res[2]["dnsHostName"]), "z")
 683         self.assertEquals(str(res[2]["lastLogon"]), "z")
 684         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
 685         self.assertTrue(not "dnsHostName" in res[3])
 686         self.assertEquals(str(res[3]["lastLogon"]), "z")
 687
 688         # Search by negated conjunction of local and remote attribute
 689         res = self.ldb.search(expression="(!(&(codePage=x)(description=x)))",
 690                               attrs=["dnsHostName", "lastLogon"])
 691         self.assertEquals(len(res), 5)
 692         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 693         self.assertTrue(not "dnsHostName" in res[0])
 694         self.assertEquals(str(res[0]["lastLogon"]), "y")
 695         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 696         self.assertTrue(not "dnsHostName" in res[1])
 697         self.assertEquals(str(res[1]["lastLogon"]), "x")
 698         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
 699         self.assertEquals(str(res[2]["dnsHostName"]), "z")
 700         self.assertEquals(str(res[2]["lastLogon"]), "z")
 701         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
 702         self.assertTrue(not "dnsHostName" in res[3])
 703         self.assertEquals(str(res[3]["lastLogon"]), "z")
 704
 705         # Search by negated disjunction of local attributes
 706         res = self.ldb.search(expression="(!(|(revision=x)(dnsHostName=x)))",
 707                               attrs=["dnsHostName", "lastLogon"])
 708         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 709         self.assertTrue(not "dnsHostName" in res[0])
 710         self.assertEquals(str(res[0]["lastLogon"]), "y")
 711         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=A"))
 712         self.assertTrue(not "dnsHostName" in res[1])
 713         self.assertEquals(str(res[1]["lastLogon"]), "x")
 714         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=Z"))
 715         self.assertEquals(str(res[2]["dnsHostName"]), "z")
 716         self.assertEquals(str(res[2]["lastLogon"]), "z")
 717         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=C"))
 718         self.assertTrue(not "dnsHostName" in res[3])
 719         self.assertEquals(str(res[3]["lastLogon"]), "z")
 720
 721         # Search by negated disjunction of remote attributes
 722         res = self.ldb.search(expression="(!(|(badPwdCount=x)(lastLogon=x)))",
 723                               attrs=["dnsHostName", "lastLogon"])
 724         self.assertEquals(len(res), 4)
 725         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=Y"))
 726         self.assertEquals(str(res[0]["dnsHostName"]), "y")
 727         self.assertEquals(str(res[0]["lastLogon"]), "y")
 728         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=Z"))
 729         self.assertEquals(str(res[1]["dnsHostName"]), "z")
 730         self.assertEquals(str(res[1]["lastLogon"]), "z")
 731         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=C"))
 732         self.assertTrue(not "dnsHostName" in res[2])
 733         self.assertEquals(str(res[2]["lastLogon"]), "z")
 734
 735         # Search by negated disjunction of local and remote attribute
 736         res = self.ldb.search(expression="(!(|(revision=x)(lastLogon=y)))",
 737                               attrs=["dnsHostName", "lastLogon"])
 738         self.assertEquals(len(res), 4)
 739         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=A"))
 740         self.assertTrue(not "dnsHostName" in res[0])
 741         self.assertEquals(str(res[0]["lastLogon"]), "x")
 742         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=Z"))
 743         self.assertEquals(str(res[1]["dnsHostName"]), "z")
 744         self.assertEquals(str(res[1]["lastLogon"]), "z")
 745         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=C"))
 746         self.assertTrue(not "dnsHostName" in res[2])
 747         self.assertEquals(str(res[2]["lastLogon"]), "z")
 748
 749         # Search by complex parse tree
 750         res = self.ldb.search(expression="(|(&(revision=x)(dnsHostName=x))(!(&(description=x)(nextRid=y)))(badPwdCount=y))", attrs=["dnsHostName", "lastLogon"])
 751         self.assertEquals(len(res), 6)
 752         self.assertEquals(str(res[0].dn), self.samba4.dn("cn=B"))
 753         self.assertTrue(not "dnsHostName" in res[0])
 754         self.assertEquals(str(res[0]["lastLogon"]), "y")
 755         self.assertEquals(str(res[1].dn), self.samba4.dn("cn=X"))
 756         self.assertEquals(str(res[1]["dnsHostName"]), "x")
 757         self.assertEquals(str(res[1]["lastLogon"]), "x")
 758         self.assertEquals(str(res[2].dn), self.samba4.dn("cn=A"))
 759         self.assertTrue(not "dnsHostName" in res[2])
 760         self.assertEquals(str(res[2]["lastLogon"]), "x")
 761         self.assertEquals(str(res[3].dn), self.samba4.dn("cn=Z"))
 762         self.assertEquals(str(res[3]["dnsHostName"]), "z")
 763         self.assertEquals(str(res[3]["lastLogon"]), "z")
 764         self.assertEquals(str(res[4].dn), self.samba4.dn("cn=C"))
 765         self.assertTrue(not "dnsHostName" in res[4])
 766         self.assertEquals(str(res[4]["lastLogon"]), "z")
 767
 768         # Clean up
 769         dns = [self.samba4.dn("cn=%s" % n) for n in ["A","B","C","X","Y","Z"]]
 770         for dn in dns:
 771             self.ldb.delete(dn)
 772
 773     def test_map_modify_local(self):
 774         """Modification of local records."""
 775         # Add local record
 776         dn = "cn=test,dc=idealx,dc=org"
 777         self.ldb.add({"dn": dn,
 778                  "cn": "test",
 779                  "foo": "bar",
 780                  "revision": "1",
 781                  "description": "test"})
 782         # Check it's there
 783         attrs = ["foo", "revision", "description"]
 784         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
 785         self.assertEquals(len(res), 1)
 786         self.assertEquals(str(res[0].dn), dn)
 787         self.assertEquals(str(res[0]["foo"]), "bar")
 788         self.assertEquals(str(res[0]["revision"]), "1")
 789         self.assertEquals(str(res[0]["description"]), "test")
 790         # Check it's not in the local db
 791         res = self.samba4.db.search(expression="(cn=test)",
 792                                     scope=SCOPE_DEFAULT, attrs=attrs)
 793         self.assertEquals(len(res), 0)
 794         # Check it's not in the remote db
 795         res = self.samba3.db.search(expression="(cn=test)",
 796                                     scope=SCOPE_DEFAULT, attrs=attrs)
 797         self.assertEquals(len(res), 0)
 798
 799         # Modify local record
 800         ldif = """
 801 dn: """ + dn + """
 802 replace: foo
 803 foo: baz
 804 replace: description
 805 description: foo
 806 """
 807         self.ldb.modify_ldif(ldif)
 808         # Check in local db
 809         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
 810         self.assertEquals(len(res), 1)
 811         self.assertEquals(str(res[0].dn), dn)
 812         self.assertEquals(str(res[0]["foo"]), "baz")
 813         self.assertEquals(str(res[0]["revision"]), "1")
 814         self.assertEquals(str(res[0]["description"]), "foo")
 815
 816         # Rename local record
 817         dn2 = "cn=toast,dc=idealx,dc=org"
 818         self.ldb.rename(dn, dn2)
 819         # Check in local db
 820         res = self.ldb.search(dn2, scope=SCOPE_BASE, attrs=attrs)
 821         self.assertEquals(len(res), 1)
 822         self.assertEquals(str(res[0].dn), dn2)
 823         self.assertEquals(str(res[0]["foo"]), "baz")
 824         self.assertEquals(str(res[0]["revision"]), "1")
 825         self.assertEquals(str(res[0]["description"]), "foo")
 826
 827         # Delete local record
 828         self.ldb.delete(dn2)
 829         # Check it's gone
 830         res = self.ldb.search(dn2, scope=SCOPE_BASE)
 831         self.assertEquals(len(res), 0)
 832
 833     def test_map_modify_remote_remote(self):
 834         """Modification of remote data of remote records"""
 835         # Add remote record
 836         dn = self.samba4.dn("cn=test")
 837         dn2 = self.samba3.dn("cn=test")
 838         self.samba3.db.add({"dn": dn2,
 839                    "cn": "test",
 840                    "description": "foo",
 841                    "sambaBadPasswordCount": "3",
 842                    "sambaNextRid": "1001"})
 843         # Check it's there
 844         res = self.samba3.db.search(dn2, scope=SCOPE_BASE,
 845                 attrs=["description", "sambaBadPasswordCount", "sambaNextRid"])
 846         self.assertEquals(len(res), 1)
 847         self.assertEquals(str(res[0].dn), dn2)
 848         self.assertEquals(str(res[0]["description"]), "foo")
 849         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "3")
 850         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
 851         # Check in mapped db
 852         attrs = ["description", "badPwdCount", "nextRid"]
 853         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs, expression="")
 854         self.assertEquals(len(res), 1)
 855         self.assertEquals(str(res[0].dn), dn)
 856         self.assertEquals(str(res[0]["description"]), "foo")
 857         self.assertEquals(str(res[0]["badPwdCount"]), "3")
 858         self.assertEquals(str(res[0]["nextRid"]), "1001")
 859         # Check in local db
 860         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
 861         self.assertEquals(len(res), 0)
 862
 863         # Modify remote data of remote record
 864         ldif = """
 865 dn: """ + dn + """
 866 replace: description
 867 description: test
 868 replace: badPwdCount
 869 badPwdCount: 4
 870 """
 871         self.ldb.modify_ldif(ldif)
 872         # Check in mapped db
 873         res = self.ldb.search(dn, scope=SCOPE_BASE,
 874                 attrs=["description", "badPwdCount", "nextRid"])
 875         self.assertEquals(len(res), 1)
 876         self.assertEquals(str(res[0].dn), dn)
 877         self.assertEquals(str(res[0]["description"]), "test")
 878         self.assertEquals(str(res[0]["badPwdCount"]), "4")
 879         self.assertEquals(str(res[0]["nextRid"]), "1001")
 880         # Check in remote db
 881         res = self.samba3.db.search(dn2, scope=SCOPE_BASE,
 882                 attrs=["description", "sambaBadPasswordCount", "sambaNextRid"])
 883         self.assertEquals(len(res), 1)
 884         self.assertEquals(str(res[0].dn), dn2)
 885         self.assertEquals(str(res[0]["description"]), "test")
 886         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
 887         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
 888
 889         # Rename remote record
 890         dn2 = self.samba4.dn("cn=toast")
 891         self.ldb.rename(dn, dn2)
 892         # Check in mapped db
 893         dn = dn2
 894         res = self.ldb.search(dn, scope=SCOPE_BASE,
 895                 attrs=["description", "badPwdCount", "nextRid"])
 896         self.assertEquals(len(res), 1)
 897         self.assertEquals(str(res[0].dn), dn)
 898         self.assertEquals(str(res[0]["description"]), "test")
 899         self.assertEquals(str(res[0]["badPwdCount"]), "4")
 900         self.assertEquals(str(res[0]["nextRid"]), "1001")
 901         # Check in remote db
 902         dn2 = self.samba3.dn("cn=toast")
 903         res = self.samba3.db.search(dn2, scope=SCOPE_BASE,
 904                 attrs=["description", "sambaBadPasswordCount", "sambaNextRid"])
 905         self.assertEquals(len(res), 1)
 906         self.assertEquals(str(res[0].dn), dn2)
 907         self.assertEquals(str(res[0]["description"]), "test")
 908         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
 909         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
 910
 911         # Delete remote record
 912         self.ldb.delete(dn)
 913         # Check in mapped db that it's removed
 914         res = self.ldb.search(dn, scope=SCOPE_BASE)
 915         self.assertEquals(len(res), 0)
 916         # Check in remote db
 917         res = self.samba3.db.search(dn2, scope=SCOPE_BASE)
 918         self.assertEquals(len(res), 0)
 919
 920     def test_map_modify_remote_local(self):
 921         """Modification of local data of remote records"""
 922         # Add remote record (same as before)
 923         dn = self.samba4.dn("cn=test")
 924         dn2 = self.samba3.dn("cn=test")
 925         self.samba3.db.add({"dn": dn2,
 926                    "cn": "test",
 927                    "description": "foo",
 928                    "sambaBadPasswordCount": "3",
 929                    "sambaNextRid": "1001"})
 930
 931         # Modify local data of remote record
 932         ldif = """
 933 dn: """ + dn + """
 934 add: revision
 935 revision: 1
 936 replace: description
 937 description: test
 938
 939 """
 940         self.ldb.modify_ldif(ldif)
 941         # Check in mapped db
 942         attrs = ["revision", "description"]
 943         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
 944         self.assertEquals(len(res), 1)
 945         self.assertEquals(str(res[0].dn), dn)
 946         self.assertEquals(str(res[0]["description"]), "test")
 947         self.assertEquals(str(res[0]["revision"]), "1")
 948         # Check in remote db
 949         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, attrs=attrs)
 950         self.assertEquals(len(res), 1)
 951         self.assertEquals(str(res[0].dn), dn2)
 952         self.assertEquals(str(res[0]["description"]), "test")
 953         self.assertTrue(not "revision" in res[0])
 954         # Check in local db
 955         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
 956         self.assertEquals(len(res), 1)
 957         self.assertEquals(str(res[0].dn), dn)
 958         self.assertTrue(not "description" in res[0])
 959         self.assertEquals(str(res[0]["revision"]), "1")
 960
 961         # Delete (newly) split record
 962         self.ldb.delete(dn)
 963
 964     def test_map_modify_split(self):
 965         """Testing modification of split records"""
 966         # Add split record
 967         dn = self.samba4.dn("cn=test")
 968         dn2 = self.samba3.dn("cn=test")
 969         self.ldb.add({
 970             "dn": dn,
 971             "cn": "test",
 972             "description": "foo",
 973             "badPwdCount": "3",
 974             "nextRid": "1001",
 975             "revision": "1"})
 976         # Check it's there
 977         attrs = ["description", "badPwdCount", "nextRid", "revision"]
 978         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
 979         self.assertEquals(len(res), 1)
 980         self.assertEquals(str(res[0].dn), dn)
 981         self.assertEquals(str(res[0]["description"]), "foo")
 982         self.assertEquals(str(res[0]["badPwdCount"]), "3")
 983         self.assertEquals(str(res[0]["nextRid"]), "1001")
 984         self.assertEquals(str(res[0]["revision"]), "1")
 985         # Check in local db
 986         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
 987         self.assertEquals(len(res), 1)
 988         self.assertEquals(str(res[0].dn), dn)
 989         self.assertTrue(not "description" in res[0])
 990         self.assertTrue(not "badPwdCount" in res[0])
 991         self.assertTrue(not "nextRid" in res[0])
 992         self.assertEquals(str(res[0]["revision"]), "1")
 993         # Check in remote db
 994         attrs = ["description", "sambaBadPasswordCount", "sambaNextRid",
 995                  "revision"]
 996         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, attrs=attrs)
 997         self.assertEquals(len(res), 1)
 998         self.assertEquals(str(res[0].dn), dn2)
 999         self.assertEquals(str(res[0]["description"]), "foo")
1000         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "3")
1001         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
1002         self.assertTrue(not "revision" in res[0])
1003
1004         # Modify of split record
1005         ldif = """
1006 dn: """ + dn + """
1007 replace: description
1008 description: test
1009 replace: badPwdCount
1010 badPwdCount: 4
1011 replace: revision
1012 revision: 2
1013 """
1014         self.ldb.modify_ldif(ldif)
1015         # Check in mapped db
1016         attrs = ["description", "badPwdCount", "nextRid", "revision"]
1017         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
1018         self.assertEquals(len(res), 1)
1019         self.assertEquals(str(res[0].dn), dn)
1020         self.assertEquals(str(res[0]["description"]), "test")
1021         self.assertEquals(str(res[0]["badPwdCount"]), "4")
1022         self.assertEquals(str(res[0]["nextRid"]), "1001")
1023         self.assertEquals(str(res[0]["revision"]), "2")
1024         # Check in local db
1025         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
1026         self.assertEquals(len(res), 1)
1027         self.assertEquals(str(res[0].dn), dn)
1028         self.assertTrue(not "description" in res[0])
1029         self.assertTrue(not "badPwdCount" in res[0])
1030         self.assertTrue(not "nextRid" in res[0])
1031         self.assertEquals(str(res[0]["revision"]), "2")
1032         # Check in remote db
1033         attrs = ["description", "sambaBadPasswordCount", "sambaNextRid",
1034                  "revision"]
1035         res = self.samba3.db.search(dn2, scope=SCOPE_BASE, attrs=attrs)
1036         self.assertEquals(len(res), 1)
1037         self.assertEquals(str(res[0].dn), dn2)
1038         self.assertEquals(str(res[0]["description"]), "test")
1039         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
1040         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
1041         self.assertTrue(not "revision" in res[0])
1042
1043         # Rename split record
1044         dn2 = self.samba4.dn("cn=toast")
1045         self.ldb.rename(dn, dn2)
1046         # Check in mapped db
1047         dn = dn2
1048         attrs = ["description", "badPwdCount", "nextRid", "revision"]
1049         res = self.ldb.search(dn, scope=SCOPE_BASE, attrs=attrs)
1050         self.assertEquals(len(res), 1)
1051         self.assertEquals(str(res[0].dn), dn)
1052         self.assertEquals(str(res[0]["description"]), "test")
1053         self.assertEquals(str(res[0]["badPwdCount"]), "4")
1054         self.assertEquals(str(res[0]["nextRid"]), "1001")
1055         self.assertEquals(str(res[0]["revision"]), "2")
1056         # Check in local db
1057         res = self.samba4.db.search(dn, scope=SCOPE_BASE, attrs=attrs)
1058         self.assertEquals(len(res), 1)
1059         self.assertEquals(str(res[0].dn), dn)
1060         self.assertTrue(not "description" in res[0])
1061         self.assertTrue(not "badPwdCount" in res[0])
1062         self.assertTrue(not "nextRid" in res[0])
1063         self.assertEquals(str(res[0]["revision"]), "2")
1064         # Check in remote db
1065         dn2 = self.samba3.dn("cn=toast")
1066         res = self.samba3.db.search(dn2, scope=SCOPE_BASE,
1067           attrs=["description", "sambaBadPasswordCount", "sambaNextRid",
1068                  "revision"])
1069         self.assertEquals(len(res), 1)
1070         self.assertEquals(str(res[0].dn), dn2)
1071         self.assertEquals(str(res[0]["description"]), "test")
1072         self.assertEquals(str(res[0]["sambaBadPasswordCount"]), "4")
1073         self.assertEquals(str(res[0]["sambaNextRid"]), "1001")
1074         self.assertTrue(not "revision" in res[0])
1075
1076         # Delete split record
1077         self.ldb.delete(dn)
1078         # Check in mapped db
1079         res = self.ldb.search(dn, scope=SCOPE_BASE)
1080         self.assertEquals(len(res), 0)
1081         # Check in local db
1082         res = self.samba4.db.search(dn, scope=SCOPE_BASE)
1083         self.assertEquals(len(res), 0)
1084         # Check in remote db
1085         res = self.samba3.db.search(dn2, scope=SCOPE_BASE)
1086         self.assertEquals(len(res), 0)