From 01e941e3eadd7a208982d20c0ca9c104142f2b91 Mon Sep 17 00:00:00 2001

From: Mark Reynolds <mreynolds@redhat.com>

Date: Wed, 10 Aug 2022 08:58:28 -0400

Subject: [PATCH 4/4] Issue 3903 - fix repl keep alive event interval

Description:  Previously we passed the interval as seconds to the

              event queue, but it is supposed to be milliseconds.

              Fixed a crash with repl logging and decoding extended

              op payload (referrals).

              Also reworked alot of the replication CI tests that

              were flaky.

relates: https://github.com/389ds/389-ds-base/issues/3903

Reviewed by: tbordaz & spichugi(Thanks!)

---

 .../suites/replication/acceptance_test.py     |  52 +-

 .../cleanallruv_abort_certify_test.py         | 136 ++++

 .../cleanallruv_abort_restart_test.py         | 146 ++++

 .../replication/cleanallruv_abort_test.py     | 123 +++

 .../replication/cleanallruv_force_test.py     | 187 +++++

 .../cleanallruv_multiple_force_test.py        | 214 +++++

 .../replication/cleanallruv_restart_test.py   | 161 ++++

 .../cleanallruv_shutdown_crash_test.py        | 123 +++

 .../replication/cleanallruv_stress_test.py    | 216 +++++

 .../suites/replication/cleanallruv_test.py    | 742 +-----------------

 .../suites/replication/regression_m2_test.py  |  13 +-

 .../replication/regression_m2c2_test.py       |   1 +

 .../plugins/replication/repl5_replica.c       |  12 +-

 ldap/servers/plugins/replication/repl_extop.c |   4 +-

 ldap/servers/slapd/task.c                     |   8 +-

 src/lib389/lib389/instance/remove.py          |   6 +

 16 files changed, 1385 insertions(+), 759 deletions(-)

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_abort_certify_test.py

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_abort_restart_test.py

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_abort_test.py

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_force_test.py

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_multiple_force_test.py

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_restart_test.py

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_shutdown_crash_test.py

 create mode 100644 dirsrvtests/tests/suites/replication/cleanallruv_stress_test.py

diff --git a/dirsrvtests/tests/suites/replication/acceptance_test.py b/dirsrvtests/tests/suites/replication/acceptance_test.py

index a5f0c4c6b..863ee2553 100644

--- a/dirsrvtests/tests/suites/replication/acceptance_test.py

+++ b/dirsrvtests/tests/suites/replication/acceptance_test.py

@@ -8,6 +8,7 @@

 #

 import pytest

 import logging

+import time

 from lib389.replica import Replicas

 from lib389.tasks import *

 from lib389.utils import *

@@ -124,12 +125,16 @@ def test_modify_entry(topo_m4, create_entry):

         8. Some time should pass

         9. The change should be present on all suppliers

     """

+    if DEBUGGING:

+        sleep_time = 8

+    else:

+        sleep_time = 2

     log.info('Modifying entry {} - add operation'.format(TEST_ENTRY_DN))

     test_user = UserAccount(topo_m4.ms["supplier1"], TEST_ENTRY_DN)

     test_user.add('mail', '{}@redhat.com'.format(TEST_ENTRY_NAME))

-    time.sleep(1)

+    time.sleep(sleep_time)

     all_user = topo_m4.all_get_dsldapobject(TEST_ENTRY_DN, UserAccount)

     for u in all_user:

@@ -137,7 +142,7 @@ def test_modify_entry(topo_m4, create_entry):

     log.info('Modifying entry {} - replace operation'.format(TEST_ENTRY_DN))

     test_user.replace('mail', '{}@greenhat.com'.format(TEST_ENTRY_NAME))

-    time.sleep(1)

+    time.sleep(sleep_time)

     all_user = topo_m4.all_get_dsldapobject(TEST_ENTRY_DN, UserAccount)

     for u in all_user:

@@ -145,7 +150,7 @@ def test_modify_entry(topo_m4, create_entry):

     log.info('Modifying entry {} - delete operation'.format(TEST_ENTRY_DN))

     test_user.remove('mail', '{}@greenhat.com'.format(TEST_ENTRY_NAME))

-    time.sleep(1)

+    time.sleep(sleep_time)

     all_user = topo_m4.all_get_dsldapobject(TEST_ENTRY_DN, UserAccount)

     for u in all_user:

@@ -167,7 +172,10 @@ def test_delete_entry(topo_m4, create_entry):

     log.info('Deleting entry {} during the test'.format(TEST_ENTRY_DN))

     topo_m4.ms["supplier1"].delete_s(TEST_ENTRY_DN)

-

+    if DEBUGGING:

+        time.sleep(8)

+    else:

+        time.sleep(1)

     entries = get_repl_entries(topo_m4, TEST_ENTRY_NAME, ["uid"])

     assert not entries, "Entry deletion {} wasn't replicated successfully".format(TEST_ENTRY_DN)

@@ -231,6 +239,11 @@ def test_modrdn_after_pause(topo_m4):

         5. The change should be present on all suppliers

     """

