selftest: Add conflict test where the single-valued link already exists

As well as testing scenarios where both variants of the link are new, we
should also check the case where the received link already exists on the
DC as an inactive (i.e. previously deleted) link.

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

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
Reviewed-by: Garming Sam <garming@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>

1 files changed, 59 insertions, 0 deletions

diff --git a/source4/torture/drs/python/link_conflicts.py b/source4/torture/drs/python/link_conflicts.py
index 95e2950dd2f..c41d03689d1 100644
--- a/source4/torture/drs/python/link_conflicts.py
+++ b/source4/torture/drs/python/link_conflicts.py
@@ -645,3 +645,62 @@ class DrsReplicaLinkConflictTestCase(drs_base.DrsBaseTestCase):
         self._test_conflict_single_valued_link_deleted_loser(sync_order=DC1_TO_DC2)
         self._test_conflict_single_valued_link_deleted_loser(sync_order=DC2_TO_DC1)
 
+    def _test_conflict_existing_single_valued_link(self, sync_order):
+        """
+        Tests a single-valued link conflict, where the conflicting link value
+        already exists (as inactive) on both DCs.
+        """
+        # create the link objects
+        src_ou = self.unique_dn("OU=src")
+        src_guid = self.add_object(self.ldb_dc1, src_ou)
+
+        target1_ou = self.unique_dn("OU=target1")
+        target2_ou = self.unique_dn("OU=target2")
+        target1_guid = self.add_object(self.ldb_dc1, target1_ou)
+        target2_guid = self.add_object(self.ldb_dc1, target2_ou)
+
+        # add the links, but then delete them
+        self.add_link_attr(self.ldb_dc1, src_ou, "managedBy", target1_ou)
+        self.del_link_attr(self.ldb_dc1, src_ou, "managedBy", target1_ou)
+        self.add_link_attr(self.ldb_dc1, src_ou, "managedBy", target2_ou)
+        self.del_link_attr(self.ldb_dc1, src_ou, "managedBy", target2_ou)
+        self.sync_DCs()
+
+        # re-add the links independently on each DC
+        self.add_link_attr(self.ldb_dc1, src_ou, "managedBy", target1_ou)
+        self.ensure_unique_timestamp()
+        self.add_link_attr(self.ldb_dc2, src_ou, "managedBy", target2_ou)
+
+        # try to sync the 2 DCs (this currently fails)
+        try:
+            self.sync_DCs(sync_order=sync_order)
+        except Exception, e:
+            self.fail("Replication could not resolve link conflict: %s" % e)
+
+        res1 = self.ldb_dc1.search(base="<GUID=%s>" % src_guid,
+                                  scope=SCOPE_BASE, attrs=["managedBy"])
+        res2 = self.ldb_dc2.search(base="<GUID=%s>" % src_guid,
+                                  scope=SCOPE_BASE, attrs=["managedBy"])
+
+        # check the object has only have one occurence of the single-valued
+        # attribute and it matches on both DCs
+        self.assert_attrs_match(res1, res2, "managedBy", 1)
+
+        # here we expect DC2 to win because it has the more recent link
+        self.assertTrue(res1[0]["managedBy"][0] == target2_ou,
+                        "Expected most recent update to win conflict")
+
+        # we can't query the deleted links over LDAP, but we can check DRS
+        # to make sure the DC kept a copy of the conflicting link
+        link1 = AbstractLink(drsuapi.DRSUAPI_ATTID_managedBy, 0,
+                             misc.GUID(src_guid), misc.GUID(target1_guid))
+        link2 = AbstractLink(drsuapi.DRSUAPI_ATTID_managedBy,
+                             drsuapi.DRSUAPI_DS_LINKED_ATTRIBUTE_FLAG_ACTIVE,
+                             misc.GUID(src_guid), misc.GUID(target2_guid))
+        self._check_replicated_links(src_ou, [link1, link2])
+
+    def test_conflict_existing_single_valued_link(self):
+        # repeat the test twice, to give each DC a chance to resolve the conflict
+        self._test_conflict_existing_single_valued_link(sync_order=DC1_TO_DC2)
+        self._test_conflict_existing_single_valued_link(sync_order=DC2_TO_DC1)
+