Changeset 988 for vendor/current/source4/dsdb/tests

Timestamp:

Nov 24, 2016, 1:14:11 PM (9 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: update vendor to version 4.4.3

Location:

vendor/current/source4/dsdb/tests/python

Files:

• acl.py (modified) (32 diffs)
• deletetest.py (modified) (21 diffs)
• dirsync.py (added)
• dsdb_schema_info.py (modified) (1 diff)
• ldap.py (modified) (91 diffs)
• ldap_schema.py (modified) (21 diffs)
• ldap_syntaxes.py (modified) (13 diffs)
• password_lockout.py (added)
• passwords.py (modified) (28 diffs)
• sam.py (modified) (25 diffs)
• sec_descriptor.py (modified) (11 diffs)
• sites.py (added)
• token_group.py (modified) (9 diffs)
• tombstone_reanimation.py (added)
• urgent_replication.py (modified) (21 diffs)
• user_account_control.py (added)

vendor/current/source4/dsdb/tests/python/acl.py

@@ -9 +9 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 
 import samba.getopt as options
@@ -21 +21 @@
 from ldb import ERR_OPERATIONS_ERROR
 from ldb import Message, MessageElement, Dn
-from ldb import FLAG_MOD_REPLACE, FLAG_MOD_ADD
+from ldb import FLAG_MOD_REPLACE, FLAG_MOD_ADD, FLAG_MOD_DELETE
 from samba.dcerpc import security, drsuapi, misc
 
@@ -27 +27 @@
 from samba import gensec, sd_utils
 from samba.samdb import SamDB
-from samba.credentials import Credentials
+from samba.credentials import Credentials, DONT_USE_KERBEROS
 import samba.tests
 from samba.tests import delete_force
-from subunit.run import SubunitTestRunner
-import unittest
+import samba.dsdb
 
 parser = optparse.OptionParser("acl.py [options] <host>")
@@ -41 +40 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
+
 opts, args = parser.parse_args()
 
@@ -67 +69 @@
     def setUp(self):
         super(AclTests, self).setUp()
-        self.ldb_admin = ldb
-        self.base_dn = ldb.domain_dn()
-        self.domain_sid = security.dom_sid(ldb.get_domain_sid())
+        self.ldb_admin = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
+        self.base_dn = self.ldb_admin.domain_dn()
+        self.domain_sid = security.dom_sid(self.ldb_admin.get_domain_sid())
         self.user_pass = "samba123@"
         self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
-        self.sd_utils = sd_utils.SDUtils(ldb)
+        self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
         #used for anonymous login
         self.creds_tmp = Credentials()
@@ -94 +96 @@
         creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
                                       | gensec.FEATURE_SEAL)
+        creds_tmp.set_kerberos_state(DONT_USE_KERBEROS) # kinit is too expensive to use in a tight loop
         ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
         return ldb_target
@@ -127 +130 @@
 
         # add admins to the Domain Admins group
-        self.ldb_admin.add_remove_group_members("Domain Admins", self.usr_admin_owner,
+        self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_owner],
                        add_members_operation=True)
-        self.ldb_admin.add_remove_group_members("Domain Admins", self.usr_admin_not_owner,
+        self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_not_owner],
                        add_members_operation=True)
 
@@ -168 +171 @@
         self.ldb_notowner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
         self.ldb_notowner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
-                                   grouptype=4)
+                                   grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         # Make sure we HAVE created the two objects -- user and group
         # !!! We should not be able to do that, but however beacuse of ACE ordering our inherited Deny ACE
@@ -187 +190 @@
             self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2)
             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
-                                   grouptype=4)
+                                   grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         except LdbError, (num, _):
             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
@@ -211 +214 @@
         try:
             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
-                                   grouptype=4)
+                                   grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         except LdbError, (num, _):
             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
@@ -235 +238 @@
         self.ldb_owner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
         self.ldb_owner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
-                                 grouptype=4)
+                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         # Make sure we have successfully created the two objects -- user and group
         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s,%s)" % ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
@@ -268 +271 @@
         self.ldb_user3 = self.get_ldb_connection(self.user_with_group_sm, self.user_pass)
         self.user_sid = self.sd_utils.get_object_sid( self.get_user_dn(self.user_with_wp))
