dirsrvtests/tests/tickets/ticket49008_test.py | 110 ++++++++++++++++++++++++++ 1 file changed, 110 insertions(+)
New commits: commit 360d797597f19c1684e5031ed30aca596cbbf63f Author: Simon Pichugin spichugi@redhat.com Date: Tue Jan 17 15:32:38 2017 +0100
Ticket 49008 - Add CI test
Description: Check that RUV doesn't change the state after failing operation (but is succeeding in plugin operation).
https://fedorahosted.org/389/ticket/49008
Author: lkrispen (Thanks!)
Reviewed by: spichugi
diff --git a/dirsrvtests/tests/tickets/ticket49008_test.py b/dirsrvtests/tests/tickets/ticket49008_test.py new file mode 100644 index 0000000..67ee7c7 --- /dev/null +++ b/dirsrvtests/tests/tickets/ticket49008_test.py @@ -0,0 +1,110 @@ +import pytest +from lib389.tasks import * +from lib389.utils import * +from lib389.topologies import topology_m3 as T + +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__) + + +def test_ticket49008(T): + A = T.ms['master1'] + B = T.ms['master2'] + C = T.ms['master3'] + + A.enableReplLogging() + B.enableReplLogging() + C.enableReplLogging() + + AtoB = A.agreement.list(suffix=DEFAULT_SUFFIX)[0].dn + AtoC = A.agreement.list(suffix=DEFAULT_SUFFIX)[1].dn + CtoA = C.agreement.list(suffix=DEFAULT_SUFFIX)[0].dn + CtoB = C.agreement.list(suffix=DEFAULT_SUFFIX)[1].dn + + # we want replication in a line A <==> B <==> C + A.agreement.pause(AtoC) + C.agreement.pause(CtoA) + + B.plugins.enable(name=PLUGIN_MEMBER_OF) + B.restart(timeout=10) + + # add a few entries allowing memberof + for i in range(1, 6): + name = "userX{}".format(i) + dn = "cn={},{}".format(name, DEFAULT_SUFFIX) + A.add_s(Entry((dn, {'objectclass': "top person inetuser".split(), + 'sn': name, 'cn': name}))) + + # add a few entries not allowing memberof + for i in range(1, 6): + name = "userY{}".format(i) + dn = "cn={},{}".format(name, DEFAULT_SUFFIX) + A.add_s(Entry((dn, {'objectclass': "top person".split(), + 'sn': name, 'cn': name}))) + + time.sleep(10) + + A_entries = A.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, + '(objectClass=person)') + B_entries = B.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, + '(objectClass=person)') + C_entries = C.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE, + '(objectClass=person)') + + assert len(A_entries) == len(B_entries) + assert len(B_entries) == len(C_entries) + + # add a group with members allowing memberof + dn = "cn=g1,{}".format(DEFAULT_SUFFIX) + A.add_s(Entry((dn, {'objectclass': "top groupOfNames".split(), + 'description': "Test Owned Group {}".format(name), + 'member': "cn=userX1,{}".format(DEFAULT_SUFFIX), + 'cn': "g1"}))) + + # check ruv on m2 before applying failing op + time.sleep(10) + B_RUV = B.search_s("cn=config", ldap.SCOPE_SUBTREE, + "(&(objectclass=nsds5replica)(nsDS5ReplicaRoot={}))".format(DEFAULT_SUFFIX), + ['nsds50ruv']) + elements = B_RUV[0].getValues('nsds50ruv') + ruv_before = 'ruv_before' + for ruv in elements: + if 'replica 2' in ruv: + ruv_before = ruv + + # add a group with members allowing memberof and members which don't + # the op will fail on M2 + dn = "cn=g2,{}".format(DEFAULT_SUFFIX) + A.add_s(Entry((dn, {'objectclass': "top groupOfNames".split(), + 'description': "Test Owned Group {}".format(name), + 'member': ["cn=userX1,{}".format(DEFAULT_SUFFIX), + "cn=userX2,{}".format(DEFAULT_SUFFIX), + "cn=userY1,{}".format(DEFAULT_SUFFIX)], + 'cn': "g2"}))) + + # check ruv on m2 after applying failing op + time.sleep(10) + B_RUV = B.search_s("cn=config", ldap.SCOPE_SUBTREE, + "(&(objectclass=nsds5replica)(nsDS5ReplicaRoot={}))".format(DEFAULT_SUFFIX), + ['nsds50ruv']) + elements = B_RUV[0].getValues('nsds50ruv') + ruv_after = 'ruv_after' + for ruv in elements: + if 'replica 2' in ruv: + ruv_after = ruv + + log.info('ruv before fail: {}'.format(ruv_before)) + log.info('ruv after fail: {}'.format(ruv_after)) + # the ruv should not have changed + assert ruv_before == ruv_after + + +if __name__ == '__main__': + # Run isolated + # -s for DEBUG mode + CURRENT_FILE = os.path.realpath(__file__) + pytest.main("-s {}".format(CURRENT_FILE))