+    if DEBUGGING:

+        sleep_time = 8

+    else:

+        sleep_time = 3

+

     newrdn_name = 'newrdn'

     newrdn_dn = 'uid={},{}'.format(newrdn_name, DEFAULT_SUFFIX)

@@ -264,7 +277,7 @@ def test_modrdn_after_pause(topo_m4):

     topo_m4.resume_all_replicas()

     log.info('Wait for replication to happen')

-    time.sleep(3)

+    time.sleep(sleep_time)

     try:

         entries_new = get_repl_entries(topo_m4, newrdn_name, ["uid"])

@@ -354,6 +367,11 @@ def test_many_attrs(topo_m4, create_entry):

     for add_name in add_list:

         test_user.add('description', add_name)

+    if DEBUGGING:

+        time.sleep(10)

+    else:

+        time.sleep(1)

+

     log.info('Check that everything was properly replicated after an add operation')

     entries = get_repl_entries(topo_m4, TEST_ENTRY_NAME, ["description"])

     for entry in entries:

@@ -363,6 +381,11 @@ def test_many_attrs(topo_m4, create_entry):

     for delete_name in delete_list:

         test_user.remove('description', delete_name)

+    if DEBUGGING:

+        time.sleep(10)

+    else:

+        time.sleep(1)

+

     log.info('Check that everything was properly replicated after a delete operation')

     entries = get_repl_entries(topo_m4, TEST_ENTRY_NAME, ["description"])

     for entry in entries:

@@ -386,12 +409,22 @@ def test_double_delete(topo_m4, create_entry):

     log.info('Deleting entry {} from supplier1'.format(TEST_ENTRY_DN))

     topo_m4.ms["supplier1"].delete_s(TEST_ENTRY_DN)

+    if DEBUGGING:

+        time.sleep(5)

+    else:

+        time.sleep(1)

+

     log.info('Deleting entry {} from supplier2'.format(TEST_ENTRY_DN))

     try:

         topo_m4.ms["supplier2"].delete_s(TEST_ENTRY_DN)

     except ldap.NO_SUCH_OBJECT:

         log.info("Entry {} wasn't found supplier2. It is expected.".format(TEST_ENTRY_DN))

+    if DEBUGGING:

+        time.sleep(5)

+    else:

+        time.sleep(1)

+        

     log.info('Make searches to check if server is alive')

     entries = get_repl_entries(topo_m4, TEST_ENTRY_NAME, ["uid"])

     assert not entries, "Entry deletion {} wasn't replicated successfully".format(TEST_ENTRY_DN)

@@ -436,6 +469,11 @@ def test_password_repl_error(topo_m4, create_entry):

     m3_conn = test_user_m3.bind(TEST_ENTRY_NEW_PASS)

     m4_conn = test_user_m4.bind(TEST_ENTRY_NEW_PASS)

+    if DEBUGGING:

+        time.sleep(5)

+    else:

+        time.sleep(1)

+

     log.info('Check the error log for the error with {}'.format(TEST_ENTRY_DN))

     assert not m2.ds_error_log.match('.*can.t add a change for uid={}.*'.format(TEST_ENTRY_NAME))

