source4/lib/ldb/tests/python/deletetest.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 import getopt
   5 import optparse
   6 import sys
   7 import os
   8
   9 sys.path.append("bin/python")
  10 sys.path.append("../lib/subunit/python")
  11 sys.path.append("../lib/testtools")
  12
  13 import samba.getopt as options
  14
  15 import ldb
  16 from samba.auth import system_session
  17 from ldb import SCOPE_SUBTREE, SCOPE_BASE, LdbError
  18 from ldb import ERR_NO_SUCH_OBJECT, ERR_ATTRIBUTE_OR_VALUE_EXISTS
  19 from samba import Ldb
  20
  21 from subunit.run import SubunitTestRunner
  22 import unittest
  23
  24 parser = optparse.OptionParser("deletetest.py [options] <host|file>")
  25 sambaopts = options.SambaOptions(parser)
  26 parser.add_option_group(sambaopts)
  27 parser.add_option_group(options.VersionOptions(parser))
  28 # use command line creds if available
  29 credopts = options.CredentialsOptions(parser)
  30 parser.add_option_group(credopts)
  31 opts, args = parser.parse_args()
  32
  33 if len(args) < 1:
  34     parser.print_usage()
  35     sys.exit(1)
  36
  37 host = args[0]
  38
  39 lp = sambaopts.get_loadparm()
  40 creds = credopts.get_credentials(lp)
  41
  42 class BasicDeleteTests(unittest.TestCase):
  43
  44     def delete_force(self, ldb_ctx, dn):
  45         try:
  46             ldb_ctx.delete(dn)
  47         except LdbError, (num, _):
  48             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
  49
  50     def GUID_string(self, guid):
  51         return self.ldb_ctx.schema_format_value("objectGUID", guid)
  52
  53     def find_basedn(self, ldb_ctx):
  54         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE,
  55                          attrs=["defaultNamingContext"])
  56         self.assertEquals(len(res), 1)
  57         return res[0]["defaultNamingContext"][0]
  58
  59     def setUp(self):
  60         self.ldb_ctx = ldb_ctx
  61         self.base_dn = self.find_basedn(ldb_ctx)
  62
  63     def search_guid(self,guid):
  64         print "SEARCH by GUID %s" % self.GUID_string(guid)
  65
  66         expression = "(objectGUID=%s)" % self.GUID_string(guid)
  67         res = ldb_ctx.search(expression=expression,
  68                          controls=["show_deleted:1"])
  69         self.assertEquals(len(res), 1)
  70         return res[0]
  71
  72     def search_dn(self,dn):
  73         print "SEARCH by DN %s" % dn
  74
  75         res = ldb_ctx.search(expression="(objectClass=*)",
  76                          base=dn,
  77                          scope=SCOPE_BASE,
  78                          controls=["show_deleted:1"])
  79         self.assertEquals(len(res), 1)
  80         return res[0]
  81
  82     def del_attr_values(self, delObj):
  83         print "Checking attributes for %s" % delObj["dn"]
  84
  85         self.assertEquals(delObj["isDeleted"][0],"TRUE")
  86         self.assertTrue(not("objectCategory" in delObj))
  87         self.assertTrue(not("sAMAccountType" in delObj))
  88
  89     def preserved_attributes_list(self, liveObj, delObj):
  90         print "Checking for preserved attributes list"
  91
  92         preserved_list = ["nTSecurityDescriptor", "attributeID", "attributeSyntax", "dNReferenceUpdate", "dNSHostName",
  93         "flatName", "governsID", "groupType", "instanceType", "lDAPDisplayName", "legacyExchangeDN",
  94         "isDeleted", "isRecycled", "lastKnownParent", "msDS-LastKnownRDN", "mS-DS-CreatorSID",
  95         "mSMQOwnerID", "nCName", "objectClass", "distinguishedName", "objectGUID", "objectSid",
  96         "oMSyntax", "proxiedObjectName", "name", "replPropertyMetaData", "sAMAccountName",
  97         "securityIdentifier", "sIDHistory", "subClassOf", "systemFlags", "trustPartner", "trustDirection",
  98         "trustType", "trustAttributes", "userAccountControl", "uSNChanged", "uSNCreated", "whenCreated"]
  99
 100         for a in liveObj:
 101             if a in preserved_list:
 102                 self.assertTrue(a in delObj)
 103
 104     def check_rdn(self, liveObj, delObj, rdnName):
 105         print "Checking for correct rDN"
 106         rdn=liveObj[rdnName][0]
 107         rdn2=delObj[rdnName][0]
 108         name2=delObj[rdnName][0]
 109         guid=liveObj["objectGUID"][0]
 110         self.assertEquals(rdn2, rdn + "\nDEL:" + self.GUID_string(guid))
 111         self.assertEquals(name2, rdn + "\nDEL:" + self.GUID_string(guid))
 112
 113     def delete_deleted(self, ldb_ctx, dn):
 114         print "Testing the deletion of the already deleted dn %s" % dn
 115
 116         try:
 117             ldb_ctx.delete(dn)
 118             self.fail()
 119         except LdbError, (num, _):
 120             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 121
 122     def find_domain_func_level(self,ldb_ctx):
 123         print "Searching for the domain functional level"
 124
 125         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["domainControllerFunctionality"])
 126         self.assertEquals(len(res), 1)
 127         return  res[0]["domainControllerFunctionality"][0]
 128
 129     def find_recycle_bin_msg(self,ldb_ctx):
 130         print "Searching for recycle bin msg"
 131
 132         res = ldb_ctx.search(base="",
 133                          expression="(&(objectClass=msDS-OptionalFeature)(msDS-OptionalFeatureGUID=766ddcd8-acd0-445e-f3b9-a7f9b6744f2a))",
 134                          scope=SCOPE_SUBTREE, attrs=["*"],controls=["search_options:1:2"])
 135         self.assertEquals(len(res), 1)
 136         return  res[0]
 137
 138     def find_ntds_settings(self,ldb_ctx):
 139         print "Searching for ntds settings dn"
 140
 141         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["dsServiceName"])
 142         self.assertEquals(len(res), 1)
 143         return  res[0]["dsServiceName"][0]
 144
 145     def get_configuration_nc(self, ldb_ctx):
 146         print "Searching for the configuration nc"
 147
 148         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["configurationNamingContext"])
 149         rootDse = res[0]
 150
 151         return rootDse["configurationNamingContext"]
 152
 153     def check_recyclebin_disabled(self, ldb_ctx):
 154         print "Testing if recycle bin is disabled"
 155
 156         ret = True
 157
 158         recycle_bin = self.find_recycle_bin_msg(ldb_ctx)
 159         ntds_settings = self.find_ntds_settings(ldb_ctx)
 160
 161         configbase = self.get_configuration_nc(ldb_ctx)
 162         partition = "CN=Partitions," + str(configbase)
 163
 164         ntds_msg = self.search_dn(ntds_settings)
 165         partition_msg = self.search_dn(partition)
 166         recycle_bin_dn = str(recycle_bin["dn"])
 167
 168         try:
 169             self.assertTrue(ntds_msg["msDS-EnabledFeature"][0] == recycle_bin_dn)
 170             self.fail()
 171         except KeyError, error:
 172             try:
 173                 self.assertEquals(error[0], 'No such element')
 174             except:
 175                 ret = False
 176         except AssertionError:
 177             ret = False
 178
 179         try:
 180             self.assertTrue(partition_msg["msDS-EnabledFeature"][0] == recycle_bin_dn)
 181             self.fail()
 182         except KeyError, error:
 183             try:
 184                 self.assertEquals(error[0], 'No such element')
 185             except:
 186                 ret = False
 187         except AssertionError:
 188             ret = False
 189
 190         try:
 191             count = 0
 192             for item in recycle_bin["msDS-EnabledFeatureBL"]:
 193                 if (item == partition) or (item == ntds_settings):
 194                     count += 1
 195             self.assertEquals(count, 0)
 196         except KeyError, error:
 197             try:
 198                 self.assertEquals(error[0], 'No such element')
 199             except:
 200                 ret = False
 201         except AssertionError:
 202                 ret = False
 203
 204         return ret
 205
 206     def test_all(self):
 207         """Basic delete tests"""
 208
 209         print self.base_dn
 210
 211         recycle_disabled = self.check_recyclebin_disabled(self.ldb_ctx)
 212         if (recycle_disabled == False):
 213             print "Recycle Bin is already enabled"
 214         else:
 215             dn1="cn=testuser,cn=users," + self.base_dn
 216             dn2="cn=testuser2,cn=users," + self.base_dn
 217             grp1="cn=testdelgroup1,cn=users," + self.base_dn
 218
 219             self.delete_force(self.ldb_ctx, dn1)
 220             self.delete_force(self.ldb_ctx, dn2)
 221             self.delete_force(self.ldb_ctx, grp1)
 222
 223             ldb_ctx.add({
 224                 "dn": dn1,
 225                 "objectclass": "user",
 226                 "cn": "testuser",
 227                 "description": "test user description",
 228                 "samaccountname": "testuser"})
 229
 230             ldb_ctx.add({
 231                 "dn": dn2,
 232                 "objectclass": "user",
 233                 "cn": "testuser2",
 234                 "description": "test user 2 description",
 235                 "samaccountname": "testuser2"})
 236
 237             ldb_ctx.add({
 238                 "dn": grp1,
 239                 "objectclass": "group",
 240                 "cn": "testdelgroup1",
 241                 "description": "test group",
 242                 "samaccountname": "testdelgroup1",
 243                 "member": [ dn1, dn2 ] })
 244
 245             objLive1 = self.search_dn(dn1)
 246             guid1=objLive1["objectGUID"][0]
 247
 248             objLive2 = self.search_dn(dn2)
 249             guid2=objLive2["objectGUID"][0]
 250
 251             objLive3 = self.search_dn(grp1)
 252             guid3=objLive3["objectGUID"][0]
 253
 254             ldb_ctx.delete(dn1)
 255             ldb_ctx.delete(dn2)
 256             ldb_ctx.delete(grp1)
 257
 258             objDeleted1 = self.search_guid(guid1)
 259             objDeleted2 = self.search_guid(guid2)
 260             objDeleted3 = self.search_guid(guid3)
 261
 262             self.del_attr_values(objDeleted1)
 263             self.del_attr_values(objDeleted2)
 264             self.del_attr_values(objDeleted3)
 265
 266             self.preserved_attributes_list(objLive1, objDeleted1)
 267             self.preserved_attributes_list(objLive2, objDeleted2)
 268
 269             self.check_rdn(objLive1, objDeleted1, "cn")
 270             self.check_rdn(objLive2, objDeleted2, "cn")
 271             self.check_rdn(objLive3, objDeleted3, "cn")
 272
 273             self.delete_deleted(ldb_ctx, dn1)
 274             self.delete_deleted(ldb_ctx, dn2)
 275             self.delete_deleted(ldb_ctx, grp1)
 276
 277
 278
 279 class RecycleBinActivationTests(unittest.TestCase):
 280
 281     def find_basedn(self, ldb_ctx):
 282         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE,
 283                          attrs=["defaultNamingContext"])
 284         self.assertEquals(len(res), 1)
 285         return res[0]["defaultNamingContext"][0]
 286
 287     def setUp(self):
 288         self.ldb_ctx = ldb_ctx
 289         self.base_dn = self.find_basedn(ldb_ctx)
 290
 291     def find_recycle_bin_msg(self,ldb_ctx):
 292         print "Searching for recycle bin msg"
 293
 294         res = ldb_ctx.search(base="",
 295                          expression="(&(objectClass=msDS-OptionalFeature)(msDS-OptionalFeatureGUID=766ddcd8-acd0-445e-f3b9-a7f9b6744f2a))",
 296                          scope=SCOPE_SUBTREE, attrs=["*"],controls=["search_options:1:2"])
 297         self.assertEquals(len(res), 1)
 298         return  res[0]
 299
 300     def find_ntds_settings(self,ldb_ctx):
 301         print "Searching for ntds settings dn"
 302
 303         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["dsServiceName"])
 304         self.assertEquals(len(res), 1)
 305         return  res[0]["dsServiceName"][0]
 306
 307     def find_domain_func_level(self,ldb_ctx):
 308         print "Searching for the domain functional level"
 309
 310         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["domainControllerFunctionality"])
 311         self.assertEquals(len(res), 1)
 312         return  res[0]["domainControllerFunctionality"][0]
 313
 314     def search_dn(self,dn):
 315         print "Searching by DN %s" % dn
 316
 317         res = ldb_ctx.search(expression="(objectClass=*)",
 318                          base=dn,
 319                          scope=SCOPE_BASE)
 320         self.assertEquals(len(res), 1)
 321         return res[0]
 322
 323     def get_configuration_nc(self, ldb_ctx):
 324         print "Searching for the configuration nc"
 325
 326         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["configurationNamingContext"])
 327         rootDse = res[0]
 328
 329         return rootDse["configurationNamingContext"]
 330
 331     def enablerecyclebin(self, ldb_ctx, partition):
 332         print "Enabling the recycle bin optional feature"
 333
 334         msg = ldb.Message()
 335         msg.dn = ldb.Dn(ldb_ctx, "")
 336         msg["enableOptionalFeature"] = ldb.MessageElement(
 337                  partition + ":766ddcd8-acd0-445e-f3b9-a7f9b6744f2a",
 338                  ldb.FLAG_MOD_ADD, "enableOptionalFeature")
 339         res = ldb_ctx.modify(msg)
 340
 341         print "Recycle Bin feature enabled"
 342
 343     def check_recyclebin_disabled(self, ldb_ctx):
 344         print "Testing if recycle bin is disabled"
 345
 346         ret = True
 347
 348         recycle_bin = self.find_recycle_bin_msg(ldb_ctx)
 349         ntds_settings = self.find_ntds_settings(ldb_ctx)
 350
 351         configbase = self.get_configuration_nc(ldb_ctx)
 352         partition = "CN=Partitions," + str(configbase)
 353
 354         ntds_msg = self.search_dn(ntds_settings)
 355         partition_msg = self.search_dn(partition)
 356         recycle_bin_dn = str(recycle_bin["dn"])
 357
 358         try:
 359             self.assertTrue(ntds_msg["msDS-EnabledFeature"][0] == recycle_bin_dn)
 360             self.fail()
 361         except KeyError, error:
 362             try:
 363                 self.assertEquals(error[0], 'No such element')
 364             except:
 365                 ret = False
 366         except AssertionError:
 367             ret = False
 368
 369         try:
 370             self.assertTrue(partition_msg["msDS-EnabledFeature"][0] == recycle_bin_dn)
 371             self.fail()
 372         except KeyError, error:
 373             try:
 374                 self.assertEquals(error[0], 'No such element')
 375             except:
 376                 ret = False
 377         except AssertionError:
 378             ret = False
 379
 380         try:
 381             count = 0
 382             for item in recycle_bin["msDS-EnabledFeatureBL"]:
 383                 if (item == partition) or (item == ntds_settings):
 384                     count += 1
 385             self.assertEquals(count, 0)
 386         except KeyError, error:
 387             try:
 388                 self.assertEquals(error[0], 'No such element')
 389             except:
 390                 ret = False
 391         except AssertionError:
 392                 ret = False
 393
 394         return ret
 395
 396     def test_all(self):
 397
 398         """Recycle Bin activation tests"""
 399
 400         print self.base_dn
 401
 402         print "Testing domain's functional level"
 403         func_level = self.find_domain_func_level(ldb_ctx)
 404         recycle_bin = self.find_recycle_bin_msg(ldb_ctx)
 405         required_func_level = recycle_bin["msDS-RequiredForestBehaviorVersion"][0]
 406
 407         recycle_disabled = self.check_recyclebin_disabled(self.ldb_ctx)
 408         if recycle_disabled == False:
 409             print "Recycle Bin is already enabled"
 410
 411         elif func_level < required_func_level:
 412             print "Functional level should be at least %s to perform this test" %required_func_level
 413
 414         else:
 415             ntds_settings = self.find_ntds_settings(ldb_ctx)
 416
 417             configbase = self.get_configuration_nc(ldb_ctx)
 418             partition = "CN=Partitions," + str(configbase)
 419
 420             ntds_msg = self.search_dn(ntds_settings)
 421             partition_msg = self.search_dn(partition)
 422             recycle_bin_dn = str(recycle_bin["dn"])
 423
 424             self.enablerecyclebin(ldb_ctx, partition)
 425
 426             print "Testing if recycle bin was enabled successfully"
 427
 428             ntds_msg = self.search_dn(ntds_settings)
 429             partition_msg = self.search_dn(partition)
 430             recycle_bin = self.find_recycle_bin_msg(ldb_ctx)
 431
 432             self.assertTrue(ntds_msg["msDS-EnabledFeature"][0] == recycle_bin_dn)
 433             self.assertTrue(partition_msg["msDS-EnabledFeature"][0] == recycle_bin_dn)
 434
 435             print "Testing the Recycle Bin msg backlinks"
 436             count = 0
 437             for item in recycle_bin["msDS-EnabledFeatureBL"]:
 438                 if (item == partition) or (item == ntds_settings):
 439                     count += 1
 440             self.assertEquals(count, 2)
 441
 442             print "Testing the activation of a previously activated recycle bin"
 443             try:
 444                 self.enablerecyclebin(ldb_ctx, partition)
 445                 self.fail()
 446             except LdbError, (num, _):
 447                 self.assertEquals(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)
 448
 449 class ThreeStageDeleteTests(unittest.TestCase):
 450
 451     def delete_force(self, ldb_ctx, dn):
 452         try:
 453             ldb_ctx.delete(dn)
 454         except LdbError, (num, _):
 455             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 456
 457     def find_basedn(self, ldb_ctx):
 458         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE,
 459                          attrs=["defaultNamingContext"])
 460         self.assertEquals(len(res), 1)
 461         return res[0]["defaultNamingContext"][0]
 462
 463     def setUp(self):
 464         self.ldb_ctx = ldb_ctx
 465         self.base_dn = self.find_basedn(ldb_ctx)
 466
 467     def GUID_string(self, guid):
 468         return self.ldb_ctx.schema_format_value("objectGUID", guid)
 469
 470     def search_dn(self,dn):
 471         print "SEARCH by DN %s" % dn
 472
 473         res = ldb_ctx.search(expression="(objectClass=*)",
 474                          base=dn,
 475                          scope=SCOPE_BASE,
 476                          controls=["show_deleted:1"])
 477         self.assertEquals(len(res), 1)
 478         return res[0]
 479
 480     def search_guid(self,guid):
 481         print "SEARCH by GUID %s" % self.GUID_string(guid)
 482
 483         expression = "(objectGUID=%s)" % self.GUID_string(guid)
 484         res = ldb_ctx.search(expression=expression,
 485                          controls=["show_deleted:1"])
 486         self.assertEquals(len(res), 1)
 487         return res[0]
 488
 489     def del_attr_values(self, delObj):
 490         print "Checking attributes for %s" % delObj["dn"]
 491
 492         self.assertEquals(delObj["isDeleted"][0],"TRUE")
 493         self.assertTrue(not("objectCategory" in delObj))
 494         self.assertTrue(not("sAMAccountType" in delObj))
 495
 496     def check_rdn(self, liveObj, delObj, rdnName):
 497         print "Checking for correct rDN"
 498         rdn=liveObj[rdnName][0]
 499         rdn2=delObj[rdnName][0]
 500         name2=delObj[rdnName][0]
 501         guid=liveObj["objectGUID"][0]
 502         self.assertEquals(rdn2, rdn + "\nDEL:" + self.GUID_string(guid))
 503         self.assertEquals(name2, rdn + "\nDEL:" + self.GUID_string(guid))
 504
 505     def get_configuration_nc(self, ldb_ctx):
 506         print "Searching for the configuration nc"
 507
 508         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["configurationNamingContext"])
 509         rootDse = res[0]
 510
 511         return rootDse["configurationNamingContext"]
 512
 513     def check_lastknownparent(self, liveObj, delObj, rdnName):
 514         print "Checking for correct lastKnownParent"
 515         liveObjCN=liveObj[rdnName][0]
 516         liveObjDN=str(liveObj["dn"])
 517         delObjLastKnownParent=delObj["lastKnownParent"][0]
 518
 519         self.assertEquals(liveObjDN, rdnName + "=" + liveObjCN + "," + delObjLastKnownParent)
 520
 521     def check_lastknownrdn(self, liveObj, delObj, rdnName):
 522         print "Checking for correct msDS-LastKnownRDN"
 523         liveObjRDN=liveObj[rdnName][0]
 524         delObjLastKnownRDN=delObj["msDS-LastKnownRDN"][0]
 525
 526         self.assertEquals(liveObjRDN,delObjLastKnownRDN)
 527
 528     def preserved_live_attributes(self, liveObj, delObj):
 529         print "Checking if all the attributes in the liveObj are in the delObj"
 530
 531         removed_attrs = ["objectCategory", "sAMAccountType"]
 532
 533         for a in liveObj:
 534             if a not in removed_attrs:
 535                 self.assertTrue(a in delObj)
 536
 537     def preserved_attributes_list(self, liveObj, delObj):
 538         print "Checking for preserved attributes list"
 539
 540         preserved_list = ["nTSecurityDescriptor", "attributeID", "attributeSyntax", "dNReferenceUpdate", "dNSHostName",
 541         "flatName", "governsID", "groupType", "instanceType", "lDAPDisplayName", "legacyExchangeDN",
 542         "isDeleted", "isRecycled", "lastKnownParent", "msDS-LastKnownRDN", "mS-DS-CreatorSID",
 543         "mSMQOwnerID", "nCName", "objectClass", "distinguishedName", "objectGUID", "objectSid",
 544         "oMSyntax", "proxiedObjectName", "name", "replPropertyMetaData", "sAMAccountName",
 545         "securityIdentifier", "sIDHistory", "subClassOf", "systemFlags", "trustPartner", "trustDirection",
 546         "trustType", "trustAttributes", "userAccountControl", "uSNChanged", "uSNCreated", "whenCreated"]
 547
 548         for a in liveObj:
 549             if a in preserved_list:
 550                 self.assertTrue(a in delObj)
 551
 552     def check_isRecycled(self, delObj):
 553         print "Checking isRecycled attribute for %s" % delObj["dn"]
 554
 555         self.assertEquals(delObj["isRecycled"][0],"TRUE")
 556
 557     def enablerecyclebin(self, ldb_ctx, partition):
 558         print "Enabling the recycle bin optional feature"
 559
 560         msg = ldb.Message()
 561         msg.dn = ldb.Dn(ldb_ctx, "")
 562         msg["enableOptionalFeature"] = ldb.MessageElement(
 563                  partition + ":766ddcd8-acd0-445e-f3b9-a7f9b6744f2a",
 564                  ldb.FLAG_MOD_ADD, "enableOptionalFeature")
 565         res = ldb_ctx.modify(msg)
 566
 567         print "Recycle Bin feature enabled"
 568
 569     def find_domain_func_level(self,ldb_ctx):
 570         print "Searching for the domain functional level"
 571
 572         res = ldb_ctx.search(base="", expression="", scope=SCOPE_BASE, attrs=["domainControllerFunctionality"])
 573         self.assertEquals(len(res), 1)
 574         return  res[0]["domainControllerFunctionality"][0]
 575
 576     def find_recycle_bin_msg(self,ldb_ctx):
 577         print "Searching for recycle bin msg"
 578
 579         res = ldb_ctx.search(base="",
 580                          expression="(&(objectClass=msDS-OptionalFeature)(msDS-OptionalFeatureGUID=766ddcd8-acd0-445e-f3b9-a7f9b6744f2a))",
 581                          scope=SCOPE_SUBTREE, attrs=["*"],controls=["search_options:1:2"])
 582         self.assertEquals(len(res), 1)
 583         return  res[0]
 584
 585     def test_all(self):
 586
 587         print "Testing domain's functional level"
 588         func_level = self.find_domain_func_level(ldb_ctx)
 589         recycle_bin = self.find_recycle_bin_msg(ldb_ctx)
 590         required_func_level = recycle_bin["msDS-RequiredForestBehaviorVersion"][0]
 591
 592         if func_level < required_func_level:
 593             print "Functional level should be at least %s to perform this test" %required_func_level
 594
 595         else:
 596             configbase = self.get_configuration_nc(ldb_ctx)
 597             partition = "CN=Partitions," + str(configbase)
 598
 599             try:
 600                 self.enablerecyclebin(ldb_ctx, partition)
 601             except LdbError, (num, _):
 602                 self.assertEquals(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)
 603                 print "Recycle Bin already enabled"
 604
 605             dn1="cn=testuser1,cn=users," + self.base_dn
 606
 607             self.delete_force(self.ldb_ctx, dn1)
 608
 609             ldb_ctx.add({
 610                 "dn": dn1,
 611                 "objectclass": "user",
 612                 "cn": "testuser1",
 613                 "description": "test user description",
 614                 "samaccountname": "testuser1"})
 615
 616             objLive1 = self.search_dn(dn1)
 617             guid1=objLive1["objectGUID"][0]
 618
 619             ldb_ctx.delete(dn1)
 620
 621             objDeleted1 = self.search_guid(guid1)
 622
 623             temp = objDeleted1
 624
 625             self.check_rdn(objLive1, objDeleted1, "cn")
 626             self.check_lastknownparent(objLive1, objDeleted1, "CN")
 627             self.check_lastknownrdn(objLive1, objDeleted1, "CN")
 628             self.del_attr_values(objDeleted1)
 629             self.preserved_live_attributes(objLive1, objDeleted1)
 630
 631             ldb_ctx.delete(objDeleted1["dn"])
 632
 633             objDeleted1 = self.search_guid(guid1)
 634
 635             self.check_rdn(objLive1, objDeleted1, "cn")
 636             self.check_lastknownparent(objLive1, objDeleted1, "CN")
 637             self.check_lastknownrdn(objLive1, objDeleted1, "CN")
 638             self.preserved_attributes_list(objLive1, objDeleted1)
 639             self.del_attr_values(objDeleted1)
 640             self.check_isRecycled(objDeleted1)
 641
 642 if not "://" in host:
 643     if os.path.isfile(host):
 644         host = "tdb://%s" % host
 645     else:
 646         host = "ldap://%s" % host
 647
 648 ldb_ctx = Ldb(host, credentials=creds, session_info=system_session(), lp=lp)
 649
 650 runner = SubunitTestRunner()
 651 rc = 0
 652 if not runner.run(unittest.makeSuite(BasicDeleteTests)).wasSuccessful():
 653     rc = 1
 654 if not runner.run(unittest.makeSuite(RecycleBinActivationTests)).wasSuccessful():
 655     rc = 1
 656 if not runner.run(unittest.makeSuite(ThreeStageDeleteTests)).wasSuccessful():
 657    rc = 1
 658
 659 sys.exit(rc)