test_simple_replication: Wait for replication to finish before checking

Add ldap_connect() method to Host to allow executing querying LDAP from tests.

Use information in the mapping tree to poll until all replication is finished
(or failing) before checking that entries replicated successfully.

ipatests/test_integration/host.py

@@ -27,6 +27,7 @@ import errno
 
 import paramiko
 
+from ipapython.ipaldap import IPAdmin
 from ipapython import ipautil
 from ipapython.ipa_log_manager import log_mgr
 
@@ -291,6 +292,14 @@ class Host(object):
             self._sftp = paramiko.SFTPClient.from_transport(transport)
             return self._sftp
 
+    def ldap_connect(self):
+        """Return an LDAPClient authenticated to this host as directory manager
+        """
+        ldap = IPAdmin(self.external_hostname)
+        ldap.do_simple_bind(self.config.dirman_dn,
+                            self.config.dirman_password)
+        return ldap
+
     def mkdir_recursive(self, path):
         """`mkdir -p` on the remote host"""
         try:

ipatests/test_integration/tasks.py

@@ -24,8 +24,13 @@ import textwrap
 import re
 import collections
 import itertools
+import time
+import StringIO
+
+from ldif import LDIFWriter
 
 from ipapython import ipautil
+from ipapython.dn import DN
 from ipapython.ipa_log_manager import log_mgr
 from ipatests.test_integration.config import env_to_script
 
@@ -379,3 +384,48 @@ def install_clients(servers, clients):
     for server, client in itertools.izip(itertools.cycle(servers), clients):
         log.info('Installing client %s on %s' % (server, client))
         install_client(server, client)
+
+
+def _entries_to_ldif(entries):
+    """Format LDAP entries as LDIF"""
+    lines = []
+    io = StringIO.StringIO()
+    writer = LDIFWriter(io)
+    for entry in entries:
+        writer.unparse(str(entry.dn), entry)
+    return io.getvalue()
+
+
+def wait_for_replication(ldap, timeout=30):
+    """Wait until updates on all replication agreements are done (or failed)
+
+    :param ldap: LDAP client
+        autenticated with necessary rights to read the mapping tree
+    :param timeout: Maximum time to wait, in seconds
+
+    Note that this waits for updates originating on this host, not those
+    coming from other hosts.
+    """
+    log.debug('Waiting for replication to finish')
+    for i in range(timeout):
+        time.sleep(1)
+        status_attr = 'nsds5replicaLastUpdateStatus'
+        progress_attr = 'nsds5replicaUpdateInProgress'
+        entries = ldap.get_entries(
+            DN(('cn', 'mapping tree'), ('cn', 'config')),
+            filter='(objectclass=nsds5replicationagreement)',
+            attrs_list=[status_attr, progress_attr])
+        log.debug('Replication agreements: \n%s', _entries_to_ldif(entries))
+        if any(not e.single_value(status_attr).startswith('0 ')
+               for e in entries):
+            log.error('Replication error')
+            break
+        in_progress = []
+        if any(e.single_value(progress_attr) == 'TRUE' for e in entries):
+            log.debug('Replication in progress (waited %s/%ss)',
+                      i, timeout)
+        else:
+            log.debug('Replication finished')
+            break
+    else:
+        log.error('Giving up wait for replication to finish')

ipatests/test_integration/test_simple_replication.py

@@ -17,9 +17,9 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import time
-
+from ipapython.dn import DN
 from ipatests.test_integration.base import IntegrationTest
+from ipatests.test_integration import tasks
 
 
 class TestSimpleReplication(IntegrationTest):
@@ -31,28 +31,31 @@ class TestSimpleReplication(IntegrationTest):
     num_replicas = 1
     topology = 'star'
 
-    def test_user_replication_to_replica(self):
-        """Test user replication master -> replica"""
-        login = 'testuser1'
-        self.master.run_command(['ipa', 'user-add', login,
+    def check_replication(self, source_host, dest_host, login):
+        source_host.run_command(['ipa', 'user-add', login,
                                  '--first', 'test',
                                  '--last', 'user'])
 
-        self.log.debug('Sleeping so replication has a chance to finish')
-        time.sleep(5)
+        ldap = dest_host.ldap_connect()
+        tasks.wait_for_replication(ldap)
 
-        result = self.replicas[0].run_command(['ipa', 'user-show', login])
+        # Check using LDAP
+        basedn = dest_host.domain.basedn
+        user_dn = DN(('uid', login), ('cn', 'users'), ('cn', 'accounts'),
+                     basedn)
+        entry = ldap.get_entry(user_dn)
+        print entry
+        assert entry.dn == user_dn
+        assert entry['uid'] == [login]
+
+        # Check using CLI
+        result = dest_host.run_command(['ipa', 'user-show', login])
         assert 'User login: %s' % login in result.stdout_text
 
+    def test_user_replication_to_replica(self):
+        """Test user replication master -> replica"""
+        self.check_replication(self.master, self.replicas[0], 'testuser1')
+
     def test_user_replication_to_master(self):
         """Test user replication replica -> master"""
-        login = 'testuser2'
-        self.replicas[0].run_command(['ipa', 'user-add', login,
-                                      '--first', 'test',
-                                      '--last', 'user'])
-
-        self.log.debug('Sleeping so replication has a chance to finish')
-        time.sleep(5)
-
-        result = self.master.run_command(['ipa', 'user-show', login])
-        assert 'User login: %s' % login in result.stdout_text
+        self.check_replication(self.replicas[0], self.master, 'testuser2')