dsdb tests: Add member/memberOf checking to delete_objects testing

Pair-Programmed-With: Stefan Metzmacher <metze@samba.org>

Signed-off-by: Andrew Bartlett <abartlet@samba.org>
Signed-off-by: Stefan Metzmacher <metze@samba.org>

1 files changed, 257 insertions, 21 deletions

diff --git a/source4/torture/drs/python/delete_object.py b/source4/torture/drs/python/delete_object.py
index f36232e0a5..3e20b6d069 100644
--- a/source4/torture/drs/python/delete_object.py
+++ b/source4/torture/drs/python/delete_object.py
@@ -30,10 +30,10 @@ import time
 
 
 from ldb import (
-    SCOPE_SUBTREE
+    SCOPE_SUBTREE,
     )
 
-import drs_base
+import drs_base, ldb
 
 
 class DrsDeleteObjectTestCase(drs_base.DrsBaseTestCase):
@@ -55,9 +55,10 @@ class DrsDeleteObjectTestCase(drs_base.DrsBaseTestCase):
     def _make_username(self):
         return "DrsDelObjUser_" + time.strftime("%s", time.gmtime())
 
-    def _check_user(self, sam_ldb, user_orig, is_deleted):
+    # now also used to check the group
+    def _check_obj(self, sam_ldb, obj_orig, is_deleted):
         # search the user by guid as it may be deleted