-        self.ldb_admin.newgroup("test_modify_group2", grouptype=4)
-        self.ldb_admin.newgroup("test_modify_group3", grouptype=4)
+        self.ldb_admin.newgroup("test_modify_group2", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+        self.ldb_admin.newgroup("test_modify_group3", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         self.ldb_admin.newuser("test_modify_user2", self.user_pass)
 
@@ -303 +306 @@
         # Second test object -- Group
         print "Testing modify on Group object"
-        self.ldb_admin.newgroup("test_modify_group1", grouptype=4)
+        self.ldb_admin.newgroup("test_modify_group1",
+                                grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
         ldif = """
@@ -361 +365 @@
         # Second test object -- Group
         print "Testing modify on Group object"
-        self.ldb_admin.newgroup("test_modify_group1", grouptype=4)
+        self.ldb_admin.newgroup("test_modify_group1",
+                                grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
         ldif = """
@@ -379 +384 @@
 replace: url
 url: www.samba.org"""
+        try:
+            self.ldb_user.modify_ldif(ldif)
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+        else:
+            # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+            self.fail()
+        # Modify on attribute you do not have rights for granted while also modifying something you do have rights for
+        ldif = """
+dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
+changetype: modify
+replace: url
+url: www.samba.org
+replace: displayName
+displayName: test_changed"""
         try:
             self.ldb_user.modify_ldif(ldif)
@@ -435 +455 @@
         # Second test object -- Group
         print "Testing modify on Group object"
-        self.ldb_admin.newgroup("test_modify_group1", grouptype=4)
+        self.ldb_admin.newgroup("test_modify_group1",
+                                grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         # Modify on attribute you do not have rights for granted
         ldif = """
@@ -601 +622 @@
     def setUp(self):
         super(AclSearchTests, self).setUp()
+        # Get the old "dSHeuristics" if it was set
+        dsheuristics = self.ldb_admin.get_dsheuristics()
+        # Reset the "dSHeuristics" as they were before
+        self.addCleanup(self.ldb_admin.set_dsheuristics, dsheuristics)
+        # Set the "dSHeuristics" to activate the correct "userPassword" behaviour
+        self.ldb_admin.set_dsheuristics("000000001")
+        # Get the old "minPwdAge"
+        minPwdAge = self.ldb_admin.get_minPwdAge()
+        # Reset the "minPwdAge" as it was before
+        self.addCleanup(self.ldb_admin.set_minPwdAge, minPwdAge)
+        # Set it temporarely to "0"
+        self.ldb_admin.set_minPwdAge("0")
+
         self.u1 = "search_u1"
         self.u2 = "search_u2"
@@ -608 +642 @@
         self.ldb_admin.newuser(self.u2, self.user_pass)
         self.ldb_admin.newuser(self.u3, self.user_pass)
-        self.ldb_admin.newgroup(self.group1, grouptype=-2147483646)
-        self.ldb_admin.add_remove_group_members(self.group1, self.u2,
+        self.ldb_admin.newgroup(self.group1, grouptype=samba.dsdb.GTYPE_SECURITY_GLOBAL_GROUP)
+        self.ldb_admin.add_remove_group_members(self.group1, [self.u2],
                                                 add_members_operation=True)
         self.ldb_user = self.get_ldb_connection(self.u1, self.user_pass)
@@ -669 +703 @@
         self.assertEquals(len(res), 1)
         #verify some of the attributes
-        #dont care about values
+        #don't care about values
         self.assertTrue("ldapServiceName" in res[0])
         self.assertTrue("namingContexts" in res[0])
@@ -695 +729 @@
             self.fail()
         try:
-            res = anonymous.search("CN=Configuration," + self.base_dn, expression="(objectClass=*)",
+            res = anonymous.search(anonymous.get_config_basedn(), expression="(objectClass=*)",
                                         scope=SCOPE_SUBTREE)
         except LdbError, (num, _):
@@ -716 +750 @@
         self.assertTrue(res[0]["dn"] == Dn(self.ldb_admin,
                                            "OU=test_search_ou2,OU=test_search_ou1," + self.base_dn))
-        res = anonymous.search("CN=Configuration," + self.base_dn, expression="(objectClass=*)",
+        res = anonymous.search(anonymous.get_config_basedn(), expression="(objectClass=*)",
                                scope=SCOPE_SUBTREE)
         self.assertEquals(len(res), 1)
@@ -1231 +1265 @@
         self.assertNotEqual(len(res), 0)
 
+    def test_rename_u9(self):
+        """Rename 'User object' cross OU, with explicit deny on sd and dc"""
+        ou1_dn = "OU=test_rename_ou1," + self.base_dn
+        ou2_dn = "OU=test_rename_ou2," + self.base_dn
+        user_dn = "CN=test_rename_user2," + ou1_dn
+        rename_user_dn = "CN=test_rename_user5," + ou2_dn
+        # Create OU structure
+        self.ldb_admin.create_ou(ou1_dn)
+        self.ldb_admin.create_ou(ou2_dn)
+        self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
+        mod = "(D;;SD;;;DA)"
+        self.sd_utils.dacl_add_ace(user_dn, mod)
+        mod = "(D;;DC;;;DA)"
+        self.sd_utils.dacl_add_ace(ou1_dn, mod)
+        # Rename 'User object' having SD and CC to AU
+        try:
+            self.ldb_admin.rename(user_dn, rename_user_dn)
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+        else:
+            self.fail()
+        #add an allow ace so we can delete this ou
+        mod = "(A;;DC;;;DA)"
+        self.sd_utils.dacl_add_ace(ou1_dn, mod)
+
+
 #tests on Control Access Rights
 class AclCARTests(AclTests):
@@ -1236 +1296 @@
     def setUp(self):
         super(AclCARTests, self).setUp()
+
+        # Get the old "dSHeuristics" if it was set
+        dsheuristics = self.ldb_admin.get_dsheuristics()
+        # Reset the "dSHeuristics" as they were before
+        self.addCleanup(self.ldb_admin.set_dsheuristics, dsheuristics)
+        # Set the "dSHeuristics" to activate the correct "userPassword" behaviour
+        self.ldb_admin.set_dsheuristics("000000001")
+        # Get the old "minPwdAge"
+        minPwdAge = self.ldb_admin.get_minPwdAge()
+        # Reset the "minPwdAge" as it was before
+        self.addCleanup(self.ldb_admin.set_minPwdAge, minPwdAge)
+        # Set it temporarely to "0"
+        self.ldb_admin.set_minPwdAge("0")
+
         self.user_with_wp = "acl_car_user1"
         self.user_with_pc = "acl_car_user2"
@@ -1517 +1591 @@
         self.ldb_admin.newuser(self.u2, self.user_pass)
         self.ldb_admin.newuser(self.u3, self.user_pass)
-        self.ldb_admin.add_remove_group_members("Domain Admins", self.u3,
+        self.ldb_admin.add_remove_group_members("Domain Admins", [self.u3],
                                                 add_members_operation=True)
         self.ldb_user1 = self.get_ldb_connection(self.u1, self.user_pass)
@@ -1542 +1616 @@
         self.sd_utils.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
         #create a group under that, grant RP to u2
-        self.ldb_user1.newgroup("ext_group1", groupou="OU=ext_ou1", grouptype=4)
+        self.ldb_user1.newgroup("ext_group1", groupou="OU=ext_ou1",
+                                grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
         mod = "(A;;RP;;;%s)" % str(self.user_sid2)
         self.sd_utils.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
@@ -1563 +1638 @@
         self.assertTrue("nTSecurityDescriptor" in res[0].keys())
 
+class AclUndeleteTests(AclTests):
+
+    def setUp(self):
+        super(AclUndeleteTests, self).setUp()
+        self.regular_user = "undeleter1"
+        self.ou1 = "OU=undeleted_ou,"
+        self.testuser1 = "to_be_undeleted1"
+        self.testuser2 = "to_be_undeleted2"
+        self.testuser3 = "to_be_undeleted3"
+        self.testuser4 = "to_be_undeleted4"
+        self.testuser5 = "to_be_undeleted5"
+        self.testuser6 = "to_be_undeleted6"
+
+        self.new_dn_ou = "CN="+ self.testuser4 + "," + self.ou1 + self.base_dn
+
+        # Create regular user
+        self.testuser1_dn = self.get_user_dn(self.testuser1)
+        self.testuser2_dn = self.get_user_dn(self.testuser2)
+        self.testuser3_dn = self.get_user_dn(self.testuser3)
+        self.testuser4_dn = self.get_user_dn(self.testuser4)
+        self.testuser5_dn = self.get_user_dn(self.testuser5)
+        self.deleted_dn1 = self.create_delete_user(self.testuser1)
+        self.deleted_dn2 = self.create_delete_user(self.testuser2)
+        self.deleted_dn3 = self.create_delete_user(self.testuser3)
+        self.deleted_dn4 = self.create_delete_user(self.testuser4)
+        self.deleted_dn5 = self.create_delete_user(self.testuser5)
+
+        self.ldb_admin.create_ou(self.ou1 + self.base_dn)
+
+        self.ldb_admin.newuser(self.regular_user, self.user_pass)
+        self.ldb_admin.add_remove_group_members("Domain Admins", [self.regular_user],
+                       add_members_operation=True)
+        self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
+        self.sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+
+    def tearDown(self):
+        super(AclUndeleteTests, self).tearDown()
+        delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
+        delete_force(self.ldb_admin, self.get_user_dn(self.testuser1))
+        delete_force(self.ldb_admin, self.get_user_dn(self.testuser2))
+        delete_force(self.ldb_admin, self.get_user_dn(self.testuser3))
+        delete_force(self.ldb_admin, self.get_user_dn(self.testuser4))
+        delete_force(self.ldb_admin, self.get_user_dn(self.testuser5))
+        delete_force(self.ldb_admin, self.new_dn_ou)
+        delete_force(self.ldb_admin, self.ou1 + self.base_dn)
+
+    def GUID_string(self, guid):
+        return ldb.schema_format_value("objectGUID", guid)
+
+    def create_delete_user(self, new_user):
+        self.ldb_admin.newuser(new_user, self.user_pass)
+
+        res = self.ldb_admin.search(expression="(objectClass=*)",
+                                    base=self.get_user_dn(new_user),
+                                    scope=SCOPE_BASE,
+                                    controls=["show_deleted:1"])
+        guid = res[0]["objectGUID"][0]
+        self.ldb_admin.delete(self.get_user_dn(new_user))
+        res = self.ldb_admin.search(base="<GUID=%s>" % self.GUID_string(guid),
+                         scope=SCOPE_BASE, controls=["show_deleted:1"])
+        self.assertEquals(len(res), 1)
+        return str(res[0].dn)
+
+    def undelete_deleted(self, olddn, newdn):
+        msg = Message()
+        msg.dn = Dn(self.ldb_user, olddn)
+        msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
+        msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
+        res = self.ldb_user.modify(msg, ["show_recycled:1"])
+
+    def undelete_deleted_with_mod(self, olddn, newdn):
+        msg = Message()
+        msg.dn = Dn(ldb, olddn)
+        msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
+        msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
+        msg["url"] = MessageElement(["www.samba.org"], FLAG_MOD_REPLACE, "url")
+        res = self.ldb_user.modify(msg, ["show_deleted:1"])
+
+    def test_undelete(self):
+        # it appears the user has to have LC on the old parent to be able to move the object
+        # otherwise we get no such object. Since only System can modify the SD on deleted object
+        # we cannot grant this permission via LDAP, and this leaves us with "negative" tests at the moment
+
+        # deny write property on rdn, should fail
+        mod = "(OD;;WP;bf967a0e-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
+        self.sd_utils.dacl_add_ace(self.deleted_dn1, mod)
+        try:
+            self.undelete_deleted(self.deleted_dn1, self.testuser1_dn)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+        # seems that permissions on isDeleted and distinguishedName are irrelevant
+        mod = "(OD;;WP;bf96798f-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
+        self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
+        mod = "(OD;;WP;bf9679e4-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
+        self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
+        self.undelete_deleted(self.deleted_dn2, self.testuser2_dn)
+
+        # attempt undelete with simultanious addition of url, WP to which is denied
+        mod = "(OD;;WP;9a9a0221-4a5b-11d1-a9c3-0000f80367c1;;%s)" % str(self.sid)
+        self.sd_utils.dacl_add_ace(self.deleted_dn3, mod)
+        try:
+            self.undelete_deleted_with_mod(self.deleted_dn3, self.testuser3_dn)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+        # undelete in an ou, in which we have no right to create children
+        mod = "(D;;CC;;;%s)" % str(self.sid)
+        self.sd_utils.dacl_add_ace(self.ou1 + self.base_dn, mod)
+        try:
+            self.undelete_deleted(self.deleted_dn4, self.new_dn_ou)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+        # delete is not required
+        mod = "(D;;SD;;;%s)" % str(self.sid)
+        self.sd_utils.dacl_add_ace(self.deleted_dn5, mod)
+        self.undelete_deleted(self.deleted_dn5, self.testuser5_dn)
+
+        # deny Reanimate-Tombstone, should fail
+        mod = "(OD;;CR;45ec5156-db7e-47bb-b53f-dbeb2d03c40f;;%s)" % str(self.sid)
+        self.sd_utils.dacl_add_ace(self.base_dn, mod)
+        try:
+            self.undelete_deleted(self.deleted_dn4, self.testuser4_dn)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 
 class AclSPNTests(AclTests):
@@ -1622 +1827 @@
     # same as for join_RODC, but do not set any SPNs
     def create_rodc(self, ctx):
+         ctx.nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+         ctx.full_nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
          ctx.krbtgt_dn = "CN=krbtgt_%s,CN=Users,%s" % (ctx.myname, ctx.base_dn)
 
@@ -1651 +1858 @@
 
     def create_dc(self, ctx):
+        ctx.nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
+        ctx.full_nc_list = [ ctx.base_dn, ctx.config_dn, ctx.schema_dn ]
         ctx.userAccountControl = samba.dsdb.UF_SERVER_TRUST_ACCOUNT | samba.dsdb.UF_TRUSTED_FOR_DELEGATION
         ctx.secure_channel_type = misc.SEC_CHAN_BDC
@@ -1678 +1887 @@
                          (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
         self.replace_spn(self.ldb_user1, ctx.acct_dn, "GC/%s.%s/%s" %
-                         (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
+                         (ctx.myname, ctx.dnsdomain, ctx.dnsforest))
         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s/%s" % (ctx.myname, netbiosdomain))
         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/%s" %
@@ -1737 +1946 @@
                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
         self.replace_spn(self.ldb_admin, self.computerdn, "GC/%s.%s/%s" %
-                         (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+                         (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s/%s" % (self.computername, netbiosdomain))
         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/ForestDnsZones.%s" %
@@ -1801 +2010 @@
         try:
             self.replace_spn(self.ldb_user1, self.computerdn, "GC/%s.%s/%s" %
-                             (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+                             (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
         except LdbError, (num, _):
             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
@@ -1825 +2034 @@
 ldb = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
 
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(AclAddTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(AclModifyTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(AclDeleteTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(AclRenameTests)).wasSuccessful():
-    rc = 1
-
-# Get the old "dSHeuristics" if it was set
-dsheuristics = ldb.get_dsheuristics()
-# Set the "dSHeuristics" to activate the correct "userPassword" behaviour
-ldb.set_dsheuristics("000000001")
-# Get the old "minPwdAge"
-minPwdAge = ldb.get_minPwdAge()
-# Set it temporarely to "0"
-ldb.set_minPwdAge("0")
-if not runner.run(unittest.makeSuite(AclCARTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(AclSearchTests)).wasSuccessful():
-    rc = 1
-# Reset the "dSHeuristics" as they were before
-ldb.set_dsheuristics(dsheuristics)
-# Reset the "minPwdAge" as it was before
-ldb.set_minPwdAge(minPwdAge)
-
-if not runner.run(unittest.makeSuite(AclExtendedTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(AclSPNTests)).wasSuccessful():
-    rc = 1
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/deletetest.py

@@ -8 +8 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 
 import samba.getopt as options
 
 from samba.auth import system_session
-from ldb import SCOPE_BASE, LdbError
-from ldb import ERR_NO_SUCH_OBJECT, ERR_NOT_ALLOWED_ON_NON_LEAF
-from ldb import ERR_UNWILLING_TO_PERFORM
+from ldb import SCOPE_BASE, LdbError, Message, MessageElement, Dn, FLAG_MOD_ADD, FLAG_MOD_DELETE, FLAG_MOD_REPLACE
+from ldb import ERR_NO_SUCH_OBJECT, ERR_NOT_ALLOWED_ON_NON_LEAF, ERR_ENTRY_ALREADY_EXISTS, ERR_ATTRIBUTE_OR_VALUE_EXISTS
+from ldb import ERR_UNWILLING_TO_PERFORM, ERR_OPERATIONS_ERROR
 from samba.samdb import SamDB
 from samba.tests import delete_force
-
-from subunit.run import SubunitTestRunner
-import unittest
 
 parser = optparse.OptionParser("deletetest.py [options] <host|file>")
@@ -30 +27 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
 opts, args = parser.parse_args()
 
@@ -41 +40 @@
 creds = credopts.get_credentials(lp)
 
-class BasicDeleteTests(unittest.TestCase):
-
+class BaseDeleteTests(samba.tests.TestCase):
 
     def GUID_string(self, guid):
@@ -48 +46 @@
 
     def setUp(self):
-        self.ldb = ldb
-        self.base_dn = ldb.domain_dn()
-        self.configuration_dn = ldb.get_config_basedn().get_linearized()
+        super(BaseDeleteTests, self).setUp()
+        self.ldb = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp)
+
+        self.base_dn = self.ldb.domain_dn()
+        self.configuration_dn = self.ldb.get_config_basedn().get_linearized()
 
     def search_guid(self, guid):
         print "SEARCH by GUID %s" % self.GUID_string(guid)
 
-        res = ldb.search(base="<GUID=%s>" % self.GUID_string(guid),
+        res = self.ldb.search(base="<GUID=%s>" % self.GUID_string(guid),
                          scope=SCOPE_BASE, controls=["show_deleted:1"])
         self.assertEquals(len(res), 1)
@@ -63 +63 @@
         print "SEARCH by DN %s" % dn
 
-        res = ldb.search(expression="(objectClass=*)",
+        res = self.ldb.search(expression="(objectClass=*)",
                          base=dn,
                          scope=SCOPE_BASE,
@@ -69 +69 @@
         self.assertEquals(len(res), 1)
         return res[0]
+
+
+class BasicDeleteTests(BaseDeleteTests):
+
+    def setUp(self):
+        super(BasicDeleteTests, self).setUp()
 
     def del_attr_values(self, delObj):
@@ -119 +125 @@
         delete_force(self.ldb, "cn=ldaptestcontainer," + self.base_dn)
 
-        ldb.add({
+        self.ldb.add({
             "dn": "cn=ldaptestcontainer," + self.base_dn,
             "objectclass": "container"})
-        ldb.add({
+        self.ldb.add({
             "dn": "cn=entry1,cn=ldaptestcontainer," + self.base_dn,
             "objectclass": "container"})
-        ldb.add({
+        self.ldb.add({
             "dn": "cn=entry2,cn=ldaptestcontainer," + self.base_dn,
             "objectclass": "container"})
 
         try:
-            ldb.delete("cn=ldaptestcontainer," + self.base_dn)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
-
-        ldb.delete("cn=ldaptestcontainer," + self.base_dn, ["tree_delete:1"])
-
-        try:
-            res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+            self.ldb.delete("cn=ldaptestcontainer," + self.base_dn)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
+
+        self.ldb.delete("cn=ldaptestcontainer," + self.base_dn, ["tree_delete:1"])
+
+        try:
+            res = self.ldb.search("cn=ldaptestcontainer," + self.base_dn,
                              scope=SCOPE_BASE, attrs=[])
             self.fail()
@@ -144 +150 @@
             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
         try:
-            res = ldb.search("cn=entry1,cn=ldaptestcontainer," + self.base_dn,
+            res = self.ldb.search("cn=entry1,cn=ldaptestcontainer," + self.base_dn,
                              scope=SCOPE_BASE, attrs=[])
             self.fail()
@@ -150 +156 @@
             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
         try:
-            res = ldb.search("cn=entry2,cn=ldaptestcontainer," + self.base_dn,
+            res = self.ldb.search("cn=entry2,cn=ldaptestcontainer," + self.base_dn,
                              scope=SCOPE_BASE, attrs=[])
             self.fail()
@@ -162 +168 @@
         # Performs some protected object delete testing
 
-        res = ldb.search(base="", expression="", scope=SCOPE_BASE,
+        res = self.ldb.search(base="", expression="", scope=SCOPE_BASE,
                          attrs=["dsServiceName", "dNSHostName"])
         self.assertEquals(len(res), 1)
@@ -168 +174 @@
         # Delete failing since DC's nTDSDSA object is protected
         try:
-            ldb.delete(res[0]["dsServiceName"][0])
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
-        res = ldb.search(self.base_dn, attrs=["rIDSetReferences"],
+            self.ldb.delete(res[0]["dsServiceName"][0])
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        res = self.ldb.search(self.base_dn, attrs=["rIDSetReferences"],
                          expression="(&(objectClass=computer)(dNSHostName=" + res[0]["dNSHostName"][0] + "))")
         self.assertEquals(len(res), 1)
@@ -179 +185 @@
         # Deletes failing since DC's rIDSet object is protected
         try:
-            ldb.delete(res[0]["rIDSetReferences"][0])
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-        try:
-            ldb.delete(res[0]["rIDSetReferences"][0], ["tree_delete:1"])
+            self.ldb.delete(res[0]["rIDSetReferences"][0])
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        try:
+            self.ldb.delete(res[0]["rIDSetReferences"][0], ["tree_delete:1"])
             self.fail()
         except LdbError, (num, _):
@@ -192 +198 @@
 
         try:
-            ldb.delete("cn=Enterprise Schema,cn=Partitions," + self.configuration_dn)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-        try:
-            ldb.delete("cn=Enterprise Schema,cn=Partitions," + self.configuration_dn, ["tree_delete:1"])
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
-        try:
-            ldb.delete("cn=Enterprise Configuration,cn=Partitions," + self.configuration_dn)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
-        try:
-            ldb.delete("cn=Enterprise Configuration,cn=Partitions," + self.configuration_dn, ["tree_delete:1"])
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
-
-        res = ldb.search("cn=Partitions," + self.configuration_dn, attrs=[],
+            self.ldb.delete("cn=Enterprise Schema,cn=Partitions," + self.configuration_dn)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        try:
+            self.ldb.delete("cn=Enterprise Schema,cn=Partitions," + self.configuration_dn, ["tree_delete:1"])
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        try:
+            self.ldb.delete("cn=Enterprise Configuration,cn=Partitions," + self.configuration_dn)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
+        try:
+            self.ldb.delete("cn=Enterprise Configuration,cn=Partitions," + self.configuration_dn, ["tree_delete:1"])
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
+
+        res = self.ldb.search("cn=Partitions," + self.configuration_dn, attrs=[],
                          expression="(nCName=%s)" % self.base_dn)
         self.assertEquals(len(res), 1)
 
         try:
-            ldb.delete(res[0].dn)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
-        try:
-            ldb.delete(res[0].dn, ["tree_delete:1"])
+            self.ldb.delete(res[0].dn)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
+        try:
+            self.ldb.delete(res[0].dn, ["tree_delete:1"])
             self.fail()
         except LdbError, (num, _):
@@ -230 +236 @@
         # Delete failing since "SYSTEM_FLAG_DISALLOW_DELETE"
         try:
-            ldb.delete("CN=Users," + self.base_dn)
+            self.ldb.delete("CN=Users," + self.base_dn)
             self.fail()
         except LdbError, (num, _):
@@ -237 +243 @@
         # Tree-delete failing since "isCriticalSystemObject"
         try:
-            ldb.delete("CN=Computers," + self.base_dn, ["tree_delete:1"])
+            self.ldb.delete("CN=Computers," + self.base_dn, ["tree_delete:1"])
             self.fail()
         except LdbError, (num, _):
@@ -247 +253 @@
         print self.base_dn
 
-        usr1="cn=testuser,cn=users," + self.base_dn
-        usr2="cn=testuser2,cn=users," + self.base_dn
-        grp1="cn=testdelgroup1,cn=users," + self.base_dn
-        sit1="cn=testsite1,cn=sites," + self.configuration_dn
-        ss1="cn=NTDS Site Settings,cn=testsite1,cn=sites," + self.configuration_dn
-        srv1="cn=Servers,cn=testsite1,cn=sites," + self.configuration_dn
-        srv2="cn=TESTSRV,cn=Servers,cn=testsite1,cn=sites," + self.configuration_dn
+        # user current time in ms to make unique objects
+        import time
+        marker = str(int(round(time.time()*1000)))
+        usr1_name = "u_" + marker
+        usr2_name = "u2_" + marker
+        grp_name = "g1_" + marker
+        site_name = "s1_" + marker
+
+        usr1 = "cn=%s,cn=users,%s" % (usr1_name, self.base_dn)
+        usr2 = "cn=%s,cn=users,%s" % (usr2_name, self.base_dn)
+        grp1 = "cn=%s,cn=users,%s" % (grp_name, self.base_dn)
+        sit1 = "cn=%s,cn=sites,%s" % (site_name, self.configuration_dn)
+        ss1 = "cn=NTDS Site Settings,cn=%s,cn=sites,%s" % (site_name, self.configuration_dn)
+        srv1 = "cn=Servers,cn=%s,cn=sites,%s" % (site_name, self.configuration_dn)
+        srv2 = "cn=TESTSRV,cn=Servers,cn=%s,cn=sites,%s" % (site_name, self.configuration_dn)
 
         delete_force(self.ldb, usr1)
@@ -263 +277 @@
         delete_force(self.ldb, sit1)
 
-        ldb.add({
+        self.ldb.add({
             "dn": usr1,
             "objectclass": "user",
             "description": "test user description",
-            "samaccountname": "testuser"})
-
-        ldb.add({
+            "samaccountname": usr1_name})
+
+        self.ldb.add({
             "dn": usr2,
             "objectclass": "user",
             "description": "test user 2 description",
-            "samaccountname": "testuser2"})
-
-        ldb.add({
+            "samaccountname": usr2_name})
+
+        self.ldb.add({
             "dn": grp1,
             "objectclass": "group",
             "description": "test group",
-            "samaccountname": "testdelgroup1",
+            "samaccountname": grp_name,
             "member": [ usr1, usr2 ],
             "isDeleted": "FALSE" })
 
-        ldb.add({
+        self.ldb.add({
             "dn": sit1,
             "objectclass": "site" })
 
-        ldb.add({
+        self.ldb.add({
             "dn": ss1,
             "objectclass": ["applicationSiteSettings", "nTDSSiteSettings"] })
 
-        ldb.add({
+        self.ldb.add({
             "dn": srv1,
             "objectclass": "serversContainer" })
 
-        ldb.add({
+        self.ldb.add({
             "dn": srv2,
             "objectClass": "server" })
@@ -320 +334 @@
         guid7=objLive7["objectGUID"][0]
 
-        ldb.delete(usr1)
-        ldb.delete(usr2)
-        ldb.delete(grp1)
-        ldb.delete(srv1, ["tree_delete:1"])
-        ldb.delete(sit1, ["tree_delete:1"])
+        self.ldb.delete(usr1)
+        self.ldb.delete(usr2)
+        self.ldb.delete(grp1)
+        self.ldb.delete(srv1, ["tree_delete:1"])
+        self.ldb.delete(sit1, ["tree_delete:1"])
 
         objDeleted1 = self.search_guid(guid1)
@@ -358 +372 @@
         self.check_rdn(objLive7, objDeleted7, "cn")
 
-        self.delete_deleted(ldb, usr1)
-        self.delete_deleted(ldb, usr2)
-        self.delete_deleted(ldb, grp1)
-        self.delete_deleted(ldb, sit1)
-        self.delete_deleted(ldb, ss1)
-        self.delete_deleted(ldb, srv1)
-        self.delete_deleted(ldb, srv2)
+        self.delete_deleted(self.ldb, usr1)
+        self.delete_deleted(self.ldb, usr2)
+        self.delete_deleted(self.ldb, grp1)
+        self.delete_deleted(self.ldb, sit1)
+        self.delete_deleted(self.ldb, ss1)
+        self.delete_deleted(self.ldb, srv1)
+        self.delete_deleted(self.ldb, srv2)
 
         self.assertTrue("CN=Deleted Objects" in str(objDeleted1.dn))
@@ -374 +388 @@
         self.assertFalse("CN=Deleted Objects" in str(objDeleted7.dn))
 
+
 if not "://" in host:
     if os.path.isfile(host):
@@ -380 +395 @@
         host = "ldap://%s" % host
 
-ldb = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp)
-
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(BasicDeleteTests)).wasSuccessful():
-    rc = 1
-
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/dsdb_schema_info.py

@@ -31 +31 @@
 
 sys.path.insert(0, "bin/python")
-import samba
-samba.ensure_external_module("testtools", "testtools")
+import samba.tests
 
 from ldb import SCOPE_BASE, LdbError

vendor/current/source4/dsdb/tests/python/ldap.py

@@ -2 +2 @@
 # -*- coding: utf-8 -*-
 # This is a port of the original in testprogs/ejs/ldap.js
+
+# Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2008-2011
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import optparse
@@ -11 +26 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
-
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 import samba.getopt as options
 
@@ -36 +49 @@
     SYSTEM_FLAG_CONFIG_ALLOW_LIMITED_MOVE)
 
-from subunit.run import SubunitTestRunner
-import unittest
-
 from samba.ndr import ndr_pack, ndr_unpack
 from samba.dcerpc import security, lsa
@@ -50 +60 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
 opts, args = parser.parse_args()
 
@@ -61 +73 @@
 creds = credopts.get_credentials(lp)
 
-class BasicTests(unittest.TestCase):
+class BasicTests(samba.tests.TestCase):
 
     def setUp(self):
@@ -71 +83 @@
         self.schema_dn = ldb.get_schema_basedn().get_linearized()
         self.domain_sid = security.dom_sid(ldb.get_domain_sid())
-
-        print "baseDN: %s\n" % self.base_dn
 
         delete_force(self.ldb, "cn=posixuser,cn=users," + self.base_dn)
@@ -97 +107 @@
         delete_force(self.ldb, "ou=testou,cn=users," + self.base_dn)
         delete_force(self.ldb, "cn=Test Secret,cn=system," + self.base_dn)
+        delete_force(self.ldb, "cn=testtimevaluesuser1,cn=users," + self.base_dn)
 
     def test_objectclasses(self):
         """Test objectClass behaviour"""
-        print "Test objectClass behaviour"""
-
-        # We cannot create LSA-specific objects (oc "secret" or "trustedDomain")
-        try:
-            self.ldb.add({
-                "dn": "cn=Test Secret,cn=system," + self.base_dn,
-                "objectClass": "secret" })
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
         # Invalid objectclass specified
         try:
@@ -149 +149 @@
             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
 
-        # We cannot instanciate from an abstract objectclass
+        # We cannot instanciate from an abstract object class ("connectionPoint"
+        # or "leaf"). In the first case we use "connectionPoint" (subclass of
+        # "leaf") to prevent a naming violation - this returns us a
+        # "ERR_UNWILLING_TO_PERFORM" since it is not structural. In the second
+        # case however we get "ERR_OBJECT_CLASS_VIOLATION" since an abstract
+        # class is also not allowed to be auxiliary.
         try:
             self.ldb.add({
@@ -157 +162 @@
         except LdbError, (num, _):
             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        try:
+            self.ldb.add({
+                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+                "objectClass": ["person", "leaf"] })
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
+
+        # Objects instanciated using "satisfied" abstract classes (concrete
+        # subclasses) are allowed
+        self.ldb.add({
+             "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+             "objectClass": ["top", "leaf", "connectionPoint", "serviceConnectionPoint"] })
+
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+
+        # Two disjoint top-most structural object classes aren't allowed
+        try:
+            self.ldb.add({
+                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+                "objectClass": ["person", "container"] })
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
 
         # Test allowed system flags
@@ -222 +251 @@
             self.assertEquals(num, ERR_NO_SUCH_ATTRIBUTE)
 
-        # The top-most structural class cannot be changed by adding another
-        # structural one
+        # We cannot add a the new top-most structural class "user" here since
+        # we are missing at least one new mandatory attribute (in this case
+        # "sAMAccountName")
         m = Message()
         m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
@@ -252 +282 @@
         ldb.modify(m)
 
-        # It's only possible to replace with the same objectclass combination.
-        # So the replace action on "objectClass" attributes is really useless.
+        # This does not work since object class "leaf" is not auxiliary nor it
+        # stands in direct relation to "person" (and it is abstract too!)
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["objectClass"] = MessageElement("leaf", FLAG_MOD_ADD,
+          "objectClass")
+        try:
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
+
+        # Objectclass replace operations can be performed as well
         m = Message()
         m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
@@ -266 +307 @@
         ldb.modify(m)
 
+        # This does not work since object class "leaf" is not auxiliary nor it
+        # stands in direct relation to "person" (and it is abstract too!)
         m = Message()
         m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
         m["objectClass"] = MessageElement(["top", "person", "bootableDevice",
-          "connectionPoint"], FLAG_MOD_REPLACE, "objectClass")
+          "leaf"], FLAG_MOD_REPLACE, "objectClass")
         try:
             ldb.modify(m)
@@ -356 +399 @@
         delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
 
+        self.ldb.add({
+             "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+             "objectClass": "user" })
+
+        # Add a new top-most structural class "container". This does not work
+        # since it stands in no direct relation to the current one.
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["objectClass"] = MessageElement("container", FLAG_MOD_ADD,
+          "objectClass")
+        try:
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
+
+        # Add a new top-most structural class "inetOrgPerson" and remove it
+        # afterwards
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["objectClass"] = MessageElement("inetOrgPerson", FLAG_MOD_ADD,
+          "objectClass")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["objectClass"] = MessageElement("inetOrgPerson", FLAG_MOD_DELETE,
+          "objectClass")
+        ldb.modify(m)
+
+        # Replace top-most structural class to "inetOrgPerson" and reset it
+        # back to "user"
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["objectClass"] = MessageElement("inetOrgPerson", FLAG_MOD_REPLACE,
+          "objectClass")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["objectClass"] = MessageElement("user", FLAG_MOD_REPLACE,
+          "objectClass")
+        ldb.modify(m)
+
+        # Add a new auxiliary object class "posixAccount" to "ldaptestuser"
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["objectClass"] = MessageElement("posixAccount", FLAG_MOD_ADD,
+          "objectClass")
+        ldb.modify(m)
+
+        # Be sure that "top" is the first and the (most) structural object class
+        # the last value of the "objectClass" attribute - MS-ADTS 3.1.1.1.4
+        res = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
+                         scope=SCOPE_BASE, attrs=["objectClass"])
+        self.assertTrue(len(res) == 1)
+        self.assertEquals(res[0]["objectClass"][0], "top")
+        self.assertEquals(res[0]["objectClass"][len(res[0]["objectClass"])-1], "user")
+
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+
     def test_system_only(self):
         """Test systemOnly objects"""
-        print "Test systemOnly objects"""
-
         try:
             self.ldb.add({
@@ -445 +547 @@
     def test_invalid_parent(self):
         """Test adding an object with invalid parent"""
-        print "Test adding an object with invalid parent"""
-
         try:
             self.ldb.add({
@@ -471 +571 @@
     def test_invalid_attribute(self):
         """Test invalid attributes on schema/objectclasses"""
-        print "Test invalid attributes on schema/objectclasses"""
-
         # attributes not in schema test
 
@@ -573 +671 @@
     def test_single_valued_attributes(self):
         """Test single-valued attributes"""
-        print "Test single-valued attributes"""
-
         try:
             self.ldb.add({
@@ -618 +714 @@
     def test_attribute_ranges(self):
         """Test attribute ranges"""
-        print "Test attribute ranges"""
-
         # Too short (min. 1)
         try:
@@ -673 +767 @@
     def test_empty_messages(self):
         """Test empty messages"""
-        print "Test empty messages"""
-
         m = Message()
         m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
@@ -694 +786 @@
     def test_empty_attributes(self):
         """Test empty attributes"""
-        print "Test empty attributes"""
-
         m = Message()
         m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
@@ -739 +829 @@
     def test_instanceType(self):
         """Tests the 'instanceType' attribute"""
-        print "Tests the 'instanceType' attribute"""
-
         # The instance type is single-valued
         try:
@@ -806 +894 @@
         delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
 
+        #only write is allowed with NC_HEAD for originating updates
+        try:
+            self.ldb.add({
+                "dn": "cn=ldaptestuser2,cn=users," + self.base_dn,
+                "objectclass": "user",
+                "instanceType": "3" })
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        delete_force(self.ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
+
     def test_distinguished_name(self):
         """Tests the 'distinguishedName' attribute"""
-        print "Tests the 'distinguishedName' attribute"""
-
         # The "dn" shortcut isn't supported
         m = Message()
@@ -858 +955 @@
             self.fail()
         except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
         m = Message()
@@ -882 +979 @@
             self.fail()
         except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
         delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
@@ -888 +985 @@
     def test_rdn_name(self):
         """Tests the RDN"""
-        print "Tests the RDN"""
-
         # Search
 
@@ -1059 +1154 @@
     def DISABLED_test_largeRDN(self):
         """Testing large rDN (limit 64 characters)"""
-        rdn = "CN=a012345678901234567890123456789012345678901234567890123456789012";
+        rdn = "CN=a012345678901234567890123456789012345678901234567890123456789012"
         delete_force(self.ldb, "%s,%s" % (rdn, self.base_dn))
         ldif = """
@@ -1068 +1163 @@
         delete_force(self.ldb, "%s,%s" % (rdn, self.base_dn))
 
-        rdn = "CN=a0123456789012345678901234567890123456789012345678901234567890120";
+        rdn = "CN=a0123456789012345678901234567890123456789012345678901234567890120"
         delete_force(self.ldb, "%s,%s" % (rdn, self.base_dn))
         try:
@@ -1083 +1178 @@
     def test_rename(self):
         """Tests the rename operation"""
-        print "Tests the rename operations"""
-
         try:
             # cannot rename to be a child of itself
@@ -1194 +1287 @@
     def test_rename_twice(self):
         """Tests the rename operation twice - this corresponds to a past bug"""
-        print "Tests the rename twice operation"""
-
         self.ldb.add({
              "dn": "cn=ldaptestuser5,cn=users," + self.base_dn,
@@ -1207 +1298 @@
         ldb.rename("cn=ldaptestuser5,cn=Users," + self.base_dn, "cn=ldaptestUSER5,cn=users," + self.base_dn)
         res = ldb.search(expression="cn=ldaptestuser5")
-        print "Found %u records" % len(res)
         self.assertEquals(len(res), 1, "Wrong number of hits for cn=ldaptestuser5")
         res = ldb.search(expression="(&(cn=ldaptestuser5)(objectclass=user))")
-        print "Found %u records" % len(res)
         self.assertEquals(len(res), 1, "Wrong number of hits for (&(cn=ldaptestuser5)(objectclass=user))")
         delete_force(self.ldb, "cn=ldaptestuser5,cn=users," + self.base_dn)
@@ -1216 +1305 @@
     def test_objectGUID(self):
         """Test objectGUID behaviour"""
-        print "Testing objectGUID behaviour\n"
-
         # The objectGUID cannot directly be set
         try:
@@ -1232 +1319 @@
             "dn": "cn=ldaptestcontainer," + self.base_dn,
             "objectClass": "container" })
-
-        res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
-                         scope=SCOPE_BASE,
-                         attrs=["objectGUID", "uSNCreated", "uSNChanged", "whenCreated", "whenChanged"])
-        self.assertTrue(len(res) == 1)
-        self.assertTrue("objectGUID" in res[0])
-        self.assertTrue("uSNCreated" in res[0])
-        self.assertTrue("uSNChanged" in res[0])
-        self.assertTrue("whenCreated" in res[0])
-        self.assertTrue("whenChanged" in res[0])
-
-        delete_force(self.ldb, "cn=ldaptestcontainer," + self.base_dn)
-
-        # All the following attributes are specificable on add operations
-        self.ldb.add({
-            "dn": "cn=ldaptestcontainer," + self.base_dn,
-            "objectClass": "container",
-            "uSNCreated" : "1",
-            "uSNChanged" : "1",
-            "whenCreated": timestring(long(time.time())),
-            "whenChanged": timestring(long(time.time())) })
-
-        res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
-                         scope=SCOPE_BASE,
-                         attrs=["objectGUID", "uSNCreated", "uSNChanged", "whenCreated", "whenChanged"])
-        self.assertTrue(len(res) == 1)
-        self.assertTrue("objectGUID" in res[0])
-        self.assertTrue("uSNCreated" in res[0])
-        self.assertFalse(res[0]["uSNCreated"][0] == "1") # these are corrected
-        self.assertTrue("uSNChanged" in res[0])
-        self.assertFalse(res[0]["uSNChanged"][0] == "1") # these are corrected
-
-        delete_force(self.ldb, "cn=ldaptestcontainer," + self.base_dn)
-
-        # All this attributes are specificable on add operations
-        self.ldb.add({
-            "dn": "cn=ldaptestcontainer," + self.base_dn,
-            "objectclass": "container",
-            "uSNCreated" : "1",
-            "uSNChanged" : "1",
-            "whenCreated": timestring(long(time.time())),
-            "whenChanged": timestring(long(time.time())) })
-
-        res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
-                         scope=SCOPE_BASE,
-                         attrs=["objectGUID", "uSNCreated", "uSNChanged", "whenCreated", "whenChanged"])
-        self.assertTrue(len(res) == 1)
-        self.assertTrue("objectGUID" in res[0])
-        self.assertTrue("uSNCreated" in res[0])
-        self.assertFalse(res[0]["uSNCreated"][0] == "1") # these are corrected
-        self.assertTrue("uSNChanged" in res[0])
-        self.assertFalse(res[0]["uSNChanged"][0] == "1") # these are corrected
-        self.assertTrue("whenCreated" in res[0])
-        self.assertTrue("whenChanged" in res[0])
 
         # The objectGUID cannot directly be changed
@@ -1303 +1336 @@
     def test_parentGUID(self):
         """Test parentGUID behaviour"""
-        print "Testing parentGUID behaviour\n"
-
         self.ldb.add({
             "dn": "cn=parentguidtest,cn=users," + self.base_dn,
             "objectclass":"user",
-            "samaccountname":"parentguidtest"});
+            "samaccountname":"parentguidtest"})
         res1 = ldb.search(base="cn=parentguidtest,cn=users," + self.base_dn, scope=SCOPE_BASE,
-                          attrs=["parentGUID", "samaccountname"]);
+                          attrs=["parentGUID", "samaccountname"])
         res2 = ldb.search(base="cn=users," + self.base_dn,scope=SCOPE_BASE,
-                          attrs=["objectGUID"]);
+                          attrs=["objectGUID"])
         res3 = ldb.search(base=self.base_dn, scope=SCOPE_BASE,
-                          attrs=["parentGUID"]);
+                          attrs=["parentGUID"])
         res4 = ldb.search(base=self.configuration_dn, scope=SCOPE_BASE,
-                          attrs=["parentGUID"]);
+                          attrs=["parentGUID"])
         res5 = ldb.search(base=self.schema_dn, scope=SCOPE_BASE,
-                          attrs=["parentGUID"]);
+                          attrs=["parentGUID"])
 
         """Check if the parentGUID is valid """
-        self.assertEquals(res1[0]["parentGUID"], res2[0]["objectGUID"]);
+        self.assertEquals(res1[0]["parentGUID"], res2[0]["objectGUID"])
 
         """Check if it returns nothing when there is no parent object - default NC"""
@@ -1329 +1360 @@
                 has_parentGUID = True
                 break
-        self.assertFalse(has_parentGUID);
+        self.assertFalse(has_parentGUID)
 
         """Check if it returns nothing when there is no parent object - configuration NC"""
@@ -1337 +1368 @@
                 has_parentGUID = True
                 break
-        self.assertFalse(has_parentGUID);
+        self.assertFalse(has_parentGUID)
 
         """Check if it returns nothing when there is no parent object - schema NC"""
@@ -1345 +1376 @@
                 has_parentGUID = True
                 break
-        self.assertFalse(has_parentGUID);
+        self.assertFalse(has_parentGUID)
 
         """Ensures that if you look for another object attribute after the constructed
@@ -1356 +1387 @@
         self.assertTrue(has_another_attribute)
         self.assertTrue(len(res1[0]["samaccountname"]) == 1)
-        self.assertEquals(res1[0]["samaccountname"][0], "parentguidtest");
-
-        print "Testing parentGUID behaviour on rename\n"
+        self.assertEquals(res1[0]["samaccountname"][0], "parentguidtest")
+
+        # Testing parentGUID behaviour on rename\
 
         self.ldb.add({
             "dn": "cn=testotherusers," + self.base_dn,
-            "objectclass":"container"});
+            "objectclass":"container"})
         res1 = ldb.search(base="cn=testotherusers," + self.base_dn,scope=SCOPE_BASE,
-                          attrs=["objectGUID"]);
+                          attrs=["objectGUID"])
         ldb.rename("cn=parentguidtest,cn=users," + self.base_dn,
-                   "cn=parentguidtest,cn=testotherusers," + self.base_dn);
+                   "cn=parentguidtest,cn=testotherusers," + self.base_dn)
         res2 = ldb.search(base="cn=parentguidtest,cn=testotherusers," + self.base_dn,
                           scope=SCOPE_BASE,
-                          attrs=["parentGUID"]);
-        self.assertEquals(res1[0]["objectGUID"], res2[0]["parentGUID"]);
+                          attrs=["parentGUID"])
+        self.assertEquals(res1[0]["objectGUID"], res2[0]["parentGUID"])
 
         delete_force(self.ldb, "cn=parentguidtest,cn=testotherusers," + self.base_dn)
         delete_force(self.ldb, "cn=testotherusers," + self.base_dn)
 
+    def test_usnChanged(self):
+        """Test usnChanged behaviour"""
+
+        self.ldb.add({
+            "dn": "cn=ldaptestcontainer," + self.base_dn,
+            "objectClass": "container" })
+
+        res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["objectGUID", "uSNCreated", "uSNChanged", "whenCreated", "whenChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertFalse("description" in res[0])
+        self.assertTrue("objectGUID" in res[0])
+        self.assertTrue("uSNCreated" in res[0])
+        self.assertTrue("uSNChanged" in res[0])
+        self.assertTrue("whenCreated" in res[0])
+        self.assertTrue("whenChanged" in res[0])
+
+        delete_force(self.ldb, "cn=ldaptestcontainer," + self.base_dn)
+
+        # All this attributes are specificable on add operations
+        self.ldb.add({
+            "dn": "cn=ldaptestcontainer," + self.base_dn,
+            "objectclass": "container",
+            "uSNCreated" : "1",
+            "uSNChanged" : "1",
+            "whenCreated": timestring(long(time.time())),
+            "whenChanged": timestring(long(time.time())) })
+
+        res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["objectGUID", "uSNCreated", "uSNChanged", "whenCreated", "whenChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertFalse("description" in res[0])
+        self.assertTrue("objectGUID" in res[0])
+        self.assertTrue("uSNCreated" in res[0])
+        self.assertFalse(res[0]["uSNCreated"][0] == "1") # these are corrected
+        self.assertTrue("uSNChanged" in res[0])
+        self.assertFalse(res[0]["uSNChanged"][0] == "1") # these are corrected
+        self.assertTrue("whenCreated" in res[0])
+        self.assertTrue("whenChanged" in res[0])
+
+        ldb.modify_ldif("""
+dn: cn=ldaptestcontainer,""" + self.base_dn + """
+changetype: modify
+replace: description
+""")
+
+        res2 = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["uSNCreated", "uSNChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertFalse("description" in res2[0])
+        self.assertEqual(res[0]["usnCreated"], res2[0]["usnCreated"])
+        self.assertEqual(res[0]["usnCreated"], res2[0]["usnChanged"])
+        self.assertEqual(res[0]["usnChanged"], res2[0]["usnChanged"])
+
+        ldb.modify_ldif("""
+dn: cn=ldaptestcontainer,""" + self.base_dn + """
+changetype: modify
+replace: description
+description: test
+""")
+
+        res3 = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["uSNCreated", "uSNChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertTrue("description" in res3[0])
+        self.assertEqual("test", str(res3[0]["description"][0]))
+        self.assertEqual(res[0]["usnCreated"], res3[0]["usnCreated"])
+        self.assertNotEqual(res[0]["usnCreated"], res3[0]["usnChanged"])
+        self.assertNotEqual(res[0]["usnChanged"], res3[0]["usnChanged"])
+
+        ldb.modify_ldif("""
+dn: cn=ldaptestcontainer,""" + self.base_dn + """
+changetype: modify
+replace: description
+description: test
+""")
+
+        res4 = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["uSNCreated", "uSNChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertTrue("description" in res4[0])
+        self.assertEqual("test", str(res4[0]["description"][0]))
+        self.assertEqual(res[0]["usnCreated"], res4[0]["usnCreated"])
+        self.assertNotEqual(res3[0]["usnCreated"], res4[0]["usnChanged"])
+        self.assertEqual(res3[0]["usnChanged"], res4[0]["usnChanged"])
+
+        ldb.modify_ldif("""
+dn: cn=ldaptestcontainer,""" + self.base_dn + """
+changetype: modify
+replace: description
+description: test2
+""")
+
+        res5 = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["uSNCreated", "uSNChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertTrue("description" in res5[0])
+        self.assertEqual("test2", str(res5[0]["description"][0]))
+        self.assertEqual(res[0]["usnCreated"], res5[0]["usnCreated"])
+        self.assertNotEqual(res3[0]["usnChanged"], res5[0]["usnChanged"])
+
+        ldb.modify_ldif("""
+dn: cn=ldaptestcontainer,""" + self.base_dn + """
+changetype: modify
+delete: description
+description: test2
+""")
+
+        res6 = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["uSNCreated", "uSNChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertFalse("description" in res6[0])
+        self.assertEqual(res[0]["usnCreated"], res6[0]["usnCreated"])
+        self.assertNotEqual(res5[0]["usnChanged"], res6[0]["usnChanged"])
+
+        ldb.modify_ldif("""
+dn: cn=ldaptestcontainer,""" + self.base_dn + """
+changetype: modify
+add: description
+description: test3
+""")
+
+        res7 = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["uSNCreated", "uSNChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertTrue("description" in res7[0])
+        self.assertEqual("test3", str(res7[0]["description"][0]))
+        self.assertEqual(res[0]["usnCreated"], res7[0]["usnCreated"])
+        self.assertNotEqual(res6[0]["usnChanged"], res7[0]["usnChanged"])
+
+        ldb.modify_ldif("""
+dn: cn=ldaptestcontainer,""" + self.base_dn + """
+changetype: modify
+delete: description
+""")
+
+        res8 = ldb.search("cn=ldaptestcontainer," + self.base_dn,
+                         scope=SCOPE_BASE,
+                         attrs=["uSNCreated", "uSNChanged", "description"])
+        self.assertTrue(len(res) == 1)
+        self.assertFalse("description" in res8[0])
+        self.assertEqual(res[0]["usnCreated"], res8[0]["usnCreated"])
+        self.assertNotEqual(res7[0]["usnChanged"], res8[0]["usnChanged"])
+
+        delete_force(self.ldb, "cn=ldaptestcontainer," + self.base_dn)
+
     def test_groupType_int32(self):
         """Test groupType (int32) behaviour (should appear to be casted to a 32 bit signed integer before comparsion)"""
-        print "Testing groupType (int32) behaviour\n"
 
         res1 = ldb.search(base=self.base_dn, scope=SCOPE_SUBTREE,
-                          attrs=["groupType"], expression="groupType=2147483653");
+                          attrs=["groupType"], expression="groupType=2147483653")
 
         res2 = ldb.search(base=self.base_dn, scope=SCOPE_SUBTREE,
-                          attrs=["groupType"], expression="groupType=-2147483643");
+                          attrs=["groupType"], expression="groupType=-2147483643")
 
         self.assertEquals(len(res1), len(res2))
@@ -1393 +1577 @@
     def test_linked_attributes(self):
         """This tests the linked attribute behaviour"""
-        print "Testing linked attribute behaviour\n"
 
         ldb.add({
@@ -1482 +1665 @@
     def test_wkguid(self):
         """Test Well known GUID behaviours (including DN+Binary)"""
-        print "Test Well known GUID behaviours (including DN+Binary)"""
 
         res = self.ldb.search(base=("<WKGUID=ab1d30f3768811d1aded00c04fd8d5cd,%s>" % self.base_dn), scope=SCOPE_BASE, attrs=[])
@@ -1499 +1681 @@
     def test_subschemasubentry(self):
         """Test subSchemaSubEntry appears when requested, but not when not requested"""
-        print "Test subSchemaSubEntry"""
 
         res = self.ldb.search(base=self.base_dn, scope=SCOPE_BASE, attrs=["subSchemaSubEntry"])
@@ -1512 +1693 @@
         """Basic tests"""
 
-        print "Testing user add"
+        # Testing user add
 
         ldb.add({
@@ -1561 +1742 @@
                  })
 
-        print "Testing ldb.search for (&(cn=ldaptestcomputer3)(objectClass=user))";
-        res = ldb.search(self.base_dn, expression="(&(cn=ldaptestcomputer3)(objectClass=user))");
+        # Testing ldb.search for (&(cn=ldaptestcomputer3)(objectClass=user))
+        res = ldb.search(self.base_dn, expression="(&(cn=ldaptestcomputer3)(objectClass=user))")
         self.assertEquals(len(res), 1, "Found only %d for (&(cn=ldaptestcomputer3)(objectClass=user))" % len(res))
 
-        self.assertEquals(str(res[0].dn), ("CN=ldaptestcomputer3,CN=Computers," + self.base_dn));
-        self.assertEquals(res[0]["cn"][0], "ldaptestcomputer3");
-        self.assertEquals(res[0]["name"][0], "ldaptestcomputer3");
-        self.assertEquals(res[0]["objectClass"][0], "top");
-        self.assertEquals(res[0]["objectClass"][1], "person");
-        self.assertEquals(res[0]["objectClass"][2], "organizationalPerson");
-        self.assertEquals(res[0]["objectClass"][3], "user");
-        self.assertEquals(res[0]["objectClass"][4], "computer");
+        self.assertEquals(str(res[0].dn), ("CN=ldaptestcomputer3,CN=Computers," + self.base_dn))
+        self.assertEquals(res[0]["cn"][0], "ldaptestcomputer3")
+        self.assertEquals(res[0]["name"][0], "ldaptestcomputer3")
+        self.assertEquals(res[0]["objectClass"][0], "top")
+        self.assertEquals(res[0]["objectClass"][1], "person")
+        self.assertEquals(res[0]["objectClass"][2], "organizationalPerson")
+        self.assertEquals(res[0]["objectClass"][3], "user")
+        self.assertEquals(res[0]["objectClass"][4], "computer")
         self.assertTrue("objectGUID" in res[0])
         self.assertTrue("whenCreated" in res[0])
-        self.assertEquals(res[0]["objectCategory"][0], ("CN=Computer,CN=Schema,CN=Configuration," + self.base_dn));
-        self.assertEquals(int(res[0]["primaryGroupID"][0]), 513);
-        self.assertEquals(int(res[0]["sAMAccountType"][0]), ATYPE_NORMAL_ACCOUNT);
-        self.assertEquals(int(res[0]["userAccountControl"][0]), UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_ACCOUNTDISABLE);
+        self.assertEquals(res[0]["objectCategory"][0], ("CN=Computer,%s" % ldb.get_schema_basedn()))
+        self.assertEquals(int(res[0]["primaryGroupID"][0]), 513)
+        self.assertEquals(int(res[0]["sAMAccountType"][0]), ATYPE_NORMAL_ACCOUNT)
+        self.assertEquals(int(res[0]["userAccountControl"][0]), UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_ACCOUNTDISABLE)
 
         delete_force(self.ldb, "cn=ldaptestcomputer3,cn=computers," + self.base_dn)
 
-        print "Testing attribute or value exists behaviour"
+        # Testing attribute or value exists behaviour
         try:
             ldb.modify_ldif("""
@@ -1614 +1795 @@
             self.assertEquals(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)
 
-        print "Testing ranged results"
+        # Testing ranged results
         ldb.modify_ldif("""
 dn: cn=ldaptest2computer,cn=computers,""" + self.base_dn + """
@@ -1660 +1841 @@
                          attrs=["servicePrincipalName;range=0-*"])
         self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
-        #print len(res[0]["servicePrincipalName;range=0-*"])
         self.assertEquals(len(res[0]["servicePrincipalName;range=0-*"]), 30)
 
         res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName;range=0-19"])
         self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
-            # print res[0]["servicePrincipalName;range=0-19"].length
         self.assertEquals(len(res[0]["servicePrincipalName;range=0-19"]), 20)
 
@@ -1690 +1869 @@
         self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
         self.assertEquals(len(res[0]["servicePrincipalName;range=11-*"]), 19)
-            # print res[0]["servicePrincipalName;range=11-*"][18]
-            # print pos_11
             # self.assertEquals((res[0]["servicePrincipalName;range=11-*"][18]), pos_11)
 
@@ -1701 +1878 @@
         res = ldb.search(self.base_dn, expression="(cn=ldaptest2computer))", scope=SCOPE_SUBTREE, attrs=["servicePrincipalName"])
         self.assertEquals(len(res), 1, "Could not find (cn=ldaptest2computer)")
-            # print res[0]["servicePrincipalName"][18]
-            # print pos_11
         self.assertEquals(len(res[0]["servicePrincipalName"]), 30)
             # self.assertEquals(res[0]["servicePrincipalName"][18], pos_11)
@@ -1714 +1889 @@
             "sn": "ldap user2"})
 
-        print "Testing Ambigious Name Resolution"
+        # Testing Ambigious Name Resolution
         # Testing ldb.search for (&(anr=ldap testy)(objectClass=user))
         res = ldb.search(expression="(&(anr=ldap testy)(objectClass=user))")
@@ -1796 +1971 @@
 #        self.assertEquals(len(res), 0, "Found (&(anr==\"testy ldap\")(objectClass=user))")
 
-        print "Testing Renames"
+        # Testing Renames
 
         attrs = ["objectGUID", "objectSid"]
-        print "Testing ldb.search for (&(cn=ldaptestUSer2)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptestUSer2)(objectClass=user))
         res_user = ldb.search(self.base_dn, expression="(&(cn=ldaptestUSer2)(objectClass=user))", scope=SCOPE_SUBTREE, attrs=attrs)
         self.assertEquals(len(res_user), 1, "Could not find (&(cn=ldaptestUSer2)(objectClass=user))")
@@ -1806 +1981 @@
         ldb.rename("<SID=" + ldb.schema_format_value("objectSID", res_user[0]["objectSID"][0]) + ">" , "cn=ldaptestUSER3,cn=users," + self.base_dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestuser3)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptestuser3)(objectClass=user))
         res = ldb.search(expression="(&(cn=ldaptestuser3)(objectClass=user))")
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestuser3)(objectClass=user))")
@@ -1834 +2009 @@
         self.assertEquals(len(res), 0, "(&(&(cn=ldaptestuser3)(userAccountControl=547))(objectClass=user))")
 
-        # This is a Samba special, and does not exist in real AD
-        #    print "Testing ldb.search for (dn=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")"
-        #    res = ldb.search("(dn=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")")
-        #    if (res.error != 0 || len(res) != 1) {
-        #        print "Could not find (dn=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")"
-        #        self.assertEquals(len(res), 1)
-        #    }
-        #    self.assertEquals(res[0].dn, ("CN=ldaptestUSER3,CN=Users," + self.base_dn))
-        #    self.assertEquals(res[0].cn, "ldaptestUSER3")
-        #    self.assertEquals(res[0].name, "ldaptestUSER3")
-
-        print "Testing ldb.search for (distinguishedName=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")"
+        # Testing ldb.search for (dn=CN=ldaptestUSER3,CN=Users," + self.base_dn + ") - should not work
+        res = ldb.search(expression="(dn=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")")
+        self.assertEquals(len(res), 0, "Could find (dn=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")")
+
+        # Testing ldb.search for (distinguishedName=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")
         res = ldb.search(expression="(distinguishedName=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")")
-        self.assertEquals(len(res), 1, "Could not find (dn=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")")
+        self.assertEquals(len(res), 1, "Could not find (distinguishedName=CN=ldaptestUSER3,CN=Users," + self.base_dn + ")")
         self.assertEquals(str(res[0].dn), ("CN=ldaptestUSER3,CN=Users," + self.base_dn))
         self.assertEquals(str(res[0]["cn"]), "ldaptestUSER3")
@@ -1884 +2052 @@
             self.assertEquals(num, ERR_ENTRY_ALREADY_EXISTS)
         try:
-            ldb.rename("cn=ldaptestuser3,cn=users," + self.base_dn, "cn=ldaptestuser3,cn=configuration," + self.base_dn)
+            ldb.rename("cn=ldaptestuser3,cn=users,%s" % self.base_dn, "cn=ldaptestuser3,%s" % ldb.get_config_basedn())
             self.fail()
         except LdbError, (num, _):
@@ -1897 +2065 @@
         ldb.rename("cn=ldaptestgroup,cn=users," + self.base_dn, "cn=ldaptestgroup2,cn=users," + self.base_dn)
 
-        print "Testing subtree renames"
+        # Testing subtree renames
 
         ldb.add({"dn": "cn=ldaptestcontainer," + self.base_dn,
@@ -1924 +2092 @@
 """)
 
-        print "Testing ldb.rename of cn=ldaptestcontainer," + self.base_dn + " to cn=ldaptestcontainer2," + self.base_dn
+        # Testing ldb.rename of cn=ldaptestcontainer," + self.base_dn + " to cn=ldaptestcontainer2," + self.base_dn
         ldb.rename("CN=ldaptestcontainer," + self.base_dn, "CN=ldaptestcontainer2," + self.base_dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestuser4)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptestuser4)(objectClass=user))
         res = ldb.search(expression="(&(cn=ldaptestuser4)(objectClass=user))")
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestuser4)(objectClass=user))")
 
-        print "Testing subtree ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in (just renamed from) cn=ldaptestcontainer," + self.base_dn
+        # Testing subtree ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in (just renamed from) cn=ldaptestcontainer," + self.base_dn
         try:
             res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
@@ -1940 +2108 @@
             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 
-        print "Testing one-level ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in (just renamed from) cn=ldaptestcontainer," + self.base_dn
+        # Testing one-level ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in (just renamed from) cn=ldaptestcontainer," + self.base_dn
         try:
             res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
@@ -1948 +2116 @@
             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 
-        print "Testing ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in renamed container"
+        # Testing ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in renamed container"
         res = ldb.search("cn=ldaptestcontainer2," + self.base_dn, expression="(&(cn=ldaptestuser4)(objectClass=user))", scope=SCOPE_SUBTREE)
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestuser4)(objectClass=user)) under cn=ldaptestcontainer2," + self.base_dn)
@@ -1957 +2125 @@
         time.sleep(4)
 
-        print "Testing ldb.search for (&(member=CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn + ")(objectclass=group)) to check subtree renames and linked attributes"
+        # Testing ldb.search for (&(member=CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn + ")(objectclass=group)) to check subtree renames and linked attributes"
         res = ldb.search(self.base_dn, expression="(&(member=CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn + ")(objectclass=group))", scope=SCOPE_SUBTREE)
         self.assertEquals(len(res), 1, "Could not find (&(member=CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn + ")(objectclass=group)), perhaps linked attributes are not consistant with subtree renames?")
 
-        print "Testing ldb.rename (into itself) of cn=ldaptestcontainer2," + self.base_dn + " to cn=ldaptestcontainer,cn=ldaptestcontainer2," + self.base_dn
+        # Testing ldb.rename (into itself) of cn=ldaptestcontainer2," + self.base_dn + " to cn=ldaptestcontainer,cn=ldaptestcontainer2," + self.base_dn
         try:
             ldb.rename("cn=ldaptestcontainer2," + self.base_dn, "cn=ldaptestcontainer,cn=ldaptestcontainer2," + self.base_dn)
@@ -1968 +2136 @@
             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
-        print "Testing ldb.rename (into non-existent container) of cn=ldaptestcontainer2," + self.base_dn + " to cn=ldaptestcontainer,cn=ldaptestcontainer3," + self.base_dn
+        # Testing ldb.rename (into non-existent container) of cn=ldaptestcontainer2," + self.base_dn + " to cn=ldaptestcontainer,cn=ldaptestcontainer3," + self.base_dn
         try:
             ldb.rename("cn=ldaptestcontainer2," + self.base_dn, "cn=ldaptestcontainer,cn=ldaptestcontainer3," + self.base_dn)
@@ -1975 +2143 @@
             self.assertTrue(num in (ERR_UNWILLING_TO_PERFORM, ERR_OTHER))
 
-        print "Testing delete (should fail, not a leaf node) of renamed cn=ldaptestcontainer2," + self.base_dn
+        # Testing delete (should fail, not a leaf node) of renamed cn=ldaptestcontainer2," + self.base_dn
         try:
             ldb.delete("cn=ldaptestcontainer2," + self.base_dn)
@@ -1982 +2150 @@
             self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
 
-        print "Testing base ldb.search for CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn
+        # Testing base ldb.search for CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn
         res = ldb.search(expression="(objectclass=*)", base=("CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn), scope=SCOPE_BASE)
         self.assertEquals(len(res), 1)
@@ -1988 +2156 @@
         self.assertEquals(len(res), 0)
 
-        print "Testing one-level ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in cn=ldaptestcontainer2," + self.base_dn
+        # Testing one-level ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in cn=ldaptestcontainer2," + self.base_dn
         res = ldb.search(expression="(&(cn=ldaptestuser4)(objectClass=user))", base=("cn=ldaptestcontainer2," + self.base_dn), scope=SCOPE_ONELEVEL)
-        # FIXME: self.assertEquals(len(res), 0)
-
-        print "Testing one-level ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in cn=ldaptestcontainer2," + self.base_dn
+        self.assertEquals(len(res), 1)
+
+        # Testing one-level ldb.search for (&(cn=ldaptestuser4)(objectClass=user)) in cn=ldaptestcontainer2," + self.base_dn
         res = ldb.search(expression="(&(cn=ldaptestuser4)(objectClass=user))", base=("cn=ldaptestcontainer2," + self.base_dn), scope=SCOPE_SUBTREE)
-        # FIXME: self.assertEquals(len(res), 0)
-
-        print "Testing delete of subtree renamed "+("CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn)
+        self.assertEquals(len(res), 1)
+
+        # Testing delete of subtree renamed "+("CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn)
         ldb.delete(("CN=ldaptestuser4,CN=ldaptestcontainer2," + self.base_dn))
-        print "Testing delete of renamed cn=ldaptestcontainer2," + self.base_dn
+        # Testing delete of renamed cn=ldaptestcontainer2," + self.base_dn
         ldb.delete("cn=ldaptestcontainer2," + self.base_dn)
 
@@ -2005 +2173 @@
         ldb.add({"dn": "cn=ldaptestutf8user2  Úùéìòà ,cn=users," + self.base_dn, "objectClass": "user"})
 
-        print "Testing ldb.search for (&(cn=ldaptestuser)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptestuser)(objectClass=user))"
         res = ldb.search(expression="(&(cn=ldaptestuser)(objectClass=user))")
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestuser)(objectClass=user))")
@@ -2015 +2183 @@
         self.assertTrue("objectGUID" in res[0])
         self.assertTrue("whenCreated" in res[0])
-        self.assertEquals(str(res[0]["objectCategory"]), ("CN=Person,CN=Schema,CN=Configuration," + self.base_dn))
+        self.assertEquals(str(res[0]["objectCategory"]), ("CN=Person,%s" % ldb.get_schema_basedn()))
         self.assertEquals(int(res[0]["sAMAccountType"][0]), ATYPE_NORMAL_ACCOUNT)
         self.assertEquals(int(res[0]["userAccountControl"][0]), UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_ACCOUNTDISABLE)
@@ -2021 +2189 @@
         self.assertEquals(len(res[0]["memberOf"]), 1)
 
-        print "Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=cn=person,cn=schema,cn=configuration," + self.base_dn + "))"
-        res2 = ldb.search(expression="(&(cn=ldaptestuser)(objectCategory=cn=person,cn=schema,cn=configuration," + self.base_dn + "))")
-        self.assertEquals(len(res2), 1, "Could not find (&(cn=ldaptestuser)(objectCategory=cn=person,cn=schema,cn=configuration," + self.base_dn + "))")
+        # Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=cn=person,%s))" % ldb.get_schema_basedn()
+        res2 = ldb.search(expression="(&(cn=ldaptestuser)(objectCategory=cn=person,%s))" % ldb.get_schema_basedn())
+        self.assertEquals(len(res2), 1, "Could not find (&(cn=ldaptestuser)(objectCategory=cn=person,%s))" % ldb.get_schema_basedn())
 
         self.assertEquals(res[0].dn, res2[0].dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=PerSon))"
+        # Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=PerSon))"
         res3 = ldb.search(expression="(&(cn=ldaptestuser)(objectCategory=PerSon))")
         self.assertEquals(len(res3), 1, "Could not find (&(cn=ldaptestuser)(objectCategory=PerSon)): matched %d" % len(res3))
@@ -2034 +2202 @@
 
         if gc_ldb is not None:
-            print "Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=PerSon)) in Global Catalog"
+            # Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=PerSon)) in Global Catalog"
             res3gc = gc_ldb.search(expression="(&(cn=ldaptestuser)(objectCategory=PerSon))")
             self.assertEquals(len(res3gc), 1)
@@ -2040 +2208 @@
             self.assertEquals(res[0].dn, res3gc[0].dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=PerSon)) in with 'phantom root' control"
+        # Testing ldb.search for (&(cn=ldaptestuser)(objectCategory=PerSon)) in with 'phantom root' control"
 
         if gc_ldb is not None:
@@ -2050 +2218 @@
         ldb.delete(res[0].dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestcomputer)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptestcomputer)(objectClass=user))"
         res = ldb.search(expression="(&(cn=ldaptestcomputer)(objectClass=user))")
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestuser)(objectClass=user))")
@@ -2060 +2228 @@
         self.assertTrue("objectGUID" in res[0])
         self.assertTrue("whenCreated" in res[0])
-        self.assertEquals(str(res[0]["objectCategory"]), ("CN=Computer,CN=Schema,CN=Configuration," + self.base_dn))
+        self.assertEquals(str(res[0]["objectCategory"]), ("CN=Computer,%s" % ldb.get_schema_basedn()))
         self.assertEquals(int(res[0]["primaryGroupID"][0]), 513)
         self.assertEquals(int(res[0]["sAMAccountType"][0]), ATYPE_NORMAL_ACCOUNT)
@@ -2067 +2235 @@
         self.assertEquals(len(res[0]["memberOf"]), 1)
 
-        print "Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=cn=computer,cn=schema,cn=configuration," + self.base_dn + "))"
-        res2 = ldb.search(expression="(&(cn=ldaptestcomputer)(objectCategory=cn=computer,cn=schema,cn=configuration," + self.base_dn + "))")
-        self.assertEquals(len(res2), 1, "Could not find (&(cn=ldaptestcomputer)(objectCategory=cn=computer,cn=schema,cn=configuration," + self.base_dn + "))")
+        # Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=cn=computer,%s))" % ldb.get_schema_basedn()
+        res2 = ldb.search(expression="(&(cn=ldaptestcomputer)(objectCategory=cn=computer,%s))" % ldb.get_schema_basedn())
+        self.assertEquals(len(res2), 1, "Could not find (&(cn=ldaptestcomputer)(objectCategory=cn=computer,%s))" % ldb.get_schema_basedn())
 
         self.assertEquals(res[0].dn, res2[0].dn)
 
         if gc_ldb is not None:
-            print "Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=cn=computer,cn=schema,cn=configuration," + self.base_dn + ")) in Global Catlog"
-            res2gc = gc_ldb.search(expression="(&(cn=ldaptestcomputer)(objectCategory=cn=computer,cn=schema,cn=configuration," + self.base_dn + "))")
-            self.assertEquals(len(res2gc), 1, "Could not find (&(cn=ldaptestcomputer)(objectCategory=cn=computer,cn=schema,cn=configuration," + self.base_dn + ")) in Global Catlog")
+            # Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=cn=computer,%s)) in Global Catalog" % gc_ldb.get_schema_basedn()
+            res2gc = gc_ldb.search(expression="(&(cn=ldaptestcomputer)(objectCategory=cn=computer,%s))" % gc_ldb.get_schema_basedn())
+            self.assertEquals(len(res2gc), 1, "Could not find (&(cn=ldaptestcomputer)(objectCategory=cn=computer,%s)) In Global Catalog" % gc_ldb.get_schema_basedn())
 
             self.assertEquals(res[0].dn, res2gc[0].dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=compuTER))"
+        # Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=compuTER))"
         res3 = ldb.search(expression="(&(cn=ldaptestcomputer)(objectCategory=compuTER))")
         self.assertEquals(len(res3), 1, "Could not find (&(cn=ldaptestcomputer)(objectCategory=compuTER))")
@@ -2087 +2255 @@
 
         if gc_ldb is not None:
-            print "Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=compuTER)) in Global Catalog"
+            # Testing ldb.search for (&(cn=ldaptestcomputer)(objectCategory=compuTER)) in Global Catalog"
             res3gc = gc_ldb.search(expression="(&(cn=ldaptestcomputer)(objectCategory=compuTER))")
             self.assertEquals(len(res3gc), 1, "Could not find (&(cn=ldaptestcomputer)(objectCategory=compuTER)) in Global Catalog")
@@ -2093 +2261 @@
             self.assertEquals(res[0].dn, res3gc[0].dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestcomp*r)(objectCategory=compuTER))"
+        # Testing ldb.search for (&(cn=ldaptestcomp*r)(objectCategory=compuTER))"
         res4 = ldb.search(expression="(&(cn=ldaptestcomp*r)(objectCategory=compuTER))")
         self.assertEquals(len(res4), 1, "Could not find (&(cn=ldaptestcomp*r)(objectCategory=compuTER))")
@@ -2099 +2267 @@
         self.assertEquals(res[0].dn, res4[0].dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestcomput*)(objectCategory=compuTER))"
+        # Testing ldb.search for (&(cn=ldaptestcomput*)(objectCategory=compuTER))"
         res5 = ldb.search(expression="(&(cn=ldaptestcomput*)(objectCategory=compuTER))")
         self.assertEquals(len(res5), 1, "Could not find (&(cn=ldaptestcomput*)(objectCategory=compuTER))")
@@ -2105 +2273 @@
         self.assertEquals(res[0].dn, res5[0].dn)
 
-        print "Testing ldb.search for (&(cn=*daptestcomputer)(objectCategory=compuTER))"
+        # Testing ldb.search for (&(cn=*daptestcomputer)(objectCategory=compuTER))"
         res6 = ldb.search(expression="(&(cn=*daptestcomputer)(objectCategory=compuTER))")
         self.assertEquals(len(res6), 1, "Could not find (&(cn=*daptestcomputer)(objectCategory=compuTER))")
@@ -2113 +2281 @@
         ldb.delete("<GUID=" + ldb.schema_format_value("objectGUID", res[0]["objectGUID"][0]) + ">")
 
-        print "Testing ldb.search for (&(cn=ldaptest2computer)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptest2computer)(objectClass=user))"
         res = ldb.search(expression="(&(cn=ldaptest2computer)(objectClass=user))")
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptest2computer)(objectClass=user))")
@@ -2123 +2291 @@
         self.assertTrue("objectGUID" in res[0])
         self.assertTrue("whenCreated" in res[0])
-        self.assertEquals(res[0]["objectCategory"][0], "CN=Computer,CN=Schema,CN=Configuration," + self.base_dn)
+        self.assertEquals(res[0]["objectCategory"][0], "CN=Computer,%s" % ldb.get_schema_basedn())
         self.assertEquals(int(res[0]["sAMAccountType"][0]), ATYPE_WORKSTATION_TRUST)
         self.assertEquals(int(res[0]["userAccountControl"][0]), UF_WORKSTATION_TRUST_ACCOUNT)
@@ -2130 +2298 @@
 
         attrs = ["cn", "name", "objectClass", "objectGUID", "objectSID", "whenCreated", "nTSecurityDescriptor", "memberOf", "allowedAttributes", "allowedAttributesEffective"]
-        print "Testing ldb.search for (&(cn=ldaptestUSer2)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptestUSer2)(objectClass=user))"
         res_user = ldb.search(self.base_dn, expression="(&(cn=ldaptestUSer2)(objectClass=user))", scope=SCOPE_SUBTREE, attrs=attrs)
         self.assertEquals(len(res_user), 1, "Could not find (&(cn=ldaptestUSer2)(objectClass=user))")
@@ -2150 +2318 @@
 
         attrs = ["cn", "name", "objectClass", "objectGUID", "objectSID", "whenCreated", "nTSecurityDescriptor", "member", "allowedAttributes", "allowedAttributesEffective"]
-        print "Testing ldb.search for (&(cn=ldaptestgroup2)(objectClass=group))"
+        # Testing ldb.search for (&(cn=ldaptestgroup2)(objectClass=group))"
         res = ldb.search(self.base_dn, expression="(&(cn=ldaptestgroup2)(objectClass=group))", scope=SCOPE_SUBTREE, attrs=attrs)
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestgroup2)(objectClass=group))")
@@ -2180 +2348 @@
         self.assertTrue(("<GUID=" + ldb.schema_format_value("objectGUID", ldaptestuser2_guid) + ">;<SID=" + ldb.schema_format_value("objectSid", ldaptestuser2_sid) + ">;CN=ldaptestuser2,CN=Users," + self.base_dn).upper() in memberUP)
 
-        print "Quicktest for linked attributes"
+        # Quicktest for linked attributes"
         ldb.modify_ldif("""
 dn: cn=ldaptestgroup2,cn=users,""" + self.base_dn + """
@@ -2243 +2411 @@
 
         attrs = ["cn", "name", "objectClass", "objectGUID", "whenCreated", "nTSecurityDescriptor", "member"]
-        print "Testing ldb.search for (&(cn=ldaptestgroup2)(objectClass=group)) to check linked delete"
+        # Testing ldb.search for (&(cn=ldaptestgroup2)(objectClass=group)) to check linked delete"
         res = ldb.search(self.base_dn, expression="(&(cn=ldaptestgroup2)(objectClass=group))", scope=SCOPE_SUBTREE, attrs=attrs)
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestgroup2)(objectClass=group)) to check linked delete")
@@ -2250 +2418 @@
         self.assertTrue("member" not in res[0])
 
-        print "Testing ldb.search for (&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))"
-# TODO UTF8 users don't seem to work fully anymore
-#        res = ldb.search(expression="(&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))")
+        # Testing ldb.search for (&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))"
+        res = ldb.search(expression="(&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))")
+        self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))")
         res = ldb.search(expression="(&(cn=ldaptestutf8user Úùéìòà )(objectclass=user))")
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))")
@@ -2263 +2431 @@
         self.assertTrue("whenCreated" in res[0])
 
+        # delete "ldaptestutf8user"
         ldb.delete(res[0].dn)
 
-        print "Testing ldb.search for (&(cn=ldaptestutf8user2*)(objectClass=user))"
+        # Testing ldb.search for (&(cn=ldaptestutf8user2*)(objectClass=user))"
         res = ldb.search(expression="(&(cn=ldaptestutf8user2*)(objectClass=user))")
         self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestutf8user2*)(objectClass=user))")
 
+        # Testing ldb.search for (&(cn=ldaptestutf8user2  ÈÙÉÌÒÀ)(objectClass=user))"
+        res = ldb.search(expression="(&(cn=ldaptestutf8user2  ÈÙÉÌÒÀ)(objectClass=user))")
+        self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestutf8user2  ÈÙÉÌÒÀ)(objectClass=user))")
+
+        # delete "ldaptestutf8user2 "
         ldb.delete(res[0].dn)
 
         ldb.delete(("CN=ldaptestgroup2,CN=Users," + self.base_dn))
 
-        print "Testing ldb.search for (&(cn=ldaptestutf8user2 ÈÙÉÌÒÀ)(objectClass=user))"
-# TODO UTF8 users don't seem to work fully anymore
-#        res = ldb.search(expression="(&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))")
-#        self.assertEquals(len(res), 1, "Could not find (&(cn=ldaptestutf8user ÈÙÉÌÒÀ)(objectClass=user))")
-
-        print "Testing that we can't get at the configuration DN from the main search base"
+        # Testing that we can't get at the configuration DN from the main search base"
         res = ldb.search(self.base_dn, expression="objectClass=crossRef", scope=SCOPE_SUBTREE, attrs=["cn"])
         self.assertEquals(len(res), 0)
 
-        print "Testing that we can get at the configuration DN from the main search base on the LDAP port with the 'phantom root' search_options control"
+        # Testing that we can get at the configuration DN from the main search base on the LDAP port with the 'phantom root' search_options control"
         res = ldb.search(self.base_dn, expression="objectClass=crossRef", scope=SCOPE_SUBTREE, attrs=["cn"], controls=["search_options:1:2"])
         self.assertTrue(len(res) > 0)
 
         if gc_ldb is not None:
-            print "Testing that we can get at the configuration DN from the main search base on the GC port with the search_options control == 0"
+            # Testing that we can get at the configuration DN from the main search base on the GC port with the search_options control == 0"
 
             res = gc_ldb.search(self.base_dn, expression="objectClass=crossRef", scope=SCOPE_SUBTREE, attrs=["cn"], controls=["search_options:1:0"])
             self.assertTrue(len(res) > 0)
 
-            print "Testing that we do find configuration elements in the global catlog"
+            # Testing that we do find configuration elements in the global catlog"
             res = gc_ldb.search(self.base_dn, expression="objectClass=crossRef", scope=SCOPE_SUBTREE, attrs=["cn"])
             self.assertTrue(len(res) > 0)
 
-            print "Testing that we do find configuration elements and user elements at the same time"
+            # Testing that we do find configuration elements and user elements at the same time"
             res = gc_ldb.search(self.base_dn, expression="(|(objectClass=crossRef)(objectClass=person))", scope=SCOPE_SUBTREE, attrs=["cn"])
             self.assertTrue(len(res) > 0)
 
-            print "Testing that we do find configuration elements in the global catlog, with the configuration basedn"
+            # Testing that we do find configuration elements in the global catlog, with the configuration basedn"
             res = gc_ldb.search(self.configuration_dn, expression="objectClass=crossRef", scope=SCOPE_SUBTREE, attrs=["cn"])
             self.assertTrue(len(res) > 0)
 
-        print "Testing that we can get at the configuration DN on the main LDAP port"
+        # Testing that we can get at the configuration DN on the main LDAP port"
         res = ldb.search(self.configuration_dn, expression="objectClass=crossRef", scope=SCOPE_SUBTREE, attrs=["cn"])
         self.assertTrue(len(res) > 0)
 
-        print "Testing objectCategory canonacolisation"
+        # Testing objectCategory canonacolisation"
         res = ldb.search(self.configuration_dn, expression="objectCategory=ntDsDSA", scope=SCOPE_SUBTREE, attrs=["cn"])
         self.assertTrue(len(res) > 0, "Didn't find any records with objectCategory=ntDsDSA")
@@ -2317 +2486 @@
         self.assertTrue(len(res) != 0)
 
-        print "Testing objectClass attribute order on "+ self.base_dn
+        # Testing objectClass attribute order on "+ self.base_dn
         res = ldb.search(expression="objectClass=domain", base=self.base_dn,
                          scope=SCOPE_BASE, attrs=["objectClass"])
@@ -2326 +2495 @@
     #  check enumeration
 
-        print "Testing ldb.search for objectCategory=person"
+        # Testing ldb.search for objectCategory=person"
         res = ldb.search(self.base_dn, expression="objectCategory=person", scope=SCOPE_SUBTREE, attrs=["cn"])
         self.assertTrue(len(res) > 0)
 
-        print "Testing ldb.search for objectCategory=person with domain scope control"
+        # Testing ldb.search for objectCategory=person with domain scope control"
         res = ldb.search(self.base_dn, expression="objectCategory=person", scope=SCOPE_SUBTREE, attrs=["cn"], controls=["domain_scope:1"])
         self.assertTrue(len(res) > 0)
 
-        print "Testing ldb.search for objectCategory=user"
+        # Testing ldb.search for objectCategory=user"
         res = ldb.search(self.base_dn, expression="objectCategory=user", scope=SCOPE_SUBTREE, attrs=["cn"])
         self.assertTrue(len(res) > 0)
 
-        print "Testing ldb.search for objectCategory=user with domain scope control"
+        # Testing ldb.search for objectCategory=user with domain scope control"
         res = ldb.search(self.base_dn, expression="objectCategory=user", scope=SCOPE_SUBTREE, attrs=["cn"], controls=["domain_scope:1"])
         self.assertTrue(len(res) > 0)
 
-        print "Testing ldb.search for objectCategory=group"
+        # Testing ldb.search for objectCategory=group"
         res = ldb.search(self.base_dn, expression="objectCategory=group", scope=SCOPE_SUBTREE, attrs=["cn"])
         self.assertTrue(len(res) > 0)
 
-        print "Testing ldb.search for objectCategory=group with domain scope control"
+        # Testing ldb.search for objectCategory=group with domain scope control"
         res = ldb.search(self.base_dn, expression="objectCategory=group", scope=SCOPE_SUBTREE, attrs=["cn"], controls=["domain_scope:1"])
         self.assertTrue(len(res) > 0)
 
-        print "Testing creating a user with the posixAccount objectClass"
+        # Testing creating a user with the posixAccount objectClass"
         self.ldb.add_ldif("""dn: cn=posixuser,CN=Users,%s
 objectClass: top
@@ -2367 +2536 @@
 description: A POSIX user"""% (self.base_dn))
 
-        print "Testing removing the posixAccount objectClass from an existing user"
+        # Testing removing the posixAccount objectClass from an existing user"
         self.ldb.modify_ldif("""dn: cn=posixuser,CN=Users,%s
 changetype: modify
@@ -2373 +2542 @@
 objectClass: posixAccount"""% (self.base_dn))
 
-        print "Testing adding the posixAccount objectClass to an existing user"
+        # Testing adding the posixAccount objectClass to an existing user"
         self.ldb.modify_ldif("""dn: cn=posixuser,CN=Users,%s
 changetype: modify
@@ -2460 +2629 @@
         delete_force(self.ldb, user_dn)
         try:
-            sddl = "O:DUG:DUD:PAI(A;;RPWP;;;AU)S:PAI"
+            sddl = "O:DUG:DUD:AI(A;;RPWP;;;AU)S:PAI"
             desc = security.descriptor.from_sddl(sddl, security.dom_sid('S-1-5-21'))
             desc_base64 = base64.b64encode( ndr_pack(desc) )
@@ -2470 +2639 @@
             res = self.ldb.search(base=user_dn, attrs=["nTSecurityDescriptor"])
             self.assertTrue("nTSecurityDescriptor" in res[0])
+            desc = res[0]["nTSecurityDescriptor"][0]
+            desc = ndr_unpack(security.descriptor, desc)
+            desc_sddl = desc.as_sddl(self.domain_sid)
+            self.assertTrue("O:S-1-5-21-513G:S-1-5-21-513D:AI(A;;RPWP;;;AU)" in desc_sddl)
         finally:
             delete_force(self.ldb, user_dn)
@@ -2632 +2805 @@
     def test_dsheuristics(self):
         """Tests the 'dSHeuristics' attribute"""
-        print "Tests the 'dSHeuristics' attribute"""
+        # Tests the 'dSHeuristics' attribute"
 
         # Get the current value to restore it later
         dsheuristics = self.ldb.get_dsheuristics()
-        # Should not be longer than 18 chars?
-        try:
-            self.ldb.set_dsheuristics("123ABC-+!1asdfg@#^12")
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-        # If it is >= 10 chars, tenthChar should be 1
-        try:
-            self.ldb.set_dsheuristics("00020000000002")
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-        # apart from the above, all char values are accepted
-        self.ldb.set_dsheuristics("123ABC-+!1asdfg@#^")
-        self.assertEquals(self.ldb.get_dsheuristics(), "123ABC-+!1asdfg@#^")
-        # restore old value
-        self.ldb.set_dsheuristics(dsheuristics)
+        # Perform the length checks: for each decade (except the 0th) we need
+        # the first index to be the number. This goes till the 9th one, beyond
+        # there does not seem to be another limitation.
+        try:
+            dshstr = ""
+            for i in range(1,11):
+                # This is in the range
+                self.ldb.set_dsheuristics(dshstr + "x")
+                self.ldb.set_dsheuristics(dshstr + "xxxxx")
+                dshstr = dshstr + "xxxxxxxxx"
+                if i < 10:
+                    # Not anymore in the range, new decade specifier needed
+                    try:
+                        self.ldb.set_dsheuristics(dshstr + "x")
+                        self.fail()
+                    except LdbError, (num, _):
+                        self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+                    dshstr = dshstr + str(i)
+                else:
+                    # There does not seem to be an upper limit
+                    self.ldb.set_dsheuristics(dshstr + "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")
+            # apart from the above, all char values are accepted
+            self.ldb.set_dsheuristics("123ABC-+!1asdfg@#^")
+            self.assertEquals(self.ldb.get_dsheuristics(), "123ABC-+!1asdfg@#^")
+        finally:
+            # restore old value
+            self.ldb.set_dsheuristics(dsheuristics)
 
     def test_ldapControlReturn(self):
@@ -2656 +2841 @@
            really return something"""
         res = self.ldb.search(attrs=["cn"],
-                              controls=["paged_result:1:10"])
+                              controls=["paged_results:1:10"])
         self.assertEquals(len(res.controls), 1)
         self.assertEquals(res.controls[0].oid, "1.2.840.113556.1.4.319")
+        s = str(res.controls[0])
 
     def test_operational(self):
         """Tests operational attributes"""
-        print "Tests operational attributes"""
+        # Tests operational attributes"
 
         res = self.ldb.search(self.base_dn, scope=SCOPE_BASE,
@@ -2675 +2861 @@
         self.assertTrue("whenChanged" in res[0])
 
-class BaseDnTests(unittest.TestCase):
+    def test_timevalues1(self):
+        """Tests possible syntax of time attributes"""
+
+        user_name = "testtimevaluesuser1"
+        user_dn = "CN=%s,CN=Users,%s" % (user_name, self.base_dn)
+
+        delete_force(self.ldb, user_dn)
+        self.ldb.add({ "dn": user_dn,
+                       "objectClass": "user",
+                       "sAMAccountName": user_name })
+
+        #
+        # We check the following values:
+        #
+        #   370101000000Z     => 20370101000000.0Z
+        # 20370102000000.*Z   => 20370102000000.0Z
+        #
+        ext = [ "Z", ".0Z", ".Z", ".000Z", ".RandomIgnoredCharacters...987654321Z" ]
+        for i in range(0, len(ext)):
+            v_raw = "203701%02d000000" % (i + 1)
+            if ext[i] == "Z":
+                v_set = v_raw[2:] + ext[i]
+            else:
+                v_set = v_raw + ext[i]
+            v_get = v_raw + ".0Z"
+
+            m = Message()
+            m.dn = Dn(ldb, user_dn)
+            m["msTSExpireDate"] = MessageElement([v_set],
+                                                 FLAG_MOD_REPLACE,
+                                                 "msTSExpireDate")
+            self.ldb.modify(m)
+
+            res = self.ldb.search(base=user_dn, scope=SCOPE_BASE, attrs=["msTSExpireDate"])
+            self.assertTrue(len(res) == 1)
+            self.assertTrue("msTSExpireDate" in res[0])
+            self.assertTrue(len(res[0]["msTSExpireDate"]) == 1)
+            self.assertEquals(res[0]["msTSExpireDate"][0], v_get)
+
+class BaseDnTests(samba.tests.TestCase):
 
     def setUp(self):
@@ -2758 +2983 @@
         self.assertEquals(int(res[0]["domainFunctionality"][0]), int(res4[0]["msDS-Behavior-Version"][0]))
 
-        res5 = self.ldb.search("cn=partitions," + str(ldb.get_config_basedn()), scope=SCOPE_BASE, attrs=["msDS-Behavior-Version"])
+        res5 = self.ldb.search("cn=partitions,%s" % ldb.get_config_basedn(), scope=SCOPE_BASE, attrs=["msDS-Behavior-Version"])
         self.assertEquals(len(res5), 1)
         self.assertEquals(len(res5[0]["msDS-Behavior-Version"]), 1)
@@ -2778 +3003 @@
         """Testing the ldap service name in rootDSE"""
         res = self.ldb.search("", scope=SCOPE_BASE,
-                              attrs=["ldapServiceName", "dNSHostName"])
+                              attrs=["ldapServiceName", "dnsHostName"])
         self.assertEquals(len(res), 1)
-
-        (hostname, _, dns_domainname) = res[0]["dNSHostName"][0].partition(".")
-        self.assertTrue(":%s$@%s" % (hostname, dns_domainname.upper())
-                        in res[0]["ldapServiceName"][0])
+        self.assertTrue("ldapServiceName" in res[0])
+        self.assertTrue("dnsHostName" in res[0])
+
+        (hostname, _, dns_domainname) = res[0]["dnsHostName"][0].partition(".")
+
+        given = res[0]["ldapServiceName"][0]
+        expected = "%s:%s$@%s" % (dns_domainname.lower(), hostname.lower(), dns_domainname.upper())
+        self.assertEquals(given, expected)
 
 if not "://" in host:
@@ -2798 +3027 @@
     gc_ldb = None
 
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(BaseDnTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(BasicTests)).wasSuccessful():
-    rc = 1
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/ldap_schema.py

@@ -2 +2 @@
 # -*- coding: utf-8 -*-
 # This is a port of the original in testprogs/ejs/ldap.js
+
+# Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2008-2011
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
 
 import optparse
@@ -11 +28 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+from samba.tests.subunitrun import TestProgram, SubunitOptions
 
 import samba.getopt as options
@@ -26 +42 @@
 from samba.dsdb import DS_DOMAIN_FUNCTION_2003
 from samba.tests import delete_force
-
-from subunit.run import SubunitTestRunner
-import unittest
+from samba.ndr import ndr_unpack
+from samba.dcerpc import drsblobs
 
 parser = optparse.OptionParser("ldap_schema.py [options] <host>")
@@ -37 +52 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
 opts, args = parser.parse_args()
 
@@ -49 +66 @@
 
 
-class SchemaTests(unittest.TestCase):
+class SchemaTests(samba.tests.TestCase):
 
     def setUp(self):
         super(SchemaTests, self).setUp()
-        self.ldb = ldb
-        self.base_dn = ldb.domain_dn()
-        self.schema_dn = ldb.get_schema_basedn().get_linearized()
+        self.ldb = SamDB(host, credentials=creds,
+            session_info=system_session(lp), lp=lp, options=ldb_options)
+        self.base_dn = self.ldb.domain_dn()
+        self.schema_dn = self.ldb.get_schema_basedn().get_linearized()
 
     def test_generated_schema(self):
@@ -68 +86 @@
     def test_generated_schema_is_operational(self):
         """Testing we don't get the generated schema via LDAP by default"""
+        # Must keep the "*" form
         res = self.ldb.search("cn=aggregate,"+self.schema_dn, scope=SCOPE_BASE,
-                attrs=["*"])
+                              attrs=["*"])
         self.assertEquals(len(res), 1)
         self.assertFalse("dITContentRules" in res[0])
@@ -95 +114 @@
 """
         self.ldb.add_ldif(ldif)
+        # We must do a schemaUpdateNow otherwise it's not 100% sure that the schema
+        # will contain the new attribute
+        ldif = """
+dn:
+changetype: modify
+add: schemaUpdateNow
+schemaUpdateNow: 1
+"""
+        self.ldb.modify_ldif(ldif)
 
         # Search for created attribute
         res = []
-        res = self.ldb.search("cn=%s,%s" % (attr_name, self.schema_dn), scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search("cn=%s,%s" % (attr_name, self.schema_dn), scope=SCOPE_BASE,
+                              attrs=["lDAPDisplayName","schemaIDGUID", "msDS-IntID"])
         self.assertEquals(len(res), 1)
         self.assertEquals(res[0]["lDAPDisplayName"][0], attr_ldap_display_name)
         self.assertTrue("schemaIDGUID" in res[0])
+        if "msDS-IntId" in res[0]:
+            msDS_IntId = int(res[0]["msDS-IntId"][0])
+            if msDS_IntId < 0:
+                msDS_IntId += (1 << 32)
+        else:
+            msDS_IntId = None
 
         class_name = "test-Class" + time.strftime("%s", time.gmtime())
@@ -152 +187 @@
         # Search for created objectclass
         res = []
-        res = self.ldb.search("cn=%s,%s" % (class_name, self.schema_dn), scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search("cn=%s,%s" % (class_name, self.schema_dn), scope=SCOPE_BASE,
+                              attrs=["lDAPDisplayName", "defaultObjectCategory", "schemaIDGUID", "distinguishedName"])
         self.assertEquals(len(res), 1)
         self.assertEquals(res[0]["lDAPDisplayName"][0], class_ldap_display_name)
@@ -184 +220 @@
 
         # Search for created object
-        res = []
-        res = self.ldb.search("cn=%s,cn=Users,%s" % (object_name, self.base_dn), scope=SCOPE_BASE, attrs=["*"])
-        self.assertEquals(len(res), 1)
+        obj_res = self.ldb.search("cn=%s,cn=Users,%s" % (object_name, self.base_dn), scope=SCOPE_BASE, attrs=["replPropertyMetaData"])
+
+        self.assertEquals(len(obj_res), 1)
+        self.assertTrue("replPropertyMetaData" in obj_res[0])
+        val = obj_res[0]["replPropertyMetaData"][0]
+        repl = ndr_unpack(drsblobs.replPropertyMetaDataBlob, str(val))
+        obj = repl.ctr
+
+        # Windows 2000 functional level won't have this.  It is too
+        # hard to work it out from the prefixmap however, so we skip
+        # this test in that case.
+        if msDS_IntId is not None:
+            found = False
+            for o in repl.ctr.array:
+                if o.attid == msDS_IntId:
+                    found = True
+                    break
+            self.assertTrue(found, "Did not find 0x%08x in replPropertyMetaData" % msDS_IntId)
         # Delete the object
         delete_force(self.ldb, "cn=%s,cn=Users,%s" % (object_name, self.base_dn))
@@ -215 +266 @@
         # Search for created objectclass
         res = []
-        res = self.ldb.search("cn=%s,%s" % (class_name, self.schema_dn), scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search("cn=%s,%s" % (class_name, self.schema_dn), scope=SCOPE_BASE,
+                              attrs=["lDAPDisplayName", "defaultObjectCategory",
+                                     "schemaIDGUID", "distinguishedName"])
         self.assertEquals(len(res), 1)
         self.assertEquals(res[0]["lDAPDisplayName"][0], class_ldap_display_name)
@@ -241 +294 @@
         # Search for created object
         res = []
-        res = self.ldb.search("ou=%s,%s" % (object_name, self.base_dn), scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search("ou=%s,%s" % (object_name, self.base_dn), scope=SCOPE_BASE, attrs=["dn"])
         self.assertEquals(len(res), 1)
         # Delete the object
@@ -247 +300 @@
 
 
-class SchemaTests_msDS_IntId(unittest.TestCase):
+class SchemaTests_msDS_IntId(samba.tests.TestCase):
 
     def setUp(self):
         super(SchemaTests_msDS_IntId, self).setUp()
-        self.ldb = ldb
-        res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
+        self.ldb = SamDB(host, credentials=creds,
+            session_info=system_session(lp), lp=lp, options=ldb_options)
+        res = self.ldb.search(base="", expression="", scope=SCOPE_BASE,
+                         attrs=["schemaNamingContext", "defaultNamingContext",
+                                "forestFunctionality"])
         self.assertEquals(len(res), 1)
         self.schema_dn = res[0]["schemaNamingContext"][0]
@@ -331 +387 @@
         # Search for created attribute
         res = []
-        res = self.ldb.search(attr_dn, scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search(attr_dn, scope=SCOPE_BASE,
+                              attrs=["lDAPDisplayName", "msDS-IntId", "systemFlags"])
         self.assertEquals(len(res), 1)
         self.assertEquals(res[0]["lDAPDisplayName"][0], attr_ldap_name)
@@ -373 +430 @@
         # Search for created attribute
         res = []
-        res = self.ldb.search(attr_dn, scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search(attr_dn, scope=SCOPE_BASE,
+                              attrs=["lDAPDisplayName", "msDS-IntId"])
         self.assertEquals(len(res), 1)
         self.assertEquals(res[0]["lDAPDisplayName"][0], attr_ldap_name)
@@ -428 +486 @@
         self._ldap_schemaUpdateNow()
 
-        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["msDS-IntId"])
         self.assertEquals(len(res), 1)
         self.assertEquals(res[0]["msDS-IntId"][0], "-1993108831")
@@ -440 +498 @@
 
         # Search for created Class
-        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["msDS-IntId"])
         self.assertEquals(len(res), 1)
         self.assertFalse("msDS-IntId" in res[0])
@@ -465 +523 @@
         self.ldb.add_ldif(ldif_add)
 
-        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["msDS-IntId"])
         self.assertEquals(len(res), 1)
         self.assertEquals(res[0]["msDS-IntId"][0], "-1993108831")
@@ -478 +536 @@
 
         # Search for created Class
-        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["msDS-IntId"])
         self.assertEquals(len(res), 1)
         self.assertFalse("msDS-IntId" in res[0])
@@ -490 +548 @@
         except LdbError, (num, _):
             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["*"])
+        res = self.ldb.search(class_dn, scope=SCOPE_BASE, attrs=["msDS-IntId"])
         self.assertEquals(len(res), 1)
         self.assertFalse("msDS-IntId" in res[0])
@@ -519 +577 @@
 
 
-class SchemaTests_msDS_isRODC(unittest.TestCase):
+class SchemaTests_msDS_isRODC(samba.tests.TestCase):
 
     def setUp(self):
         super(SchemaTests_msDS_isRODC, self).setUp()
-        self.ldb = ldb
-        res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
+        self.ldb =  SamDB(host, credentials=creds,
+            session_info=system_session(lp), lp=lp, options=ldb_options)
+        res = self.ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["defaultNamingContext"])
         self.assertEquals(len(res), 1)
         self.base_dn = res[0]["defaultNamingContext"][0]
@@ -568 +627 @@
     ldb_options = ["modules:paged_searches"]
 
-ldb = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp, options=ldb_options)
-
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(SchemaTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(SchemaTests_msDS_IntId)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(SchemaTests_msDS_isRODC)).wasSuccessful():
-    rc = 1
-
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/ldap_syntaxes.py

@@ -11 +11 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 
 import samba.getopt as options
@@ -21 +21 @@
 from ldb import ERR_INVALID_ATTRIBUTE_SYNTAX
 from ldb import ERR_ENTRY_ALREADY_EXISTS
-
-from subunit.run import SubunitTestRunner
-import unittest
 
 import samba.tests
@@ -34 +31 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
 opts, args = parser.parse_args()
 
@@ -45 +44 @@
 
 
-class SyntaxTests(unittest.TestCase):
+class SyntaxTests(samba.tests.TestCase):
 
     def setUp(self):
         super(SyntaxTests, self).setUp()
-        self.ldb = ldb
-        self.base_dn = ldb.domain_dn()
-        self.schema_dn = ldb.get_schema_basedn().get_linearized()
+        self.ldb = samba.tests.connect_samdb(host, credentials=creds,
+            session_info=system_session(lp), lp=lp)
+        self.base_dn = self.ldb.domain_dn()
+        self.schema_dn = self.ldb.get_schema_basedn().get_linearized()
         self._setup_dn_string_test()
         self._setup_dn_binary_test()
@@ -193 +193 @@
 
     def test_dn_string(self):
-        # add obeject with correct value
+        # add object with correct value
         object_name1 = "obj-DN-String1" + time.strftime("%s", time.gmtime())
         ldif = self._get_object_ldif(object_name1, self.dn_string_class_name, self.dn_string_class_ldap_display_name,
@@ -200 +200 @@
 
         # search by specifying the DN part only
-        res = ldb.search(base=self.base_dn,
+        res = self.ldb.search(base=self.base_dn,
                          scope=SCOPE_SUBTREE,
                          expression="(%s=%s)" % (self.dn_string_attribute, self.base_dn))
@@ -206 +206 @@
 
         # search by specifying the string part only
-        res = ldb.search(base=self.base_dn,
+        res = self.ldb.search(base=self.base_dn,
                          scope=SCOPE_SUBTREE,
                          expression="(%s=S:5:ABCDE)" % self.dn_string_attribute)
@@ -212 +212 @@
 
         # search by DN+Stirng
-        res = ldb.search(base=self.base_dn,
+        res = self.ldb.search(base=self.base_dn,
                          scope=SCOPE_SUBTREE,
                          expression="(%s=S:5:ABCDE:%s)" % (self.dn_string_attribute, self.base_dn))
@@ -284 +284 @@
         except LdbError, (num, _):
             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-        pass
 
     def test_dn_binary(self):
@@ -294 +293 @@
 
         # search by specifyingthe DN part
-        res = ldb.search(base=self.base_dn,
+        res = self.ldb.search(base=self.base_dn,
                          scope=SCOPE_SUBTREE,
                          expression="(%s=%s)" % (self.dn_binary_attribute, self.base_dn))
@@ -300 +299 @@
 
         # search by specifying the binary part
-        res = ldb.search(base=self.base_dn,
+        res = self.ldb.search(base=self.base_dn,
                          scope=SCOPE_SUBTREE,
                          expression="(%s=B:4:1234)" % self.dn_binary_attribute)
@@ -306 +305 @@
 
         # search by DN+Binary
-        res = ldb.search(base=self.base_dn,
+        res = self.ldb.search(base=self.base_dn,
                          scope=SCOPE_SUBTREE,
                          expression="(%s=B:4:1234:%s)" % (self.dn_binary_attribute, self.base_dn))
@@ -370 +369 @@
         except LdbError, (num, _):
             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-        pass
-
-ldb = samba.tests.connect_samdb(host, credentials=creds, session_info=system_session(lp), lp=lp)
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(SyntaxTests)).wasSuccessful():
-    rc = 1
-
-sys.exit(rc)
+
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/passwords.py

@@ -17 +17 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 
 import samba.getopt as options
@@ -35 +35 @@
 import samba.tests
 from samba.tests import delete_force
-from subunit.run import SubunitTestRunner
-import unittest
 
 parser = optparse.OptionParser("passwords.py [options] <host>")
@@ -45 +43 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
+
 opts, args = parser.parse_args()
 
@@ -67 +68 @@
     def setUp(self):
         super(PasswordTests, self).setUp()
-        self.ldb = ldb
-        self.base_dn = ldb.domain_dn()
+        self.ldb = SamDB(url=host, session_info=system_session(lp), credentials=creds, lp=lp)
+
+        # Gets back the basedn
+        base_dn = self.ldb.domain_dn()
+
+        # Gets back the configuration basedn
+        configuration_dn = self.ldb.get_config_basedn().get_linearized()
+
+        # Get the old "dSHeuristics" if it was set
+        dsheuristics = self.ldb.get_dsheuristics()
+
+        # Set the "dSHeuristics" to activate the correct "userPassword" behaviour
+        self.ldb.set_dsheuristics("000000001")
+
+        # Reset the "dSHeuristics" as they were before
+        self.addCleanup(self.ldb.set_dsheuristics, dsheuristics)
+
+        # Get the old "minPwdAge"
+        minPwdAge = self.ldb.get_minPwdAge()
+
+        # Set it temporarely to "0"
+        self.ldb.set_minPwdAge("0")
+        self.base_dn = self.ldb.domain_dn()
+
+        # Reset the "minPwdAge" as it was before
+        self.addCleanup(self.ldb.set_minPwdAge, minPwdAge)
 
         # (Re)adds the test user "testuser" with no password atm
@@ -137 +162 @@
 
     def test_unicodePwd_hash_set(self):
-        print "Performs a password hash set operation on 'unicodePwd' which should be prevented"
+        """Performs a password hash set operation on 'unicodePwd' which should be prevented"""
         # Notice: Direct hash password sets should never work
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["unicodePwd"] = MessageElement("XXXXXXXXXXXXXXXX", FLAG_MOD_REPLACE,
           "unicodePwd")
         try:
-            ldb.modify(m)
+            self.ldb.modify(m)
             self.fail()
         except LdbError, (num, _):
@@ -151 +176 @@
 
     def test_unicodePwd_hash_change(self):
-        print "Performs a password hash change operation on 'unicodePwd' which should be prevented"
+        """Performs a password hash change operation on 'unicodePwd' which should be prevented"""
         # Notice: Direct hash password changes should never work
 
@@ -169 +194 @@
 
     def test_unicodePwd_clear_set(self):
-        print "Performs a password cleartext set operation on 'unicodePwd'"
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        """Performs a password cleartext set operation on 'unicodePwd'"""
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["unicodePwd"] = MessageElement("\"thatsAcomplPASS2\"".encode('utf-16-le'),
           FLAG_MOD_REPLACE, "unicodePwd")
-        ldb.modify(m)
+        self.ldb.modify(m)
 
     def test_unicodePwd_clear_change(self):
-        print "Performs a password cleartext change operation on 'unicodePwd'"
+        """Performs a password cleartext change operation on 'unicodePwd'"""
 
         self.ldb2.modify_ldif("""
@@ -220 +245 @@
 
     def test_dBCSPwd_hash_set(self):
-        print "Performs a password hash set operation on 'dBCSPwd' which should be prevented"
+        """Performs a password hash set operation on 'dBCSPwd' which should be prevented"""
         # Notice: Direct hash password sets should never work
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["dBCSPwd"] = MessageElement("XXXXXXXXXXXXXXXX", FLAG_MOD_REPLACE,
           "dBCSPwd")
         try:
-            ldb.modify(m)
+            self.ldb.modify(m)
             self.fail()
         except LdbError, (num, _):
@@ -234 +259 @@
 
     def test_dBCSPwd_hash_change(self):
-        print "Performs a password hash change operation on 'dBCSPwd' which should be prevented"
+        """Performs a password hash change operation on 'dBCSPwd' which should be prevented"""
         # Notice: Direct hash password changes should never work
 
@@ -251 +276 @@
 
     def test_userPassword_clear_set(self):
-        print "Performs a password cleartext set operation on 'userPassword'"
+        """Performs a password cleartext set operation on 'userPassword'"""
         # Notice: This works only against Windows if "dSHeuristics" has been set
         # properly
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement("thatsAcomplPASS2", FLAG_MOD_REPLACE,
           "userPassword")
-        ldb.modify(m)
+        self.ldb.modify(m)
 
     def test_userPassword_clear_change(self):
-        print "Performs a password cleartext change operation on 'userPassword'"
+        """Performs a password cleartext change operation on 'userPassword'"""
         # Notice: This works only against Windows if "dSHeuristics" has been set
         # properly
@@ -306 +331 @@
 
     def test_clearTextPassword_clear_set(self):
-        print "Performs a password cleartext set operation on 'clearTextPassword'"
+        """Performs a password cleartext set operation on 'clearTextPassword'"""
         # Notice: This never works against Windows - only supported by us
 
         try:
             m = Message()
-            m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
             m["clearTextPassword"] = MessageElement("thatsAcomplPASS2".encode('utf-16-le'),
               FLAG_MOD_REPLACE, "clearTextPassword")
-            ldb.modify(m)
+            self.ldb.modify(m)
             # this passes against s4
         except LdbError, (num, msg):
@@ -322 +347 @@
 
     def test_clearTextPassword_clear_change(self):
-        print "Performs a password cleartext change operation on 'clearTextPassword'"
+        """Performs a password cleartext change operation on 'clearTextPassword'"""
         # Notice: This never works against Windows - only supported by us
 
@@ -375 +400 @@
 
     def test_failures(self):
-        print "Performs some failure testing"
-
-        try:
-            ldb.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-userPassword: thatsAcomplPASS1
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            self.ldb2.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-userPassword: thatsAcomplPASS1
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            ldb.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            self.ldb2.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            ldb.modify_ldif("""
+        """Performs some failure testing"""
+
+        try:
+            self.ldb.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb2.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb2.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb.modify_ldif("""
 dn: cn=testuser,cn=users,""" + self.base_dn + """
 changetype: modify
@@ -442 +467 @@
 
         try:
-            ldb.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-userPassword: thatsAcomplPASS1
-add: userPassword
-userPassword: thatsAcomplPASS2
-userPassword: thatsAcomplPASS2
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            self.ldb2.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-userPassword: thatsAcomplPASS1
-add: userPassword
-userPassword: thatsAcomplPASS2
-userPassword: thatsAcomplPASS2
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            ldb.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-userPassword: thatsAcomplPASS1
-userPassword: thatsAcomplPASS1
-add: userPassword
-userPassword: thatsAcomplPASS2
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            self.ldb2.modify_ldif("""
-dn: cn=testuser,cn=users,""" + self.base_dn + """
-changetype: modify
-delete: userPassword
-userPassword: thatsAcomplPASS1
-userPassword: thatsAcomplPASS1
-add: userPassword
-userPassword: thatsAcomplPASS2
-""")
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        try:
-            ldb.modify_ldif("""
+            self.ldb.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+add: userPassword
+userPassword: thatsAcomplPASS2
+userPassword: thatsAcomplPASS2
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb2.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+add: userPassword
+userPassword: thatsAcomplPASS2
+userPassword: thatsAcomplPASS2
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+userPassword: thatsAcomplPASS1
+add: userPassword
+userPassword: thatsAcomplPASS2
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb2.modify_ldif("""
+dn: cn=testuser,cn=users,""" + self.base_dn + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+userPassword: thatsAcomplPASS1
+add: userPassword
+userPassword: thatsAcomplPASS2
+""")
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        try:
+            self.ldb.modify_ldif("""
 dn: cn=testuser,cn=users,""" + self.base_dn + """
 changetype: modify
@@ -528 +553 @@
 
         try:
-            ldb.modify_ldif("""
+            self.ldb.modify_ldif("""
 dn: cn=testuser,cn=users,""" + self.base_dn + """
 changetype: modify
@@ -558 +583 @@
 
         try:
-            ldb.modify_ldif("""
+            self.ldb.modify_ldif("""
 dn: cn=testuser,cn=users,""" + self.base_dn + """
 changetype: modify
@@ -624 +649 @@
 
         # Several password changes at once are allowed
-        ldb.modify_ldif("""
+        self.ldb.modify_ldif("""
 dn: cn=testuser,cn=users,""" + self.base_dn + """
 changetype: modify
@@ -633 +658 @@
 
         # Several password changes at once are allowed
-        ldb.modify_ldif("""
+        self.ldb.modify_ldif("""
 dn: cn=testuser,cn=users,""" + self.base_dn + """
 changetype: modify
@@ -702 +727 @@
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["unicodePwd"] = MessageElement([], FLAG_MOD_ADD, "unicodePwd")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["dBCSPwd"] = MessageElement([], FLAG_MOD_ADD, "dBCSPwd")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement([], FLAG_MOD_ADD, "userPassword")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["clearTextPassword"] = MessageElement([], FLAG_MOD_ADD, "clearTextPassword")
         try:
-            ldb.modify(m)
+            self.ldb.modify(m)
             self.fail()
         except LdbError, (num, _):
@@ -739 +764 @@
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["unicodePwd"] = MessageElement([], FLAG_MOD_REPLACE, "unicodePwd")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["dBCSPwd"] = MessageElement([], FLAG_MOD_REPLACE, "dBCSPwd")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement([], FLAG_MOD_REPLACE, "userPassword")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["clearTextPassword"] = MessageElement([], FLAG_MOD_REPLACE, "clearTextPassword")
         try:
-            ldb.modify(m)
+            self.ldb.modify(m)
             self.fail()
         except LdbError, (num, _):
@@ -776 +801 @@
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["unicodePwd"] = MessageElement([], FLAG_MOD_DELETE, "unicodePwd")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["dBCSPwd"] = MessageElement([], FLAG_MOD_DELETE, "dBCSPwd")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement([], FLAG_MOD_DELETE, "userPassword")
         try:
-            ldb.modify(m)
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+            self.ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["clearTextPassword"] = MessageElement([], FLAG_MOD_DELETE, "clearTextPassword")
         try:
-            ldb.modify(m)
+            self.ldb.modify(m)
             self.fail()
         except LdbError, (num, _):
@@ -816 +841 @@
 
         # Delete the "dSHeuristics"
-        ldb.set_dsheuristics(None)
+        self.ldb.set_dsheuristics(None)
 
         time.sleep(1) # This switching time is strictly needed!
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement("myPassword", FLAG_MOD_ADD,
           "userPassword")
-        ldb.modify(m)
-
-        res = ldb.search("cn=testuser,cn=users," + self.base_dn,
+        self.ldb.modify(m)
+
+        res = self.ldb.search("cn=testuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["userPassword"])
         self.assertTrue(len(res) == 1)
@@ -833 +858 @@
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement("myPassword2", FLAG_MOD_REPLACE,
           "userPassword")
-        ldb.modify(m)
-
-        res = ldb.search("cn=testuser,cn=users," + self.base_dn,
+        self.ldb.modify(m)
+
+        res = self.ldb.search("cn=testuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["userPassword"])
         self.assertTrue(len(res) == 1)
@@ -845 +870 @@
 
         m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement([], FLAG_MOD_DELETE,
           "userPassword")
-        ldb.modify(m)
-
-        res = ldb.search("cn=testuser,cn=users," + self.base_dn,
+        self.ldb.modify(m)
+
+        res = self.ldb.search("cn=testuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["userPassword"])
         self.assertTrue(len(res) == 1)
@@ -856 +881 @@
 
         # Set the test "dSHeuristics" to deactivate "userPassword" pwd changes
-        ldb.set_dsheuristics("000000000")
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        self.ldb.set_dsheuristics("000000000")
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement("myPassword3", FLAG_MOD_REPLACE,
           "userPassword")
-        ldb.modify(m)
-
-        res = ldb.search("cn=testuser,cn=users," + self.base_dn,
+        self.ldb.modify(m)
+
+        res = self.ldb.search("cn=testuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["userPassword"])
         self.assertTrue(len(res) == 1)
@@ -871 +896 @@
 
         # Set the test "dSHeuristics" to deactivate "userPassword" pwd changes
-        ldb.set_dsheuristics("000000002")
-
-        m = Message()
-        m.dn = Dn(ldb, "cn=testuser,cn=users," + self.base_dn)
+        self.ldb.set_dsheuristics("000000002")
+
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=testuser,cn=users," + self.base_dn)
         m["userPassword"] = MessageElement("myPassword4", FLAG_MOD_REPLACE,
           "userPassword")
-        ldb.modify(m)
-
-        res = ldb.search("cn=testuser,cn=users," + self.base_dn,
+        self.ldb.modify(m)
+
+        res = self.ldb.search("cn=testuser,cn=users," + self.base_dn,
                          scope=SCOPE_BASE, attrs=["userPassword"])
         self.assertTrue(len(res) == 1)
@@ -886 +911 @@
 
         # Reset the test "dSHeuristics" (reactivate "userPassword" pwd changes)
-        ldb.set_dsheuristics("000000001")
+        self.ldb.set_dsheuristics("000000001")
 
     def test_zero_length(self):
         # Get the old "minPwdLength"
-        minPwdLength = ldb.get_minPwdLength()
+        minPwdLength = self.ldb.get_minPwdLength()
         # Set it temporarely to "0"
-        ldb.set_minPwdLength("0")
+        self.ldb.set_minPwdLength("0")
 
         # Get the old "pwdProperties"
-        pwdProperties = ldb.get_pwdProperties()
+        pwdProperties = self.ldb.get_pwdProperties()
         # Set them temporarely to "0" (to deactivate eventually the complexity)
-        ldb.set_pwdProperties("0")
-
-        ldb.setpassword("(sAMAccountName=testuser)", "")
+        self.ldb.set_pwdProperties("0")
+
+        self.ldb.setpassword("(sAMAccountName=testuser)", "")
 
         # Reset the "pwdProperties" as they were before
-        ldb.set_pwdProperties(pwdProperties)
+        self.ldb.set_pwdProperties(pwdProperties)
 
         # Reset the "minPwdLength" as it was before
-        ldb.set_minPwdLength(minPwdLength)
+        self.ldb.set_minPwdLength(minPwdLength)
 
     def tearDown(self):
@@ -920 +945 @@
         host = "ldap://%s" % host
 
-ldb = SamDB(url=host, session_info=system_session(lp), credentials=creds, lp=lp)
-
-# Gets back the basedn
-base_dn = ldb.domain_dn()
-
-# Gets back the configuration basedn
-configuration_dn = ldb.get_config_basedn().get_linearized()
-
-# Get the old "dSHeuristics" if it was set
-dsheuristics = ldb.get_dsheuristics()
-
-# Set the "dSHeuristics" to activate the correct "userPassword" behaviour
-ldb.set_dsheuristics("000000001")
-
-# Get the old "minPwdAge"
-minPwdAge = ldb.get_minPwdAge()
-
-# Set it temporarely to "0"
-ldb.set_minPwdAge("0")
-
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(PasswordTests)).wasSuccessful():
-    rc = 1
-
-# Reset the "dSHeuristics" as they were before
-ldb.set_dsheuristics(dsheuristics)
-
-# Reset the "minPwdAge" as it was before
-ldb.set_minPwdAge(minPwdAge)
-
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/sam.py

@@ -9 +9 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 
 import samba.getopt as options
@@ -22 +21 @@
 from ldb import ERR_CONSTRAINT_VIOLATION
 from ldb import ERR_UNDEFINED_ATTRIBUTE_TYPE
+from ldb import ERR_INSUFFICIENT_ACCESS_RIGHTS
 from ldb import Message, MessageElement, Dn
 from ldb import FLAG_MOD_ADD, FLAG_MOD_REPLACE, FLAG_MOD_DELETE
@@ -28 +28 @@
     UF_WORKSTATION_TRUST_ACCOUNT, UF_SERVER_TRUST_ACCOUNT,
     UF_PARTIAL_SECRETS_ACCOUNT, UF_TEMP_DUPLICATE_ACCOUNT,
-    UF_PASSWD_NOTREQD, ATYPE_NORMAL_ACCOUNT,
+    UF_INTERDOMAIN_TRUST_ACCOUNT,
+    UF_PASSWD_NOTREQD, UF_LOCKOUT, UF_PASSWORD_EXPIRED, ATYPE_NORMAL_ACCOUNT,
     GTYPE_SECURITY_BUILTIN_LOCAL_GROUP, GTYPE_SECURITY_DOMAIN_LOCAL_GROUP,
     GTYPE_SECURITY_GLOBAL_GROUP, GTYPE_SECURITY_UNIVERSAL_GROUP,
@@ -37 +38 @@
     ATYPE_DISTRIBUTION_UNIVERSAL_GROUP, ATYPE_DISTRIBUTION_LOCAL_GROUP,
     ATYPE_WORKSTATION_TRUST)
-from samba.dcerpc.security import (DOMAIN_RID_USERS, DOMAIN_RID_DOMAIN_MEMBERS,
-    DOMAIN_RID_DCS, DOMAIN_RID_READONLY_DCS)
-
-from subunit.run import SubunitTestRunner
-import unittest
+from samba.dcerpc.security import (DOMAIN_RID_USERS, DOMAIN_RID_ADMINS,
+    DOMAIN_RID_DOMAIN_MEMBERS, DOMAIN_RID_DCS, DOMAIN_RID_READONLY_DCS)
 
 from samba.dcerpc import security
@@ -53 +51 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
 opts, args = parser.parse_args()
 
@@ -64 +64 @@
 creds = credopts.get_credentials(lp)
 
-class SamTests(unittest.TestCase):
+class SamTests(samba.tests.TestCase):
 
     def setUp(self):
@@ -587 +587 @@
     def test_sam_attributes(self):
         """Test the behaviour of special attributes of SAM objects"""
-        print "Testing the behaviour of special attributes of SAM objects\n"""
+        print "Testing the behaviour of special attributes of SAM objects\n"
 
         ldb.add({
@@ -1426 +1426 @@
         # With SYSTEM rights you can set a interdomain trust account.
 
-        # Invalid attribute
-        try:
-            ldb.add({
-                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
-                "objectclass": "user",
-                "userAccountControl": "0"})
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        ldb.add({
+            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+            "objectclass": "user",
+            "userAccountControl": "0"})
+
+        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_PASSWD_NOTREQD == 0)
         delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
 
-# This has to wait until s4 supports it (needs a password module change)
-#        try:
-#            ldb.add({
-#                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
-#                "objectclass": "user",
-#                "userAccountControl": str(UF_NORMAL_ACCOUNT)})
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-#        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        ldb.add({
+            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+            "objectclass": "user",
+            "userAccountControl": str(UF_NORMAL_ACCOUNT)})
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
 
         ldb.add({
@@ -1460 +1459 @@
           ATYPE_NORMAL_ACCOUNT)
         self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+
+        ldb.add({
+            "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+            "objectclass": "user",
+            "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_LOCKOUT | UF_PASSWORD_EXPIRED)})
+
+        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
+        self.assertFalse("lockoutTime" in res1[0])
+        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)
         delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
 
@@ -1472 +1487 @@
         delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
 
-# This isn't supported yet in s4
-#        try:
-#            ldb.add({
-#                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
-#                "objectclass": "user",
-#                "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT)})
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
-#        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
-#
-#        try:
-#            ldb.add({
-#                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
-#                "objectclass": "user",
-#                "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT)})
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
-#        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
-
-# This isn't supported yet in s4 - needs ACL module adaption
-#        try:
-#            ldb.add({
-#                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
-#                "objectclass": "user",
-#                "userAccountControl": str(UF_INTERDOMAIN_TRUST_ACCOUNT)})
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
-#        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        try:
+            ldb.add({
+                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+                "objectclass": "user",
+                "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT)})
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+
+        try:
+            ldb.add({
+                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+                "objectclass": "user",
+                "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT)})
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+
+        try:
+            ldb.add({
+                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+                "objectclass": "user",
+                "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT)})
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_OBJECT_CLASS_VIOLATION)
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+
+        try:
+            ldb.add({
+                "dn": "cn=ldaptestuser,cn=users," + self.base_dn,
+                "objectclass": "user",
+                "userAccountControl": str(UF_INTERDOMAIN_TRUST_ACCOUNT)})
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
 
         # Modify operation
@@ -1534 +1556 @@
             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
-# This has to wait until s4 supports it (needs a password module change)
-#        try:
-#            m = Message()
-#            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
-#            m["userAccountControl"] = MessageElement(
-#              str(UF_NORMAL_ACCOUNT),
-#              FLAG_MOD_REPLACE, "userAccountControl")
-#            ldb.modify(m)
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        try:
+            m = Message()
+            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+            m["userAccountControl"] = MessageElement(
+              str(UF_NORMAL_ACCOUNT),
+              FLAG_MOD_REPLACE, "userAccountControl")
+            ldb.modify(m)
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
         m = Message()
@@ -1559 +1580 @@
           ATYPE_NORMAL_ACCOUNT)
         self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["userAccountControl"] = MessageElement(
+          str(UF_ACCOUNTDISABLE),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE != 0)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["lockoutTime"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "lockoutTime")
+        m["pwdLastSet"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "pwdLastSet")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        m["userAccountControl"] = MessageElement(
+          str(UF_LOCKOUT | UF_PASSWORD_EXPIRED),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
+        self.assertTrue(int(res1[0]["lockoutTime"][0]) == 0)
+        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)
 
         try:
@@ -1571 +1632 @@
             self.assertEquals(num, ERR_OTHER)
 
-# This isn't supported yet in s4
-#        try:
-#            m = Message()
-#            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
-#            m["userAccountControl"] = MessageElement(
-#              str(UF_SERVER_TRUST_ACCOUNT),
-#              FLAG_MOD_REPLACE, "userAccountControl")
-#            ldb.modify(m)
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        try:
+            m = Message()
+            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+            m["userAccountControl"] = MessageElement(
+              str(UF_SERVER_TRUST_ACCOUNT),
+              FLAG_MOD_REPLACE, "userAccountControl")
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
         m = Message()
@@ -1590 +1650 @@
         ldb.modify(m)
 
+        try:
+            m = Message()
+            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+            m["userAccountControl"] = MessageElement(
+              str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT),
+              FLAG_MOD_REPLACE, "userAccountControl")
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
         res1 = ldb.search("cn=ldaptestuser,cn=users," + self.base_dn,
                           scope=SCOPE_BASE, attrs=["sAMAccountType"])
@@ -1609 +1680 @@
           ATYPE_NORMAL_ACCOUNT)
 
-# This isn't supported yet in s4 - needs ACL module adaption
-#        try:
-#            m = Message()
-#            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
-#            m["userAccountControl"] = MessageElement(
-#              str(UF_INTERDOMAIN_TRUST_ACCOUNT),
-#              FLAG_MOD_REPLACE, "userAccountControl")
-#            ldb.modify(m)
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+        try:
+            m = Message()
+            m.dn = Dn(ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+            m["userAccountControl"] = MessageElement(
+              str(UF_INTERDOMAIN_TRUST_ACCOUNT),
+              FLAG_MOD_REPLACE, "userAccountControl")
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 
         # With a computer object
@@ -1630 +1700 @@
         # With SYSTEM rights you can set a interdomain trust account.
 
-        # Invalid attribute
-        try:
-            ldb.add({
-                "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
-                "objectclass": "computer",
-                "userAccountControl": "0"})
-            self.fail()
-        except LdbError, (num, _):
-            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        ldb.add({
+            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+            "objectclass": "computer",
+            "userAccountControl": "0"})
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_PASSWD_NOTREQD == 0)
         delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
 
-# This has to wait until s4 supports it (needs a password module change)
-#        try:
-#            ldb.add({
-#                "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
-#                "objectclass": "computer",
-#                "userAccountControl": str(UF_NORMAL_ACCOUNT)})
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
-#        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        ldb.add({
+            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+            "objectclass": "computer",
+            "userAccountControl": str(UF_NORMAL_ACCOUNT)})
+        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
 
         ldb.add({
@@ -1664 +1733 @@
           ATYPE_NORMAL_ACCOUNT)
         self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
+        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+
+        ldb.add({
+            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+            "objectclass": "computer",
+            "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_LOCKOUT | UF_PASSWORD_EXPIRED)})
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
+        self.assertFalse("lockoutTime" in res1[0])
+        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)
         delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
 
@@ -1697 +1782 @@
         delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
 
-# This isn't supported yet in s4 - needs ACL module adaption
-#        try:
-#            ldb.add({
-#                "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
-#                "objectclass": "computer",
-#                "userAccountControl": str(UF_INTERDOMAIN_TRUST_ACCOUNT)})
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
-#        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        try:
+            ldb.add({
+                "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                "objectclass": "computer",
+                "userAccountControl": str(UF_INTERDOMAIN_TRUST_ACCOUNT)})
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
 
         # Modify operation
@@ -1740 +1824 @@
             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
-# This has to wait until s4 supports it (needs a password module change)
-#        try:
-#            m = Message()
-#            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
-#            m["userAccountControl"] = MessageElement(
-#              str(UF_NORMAL_ACCOUNT),
-#              FLAG_MOD_REPLACE, "userAccountControl")
-#            ldb.modify(m)
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+        try:
+            m = Message()
+            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+            m["userAccountControl"] = MessageElement(
+              str(UF_NORMAL_ACCOUNT),
+              FLAG_MOD_REPLACE, "userAccountControl")
+            ldb.modify(m)
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 
         m = Message()
@@ -1765 +1848 @@
           ATYPE_NORMAL_ACCOUNT)
         self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(
+          str(UF_ACCOUNTDISABLE),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE != 0)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["lockoutTime"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "lockoutTime")
+        m["pwdLastSet"] = MessageElement(str(samba.unix2nttime(0)), FLAG_MOD_REPLACE, "pwdLastSet")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(
+          str(UF_LOCKOUT | UF_PASSWORD_EXPIRED),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["sAMAccountType", "userAccountControl", "lockoutTime", "pwdLastSet"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["sAMAccountType"][0]),
+          ATYPE_NORMAL_ACCOUNT)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_NORMAL_ACCOUNT != 0)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & (UF_LOCKOUT | UF_PASSWORD_EXPIRED) == 0)
+        self.assertTrue(int(res1[0]["lockoutTime"][0]) == 0)
+        self.assertTrue(int(res1[0]["pwdLastSet"][0]) == 0)
 
         try:
@@ -1855 +1978 @@
           ATYPE_WORKSTATION_TRUST)
 
-# This isn't supported yet in s4 - needs ACL module adaption
-#        try:
-#            m = Message()
-#            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
-#            m["userAccountControl"] = MessageElement(
-#              str(UF_INTERDOMAIN_TRUST_ACCOUNT),
-#              FLAG_MOD_REPLACE, "userAccountControl")
-#            ldb.modify(m)
-#            self.fail()
-#        except LdbError, (num, _):
-#            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+        try:
+            m = Message()
+            m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+            m["userAccountControl"] = MessageElement(
+              str(UF_INTERDOMAIN_TRUST_ACCOUNT),
+              FLAG_MOD_REPLACE, "userAccountControl")
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+        # "primaryGroupID" does not change if account type remains the same
+
+        # For a user account
+
+        ldb.add({
+            "dn": "cn=ldaptestuser2,cn=users," + self.base_dn,
+            "objectclass": "user",
+            "userAccountControl": str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_ACCOUNTDISABLE)})
+
+        res1 = ldb.search("cn=ldaptestuser2,cn=users," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["userAccountControl"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["userAccountControl"][0]),
+           UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD | UF_ACCOUNTDISABLE)
+
+        m = Message()
+        m.dn = Dn(ldb, "<SID=" + ldb.get_domain_sid() + "-" + str(DOMAIN_RID_ADMINS) + ">")
+        m["member"] = MessageElement(
+          "cn=ldaptestuser2,cn=users," + self.base_dn, FLAG_MOD_ADD, "member")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
+        m["primaryGroupID"] = MessageElement(str(DOMAIN_RID_ADMINS),
+          FLAG_MOD_REPLACE, "primaryGroupID")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
+        m["userAccountControl"] = MessageElement(
+          str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestuser2,cn=users," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["userAccountControl", "primaryGroupID"])
+        self.assertTrue(len(res1) == 1)
+        self.assertTrue(int(res1[0]["userAccountControl"][0]) & UF_ACCOUNTDISABLE == 0)
+        self.assertEquals(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_ADMINS)
+
+        # For a workstation account
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["primaryGroupID"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_DOMAIN_MEMBERS)
+
+        m = Message()
+        m.dn = Dn(ldb, "<SID=" + ldb.get_domain_sid() + "-" + str(DOMAIN_RID_USERS) + ">")
+        m["member"] = MessageElement(
+          "cn=ldaptestcomputer,cn=computers," + self.base_dn, FLAG_MOD_ADD, "member")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["primaryGroupID"] = MessageElement(str(DOMAIN_RID_USERS),
+          FLAG_MOD_REPLACE, "primaryGroupID")
+        ldb.modify(m)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(
+          str(UF_WORKSTATION_TRUST_ACCOUNT),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["primaryGroupID"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(int(res1[0]["primaryGroupID"][0]), DOMAIN_RID_USERS)
 
         delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+        delete_force(self.ldb, "cn=ldaptestuser2,cn=users," + self.base_dn)
+        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+
+    def test_isCriticalSystemObject(self):
+        """Test the isCriticalSystemObject behaviour"""
+        print "Testing isCriticalSystemObject behaviour\n"
+
+        # Add tests
+
+        ldb.add({
+            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+            "objectclass": "computer"})
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertTrue("isCriticalSystemObject" not in res1[0])
+
+        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+
+        ldb.add({
+            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+            "objectclass": "computer",
+            "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT)})
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "FALSE")
+
+        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+
+        ldb.add({
+            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+            "objectclass": "computer",
+            "userAccountControl": str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT)})
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "TRUE")
+
+        delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+
+        ldb.add({
+            "dn": "cn=ldaptestcomputer,cn=computers," + self.base_dn,
+            "objectclass": "computer",
+            "userAccountControl": str(UF_SERVER_TRUST_ACCOUNT)})
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "TRUE")
+
+        # Modification tests
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "TRUE")
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(str(UF_WORKSTATION_TRUST_ACCOUNT),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "FALSE")
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(
+          str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PARTIAL_SECRETS_ACCOUNT),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "TRUE")
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(str(UF_NORMAL_ACCOUNT | UF_PASSWD_NOTREQD),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "TRUE")
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(str(UF_SERVER_TRUST_ACCOUNT),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "TRUE")
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["userAccountControl"] = MessageElement(str(UF_WORKSTATION_TRUST_ACCOUNT),
+          FLAG_MOD_REPLACE, "userAccountControl")
+        ldb.modify(m)
+
+        res1 = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                          scope=SCOPE_BASE,
+                          attrs=["isCriticalSystemObject"])
+        self.assertTrue(len(res1) == 1)
+        self.assertEquals(res1[0]["isCriticalSystemObject"][0], "FALSE")
+
         delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
 
@@ -2207 +2537 @@
         self.assertEquals(res[0]["dNSHostName"][0], "testname2.testdom")
         self.assertEquals(res[0]["sAMAccountName"][0], "testname2$")
-        self.assertTrue(res[0]["servicePrincipalName"][0] == "HOST/testname2" or
-                        res[0]["servicePrincipalName"][1] == "HOST/testname2")
-        self.assertTrue(res[0]["servicePrincipalName"][0] == "HOST/testname2.testdom" or
-                        res[0]["servicePrincipalName"][1] == "HOST/testname2.testdom")
+        self.assertTrue(len(res[0]["servicePrincipalName"]) == 2)
+        self.assertTrue("HOST/testname2" in res[0]["servicePrincipalName"])
+        self.assertTrue("HOST/testname2.testdom" in res[0]["servicePrincipalName"])
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["servicePrincipalName"] = MessageElement("HOST/testname2.testdom",
+                                                   FLAG_MOD_ADD, "servicePrincipalName")
+        try:
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_ATTRIBUTE_OR_VALUE_EXISTS)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["servicePrincipalName"] = MessageElement("HOST/testname3",
+                                                   FLAG_MOD_ADD, "servicePrincipalName")
+        ldb.modify(m)
+
+        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                         scope=SCOPE_BASE, attrs=["dNSHostName", "sAMAccountName", "servicePrincipalName"])
+        self.assertTrue(len(res) == 1)
+        self.assertEquals(res[0]["dNSHostName"][0], "testname2.testdom")
+        self.assertEquals(res[0]["sAMAccountName"][0], "testname2$")
+        self.assertTrue(len(res[0]["servicePrincipalName"]) == 3)
+        self.assertTrue("HOST/testname2" in res[0]["servicePrincipalName"])
+        self.assertTrue("HOST/testname3" in res[0]["servicePrincipalName"])
+        self.assertTrue("HOST/testname2.testdom" in res[0]["servicePrincipalName"])
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
+        m["dNSHostName"] = MessageElement("testname3.testdom",
+                                          FLAG_MOD_REPLACE, "dNSHostName")
+        m["servicePrincipalName"] = MessageElement("HOST/testname3.testdom",
+                                                   FLAG_MOD_ADD, "servicePrincipalName")
+        ldb.modify(m)
+
+        res = ldb.search("cn=ldaptestcomputer,cn=computers," + self.base_dn,
+                         scope=SCOPE_BASE, attrs=["dNSHostName", "sAMAccountName", "servicePrincipalName"])
+        self.assertTrue(len(res) == 1)
+        self.assertEquals(res[0]["dNSHostName"][0], "testname3.testdom")
+        self.assertEquals(res[0]["sAMAccountName"][0], "testname2$")
+        self.assertTrue(len(res[0]["servicePrincipalName"]) == 3)
+        self.assertTrue("HOST/testname2" in res[0]["servicePrincipalName"])
+        self.assertTrue("HOST/testname3" in res[0]["servicePrincipalName"])
+        self.assertTrue("HOST/testname3.testdom" in res[0]["servicePrincipalName"])
 
         delete_force(self.ldb, "cn=ldaptestcomputer,cn=computers," + self.base_dn)
@@ -2216 +2589 @@
     def test_sam_description_attribute(self):
         """Test SAM description attribute"""
-        print "Test SAM description attribute"""
+        print "Test SAM description attribute"
 
         self.ldb.add({
             "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
-            "description": "desc2",
             "objectclass": "group",
-            "description": "desc1"})
+            "description": "desc1"
+        })
 
         res = ldb.search("cn=ldaptestgroup,cn=users," + self.base_dn,
@@ -2382 +2755 @@
 
 
+    def test_fSMORoleOwner_attribute(self):
+        """Test fSMORoleOwner attribute"""
+        print "Test fSMORoleOwner attribute"
+
+        ds_service_name = self.ldb.get_dsServiceName()
+
+        # The "fSMORoleOwner" attribute can only be set to "nTDSDSA" entries,
+        # invalid DNs return ERR_UNWILLING_TO_PERFORM
+
+        try:
+            self.ldb.add({
+                "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
+                "objectclass": "group",
+                "fSMORoleOwner": self.base_dn})
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        try:
+            self.ldb.add({
+                "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
+                "objectclass": "group",
+                "fSMORoleOwner": [] })
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
+
+        # We are able to set it to a valid "nTDSDSA" entry if the server is
+        # capable of handling the role
+
+        self.ldb.add({
+            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
+            "objectclass": "group",
+            "fSMORoleOwner": ds_service_name })
+
+        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
+
+        self.ldb.add({
+            "dn": "cn=ldaptestgroup,cn=users," + self.base_dn,
+            "objectclass": "group" })
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
+        m.add(MessageElement(self.base_dn, FLAG_MOD_REPLACE, "fSMORoleOwner"))
+        try:
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
+        m.add(MessageElement([], FLAG_MOD_REPLACE, "fSMORoleOwner"))
+        try:
+            ldb.modify(m)
+            self.fail()
+        except LdbError, (num, _):
+            self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
+
+        # We are able to set it to a valid "nTDSDSA" entry if the server is
+        # capable of handling the role
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
+        m.add(MessageElement(ds_service_name, FLAG_MOD_REPLACE, "fSMORoleOwner"))
+        ldb.modify(m)
+
+        # A clean-out works on plain entries, not master (schema, PDC...) DNs
+
+        m = Message()
+        m.dn = Dn(ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
+        m.add(MessageElement([], FLAG_MOD_DELETE, "fSMORoleOwner"))
+        ldb.modify(m)
+
+        delete_force(self.ldb, "cn=ldaptestgroup,cn=users," + self.base_dn)
+
+    def test_protected_sid_objects(self):
+        """Test deletion of objects with RID < 1000"""
+        # a list of some well-known sids
+        # objects in Builtin are aready covered by objectclass
+        protected_list = [
+            ["CN=Domain Admins","CN=Users,"],
+            ["CN=Schema Admins","CN=Users,"],
+            ["CN=Enterprise Admins","CN=Users,"],
+            ["CN=Administrator","CN=Users,"],
+            ["CN=Domain Controllers","CN=Users,"],
+            ]
+
+
+
+        for pr_object in protected_list:
+            try:
+                self.ldb.delete(pr_object[0] + "," + pr_object[1] + self.base_dn)
+            except LdbError, (num, _):
+                self.assertEquals(num, ERR_OTHER)
+            else:
+                self.fail("Deleted " + pr_object[0])
+
+            try:
+                self.ldb.rename(pr_object[0] + "," + pr_object[1] + self.base_dn,
+                                pr_object[0] + "2," + pr_object[1] + self.base_dn)
+            except LdbError, (num, _):
+                self.fail("Could not rename " + pr_object[0])
+
+            self.ldb.rename(pr_object[0] + "2," + pr_object[1] + self.base_dn,
+                            pr_object[0] + "," + pr_object[1] + self.base_dn)
+
+    def test_new_user_default_attributes(self):
+        """Test default attributes for new user objects"""
+        print "Test default attributes for new User objects\n"
+
+        user_name = "ldaptestuser"
+        user_dn = "CN=%s,CN=Users,%s" % (user_name, self.base_dn)
+        ldb.add({
+            "dn": user_dn,
+            "objectclass": "user",
+            "sAMAccountName": user_name})
+
+        res = ldb.search(user_dn, scope=SCOPE_BASE)
+        self.assertTrue(len(res) == 1)
+        user_obj = res[0]
+
+        expected_attrs = {"primaryGroupID": MessageElement(["513"]),
+                          "logonCount": MessageElement(["0"]),
+                          "cn": MessageElement([user_name]),
+                          "countryCode": MessageElement(["0"]),
+                          "objectClass": MessageElement(["top","person","organizationalPerson","user"]),
+                          "instanceType": MessageElement(["4"]),
+                          "distinguishedName": MessageElement([user_dn]),
+                          "sAMAccountType": MessageElement(["805306368"]),
+                          "objectSid": "**SKIP**",
+                          "whenCreated": "**SKIP**",
+                          "uSNCreated": "**SKIP**",
+                          "badPasswordTime": MessageElement(["0"]),
+                          "dn": Dn(ldb, user_dn),
+                          "pwdLastSet": MessageElement(["0"]),
+                          "sAMAccountName": MessageElement([user_name]),
+                          "objectCategory": MessageElement(["CN=Person,%s" % ldb.get_schema_basedn().get_linearized()]),
+                          "objectGUID": "**SKIP**",
+                          "whenChanged": "**SKIP**",
+                          "badPwdCount": MessageElement(["0"]),
+                          "accountExpires": MessageElement(["9223372036854775807"]),
+                          "name": MessageElement([user_name]),
+                          "codePage": MessageElement(["0"]),
+                          "userAccountControl": MessageElement(["546"]),
+                          "lastLogon": MessageElement(["0"]),
+                          "uSNChanged": "**SKIP**",
+                          "lastLogoff": MessageElement(["0"])}
+        # assert we have expected attribute names
+        actual_names = set(user_obj.keys())
+        # Samba does not use 'dSCorePropagationData', so skip it
+        actual_names -= set(['dSCorePropagationData'])
+        self.assertEqual(set(expected_attrs.keys()), actual_names, "Actual object does not have expected attributes")
+        # check attribute values
+        for name in expected_attrs.keys():
+            actual_val = user_obj.get(name)
+            self.assertFalse(actual_val is None, "No value for attribute '%s'" % name)
+            expected_val = expected_attrs[name]
+            if expected_val == "**SKIP**":
+                # "**ANY**" values means "any"
+                continue
+            self.assertEqual(expected_val, actual_val,
+                             "Unexpected value for '%s'" % name)
+        # clean up
+        delete_force(self.ldb, "cn=ldaptestuser,cn=users," + self.base_dn)
+
+
 if not "://" in host:
     if os.path.isfile(host):
@@ -2390 +2930 @@
 ldb = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp)
 
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(SamTests)).wasSuccessful():
-    rc = 1
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/sec_descriptor.py

@@ -11 +11 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 
 import samba.getopt as options
@@ -20 +20 @@
 
 # For running the test unit
-from samba.ndr import ndr_pack
+from samba.ndr import ndr_pack, ndr_unpack
 from samba.dcerpc import security
 
 from samba import gensec, sd_utils
 from samba.samdb import SamDB
-from samba.credentials import Credentials
+from samba.credentials import Credentials, DONT_USE_KERBEROS
 from samba.auth import system_session
 from samba.dsdb import DS_DOMAIN_FUNCTION_2008
 from samba.dcerpc.security import (
     SECINFO_OWNER, SECINFO_GROUP, SECINFO_DACL, SECINFO_SACL)
-from subunit.run import SubunitTestRunner
 import samba.tests
 from samba.tests import delete_force
-import unittest
 
 parser = optparse.OptionParser("sec_descriptor.py [options] <host>")
@@ -43 +41 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
+
 opts, args = parser.parse_args()
 
@@ -137 +138 @@
         creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
                                       | gensec.FEATURE_SEAL)
+        creds_tmp.set_kerberos_state(DONT_USE_KERBEROS) # kinit is too expensive to use in a tight loop
         ldb_target = SamDB(url=host, credentials=creds_tmp, lp=lp)
         return ldb_target
@@ -142 +144 @@
     def setUp(self):
         super(DescriptorTests, self).setUp()
-        self.ldb_admin = ldb
-        self.base_dn = ldb.domain_dn()
+        self.ldb_admin = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp,
+            options=ldb_options)
+        self.base_dn = self.ldb_admin.domain_dn()
         self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
         self.schema_dn = self.ldb_admin.get_schema_basedn().get_linearized()
         self.domain_sid = security.dom_sid(self.ldb_admin.get_domain_sid())
-        self.sd_utils = sd_utils.SDUtils(ldb)
+        self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
         print "baseDN: %s" % self.base_dn
 
@@ -200 +203 @@
 
         self.ldb_admin.add_remove_group_members("Enterprise Admins",
-                                                "testuser1,testuser5,testuser6,testuser8",
+                                                ["testuser1", "testuser5", "testuser6", "testuser8"],
                                                 add_members_operation=True)
         self.ldb_admin.add_remove_group_members("Domain Admins",
-                                                "testuser2,testuser5,testuser6,testuser7",
+                                                ["testuser2","testuser5","testuser6","testuser7"],
                                                 add_members_operation=True)
         self.ldb_admin.add_remove_group_members("Schema Admins",
-                                                "testuser3,testuser6,testuser7,testuser8",
+                                                ["testuser3","testuser6","testuser7","testuser8"],
                                                 add_members_operation=True)
 
@@ -313 +316 @@
             },
         }
-        # Discover 'msDS-Behavior-Version'
-        res = self.ldb_admin.search(base=self.base_dn, expression="distinguishedName=%s" % self.base_dn, \
-                attrs=['msDS-Behavior-Version'])
-        res = int(res[0]['msDS-Behavior-Version'][0])
+        # Discover 'domainControllerFunctionality'
+        res = self.ldb_admin.search(base="", scope=SCOPE_BASE,
+                                    attrs=['domainControllerFunctionality'])
+        res = int(res[0]['domainControllerFunctionality'][0])
         if res < DS_DOMAIN_FUNCTION_2008:
             self.DS_BEHAVIOR = "ds_behavior_win2003"
@@ -1848 +1851 @@
         self.assertFalse("G:" in desc_sddl)
 
+    def test_311(self):
+        sd_flags = (SECINFO_OWNER |
+                    SECINFO_GROUP |
+                    SECINFO_DACL |
+                    SECINFO_SACL)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                [], controls=None)
+        self.assertFalse("nTSecurityDescriptor" in res[0])
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["name"], controls=None)
+        self.assertFalse("nTSecurityDescriptor" in res[0])
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["name"], controls=["sd_flags:1:%d" % (sd_flags)])
+        self.assertFalse("nTSecurityDescriptor" in res[0])
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                [], controls=["sd_flags:1:%d" % (sd_flags)])
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["*"], controls=["sd_flags:1:%d" % (sd_flags)])
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["nTSecurityDescriptor", "*"], controls=["sd_flags:1:%d" % (sd_flags)])
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["*", "nTSecurityDescriptor"], controls=["sd_flags:1:%d" % (sd_flags)])
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["nTSecurityDescriptor", "name"], controls=["sd_flags:1:%d" % (sd_flags)])
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["name", "nTSecurityDescriptor"], controls=["sd_flags:1:%d" % (sd_flags)])
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["nTSecurityDescriptor"], controls=None)
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["name", "nTSecurityDescriptor"], controls=None)
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None,
+                ["nTSecurityDescriptor", "name"], controls=None)
+        self.assertTrue("nTSecurityDescriptor" in res[0])
+        tmp = res[0]["nTSecurityDescriptor"][0]
+        sd = ndr_unpack(security.descriptor, tmp)
+        sddl = sd.as_sddl(self.sd_utils.domain_sid)
+        self.assertTrue("O:" in sddl)
+        self.assertTrue("G:" in sddl)
+        self.assertTrue("D:" in sddl)
+        self.assertTrue("S:" in sddl)
+
+    def test_312(self):
+        """This search is done by the windows dc join..."""
+
+        res = self.ldb_admin.search(self.base_dn, SCOPE_BASE, None, ["1.1"],
+                controls=["extended_dn:1:0", "sd_flags:1:0", "search_options:1:1"])
+        self.assertFalse("nTSecurityDescriptor" in res[0])
 
 class RightsAttributesTests(DescriptorTests):
@@ -1865 +1991 @@
         self.ldb_admin.newuser("testuser_attr2", "samba123@")
         self.ldb_admin.add_remove_group_members("Domain Admins",
-                                                "testuser_attr2",
+                                                ["testuser_attr2"],
                                                 add_members_operation=True)
 
@@ -1960 +2086 @@
         self.assertTrue("managedBy" in res[0]["allowedAttributesEffective"])
 
+class SdAutoInheritTests(DescriptorTests):
+    def deleteAll(self):
+        delete_force(self.ldb_admin, self.sub_dn)
+        delete_force(self.ldb_admin, self.ou_dn)
+
+    def setUp(self):
+        super(SdAutoInheritTests, self).setUp()
+        self.ou_dn = "OU=test_SdAutoInherit_ou," + self.base_dn
+        self.sub_dn = "OU=test_sub," + self.ou_dn
+        self.deleteAll()
+
+    def test_301(self):
+        """ Modify a descriptor with OWNER_SECURITY_INFORMATION set.
+            See that only the owner has been changed.
+        """
+        attrs = ["nTSecurityDescriptor", "replPropertyMetaData", "uSNChanged"]
+        controls=["sd_flags:1:%d" % (SECINFO_DACL)]
+        ace = "(A;CI;CC;;;NU)"
+        sub_ace = "(A;CIID;CC;;;NU)"
+        sd_sddl = "O:BAG:BAD:P(A;CI;0x000f01ff;;;AU)"
+        sd = security.descriptor.from_sddl(sd_sddl, self.domain_sid)
+
+        self.ldb_admin.create_ou(self.ou_dn,sd=sd)
+        self.ldb_admin.create_ou(self.sub_dn)
+
+        ou_res0 = self.sd_utils.ldb.search(self.ou_dn, SCOPE_BASE,
+                                           None, attrs, controls=controls)
+        sub_res0 = self.sd_utils.ldb.search(self.sub_dn, SCOPE_BASE,
+                                            None, attrs, controls=controls)
+
+        ou_sd0 = ndr_unpack(security.descriptor, ou_res0[0]["nTSecurityDescriptor"][0])
+        sub_sd0 = ndr_unpack(security.descriptor, sub_res0[0]["nTSecurityDescriptor"][0])
+
+        ou_sddl0 = ou_sd0.as_sddl(self.domain_sid)
+        sub_sddl0 = sub_sd0.as_sddl(self.domain_sid)
+
+        self.assertFalse(ace in ou_sddl0)
+        self.assertFalse(ace in sub_sddl0)
+
+        ou_sddl1 = (ou_sddl0[:ou_sddl0.index("(")] + ace +
+                    ou_sddl0[ou_sddl0.index("("):])
+
+        sub_sddl1 = (sub_sddl0[:sub_sddl0.index("(")] + ace +
+                     sub_sddl0[sub_sddl0.index("("):])
+
+        self.sd_utils.modify_sd_on_dn(self.ou_dn, ou_sddl1, controls=controls)
+
+        sub_res2 = self.sd_utils.ldb.search(self.sub_dn, SCOPE_BASE,
+                                            None, attrs, controls=controls)
+        ou_res2 = self.sd_utils.ldb.search(self.ou_dn, SCOPE_BASE,
+                                           None, attrs, controls=controls)
+
+        ou_sd2 = ndr_unpack(security.descriptor, ou_res2[0]["nTSecurityDescriptor"][0])
+        sub_sd2 = ndr_unpack(security.descriptor, sub_res2[0]["nTSecurityDescriptor"][0])
+
+        ou_sddl2 = ou_sd2.as_sddl(self.domain_sid)
+        sub_sddl2 = sub_sd2.as_sddl(self.domain_sid)
+
+        self.assertFalse(ou_sddl2 == ou_sddl0)
+        self.assertFalse(sub_sddl2 == sub_sddl0)
+
+        if ace not in ou_sddl2:
+            print "ou0: %s" % ou_sddl0
+            print "ou2: %s" % ou_sddl2
+
+        if sub_ace not in sub_sddl2:
+            print "sub0: %s" % sub_sddl0
+            print "sub2: %s" % sub_sddl2
+
+        self.assertTrue(ace in ou_sddl2)
+        self.assertTrue(sub_ace in sub_sddl2)
+
+        ou_usn0 = int(ou_res0[0]["uSNChanged"][0])
+        ou_usn2 = int(ou_res2[0]["uSNChanged"][0])
+        self.assertTrue(ou_usn2 > ou_usn0)
+
+        sub_usn0 = int(sub_res0[0]["uSNChanged"][0])
+        sub_usn2 = int(sub_res2[0]["uSNChanged"][0])
+        self.assertTrue(sub_usn2 == sub_usn0)
+
 if not "://" in host:
     if os.path.isfile(host):
@@ -1970 +2176 @@
     ldb_options = ["modules:paged_searches"]
 
-ldb = SamDB(host,
-            credentials=creds,
-            session_info=system_session(lp),
-            lp=lp,
-            options=ldb_options)
-
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(OwnerGroupDescriptorTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(DaclDescriptorTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(SdFlagsDescriptorTests)).wasSuccessful():
-    rc = 1
-if not runner.run(unittest.makeSuite(RightsAttributesTests)).wasSuccessful():
-    rc = 1
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/token_group.py

@@ -9 +9 @@
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
+
+from samba.tests.subunitrun import SubunitOptions, TestProgram
 
 import samba.getopt as options
 
 from samba.auth import system_session
-from samba import ldb
+from samba import ldb, dsdb
 from samba.samdb import SamDB
 from samba.auth import AuthContext
-from samba.ndr import ndr_pack, ndr_unpack
+from samba.ndr import ndr_unpack
 from samba import gensec
-from samba.credentials import Credentials
-
-from subunit.run import SubunitTestRunner
-import unittest
+from samba.credentials import Credentials, DONT_USE_KERBEROS
+from samba.dsdb import GTYPE_SECURITY_GLOBAL_GROUP, GTYPE_SECURITY_UNIVERSAL_GROUP
 import samba.tests
-
-from samba.dcerpc import security
+from samba.tests import delete_force
+
 from samba.auth import AUTH_SESSION_INFO_DEFAULT_GROUPS, AUTH_SESSION_INFO_AUTHENTICATED, AUTH_SESSION_INFO_SIMPLE_PRIVILEGES
 
@@ -37 +35 @@
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
 opts, args = parser.parse_args()
 
@@ -47 +47 @@
 lp = sambaopts.get_loadparm()
 creds = credopts.get_credentials(lp)
-
-class TokenTest(samba.tests.TestCase):
+creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
+
+def closure(vSet, wSet, aSet):
+    for edge in aSet:
+        start, end = edge
+        if start in wSet:
+            if end not in wSet and end in vSet:
+                wSet.add(end)
+                closure(vSet, wSet, aSet)
+
+class StaticTokenTest(samba.tests.TestCase):
 
     def setUp(self):
-        super(TokenTest, self).setUp()
-        self.ldb = samdb
-        self.base_dn = samdb.domain_dn()
+        super(StaticTokenTest, self).setUp()
+        self.ldb = SamDB(url, credentials=creds, session_info=system_session(lp), lp=lp)
+        self.base_dn = self.ldb.domain_dn()
 
         res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
@@ -79 +88 @@
         self.assertEquals(len(res), 1)
 
-        print("Geting tokenGroups from rootDSE")
+        print("Getting tokenGroups from rootDSE")
         tokengroups = []
         for sid in res[0]['tokenGroups']:
@@ -89 +98 @@
             print("token sids don't match")
             print("tokengroups: %s" % tokengroups)
-            print("calculated : %s" % self.user_sids);
+            print("calculated : %s" % self.user_sids)
             print("difference : %s" % sidset1.difference(sidset2))
             self.fail(msg="calculated groups don't match against rootDSE tokenGroups")
 
     def test_dn_tokenGroups(self):
-        print("Geting tokenGroups from user DN")
+        print("Getting tokenGroups from user DN")
         res = self.ldb.search(self.user_sid_dn, scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
         self.assertEquals(len(res), 1)
@@ -106 +115 @@
         if len(sidset1.difference(sidset2)):
             print("token sids don't match")
-            print("tokengroups: %s" % tokengroups)
-            print("calculated : %s" % sids);
             print("difference : %s" % sidset1.difference(sidset2))
             self.fail(msg="calculated groups don't match against user DN tokenGroups")
-        
+
     def test_pac_groups(self):
         settings = {}
@@ -135 +142 @@
         server_finished = False
         server_to_client = ""
-        
-        """Run the actual call loop"""
+
+        # Run the actual call loop.
         while client_finished == False and server_finished == False:
             if not client_finished:
@@ -156 +163 @@
         if len(sidset1.difference(sidset2)):
             print("token sids don't match")
+            print("difference : %s" % sidset1.difference(sidset2))
+            self.fail(msg="calculated groups don't match against user PAC tokenGroups")
+
+class DynamicTokenTest(samba.tests.TestCase):
+
+    def get_creds(self, target_username, target_password):
+        creds_tmp = Credentials()
+        creds_tmp.set_username(target_username)
+        creds_tmp.set_password(target_password)
+        creds_tmp.set_domain(creds.get_domain())
+        creds_tmp.set_realm(creds.get_realm())
+        creds_tmp.set_workstation(creds.get_workstation())
+        creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
+                                      | gensec.FEATURE_SEAL)
+        return creds_tmp
+
+    def get_ldb_connection(self, target_username, target_password):
+        creds_tmp = self.get_creds(target_username, target_password)
+        ldb_target = SamDB(url=url, credentials=creds_tmp, lp=lp)
+        return ldb_target
+
+    def setUp(self):
+        super(DynamicTokenTest, self).setUp()
+        self.admin_ldb = SamDB(url, credentials=creds, session_info=system_session(lp), lp=lp)
+
+        self.base_dn = self.admin_ldb.domain_dn()
+
+        self.test_user = "tokengroups_user1"
+        self.test_user_pass = "samba123@"
+        self.admin_ldb.newuser(self.test_user, self.test_user_pass)
+        self.test_group0 = "tokengroups_group0"
+        self.admin_ldb.newgroup(self.test_group0, grouptype=dsdb.GTYPE_SECURITY_DOMAIN_LOCAL_GROUP)
+        res = self.admin_ldb.search(base="cn=%s,cn=users,%s" % (self.test_group0, self.base_dn),
+                                    attrs=["objectSid"], scope=ldb.SCOPE_BASE)
+        self.test_group0_sid = ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["objectSid"][0])
+
+        self.admin_ldb.add_remove_group_members(self.test_group0, [self.test_user],
+                                       add_members_operation=True)
+
+        self.test_group1 = "tokengroups_group1"
+        self.admin_ldb.newgroup(self.test_group1, grouptype=dsdb.GTYPE_SECURITY_GLOBAL_GROUP)
+        res = self.admin_ldb.search(base="cn=%s,cn=users,%s" % (self.test_group1, self.base_dn),
+                                    attrs=["objectSid"], scope=ldb.SCOPE_BASE)
+        self.test_group1_sid = ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["objectSid"][0])
+
+        self.admin_ldb.add_remove_group_members(self.test_group1, [self.test_user],
+                                       add_members_operation=True)
+
+        self.test_group2 = "tokengroups_group2"
+        self.admin_ldb.newgroup(self.test_group2, grouptype=dsdb.GTYPE_SECURITY_UNIVERSAL_GROUP)
+
+        res = self.admin_ldb.search(base="cn=%s,cn=users,%s" % (self.test_group2, self.base_dn),
+                                    attrs=["objectSid"], scope=ldb.SCOPE_BASE)
+        self.test_group2_sid = ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["objectSid"][0])
+
+        self.admin_ldb.add_remove_group_members(self.test_group2, [self.test_user],
+                                       add_members_operation=True)
+
+        self.ldb = self.get_ldb_connection(self.test_user, self.test_user_pass)
+
+        res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
+        self.assertEquals(len(res), 1)
+
+        self.user_sid_dn = "<SID=%s>" % str(ndr_unpack(samba.dcerpc.security.dom_sid, res[0]["tokenGroups"][0]))
+
+        res = self.ldb.search(self.user_sid_dn, scope=ldb.SCOPE_BASE, attrs=[])
+        self.assertEquals(len(res), 1)
+
+        self.test_user_dn = res[0].dn
+
+        session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS |
+                               AUTH_SESSION_INFO_AUTHENTICATED |
+                               AUTH_SESSION_INFO_SIMPLE_PRIVILEGES)
+        session = samba.auth.user_session(self.ldb, lp_ctx=lp, dn=self.user_sid_dn,
+                                          session_info_flags=session_info_flags)
+
+        token = session.security_token
+        self.user_sids = []
+        for s in token.sids:
+            self.user_sids.append(str(s))
+
+    def tearDown(self):
+        super(DynamicTokenTest, self).tearDown()
+        delete_force(self.admin_ldb, "CN=%s,%s,%s" %
+                          (self.test_user, "cn=users", self.base_dn))
+        delete_force(self.admin_ldb, "CN=%s,%s,%s" %
+                          (self.test_group0, "cn=users", self.base_dn))
+        delete_force(self.admin_ldb, "CN=%s,%s,%s" %
+                          (self.test_group1, "cn=users", self.base_dn))
+        delete_force(self.admin_ldb, "CN=%s,%s,%s" %
+                          (self.test_group2, "cn=users", self.base_dn))
+
+    def test_rootDSE_tokenGroups(self):
+        """Testing rootDSE tokengroups against internal calculation"""
+        if not url.startswith("ldap"):
+            self.fail(msg="This test is only valid on ldap")
+
+        res = self.ldb.search("", scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
+        self.assertEquals(len(res), 1)
+
+        print("Getting tokenGroups from rootDSE")
+        tokengroups = []
+        for sid in res[0]['tokenGroups']:
+            tokengroups.append(str(ndr_unpack(samba.dcerpc.security.dom_sid, sid)))
+
+        sidset1 = set(tokengroups)
+        sidset2 = set(self.user_sids)
+        if len(sidset1.difference(sidset2)):
+            print("token sids don't match")
             print("tokengroups: %s" % tokengroups)
-            print("calculated : %s" % sids);
+            print("calculated : %s" % self.user_sids)
+            print("difference : %s" % sidset1.difference(sidset2))
+            self.fail(msg="calculated groups don't match against rootDSE tokenGroups")
+
+    def test_dn_tokenGroups(self):
+        print("Getting tokenGroups from user DN")
+        res = self.ldb.search(self.user_sid_dn, scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
+        self.assertEquals(len(res), 1)
+
+        dn_tokengroups = []
+        for sid in res[0]['tokenGroups']:
+            dn_tokengroups.append(str(ndr_unpack(samba.dcerpc.security.dom_sid, sid)))
+
+        sidset1 = set(dn_tokengroups)
+        sidset2 = set(self.user_sids)
+        if len(sidset1.difference(sidset2)):
+            print("token sids don't match")
+            print("difference : %s" % sidset1.difference(sidset2))
+            self.fail(msg="calculated groups don't match against user DN tokenGroups")
+
+    def test_pac_groups(self):
+        settings = {}
+        settings["lp_ctx"] = lp
+        settings["target_hostname"] = lp.get("netbios name")
+
+        gensec_client = gensec.Security.start_client(settings)
+        gensec_client.set_credentials(self.get_creds(self.test_user, self.test_user_pass))
+        gensec_client.want_feature(gensec.FEATURE_SEAL)
+        gensec_client.start_mech_by_sasl_name("GSSAPI")
+
+        auth_context = AuthContext(lp_ctx=lp, ldb=self.ldb, methods=[])
+
+        gensec_server = gensec.Security.start_server(settings, auth_context)
+        machine_creds = Credentials()
+        machine_creds.guess(lp)
+        machine_creds.set_machine_account(lp)
+        gensec_server.set_credentials(machine_creds)
+
+        gensec_server.want_feature(gensec.FEATURE_SEAL)
+        gensec_server.start_mech_by_sasl_name("GSSAPI")
+
+        client_finished = False
+        server_finished = False
+        server_to_client = ""
+
+        # Run the actual call loop.
+        while client_finished == False and server_finished == False:
+            if not client_finished:
+                print "running client gensec_update"
+                (client_finished, client_to_server) = gensec_client.update(server_to_client)
+            if not server_finished:
+                print "running server gensec_update"
+                (server_finished, server_to_client) = gensec_server.update(client_to_server)
+
+        session = gensec_server.session_info()
+
+        token = session.security_token
+        pac_sids = []
+        for s in token.sids:
+            pac_sids.append(str(s))
+
+        sidset1 = set(pac_sids)
+        sidset2 = set(self.user_sids)
+        if len(sidset1.difference(sidset2)):
+            print("token sids don't match")
             print("difference : %s" % sidset1.difference(sidset2))
             self.fail(msg="calculated groups don't match against user PAC tokenGroups")
 
+
+    def test_tokenGroups_manual(self):
+        # Manually run the tokenGroups algorithm from MS-ADTS 3.1.1.4.5.19 and MS-DRSR 4.1.8.3
+        # and compare the result
+        res = self.admin_ldb.search(base=self.base_dn, scope=ldb.SCOPE_SUBTREE,
+                                    expression="(|(objectclass=user)(objectclass=group))",
+                                    attrs=["memberOf"])
+        aSet = set()
+        aSetR = set()
+        vSet = set()
+        for obj in res:
+            if "memberOf" in obj:
+                for dn in obj["memberOf"]:
+                    first = obj.dn.get_casefold()
+                    second = ldb.Dn(self.admin_ldb, dn).get_casefold()
+                    aSet.add((first, second))
+                    aSetR.add((second, first))
+                    vSet.add(first)
+                    vSet.add(second)
+
+        res = self.admin_ldb.search(base=self.base_dn, scope=ldb.SCOPE_SUBTREE,
+                                    expression="(objectclass=user)",
+                                    attrs=["primaryGroupID"])
+        for obj in res:
+            if "primaryGroupID" in obj:
+                sid = "%s-%d" % (self.admin_ldb.get_domain_sid(), int(obj["primaryGroupID"][0]))
+                res2 = self.admin_ldb.search(base="<SID=%s>" % sid, scope=ldb.SCOPE_BASE,
+                                             attrs=[])
+                first = obj.dn.get_casefold()
+                second = res2[0].dn.get_casefold()
+
+                aSet.add((first, second))
+                aSetR.add((second, first))
+                vSet.add(first)
+                vSet.add(second)
+
+        wSet = set()
+        wSet.add(self.test_user_dn.get_casefold())
+        closure(vSet, wSet, aSet)
+        wSet.remove(self.test_user_dn.get_casefold())
+
+        tokenGroupsSet = set()
+
+        res = self.ldb.search(self.user_sid_dn, scope=ldb.SCOPE_BASE, attrs=["tokenGroups"])
+        self.assertEquals(len(res), 1)
+
+        dn_tokengroups = []
+        for sid in res[0]['tokenGroups']:
+            sid = ndr_unpack(samba.dcerpc.security.dom_sid, sid)
+            res3 = self.admin_ldb.search(base="<SID=%s>" % sid, scope=ldb.SCOPE_BASE,
+                                         attrs=[])
+            tokenGroupsSet.add(res3[0].dn.get_casefold())
+
+        if len(wSet.difference(tokenGroupsSet)):
+            self.fail(msg="additional calculated: %s" % wSet.difference(tokenGroupsSet))
+
+        if len(tokenGroupsSet.difference(wSet)):
+            self.fail(msg="additional tokenGroups: %s" % tokenGroupsSet.difference(wSet))
+
+
+    def filtered_closure(self, wSet, filter_grouptype):
+        res = self.admin_ldb.search(base=self.base_dn, scope=ldb.SCOPE_SUBTREE,
+                                    expression="(|(objectclass=user)(objectclass=group))",
+                                    attrs=["memberOf"])
+        aSet = set()
+        aSetR = set()
+        vSet = set()
+        for obj in res:
+            vSet.add(obj.dn.get_casefold())
+            if "memberOf" in obj:
+                for dn in obj["memberOf"]:
+                    first = obj.dn.get_casefold()
+                    second = ldb.Dn(self.admin_ldb, dn).get_casefold()
+                    aSet.add((first, second))
+                    aSetR.add((second, first))
+                    vSet.add(first)
+                    vSet.add(second)
+
+        res = self.admin_ldb.search(base=self.base_dn, scope=ldb.SCOPE_SUBTREE,
+                                    expression="(objectclass=user)",
+                                    attrs=["primaryGroupID"])
+        for obj in res:
+            if "primaryGroupID" in obj:
+                sid = "%s-%d" % (self.admin_ldb.get_domain_sid(), int(obj["primaryGroupID"][0]))
+                res2 = self.admin_ldb.search(base="<SID=%s>" % sid, scope=ldb.SCOPE_BASE,
+                                             attrs=[])
+                first = obj.dn.get_casefold()
+                second = res2[0].dn.get_casefold()
+
+                aSet.add((first, second))
+                aSetR.add((second, first))
+                vSet.add(first)
+                vSet.add(second)
+
+        uSet = set()
+        for v in vSet:
+            res_group = self.admin_ldb.search(base=v, scope=ldb.SCOPE_BASE,
+                                              attrs=["groupType"],
+                                              expression="objectClass=group")
+            if len(res_group) == 1:
+                if hex(int(res_group[0]["groupType"][0]) & 0x00000000FFFFFFFF) == hex(filter_grouptype):
+                    uSet.add(v)
+            else:
+                uSet.add(v)
+
+        closure(uSet, wSet, aSet)
+
+
+    def test_tokenGroupsGlobalAndUniversal_manual(self):
+        # Manually run the tokenGroups algorithm from MS-ADTS 3.1.1.4.5.19 and MS-DRSR 4.1.8.3
+        # and compare the result
+
+        # The variable names come from MS-ADTS May 15, 2014
+
+        S = set()
+        S.add(self.test_user_dn.get_casefold())
+
+        self.filtered_closure(S, GTYPE_SECURITY_GLOBAL_GROUP)
+
+        T = set()
+        # Not really a SID, we do this on DNs...
+        for sid in S:
+            X = set()
+            X.add(sid)
+            self.filtered_closure(X, GTYPE_SECURITY_UNIVERSAL_GROUP)
+
+            T = T.union(X)
+
+        T.remove(self.test_user_dn.get_casefold())
+
+        tokenGroupsSet = set()
+
+        res = self.ldb.search(self.user_sid_dn, scope=ldb.SCOPE_BASE, attrs=["tokenGroupsGlobalAndUniversal"])
+        self.assertEquals(len(res), 1)
+
+        dn_tokengroups = []
+        for sid in res[0]['tokenGroupsGlobalAndUniversal']:
+            sid = ndr_unpack(samba.dcerpc.security.dom_sid, sid)
+            res3 = self.admin_ldb.search(base="<SID=%s>" % sid, scope=ldb.SCOPE_BASE,
+                                         attrs=[])
+            tokenGroupsSet.add(res3[0].dn.get_casefold())
+
+        if len(T.difference(tokenGroupsSet)):
+            self.fail(msg="additional calculated: %s" % T.difference(tokenGroupsSet))
+
+        if len(tokenGroupsSet.difference(T)):
+            self.fail(msg="additional tokenGroupsGlobalAndUniversal: %s" % tokenGroupsSet.difference(T))
 
 if not "://" in url:
@@ -168 +495 @@
         url = "ldap://%s" % url
 
-samdb = SamDB(url, credentials=creds, session_info=system_session(lp), lp=lp)
-
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(TokenTest)).wasSuccessful():
-    rc = 1
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)

vendor/current/source4/dsdb/tests/python/urgent_replication.py

@@ -4 +4 @@
 import optparse
 import sys
-import os
-
 sys.path.insert(0, "bin/python")
 import samba
-samba.ensure_external_module("testtools", "testtools")
-samba.ensure_external_module("subunit", "subunit/python")
-
-import samba.getopt as options
-
-from samba.auth import system_session
+from samba.tests.subunitrun import TestProgram, SubunitOptions
+
 from ldb import (LdbError, ERR_NO_SUCH_OBJECT, Message,
     MessageElement, Dn, FLAG_MOD_REPLACE)
-from samba.samdb import SamDB
 import samba.tests
 import samba.dsdb as dsdb
-
-from subunit.run import SubunitTestRunner
-import unittest
+import samba.getopt as options
 
 parser = optparse.OptionParser("urgent_replication.py [options] <host>")
@@ -27 +18 @@
 parser.add_option_group(sambaopts)
 parser.add_option_group(options.VersionOptions(parser))
+
 # use command line creds if available
 credopts = options.CredentialsOptions(parser)
 parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
 opts, args = parser.parse_args()
 
@@ -38 +32 @@
 host = args[0]
 
-lp = sambaopts.get_loadparm()
-creds = credopts.get_credentials(lp)
 
 class UrgentReplicationTests(samba.tests.TestCase):
@@ -51 +43 @@
     def setUp(self):
         super(UrgentReplicationTests, self).setUp()
-        self.ldb = ldb
-        self.base_dn = ldb.domain_dn()
+        self.ldb = samba.tests.connect_samdb(host, global_schema=False)
+        self.base_dn = self.ldb.domain_dn()
 
         print "baseDN: %s\n" % self.base_dn
 
     def test_nonurgent_object(self):
-        """Test if the urgent replication is not activated
-           when handling a non urgent object"""
+        """Test if the urgent replication is not activated when handling a non urgent object."""
         self.ldb.add({
             "dn": "cn=nonurgenttest,cn=users," + self.base_dn,
@@ -65 +56 @@
             "description":"nonurgenttest description"})
 
-        # urgent replication should not be enabled when creating 
+        # urgent replication should not be enabled when creating
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -71 +62 @@
         # urgent replication should not be enabled when modifying
         m = Message()
-        m.dn = Dn(ldb, "cn=nonurgenttest,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=nonurgenttest,cn=users," + self.base_dn)
         m["description"] = MessageElement("new description", FLAG_MOD_REPLACE,
           "description")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -83 +74 @@
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
 
-
     def test_nTDSDSA_object(self):
-        '''Test if the urgent replication is activated
-           when handling a nTDSDSA object'''
-        self.ldb.add({
-            "dn": "cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,cn=Configuration," + self.base_dn,
+        """Test if the urgent replication is activated when handling a nTDSDSA object."""
+        self.ldb.add({
+            "dn": "cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,%s" %
+                self.ldb.get_config_basedn(),
             "objectclass":"server",
             "cn":"test server",
@@ -108 +98 @@
         # urgent replication should NOT be enabled when modifying
         m = Message()
-        m.dn = Dn(ldb, "cn=NTDS Settings test,cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,cn=Configuration," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=NTDS Settings test,cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,cn=Configuration," + self.base_dn)
         m["options"] = MessageElement("0", FLAG_MOD_REPLACE,
           "options")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn("cn=Configuration," + self.base_dn)
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -122 +112 @@
         self.delete_force(self.ldb, "cn=test server,cn=Servers,cn=Default-First-Site-Name,cn=Sites,cn=Configuration," + self.base_dn)
 
-
     def test_crossRef_object(self):
-        '''Test if the urgent replication is activated
-           when handling a crossRef object'''
+        """Test if the urgent replication is activated when handling a crossRef object."""
         self.ldb.add({
                       "dn": "CN=test crossRef,CN=Partitions,CN=Configuration,"+ self.base_dn,
                       "objectClass": "crossRef",
                       "cn": "test crossRef",
-                      "dnsRoot": lp.get("realm").lower(),
+                      "dnsRoot": self.get_loadparm().get("realm").lower(),
                       "instanceType": "4",
                       "nCName": self.base_dn,
@@ -143 +131 @@
         # urgent replication should NOT be enabled when modifying
         m = Message()
-        m.dn = Dn(ldb, "cn=test crossRef,CN=Partitions,CN=Configuration," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=test crossRef,CN=Partitions,CN=Configuration," + self.base_dn)
         m["systemFlags"] = MessageElement("0", FLAG_MOD_REPLACE,
           "systemFlags")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn("cn=Configuration," + self.base_dn)
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -156 +144 @@
         self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
 
-
-
     def test_attributeSchema_object(self):
-        '''Test if the urgent replication is activated
-           when handling an attributeSchema object'''
+        """Test if the urgent replication is activated when handling an attributeSchema object"""
 
         try:
@@ -187 +172 @@
             print "Not testing urgent replication when creating attributeSchema object ...\n"
 
-        # urgent replication should be enabled when modifying 
-        m = Message()
-        m.dn = Dn(ldb, "CN=test attributeSchema,CN=Schema,CN=Configuration," + self.base_dn)
+        # urgent replication should be enabled when modifying
+        m = Message()
+        m.dn = Dn(self.ldb, "CN=test attributeSchema,CN=Schema,CN=Configuration," + self.base_dn)
         m["lDAPDisplayName"] = MessageElement("updated test attributeSchema", FLAG_MOD_REPLACE,
           "lDAPDisplayName")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn("cn=Schema,cn=Configuration," + self.base_dn)
         self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
 
-
     def test_classSchema_object(self):
-        '''Test if the urgent replication is activated
-           when handling a classSchema object'''
+        """Test if the urgent replication is activated when handling a classSchema object."""
         try:
             self.ldb.add_ldif(
@@ -232 +215 @@
         # urgent replication should be enabled when modifying 
         m = Message()
-        m.dn = Dn(ldb, "CN=test classSchema,CN=Schema,CN=Configuration," + self.base_dn)
+        m.dn = Dn(self.ldb, "CN=test classSchema,CN=Schema,CN=Configuration," + self.base_dn)
         m["lDAPDisplayName"] = MessageElement("updated test classSchema", FLAG_MOD_REPLACE,
           "lDAPDisplayName")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn("cn=Schema,cn=Configuration," + self.base_dn)
         self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
 
-
     def test_secret_object(self):
-        '''Test if the urgent replication is activated
-           when handling a secret object'''
+        """Test if the urgent replication is activated when handling a secret object."""
 
         self.ldb.add({
@@ -257 +238 @@
         # urgent replication should be enabled when modifying
         m = Message()
-        m.dn = Dn(ldb, "cn=test secret,cn=System," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=test secret,cn=System," + self.base_dn)
         m["currentValue"] = MessageElement("yyyyyyyy", FLAG_MOD_REPLACE,
           "currentValue")
-        ldb.modify(m)
-        res = self.ldb.load_partition_usn(self.base_dn)
-        self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
-
-        # urgent replication should NOT be enabled when deleting 
+        self.ldb.modify(m)
+        res = self.ldb.load_partition_usn(self.base_dn)
+        self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
+
+        # urgent replication should NOT be enabled when deleting
         self.delete_force(self.ldb, "cn=test secret,cn=System," + self.base_dn)
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
 
-
     def test_rIDManager_object(self):
-        '''Test if the urgent replication is activated
-            when handling a rIDManager object'''
+        """Test if the urgent replication is activated when handling a rIDManager object."""
         self.ldb.add_ldif(
             """dn: CN=RID Manager test,CN=System,%s""" % self.base_dn + """
@@ -290 +269 @@
         # urgent replication should be enabled when modifying
         m = Message()
-        m.dn = Dn(ldb, "CN=RID Manager test,CN=System," + self.base_dn)
+        m.dn = Dn(self.ldb, "CN=RID Manager test,CN=System," + self.base_dn)
         m["systemFlags"] = MessageElement("0", FLAG_MOD_REPLACE,
           "systemFlags")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -302 +281 @@
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
 
-
     def test_urgent_attributes(self):
-        '''Test if the urgent replication is activated
-            when handling urgent attributes of an object'''
+        """Test if the urgent replication is activated when handling urgent attributes of an object."""
 
         self.ldb.add({
@@ -320 +297 @@
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
 
-        # urgent replication should be enabled when modifying userAccountControl 
-        m = Message()
-        m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
+        # urgent replication should be enabled when modifying userAccountControl
+        m = Message()
+        m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
         m["userAccountControl"] = MessageElement(str(dsdb.UF_NORMAL_ACCOUNT+dsdb.UF_SMARTCARD_REQUIRED), FLAG_MOD_REPLACE,
           "userAccountControl")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -331 +308 @@
         # urgent replication should be enabled when modifying lockoutTime
         m = Message()
-        m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
         m["lockoutTime"] = MessageElement("1", FLAG_MOD_REPLACE,
           "lockoutTime")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -340 +317 @@
         # urgent replication should be enabled when modifying pwdLastSet
         m = Message()
-        m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
         m["pwdLastSet"] = MessageElement("1", FLAG_MOD_REPLACE,
           "pwdLastSet")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -350 +327 @@
         # attribute
         m = Message()
-        m.dn = Dn(ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
+        m.dn = Dn(self.ldb, "cn=user UrgAttr test,cn=users," + self.base_dn)
         m["description"] = MessageElement("updated urgent attributes test description",
                                           FLAG_MOD_REPLACE, "description")
-        ldb.modify(m)
+        self.ldb.modify(m)
         res = self.ldb.load_partition_usn(self.base_dn)
         self.assertNotEquals(res["uSNHighest"], res["uSNUrgent"])
@@ -363 +340 @@
 
 
-if not "://" in host:
-    if os.path.isfile(host):
-        host = "tdb://%s" % host
-    else:
-        host = "ldap://%s" % host
-
-
-ldb = SamDB(host, credentials=creds, session_info=system_session(lp), lp=lp,
-            global_schema=False)
-
-runner = SubunitTestRunner()
-rc = 0
-if not runner.run(unittest.makeSuite(UrgentReplicationTests)).wasSuccessful():
-    rc = 1
-sys.exit(rc)
+TestProgram(module=__name__, opts=subunitopts)