selftest: Add test for a re-animated object conflict

Added a test to simulate a user accidentally being deleted and 2
different admins trying to resolve the problem simultaneously - one by
re-animating the object and one by just creating a new object with
the same name.

Currently this test fails on Samba because it chooses the higher
version
number as the winner instead of the latest change.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13039

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
Reviewed-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>

1 files changed, 63 insertions, 0 deletions

diff --git a/source4/torture/drs/python/replica_sync.py b/source4/torture/drs/python/replica_sync.py
index c8944b0f3ca..93407df3962 100644
--- a/source4/torture/drs/python/replica_sync.py
+++ b/source4/torture/drs/python/replica_sync.py
@@ -30,6 +30,7 @@
 import drs_base
 import samba.tests
 import time
+import ldb
 
 from ldb import (
     SCOPE_BASE, LdbError, ERR_NO_SUCH_OBJECT)
@@ -673,3 +674,65 @@ objectClass: organizationalUnit
         self._check_deleted(self.ldb_dc2, ou1_child)
         self._check_deleted(self.ldb_dc2, ou2_child)
         self._check_deleted(self.ldb_dc2, ou3_child)
+
+    def reanimate_object(self, samdb, guid, new_dn):
+        """Re-animates a deleted object"""
+        res = samdb.search(base="<GUID=%s>" % guid, attrs=["isDeleted"],
+                           controls=['show_deleted:1'], scope=SCOPE_BASE)
+        if len(res) != 1:
+            return
+
+        msg = ldb.Message()
+        msg.dn = res[0].dn
+        msg["isDeleted"] = ldb.MessageElement([], ldb.FLAG_MOD_DELETE, "isDeleted")
+        msg["distinguishedName"] = ldb.MessageElement([new_dn], ldb.FLAG_MOD_REPLACE, "distinguishedName")
+        samdb.modify(msg, ["show_deleted:1"])
+
+    def test_ReplReanimationConflict(self):
+        """
+        Checks that if a reanimated object conflicts with a new object, then
+        the conflict is resolved correctly.
+        """
+
+        self._disable_inbound_repl(self.dnsname_dc1)
+        self._disable_inbound_repl(self.dnsname_dc2)
+
+        # create an object, "accidentally" delete it, and replicate the changes to both DCs
+        self.ou1 = self._create_ou(self.ldb_dc2, "OU=Conflict object")
+        self.ldb_dc2.delete('<GUID=%s>' % self.ou1)
+        self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True, full_sync=False)
+
+        # Now pretend that the admin for one DC resolves the problem by
+        # re-animating the object...
+        self.reanimate_object(self.ldb_dc1, self.ou1, "OU=Conflict object,%s" % self.domain_dn)
+
+        # ...whereas another admin just creates a user with the same name
+        # again on a different DC
+        time.sleep(1)
+        self.ou2 = self._create_ou(self.ldb_dc2, "OU=Conflict object")
+
+        # Now sync the DCs to resolve the conflict
+        self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True, full_sync=False)
+
+        # Check the latest change won and SELF.OU1 was made into a conflict
+        res1 = self.ldb_dc1.search(base="<GUID=%s>" % self.ou1,
+                                  scope=SCOPE_BASE, attrs=["name"])
+        res2 = self.ldb_dc1.search(base="<GUID=%s>" % self.ou2,
+                                  scope=SCOPE_BASE, attrs=["name"])
+        print res1[0]["name"][0]
+        print res2[0]["name"][0]
+        self.assertTrue('CNF:%s' % self.ou1 in str(res1[0]["name"][0]))
+        self.assertFalse('CNF:%s' % self.ou2 in str(res2[0]["name"][0]))
+
+        # Delete both objects by GUID on DC1
+        self.ldb_dc1.delete('<GUID=%s>' % self.ou1)
+        self.ldb_dc1.delete('<GUID=%s>' % self.ou2)
+
+        self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True, full_sync=False)
+
+        self._check_deleted(self.ldb_dc1, self.ou1)
+        self._check_deleted(self.ldb_dc1, self.ou2)
+        # Check deleted on DC2
+        self._check_deleted(self.ldb_dc2, self.ou1)
+        self._check_deleted(self.ldb_dc2, self.ou2)
+