source4/scripting/python/samba/tests/samba3sam.py

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