-        guid_str = self._GUID_string(user_orig["objectGUID"][0])
+        guid_str = self._GUID_string(obj_orig["objectGUID"][0])
         expression = "(objectGUID=%s)" % guid_str
         res = sam_ldb.search(base=self.domain_dn,
                              expression=expression,
@@ -67,29 +68,38 @@ class DrsDeleteObjectTestCase(drs_base.DrsBaseTestCase):
         # Deleted Object base DN
         dodn = self._deleted_objects_dn(sam_ldb)
         # now check properties of the user
-        name_orig = user_orig["cn"][0]
+        name_orig = obj_orig["cn"][0]
         name_cur  = user_cur["cn"][0]
         if is_deleted:
             self.assertEquals(user_cur["isDeleted"][0],"TRUE")
-            self.assertTrue(not("objectCategory" in user_cur))
-            self.assertTrue(not("sAMAccountType" in user_cur))
+            self.assertFalse("objectCategory" in user_cur)
+            self.assertFalse("sAMAccountType" in user_cur)
+            self.assertFalse("description" in user_cur)
+            self.assertFalse("memberOf" in user_cur)
+            self.assertFalse("member" in user_cur)
             self.assertTrue(dodn in str(user_cur["dn"]),
                             "User %s is deleted but it is not located under %s (found at %s)!" % (name_orig, dodn, user_cur["dn"]))
             self.assertEquals(name_cur, name_orig + "\nDEL:" + guid_str)
         else:
-            self.assertTrue(not("isDeleted" in user_cur))
+            self.assertFalse("isDeleted" in user_cur)
             self.assertEquals(name_cur, name_orig)
-            self.assertEquals(user_orig["dn"], user_cur["dn"])
+            self.assertEquals(obj_orig["dn"], user_cur["dn"])
             self.assertTrue(dodn not in str(user_cur["dn"]))
+        return user_cur
 
-    def test_ReplicateDeteleteObject(self):
+    def test_ReplicateDeletedObject1(self):
         """Verifies how a deleted-object is replicated between two DCs.
            This test should verify that:
             - deleted-object is replicated properly
-           TODO: We should verify that after replication,
-                 object's state to conform to a deleted-object state
-                 or tombstone -object, depending on DC's features
-                 It will also be great if check replPropertyMetaData."""
+            - We verify that after replication,
+              object's state to conform to a tombstone-object state
+            - This test replicates the object modifications to
+              the server with the user deleted first
+
+           TODO:  It will also be great if check replPropertyMetaData.
+           TODO:  Check for deleted-object state, depending on DC's features
+                  when recycle-bin is enabled
+           """
         # work-out unique username to test with
         username = self._make_username()
 
@@ -104,7 +114,7 @@ class DrsDeleteObjectTestCase(drs_base.DrsBaseTestCase):
 
         # check user info on DC1
         print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
-        self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=False)
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
 
         # trigger replication from DC1 to DC2
         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
@@ -112,23 +122,249 @@ class DrsDeleteObjectTestCase(drs_base.DrsBaseTestCase):
         # delete user on DC1
         self.ldb_dc1.delete(user_dn)
         # check user info on DC1 - should be deleted
-        self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
+        # check user info on DC2 - should be valid user
+        user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
+
+        # The user should not have a description or memberOf yet
+        self.assertFalse("description" in user_cur)
+        self.assertFalse("memberOf" in user_cur)
+
+        self.ldb_dc2.newgroup("group_%s" % username)
+
+        self.ldb_dc2.newgroup("group2_%s" % username)
+
+        ldb_res = self.ldb_dc2.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+        self.assertTrue("sAMAccountName" in ldb_res[0])
+        group_orig = ldb_res[0]
+        group_dn = ldb_res[0]["dn"]
+
+        # modify user on DC2 to have a description and be a member of the group
+        m = ldb.Message()
+        m.dn = user_dn
+        m["description"] = ldb.MessageElement("a description",
+                                              ldb.FLAG_MOD_ADD, "description")
+        self.ldb_dc2.modify(m)
+        m = ldb.Message()
+        m.dn = group_dn
+        m["member"] = ldb.MessageElement(str(user_dn),
+                                           ldb.FLAG_MOD_ADD, "member")
+        self.ldb_dc2.modify(m)
+
+        ldb_res = self.ldb_dc2.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group2_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+        self.assertTrue("sAMAccountName" in ldb_res[0])
+        group2_dn = ldb_res[0]["dn"]
+        group2_orig = ldb_res[0]
+
+        m = ldb.Message()
+        m.dn = group2_dn
+        m["member"] = ldb.MessageElement(str(group_dn),
+                                         ldb.FLAG_MOD_ADD, "member")
+        self.ldb_dc2.modify(m)
+
         # check user info on DC2 - should be valid user
-        self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=False)
+        user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
+
+        # The user should not have a description yet
+        self.assertTrue("description" in user_cur)
+        self.assertTrue("memberOf" in user_cur)
+
+        ldb_res = self.ldb_dc2.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+
+        # This group is a member of another group
+        self.assertTrue("memberOf" in ldb_res[0])
+
+        # The user was deleted on DC1, but check the modify we just did on DC2
+        self.assertTrue("member" in ldb_res[0])
 
         # trigger replication from DC2 to DC1
         # to check if deleted object gets restored
         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
         # check user info on DC1 - should be deleted
-        self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
         # check user info on DC2 - should be valid user
-        self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=False)
+        self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
+
+        ldb_res = self.ldb_dc1.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+
+        # This group is a member of another group
+        self.assertTrue("memberOf" in ldb_res[0])
+
+        # The user was deleted on DC1, but the modify we did on DC2, check it never replicated in
+        self.assertFalse("member" in ldb_res[0])
 
         # trigger replication from DC1 to DC2
         # to check if deleted object is replicated
         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
         # check user info on DC1 - should be deleted
-        self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
         # check user info on DC2 - should be deleted
-        self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=True)
+        self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=True)
+
+        # delete group on DC1
+        self.ldb_dc1.delete(group_dn)
+
+        # trigger replication from DC1 to DC2
+        # to check if deleted object is replicated
+        self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
+
+        # check group info on DC1 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=group_orig, is_deleted=True)
+        # check group info on DC2 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=group_orig, is_deleted=True)
+
+        ldb_res = self.ldb_dc2.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group2_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+        self.assertFalse("member" in ldb_res[0])
+
+        # delete group on DC1
+        self.ldb_dc1.delete(group2_dn)
+
+        # trigger replication from DC1 to DC2
+        # to check if deleted object is replicated
+        self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
+
+        # check group info on DC1 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=group2_orig, is_deleted=True)
+        # check group info on DC2 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=group2_orig, is_deleted=True)
+
+    def test_ReplicateDeletedObject2(self):
+        """Verifies how a deleted-object is replicated between two DCs.
+           This test should verify that:
+            - deleted-object is replicated properly
+            - We verify that after replication,
+              object's state to conform to a tombstone-object state
+            - This test replicates the delete to the server with the
+              object modifications first
+
+           TODO:  It will also be great if check replPropertyMetaData.
+           TODO:  Check for deleted-object state, depending on DC's features
+                  when recycle-bin is enabled
+           """
+        # work-out unique username to test with
+        username = self._make_username()
+
+        # create user on DC1
+        self.ldb_dc1.newuser(username=username, password="P@sswOrd!")
+        ldb_res = self.ldb_dc1.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=%s)" % username)
+        self.assertEquals(len(ldb_res), 1)
+        user_orig = ldb_res[0]
+        user_dn   = ldb_res[0]["dn"]
+
+        # check user info on DC1
+        print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
+
+        # trigger replication from DC1 to DC2
+        self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
+
+        # delete user on DC1
+        self.ldb_dc1.delete(user_dn)
+        # check user info on DC1 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
+        # check user info on DC2 - should be valid user
+        user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
+
+        # The user should not have a description or memberOf yet
+        self.assertFalse("description" in user_cur)
+        self.assertFalse("memberOf" in user_cur)
+
+        self.ldb_dc2.newgroup("group_%s" % username)
+
+        self.ldb_dc2.newgroup("group2_%s" % username)
+
+        ldb_res = self.ldb_dc2.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+        self.assertTrue("sAMAccountName" in ldb_res[0])
+        group_orig = ldb_res[0]
+        group_dn = ldb_res[0]["dn"]
+
+        # modify user on DC2 to have a description and be a member of the group
+        m = ldb.Message()
+        m.dn = user_dn
+        m["description"] = ldb.MessageElement("a description",
+                                              ldb.FLAG_MOD_ADD, "description")
+        self.ldb_dc2.modify(m)
+        m = ldb.Message()
+        m.dn = group_dn
+        m["member"] = ldb.MessageElement(str(user_dn),
+                                           ldb.FLAG_MOD_ADD, "member")
+        self.ldb_dc2.modify(m)
+
+        ldb_res = self.ldb_dc2.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group2_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+        self.assertTrue("sAMAccountName" in ldb_res[0])
+        group2_dn = ldb_res[0]["dn"]
+        group2_orig = ldb_res[0]
+
+        m = ldb.Message()
+        m.dn = group2_dn
+        m["member"] = ldb.MessageElement(str(group_dn),
+                                         ldb.FLAG_MOD_ADD, "member")
+        self.ldb_dc2.modify(m)
+
+        # check user info on DC2 - should be valid user
+        user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
+
+        # The user should not have a description yet
+        self.assertTrue("description" in user_cur)
+        self.assertTrue("memberOf" in user_cur)
+
+        # trigger replication from DC1 to DC2
+        # to check if deleted object gets restored
+        self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
+        # check user info on DC1 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
+        # check user info on DC2 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=True)
+
+        ldb_res = self.ldb_dc2.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+        self.assertTrue("memberOf" in ldb_res[0])
+        self.assertFalse("member" in ldb_res[0])
+
+        # trigger replication from DC2 to DC1
+        # to check if deleted object is replicated
+        self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
+        # check user info on DC1 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
+        # check user info on DC2 - should be deleted
+        self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=True)
+
+        ldb_res = self.ldb_dc1.search(base=self.domain_dn,
+                                      scope=SCOPE_SUBTREE,
+                                      expression="(samAccountName=group_%s)" % username)
+        self.assertTrue(len(ldb_res) == 1)
+        self.assertTrue("memberOf" in ldb_res[0])
+        self.assertFalse("member" in ldb_res[0])
+
+        # delete group on DC1
+        self.ldb_dc1.delete(group_dn)
+        self.ldb_dc1.delete(group2_dn)
+
+        # trigger replication from DC1 to DC2, for cleanup
+        self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)