Hi list,
Here are 2 integration tests for topology plugin. Plus I got rid of
using nose in test_ipaserver/test_topology_plugin.py: I am using
@pytest.mark.skipif instead
--
Oleg Fayans
Quality Engineer
FreeIPA team
RedHat.
From ac28eb85f92d730134b9e8b047dd49d54fb07d3a Mon Sep 17 00:00:00 2001
From: Oleg Fayans <ofay...@redhat.com>
Date: Mon, 10 Aug 2015 08:14:41 +0200
Subject: [PATCH] Added first part of integration tests for topology plugin,
got rid of using nose in the ipaserver topology test
---
ipatests/test_integration/tasks.py | 37 ++++++-
ipatests/test_integration/test_topology.py | 124 ++++++++++++++++++++++++
ipatests/test_ipaserver/test_topology_plugin.py | 12 +--
3 files changed, 165 insertions(+), 8 deletions(-)
create mode 100644 ipatests/test_integration/test_topology.py
diff --git a/ipatests/test_integration/tasks.py b/ipatests/test_integration/tasks.py
index 2b83f274619fcb18bcb87118d7df35a85ce52c1a..d99b99b9994336d4beefa30219276c3d65887532 100644
--- a/ipatests/test_integration/tasks.py
+++ b/ipatests/test_integration/tasks.py
@@ -243,7 +243,6 @@ def install_replica(master, replica, setup_ca=True, setup_dns=False):
'--forwarder', replica.config.dns_forwarder
])
replica.run_command(args)
-
enable_replication_debugging(replica)
setup_sssd_debugging(replica)
@@ -507,7 +506,8 @@ def uninstall_master(host):
paths.SYSCONFIG_PKI_TOMCAT,
paths.SYSCONFIG_PKI_TOMCAT_PKI_TOMCAT_DIR,
paths.VAR_LIB_PKI_TOMCAT_DIR,
- paths.PKI_TOMCAT],
+ paths.PKI_TOMCAT,
+ paths.REPLICA_INFO_GPG_TEMPLATE % host.hostname],
raiseonerr=False)
unapply_fixes(host)
@@ -519,6 +519,39 @@ def uninstall_client(host):
raiseonerr=False)
unapply_fixes(host)
+def clean_replication_agreement(master, replica):
+ """
+ clean_replication_agreement(master, replica):
+ Performs `ipa-replica-manage del replica_hostname --force`.
+ Both arguments should be objects of class Host
+ """
+ master.run_command(['ipa-replica-manage', 'del', replica.hostname, '--force'])
+
+def create_segment(master, leftnode, rightnode):
+ """create_segment(master, leftnode, rightnode)
+ creates a topology segment. The first argument is a node to run the command on
+ The first 3 arguments should be objects of class Host
+ Returns a hash object containing segment's name, leftnode, rightnode information
+ """
+ kinit_admin(master)
+ lefthost = leftnode.hostname
+ righthost = rightnode.hostname
+ segment_name = "%s-to-%s" % (lefthost, righthost)
+ result = master.run_command(["ipa", "topologysegment-add", "realm", segment_name,
+ "--leftnode=%s" % lefthost,
+ "--rightnode=%s" % righthost])
+ return {'leftnode': lefthost,
+ 'rightnode': righthost,
+ 'name': segment_name}
+
+def destroy_segment(master, segment_name):
+ """
+ destroy_segment(master, segment_name)
+ Destroys topology segment. First argument should be object of class Host
+ """
+ kinit_admin(master)
+ master.run_command(["ipa", "topologysegment-del", "realm", segment_name])
+
def get_topo(name_or_func):
"""Get a topology function by name
diff --git a/ipatests/test_integration/test_topology.py b/ipatests/test_integration/test_topology.py
new file mode 100644
index 0000000000000000000000000000000000000000..f93fe9c82f3912c3dcdadf631ea10f6957bc0b08
--- /dev/null
+++ b/ipatests/test_integration/test_topology.py
@@ -0,0 +1,124 @@
+#
+# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
+#
+
+import re
+import time
+from ipatests.test_integration.base import IntegrationTest
+from ipatests.test_integration import tasks
+from ipatests.util import assert_deepequal
+import rpdb
+
+
+
+
+class TestTopologyOptions(IntegrationTest):
+ num_replicas = 2
+ topology = 'star'
+ segname_re = re.compile("Segment name: ([\w\.\-]+)")
+ noentries_re = re.compile("Number of entries returned (\d+)")
+ leftnode_re = re.compile("Left node: ([\w\.]+)")
+ rightnode_re = re.compile("Right node: ([\w\.]+)")
+ connectivity_re = re.compile("Connectivity: ([\w\-]+)")
+
+ @classmethod
+ def install(cls, mh):
+ tasks.install_topo(cls.topology, cls.master,
+ cls.replicas[:len(cls.replicas) - 1],
+ cls.clients)
+
+ def tokenize_topologies(self, command_output):
+ """
+ tokenize_topologies(command_output)
+ takes an output of `ipa topologysegment-find` and returns an array of
+ segment hashes
+ """
+ segments = command_output.split("-----------------")[2]
+ raw_segments = segments.split('\n\n')
+ result = []
+ for i in raw_segments:
+ if self.segname_re.findall(i):
+ result.append({'leftnode': self.leftnode_re.findall(i)[0],
+ 'rightnode': self.rightnode_re.findall(i)[0],
+ 'name': self.segname_re.findall(i)[0],
+ 'connectivity': self.connectivity_re.findall(i)[0]})
+ return result
+
+ def test_topology_updated_on_replica_install_remove(self):
+ """
+ Install and remove a replica and make sure topology information is
+ updated on all other replicas
+ """
+ tasks.kinit_admin(self.master)
+ result1 = self.master.run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ first_segment_name = "%s-to-%s" % (self.master.hostname,
+ self.replicas[0].hostname)
+ output1 = result1.stdout_text
+ firstsegment = self.tokenize_topologies(output1)[0]
+ assert(firstsegment['name'] == first_segment_name)
+ assert(self.noentries_re.findall(output1)[0] == "1")
+ assert(firstsegment['leftnode'] == self.master.hostname)
+ assert(firstsegment['rightnode'] == self.replicas[0].hostname)
+ tasks.install_replica(self.master, self.replicas[1], setup_ca=False,
+ setup_dns=False)
+ # We need to make sure topology information is consistent across all
+ # replicas
+ result2 = self.master.run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ result3 = self.replicas[0].run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ result4 = self.replicas[1].run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ segments = self.tokenize_topologies(result2.stdout_text)
+ assert(len(segments) == 2)
+ assert(result2.stdout_text == result3.stdout_text)
+ assert(result3.stdout_text == result4.stdout_text)
+ # Now let's check that uninstalling the replica will update the topology
+ # info on the rest of replicas.
+ tasks.uninstall_master(self.replicas[1])
+ tasks.clean_replication_agreement(self.master, self.replicas[1])
+ result5 = self.master.run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ rpdb.set_trace()
+ assert(self.noentries_re.findall(result5.stdout_text)[0] == "1")
+
+ def test_add_remove_segment(self):
+ """
+ Make sure a topology segment can be manually created and deleted
+ with the influence on the real topology
+ The corresponding testcase can be found at
+ http://www.freeipa.org/page/V4/Manage_replication_topology/Test_plan#Test_case:_Basic_CRUD_test
+ """
+ tasks.kinit_admin(self.master)
+ # Install the second replica
+ tasks.install_replica(self.master, self.replicas[1], setup_ca=False,
+ setup_dns=False)
+ result1 = self.master.run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ # turn a star into a ring
+ segment = tasks.create_segment(self.master,
+ self.replicas[0],
+ self.replicas[1])
+
+ # Make sure the new segment is shown by `ipa topologysegment-find`
+ result2 = self.master.run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ assert(result2.stdout_text.find(segment['name']) > 0)
+ # Remove master <-> replica2 segment and make sure that the changes get
+ # there through replica1
+ deleteme = self.segname_re.findall(result1.stdout_text)[1]
+ tasks.destroy_segment(self.master, deleteme)
+
+ # make sure replica1 does not have segment that was deleted on master
+ result3 = self.replicas[0].run_command(['ipa', 'topologysegment-find',
+ 'realm'])
+ assert(result3.stdout_text.find(deleteme) < 0)
+ self.master.run_command(['ipa', 'user-add', 'someuser',
+ '--first', 'test',
+ '--last', 'user'])
+ time.sleep(60) # replication requires some time
+ users_on_replica2 = self.replicas[1].run_command(['ipa',
+ 'user-find'])
+ assert(users_on_replica2.find('someuser') > 0)
+ # We end up having a line topology: master <-> replica1 <-> replica2
\ No newline at end of file
diff --git a/ipatests/test_ipaserver/test_topology_plugin.py b/ipatests/test_ipaserver/test_topology_plugin.py
index 6678974993cb1762abb01e89a30caa3ebd94e3d0..a734376fb16fb362594533b6ff93ea01af0450e5 100644
--- a/ipatests/test_ipaserver/test_topology_plugin.py
+++ b/ipatests/test_ipaserver/test_topology_plugin.py
@@ -7,13 +7,14 @@ from ipaserver.plugins.ldap2 import ldap2
from ipalib import api
from ipapython import ipautil
from ipapython.dn import DN
-import nose
+import pytest
class TestTopologyPlugin(object):
"""
Test Topology plugin from the DS point of view
"""
+ pwfile = os.path.join(api.env.dot_ipa, ".dmpw")
def setup(self):
"""
@@ -25,6 +26,8 @@ class TestTopologyPlugin(object):
if self.conn and self.conn.isconnected():
self.conn.disconnect()
+ @pytest.mark.skipif(ipautil.file_exists(pwfile) == False,
+ reason = "You did not provide a .dmpw file with the DM password")
def test_topologyplugin(self):
pluginattrs = {
u'nsslapd-pluginPath': [u'libtopology'],
@@ -56,11 +59,8 @@ class TestTopologyPlugin(object):
('cn', 'plugins'),
('cn', 'config'))
pwfile = os.path.join(api.env.dot_ipa, ".dmpw")
- if ipautil.file_exists(pwfile):
- with open(pwfile, "r") as f:
- dm_password = f.read().rstrip()
- else:
- raise nose.SkipTest("No directory manager password in %s" % pwfile)
+ with open(pwfile, "r") as f:
+ dm_password = f.read().rstrip()
self.conn = ldap2(api)
self.conn.connect(bind_dn=DN(('cn', 'directory manager')),
bind_pw=dm_password)
--
2.4.3
--
Manage your subscription for the Freeipa-devel mailing list:
https://www.redhat.com/mailman/listinfo/freeipa-devel
Contribute to FreeIPA: http://www.freeipa.org/page/Contribute/Code