@@ -552,7 +590,7 @@ def test_csnpurge_large_valueset(topo_m2):

     replica = replicas.list()[0]

     log.info('nsds5ReplicaPurgeDelay to 5')

     replica.set('nsds5ReplicaPurgeDelay', '5')

-    time.sleep(6)

+    time.sleep(10)

     # add some new values to the valueset containing entries that should be purged

     for i in range(21,25):

@@ -612,7 +650,7 @@ def test_urp_trigger_substring_search(topo_m2):

             break

         else:

             log.info('Entry not yet replicated on M2, wait a bit')

-            time.sleep(2)

+            time.sleep(3)

     # check that M2 access logs does not "(&(objectclass=nstombstone)(nscpentrydn=uid=asterisk_*_in_value,dc=example,dc=com))"

     log.info('Check that on M2, URP as not triggered such internal search')

diff --git a/dirsrvtests/tests/suites/replication/cleanallruv_abort_certify_test.py b/dirsrvtests/tests/suites/replication/cleanallruv_abort_certify_test.py

new file mode 100644

index 000000000..603693b9e

--- /dev/null

+++ b/dirsrvtests/tests/suites/replication/cleanallruv_abort_certify_test.py

@@ -0,0 +1,136 @@

+# --- BEGIN COPYRIGHT BLOCK ---

+# Copyright (C) 2022 Red Hat, Inc.

+# All rights reserved.

+#

+# License: GPL (version 3 or any later version).

+# See LICENSE for details.

+# --- END COPYRIGHT BLOCK ---

+#

+import logging

+import pytest

+import os

+import time

+from lib389._constants import DEFAULT_SUFFIX

+from lib389.topologies import topology_m4

+from lib389.tasks import CleanAllRUVTask

+from lib389.replica import ReplicationManager, Replicas

+

+log = logging.getLogger(__name__)

+

+

+def remove_supplier4_agmts(msg, topology_m4):

+    """Remove all the repl agmts to supplier4. """

+

+    log.info('%s: remove all the agreements to supplier 4...' % msg)

+    repl = ReplicationManager(DEFAULT_SUFFIX)

+    # This will delete m4 from the topo *and* remove all incoming agreements

+    # to m4.

+    repl.remove_supplier(topology_m4.ms["supplier4"],

+        [topology_m4.ms["supplier1"], topology_m4.ms["supplier2"], topology_m4.ms["supplier3"]])

+

+def task_done(topology_m4, task_dn, timeout=60):

+    """Check if the task is complete"""

+

+    attrlist = ['nsTaskLog', 'nsTaskStatus', 'nsTaskExitCode',

+                'nsTaskCurrentItem', 'nsTaskTotalItems']

+    done = False

+    count = 0

+

+    while not done and count < timeout:

+        try:

+            entry = topology_m4.ms["supplier1"].getEntry(task_dn, attrlist=attrlist)

+            if entry is not None:

+                if entry.hasAttr('nsTaskExitCode'):

+                    done = True

+                    break

+            else:

+                done = True

+                break

+        except ldap.NO_SUCH_OBJECT:

+            done = True

+            break

+        except ldap.LDAPError:

+            break

+        time.sleep(1)

+        count += 1

+

+    return done

+

+@pytest.mark.flaky(max_runs=2, min_passes=1)

+def test_abort_certify(topology_m4):

+    """Test the abort task with a replica-certify-all option

+

+    :id: 78959966-d644-44a8-b98c-1fcf21b45eb0

+    :setup: Replication setup with four suppliers

+    :steps:

+        1. Disable replication on supplier 4

+        2. Remove agreements to supplier 4 from other suppliers

+        3. Stop supplier 2

+        4. Run a cleanallruv task on supplier 1

+        5. Run a cleanallruv abort task on supplier 1 with a replica-certify-all option

+    :expectedresults: No hanging tasks left

+        1. Replication on supplier 4 should be disabled

+        2. Agreements to supplier 4 should be removed

+        3. Supplier 2 should be stopped

+        4. Operation should be successful

+        5. Operation should be successful

+    """

+

+    log.info('Running test_abort_certify...')

+

+    # Remove the agreements from the other suppliers that point to supplier 4

