summaryrefslogtreecommitdiff
path: root/source4/torture/drs/python
diff options
context:
space:
mode:
Diffstat (limited to 'source4/torture/drs/python')
-rw-r--r--source4/torture/drs/python/delete_object.py278
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)