dirsrvtests/tests/tickets/ticket49008_test.py | 110 ++++++++++++++++++++++++++
1 file changed, 110 insertions(+)
New commits:
commit 360d797597f19c1684e5031ed30aca596cbbf63f
Author: Simon Pichugin <spichugi(a)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))