+    repl = ReplicationManager(DEFAULT_SUFFIX)

+    m4rid = repl.get_rid(topology_m4.ms["supplier4"])

+    remove_supplier4_agmts("test_abort_certify", topology_m4)

+

+    # Stop supplier 2

+    log.info('test_abort_certify: stop supplier 2 to freeze the cleanAllRUV task...')

+    topology_m4.ms["supplier2"].stop()

+

+    # Run the task

+    log.info('test_abort_certify: add the cleanAllRUV task...')

+    cruv_task = CleanAllRUVTask(topology_m4.ms["supplier1"])

+    cruv_task.create(properties={

+        'replica-id': m4rid,

+        'replica-base-dn': DEFAULT_SUFFIX,

+        'replica-force-cleaning': 'no',

+        'replica-certify-all': 'yes'

+        })

+    # Wait a bit

+    time.sleep(2)

+

+    # Abort the task

+    log.info('test_abort_certify: abort the cleanAllRUV task...')

+    abort_task = cruv_task.abort(certify=True)

+

+    # Wait a while and make sure the abort task is still running

+    log.info('test_abort_certify...')

+

+    if task_done(topology_m4, abort_task.dn, 10):

+        log.fatal('test_abort_certify: abort task incorrectly finished')

+        assert False

+

+    # Now start supplier 2 so it can be aborted

+    log.info('test_abort_certify: start supplier 2 to allow the abort task to finish...')

+    topology_m4.ms["supplier2"].start()

+

+    # Wait for the abort task to stop

+    if not task_done(topology_m4, abort_task.dn, 90):

+        log.fatal('test_abort_certify: The abort CleanAllRUV task was not aborted')

+        assert False

+

+    # Check supplier 1 does not have the clean task running

+    log.info('test_abort_certify: check supplier 1 no longer has a cleanAllRUV task...')

+    if not task_done(topology_m4, cruv_task.dn):

+        log.fatal('test_abort_certify: CleanAllRUV task was not aborted')

+        assert False

+

+    log.info('test_abort_certify PASSED')

+

+

+if __name__ == '__main__':

+    # Run isolated

+    # -s for DEBUG mode

+    CURRENT_FILE = os.path.realpath(__file__)

+    pytest.main(["-s", CURRENT_FILE])

+

diff --git a/dirsrvtests/tests/suites/replication/cleanallruv_abort_restart_test.py b/dirsrvtests/tests/suites/replication/cleanallruv_abort_restart_test.py

new file mode 100644

index 000000000..1406c6553

--- /dev/null

+++ b/dirsrvtests/tests/suites/replication/cleanallruv_abort_restart_test.py

@@ -0,0 +1,146 @@

+# --- BEGIN COPYRIGHT BLOCK ---

+# Copyright (C) 2022 Red Hat, Inc.

+# All rights reserved.

+#

+# License: GPL (version 3 or any later version).

+# See LICENSE for details.

+# --- END COPYRIGHT BLOCK ---

+#

+import logging

+import pytest

+import os

+import time

+from lib389._constants import DEFAULT_SUFFIX

+from lib389.topologies import topology_m4

+from lib389.tasks import CleanAllRUVTask

+from lib389.replica import ReplicationManager

+

+log = logging.getLogger(__name__)

+

+

+def remove_supplier4_agmts(msg, topology_m4):

+    """Remove all the repl agmts to supplier4. """

+

+    log.info('%s: remove all the agreements to supplier 4...' % msg)

+    repl = ReplicationManager(DEFAULT_SUFFIX)

+    # This will delete m4 from the topo *and* remove all incoming agreements

+    # to m4.

+    repl.remove_supplier(topology_m4.ms["supplier4"],

+        [topology_m4.ms["supplier1"], topology_m4.ms["supplier2"], topology_m4.ms["supplier3"]])

+

+def task_done(topology_m4, task_dn, timeout=60):

+    """Check if the task is complete"""

+

+    attrlist = ['nsTaskLog', 'nsTaskStatus', 'nsTaskExitCode',

+                'nsTaskCurrentItem', 'nsTaskTotalItems']

+    done = False

+    count = 0

+

+    while not done and count < timeout:

