import logging import pytest import os import ldap import time from lib389 import Entry from lib389._constants import DEFAULT_SUFFIX from lib389.idm.user import UserAccounts, TEST_USER_PROPERTIES from lib389.topologies import topology_m3 as topo DEBUGGING = os.getenv("DEBUGGING", default=False) if DEBUGGING: logging.getLogger(__name__).setLevel(logging.DEBUG) else: logging.getLogger(__name__).setLevel(logging.INFO) log = logging.getLogger(__name__) TEST_ENTRY_NAME = 'testuser' USER_UID='user_' def _user_get_dn(no): uid = '%s%d' % (USER_UID, no) dn = 'uid=%s,ou=people,%s' % (uid, DEFAULT_SUFFIX) return (uid, dn) def add_user(server, no, init_val): (uid, dn) = _user_get_dn(no) log.fatal('Adding user (%s): ' % dn) server.add_s(Entry((dn, {'objectclass': ['top', 'person', 'organizationalPerson', 'inetOrgPerson'], 'uid': [uid], 'sn' : [uid], 'cn' : [uid], 'employeeNumber': init_val}))) return dn def test_ticket49658(topo): """Specify a test case purpose or name here :id: 0 :setup: 3 Master Instances :steps: 1. Create 3 suppliers 2. Create on M3 a test entry having a single-value attribute employeeNumber=11 and update it MOD_DEL 11 + MOD_ADD 12 3. Check it is replicated on M1 and M2 4. Isolate M1 and M2 by pausing the replication agreements 5. On M1 do MOD_DEL 12 + MOD_ADD_13 6. On M2 do MOD_DEL 12 + MOD_ADD_13 7. Enable replication agreement M2 -> M3, so that update step 6 is replicated first 8. Enable replication agreement M1 -> M3, so that update step 5 is replicated second 9. Check that the employeeNumber is 13 on all servers :expectedresults: 1. Fill in the result that is expected 2. For each test step """ # If you need any test suite initialization, # please, write additional fixture for that (including finalizer). # Topology for suites are predefined in lib389/topologies.py. # If you need host, port or any other data about instance, # Please, use the instance object attributes for that (for example, topo.ms["master1"].serverid) if DEBUGGING: # Add debugging steps(if any)... pass M1 = topo.ms["master1"] M2 = topo.ms["master2"] M3 = topo.ms["master3"] value_11 = '11'.encode() value_12 = '12'.encode() value_13 = '13'.encode() value_14 = '14'.encode() # Step 2 test_user_dn= add_user(M3, 0, value_11) log.info('Adding %s on M3' % test_user_dn) M3.modify_s(test_user_dn, [(ldap.MOD_DELETE, 'employeeNumber', value_11), (ldap.MOD_ADD, 'employeeNumber', value_12)]) ents = M3.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=*)') assert len(ents) == 1 # Step 3 # Check the entry is replicated on M1 for j in range(30): try: ent = M1.getEntry(test_user_dn, ldap.SCOPE_BASE,) if not ent.hasAttr('employeeNumber'): # wait for the MOD log.info('M1 waiting for employeeNumber') time.sleep(1) continue; break; except ldap.NO_SUCH_OBJECT: time.sleep(1) pass time.sleep(1) ents = M1.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=*)') assert len(ents) == 1 # Check the entry is replicated on M2 for j in range(30): try: ent = M2.getEntry(test_user_dn, ldap.SCOPE_BASE,) if not ent.hasAttr('employeeNumber'): # wait for the MOD log.info('M2 waiting for employeeNumber') time.sleep(1) continue; break; except ldap.NO_SUCH_OBJECT: time.sleep(1) pass time.sleep(1) ents = M2.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=*)') assert len(ents) == 1 # # Step 4 # # disable all RA from M1 and M2 # only M3 can replicate the update # agreement_m1_m2 = M1.agreement.list(suffix=DEFAULT_SUFFIX, consumer_host=M2.host, consumer_port=M2.port) agreement_m1_m3 = M1.agreement.list(suffix=DEFAULT_SUFFIX, consumer_host=M3.host, consumer_port=M3.port) agreement_m2_m1 = M2.agreement.list(suffix=DEFAULT_SUFFIX, consumer_host=M1.host, consumer_port=M1.port) agreement_m2_m3 = M2.agreement.list(suffix=DEFAULT_SUFFIX, consumer_host=M3.host, consumer_port=M3.port) M1.agreement.pause(agreement_m1_m2[0].dn) M1.agreement.pause(agreement_m1_m3[0].dn) M2.agreement.pause(agreement_m2_m1[0].dn) M2.agreement.pause(agreement_m2_m3[0].dn) # Step 5 # Oldest update # check that the entry on M1 contains employeeNumber=13 M1.modify_s(test_user_dn, [(ldap.MOD_DELETE, 'employeeNumber', value_12), (ldap.MOD_ADD, 'employeeNumber', value_13)]) ents = M1.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=13)') assert len(ents) == 1 time.sleep(1) # Step 6 # More recent update # check that the entry on M2 contains employeeNumber=13 M2.modify_s(test_user_dn, [(ldap.MOD_DELETE, 'employeeNumber', value_12), (ldap.MOD_ADD, 'employeeNumber', value_13)]) ents = M1.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=13)') assert len(ents) == 1 # Step 7 # Renable M2 before M1 so that on M3, the most recent update is replicated before for ra in agreement_m2_m1, agreement_m2_m3: M2.agreement.resume(ra[0].dn) # Step 8 # Renable M1 so that on M3 oldest update is now replicated time.sleep(4) for ra in agreement_m1_m2, agreement_m1_m3: M1.agreement.resume(ra[0].dn) # Step 9 # Check that M1 still contains employeeNumber=13 ents = M1.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=13)') assert len(ents) == 1 # Check that M2 still contains employeeNumber=13 ents = M2.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=13)') assert len(ents) == 1 # Check that M3 still contain employeeNumber and it contains employeeNumber=13 ents = M3.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=*)') assert len(ents) == 1 ents = M3.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, '(employeeNumber=13)') assert len(ents) == 1 if __name__ == '__main__': # Run isolated # -s for DEBUG mode CURRENT_FILE = os.path.realpath(__file__) pytest.main(["-s", CURRENT_FILE])