+        try:

+            entry = topology_m4.ms["supplier1"].getEntry(task_dn, attrlist=attrlist)

+            if entry is not None:

+                if entry.hasAttr('nsTaskExitCode'):

+                    done = True

+                    break

+            else:

+                done = True

+                break

+        except ldap.NO_SUCH_OBJECT:

+            done = True

+            break

+        except ldap.LDAPError:

+            break

+        time.sleep(1)

+        count += 1

+

+    return done

+   

+

+@pytest.mark.flaky(max_runs=2, min_passes=1)

+def test_abort_restart(topology_m4):

+    """Test the abort task can handle a restart, and then resume

+

+    :id: b66e33d4-fe85-4e1c-b882-75da80f70ab3

+    :setup: Replication setup with four suppliers

+    :steps:

+        1. Disable replication on supplier 4

+        2. Remove agreements to supplier 4 from other suppliers

+        3. Stop supplier 3

+        4. Run a cleanallruv task on supplier 1

+        5. Run a cleanallruv abort task on supplier 1

+        6. Restart supplier 1

+        7. Make sure that no crash happened

+        8. Start supplier 3

+        9. Check supplier 1 does not have the clean task running

+        10. Check that errors log doesn't have 'Aborting abort task' message

+    :expectedresults:

+        1. Replication on supplier 4 should be disabled

+        2. Agreements to supplier 4 should be removed

+        3. Supplier 3 should be stopped

+        4. Operation should be successful

+        5. Operation should be successful

+        6. Supplier 1 should be restarted

+        7. No crash should happened

+        8. Supplier 3 should be started

+        9. Check supplier 1 shouldn't have the clean task running

+        10. Errors log shouldn't have 'Aborting abort task' message

+    """

+

+    log.info('Running test_abort_restart...')

+    # Remove the agreements from the other suppliers that point to supplier 4

+    repl = ReplicationManager(DEFAULT_SUFFIX)

+    m4rid = repl.get_rid(topology_m4.ms["supplier4"])

+    remove_supplier4_agmts("test_abort", topology_m4)

+

+    # Stop supplier 3

+    log.info('test_abort_restart: stop supplier 3 to freeze the cleanAllRUV task...')

+    topology_m4.ms["supplier3"].stop()

+

+    # Run the task

+    log.info('test_abort_restart: add the cleanAllRUV task...')

+    cruv_task = CleanAllRUVTask(topology_m4.ms["supplier1"])

+    cruv_task.create(properties={

+        'replica-id': m4rid,

+        'replica-base-dn': DEFAULT_SUFFIX,

+        'replica-force-cleaning': 'no',

+        'replica-certify-all': 'yes'

+        })

+    # Wait a bit

+    time.sleep(2)

+

+    # Abort the task

+    cruv_task.abort(certify=True)

+

+    # Check supplier 1 does not have the clean task running

+    log.info('test_abort_abort: check supplier 1 no longer has a cleanAllRUV task...')

+    if not task_done(topology_m4, cruv_task.dn):

+        log.fatal('test_abort_restart: CleanAllRUV task was not aborted')

+        assert False

+

+    # Now restart supplier 1, and make sure the abort process completes

+    topology_m4.ms["supplier1"].restart()

+    if topology_m4.ms["supplier1"].detectDisorderlyShutdown():

+        log.fatal('test_abort_restart: Supplier 1 previously crashed!')

+        assert False

+

+    # Start supplier 3

+    topology_m4.ms["supplier3"].start()

+

+    # Need to wait 5 seconds before server processes any leftover tasks

+    time.sleep(6)

+

+    # Check supplier 1 tried to run abort task.  We expect the abort task to be aborted.

+    if not topology_m4.ms["supplier1"].searchErrorsLog('Aborting abort task'):

+        log.fatal('test_abort_restart: Abort task did not restart')

+        assert False

+

+    log.info('test_abort_restart PASSED')

+    

+

+if __name__ == '__main__':

+    # Run isolated

+    # -s for DEBUG mode

+    CURRENT_FILE = os.path.realpath(__file__)

+    pytest.main(["-s", CURRENT_FILE])

+

diff --git a/dirsrvtests/tests/suites/replication/cleanallruv_abort_test.py b/dirsrvtests/tests/suites/replication/cleanallruv_abort_test.py

new file mode 100644

index 000000000..f89188165

--- /dev/null

+++ b/dirsrvtests/tests/suites/replication/cleanallruv_abort_test.py

@@ -0,0 +1,123 @@

+# --- BEGIN COPYRIGHT BLOCK ---

+# Copyright (C) 2022 Red Hat, Inc.

+# All rights reserved.

+#

+# License: GPL (version 3 or any later version).

+# See LICENSE for details.

+# --- END COPYRIGHT BLOCK ---

+#

+import logging

+import pytest

+import os

+import time

+from lib389._constants import DEFAULT_SUFFIX

+from lib389.topologies import topology_m4

+from lib389.tasks import CleanAllRUVTask

+from lib389.replica import ReplicationManager

+

+log = logging.getLogger(__name__)

+

+

+def remove_supplier4_agmts(msg, topology_m4):

+    """Remove all the repl agmts to supplier4. """

+

+    log.info('%s: remove all the agreements to supplier 4...' % msg)

+    repl = ReplicationManager(DEFAULT_SUFFIX)

+    # This will delete m4 from the topo *and* remove all incoming agreements

+    # to m4.

+    repl.remove_supplier(topology_m4.ms["supplier4"],

+        [topology_m4.ms["supplier1"], topology_m4.ms["supplier2"], topology_m4.ms["supplier3"]])

+

+def task_done(topology_m4, task_dn, timeout=60):

+    """Check if the task is complete"""

+

+    attrlist = ['nsTaskLog', 'nsTaskStatus', 'nsTaskExitCode',

+                'nsTaskCurrentItem', 'nsTaskTotalItems']

+    done = False

+    count = 0

+

+    while not done and count < timeout:

+        try:

+            entry = topology_m4.ms["supplier1"].getEntry(task_dn, attrlist=attrlist)

+            if entry is not None:

+                if entry.hasAttr('nsTaskExitCode'):

+                    done = True

+                    break

+            else:

+                done = True

+                break

+        except ldap.NO_SUCH_OBJECT:

+            done = True

+            break

+        except ldap.LDAPError:

+            break

+        time.sleep(1)

+        count += 1

+

+    return done

+

+    

+@pytest.mark.flaky(max_runs=2, min_passes=1)

+def test_abort(topology_m4):

+    """Test the abort task basic functionality

+

+    :id: b09a6887-8de0-4fac-8e41-73ccbaaf7a08

+    :setup: Replication setup with four suppliers

+    :steps:

+        1. Disable replication on supplier 4

+        2. Remove agreements to supplier 4 from other suppliers

+        3. Stop supplier 2

+        4. Run a cleanallruv task on supplier 1

+        5. Run a cleanallruv abort task on supplier 1

+    :expectedresults: No hanging tasks left

+        1. Replication on supplier 4 should be disabled

+        2. Agreements to supplier 4 should be removed

+        3. Supplier 2 should be stopped

+        4. Operation should be successful

+        5. Operation should be successful

+    """

+

+    log.info('Running test_abort...')

+    # Remove the agreements from the other suppliers that point to supplier 4

+    repl = ReplicationManager(DEFAULT_SUFFIX)

+    m4rid = repl.get_rid(topology_m4.ms["supplier4"])

+    remove_supplier4_agmts("test_abort", topology_m4)

+

+    # Stop supplier 2

+    log.info('test_abort: stop supplier 2 to freeze the cleanAllRUV task...')

+    topology_m4.ms["supplier2"].stop()

+

+    # Run the task

+    log.info('test_abort: add the cleanAllRUV task...')

+    cruv_task = CleanAllRUVTask(topology_m4.ms["supplier1"])

+    cruv_task.create(properties={

+        'replica-id': m4rid,

+        'replica-base-dn': DEFAULT_SUFFIX,

+        'replica-force-cleaning': 'no',

+        'replica-certify-all': 'yes'

+        })

+    # Wait a bit

+    time.sleep(2)

+

+    # Abort the task

+    cruv_task.abort()

+

+    # Check supplier 1 does not have the clean task running

+    log.info('test_abort: check supplier 1 no longer has a cleanAllRUV task...')

+    if not task_done(topology_m4, cruv_task.dn):

+        log.fatal('test_abort: CleanAllRUV task was not aborted')

+        assert False

+

+    # Start supplier 2

+    log.info('test_abort: start supplier 2 to begin the restore process...')

+    topology_m4.ms["supplier2"].start()

+

+    log.info('test_abort PASSED')

+

+

+if __name__ == '__main__':

+    # Run isolated

+    # -s for DEBUG mode

+    CURRENT_FILE = os.path.realpath(__file__)

+    pytest.main(["-s", CURRENT_FILE])

+

diff --git a/dirsrvtests/tests/suites/replication/cleanallruv_force_test.py b/dirsrvtests/tests/suites/replication/cleanallruv_force_test.py

new file mode 100644

index 000000000..d5b930584

--- /dev/null

+++ b/dirsrvtests/tests/suites/replication/cleanallruv_force_test.py

@@ -0,0 +1,187 @@

+# --- BEGIN COPYRIGHT BLOCK ---

+# Copyright (C) 2022 Red Hat, Inc.

+# All rights reserved.

+#

+# License: GPL (version 3 or any later version).

+# See LICENSE for details.

+# --- END COPYRIGHT BLOCK ---

+#

+import logging

+import pytest

+import os

+import time

+import random

+import threading

+from lib389._constants import DEFAULT_SUFFIX

+from lib389.topologies import topology_m4

+from lib389.tasks import CleanAllRUVTask

+from lib389.replica import Replicas, ReplicationManager

+from lib389.idm.directorymanager import DirectoryManager

+from lib389.idm.user import UserAccounts

+

+log = logging.getLogger(__name__)

+

+

+class AddUsers(threading.Thread):

+    def __init__(self, inst, num_users):

+        threading.Thread.__init__(self)

+        self.daemon = True

+        self.inst = inst

+        self.num_users = num_users

+

+    def run(self):

+        """Start adding users"""

+

+        dm = DirectoryManager(self.inst)

+        conn = dm.bind()

+

+        users = UserAccounts(conn, DEFAULT_SUFFIX)

+

+        u_range = list(range(self.num_users))

+        random.shuffle(u_range)

+

+        for idx in u_range:

+            try:

+                users.create(properties={

+                    'uid': 'testuser%s' % idx,

+                    'cn' : 'testuser%s' % idx,

+                    'sn' : 'user%s' % idx,

+                    'uidNumber' : '%s' % (1000 + idx),

+                    'gidNumber' : '%s' % (1000 + idx),

+                    'homeDirectory' : '/home/testuser%s' % idx

+                })

+            # One of the suppliers was probably put into read only mode - just break out

+            except ldap.UNWILLING_TO_PERFORM:

+                break

+            except ldap.ALREADY_EXISTS:

+                pass

+        conn.close()

+

+def remove_some_supplier4_agmts(msg, topology_m4):

+    """Remove all the repl agmts to supplier4 except from supplier3.  Used by

+    the force tests."""

+

+    log.info('%s: remove the agreements to supplier 4...' % msg)

+    repl = ReplicationManager(DEFAULT_SUFFIX)

+    # This will delete m4 from the topo *and* remove all incoming agreements

+    # to m4.

+    repl.remove_supplier(topology_m4.ms["supplier4"],

+        [topology_m4.ms["supplier1"], topology_m4.ms["supplier2"]])

+

+def task_done(topology_m4, task_dn, timeout=60):

+    """Check if the task is complete"""

+

+    attrlist = ['nsTaskLog', 'nsTaskStatus', 'nsTaskExitCode',

+                'nsTaskCurrentItem', 'nsTaskTotalItems']

+    done = False

+    count = 0

+

+    while not done and count < timeout:

+        try:

+            entry = topology_m4.ms["supplier1"].getEntry(task_dn, attrlist=attrlist)

+            if entry is not None:

+                if entry.hasAttr('nsTaskExitCode'):

+                    done = True

+                    break

+            else:

+                done = True

+                break