def test_3_repeated_failure_recovery( self ):
     util.print_frame()
     for i in range( 0, 3 ):
         self.failure_recovery( 'master' )
         util.log( 'succeeded to failure_recovery, role:master, cnt:%d' % i )
         self.failure_recovery( 'slave' )
         util.log( 'succeeded to failure_recovery, role:slave, cnt:%d' % i )

    def test_2_role_change_with_hanging_pgs(self):
        util.print_frame()

        i = 0
        while i < 5:
            util.log('')
            util.log('Loop:%d' % i)

            # get master, slave1, slave2
            m, s1, s2 = util.get_mss(self.cluster)
            self.assertNotEqual(m, None, 'master is None.')
            self.assertNotEqual(s1, None, 'slave1 is None.')
            self.assertNotEqual(s2, None, 'slave2 is None.')

            hang = random.choice(self.cluster['servers'])
            if hang == m:
                hanging_servers = [m]
                running_servers = [s1, s2]
                type = 'master'
            else:
                hanging_servers = [s1]
                running_servers = [m, s2]
                type = 'slave'
            s = random.choice([s1, s2])

            util.log('hanging pgs(id:%d, type:%s), expected_master_id:%d' % (hang['id'], type, s['id']))
            self.role_change_with_hanging_pgs(hanging_servers, running_servers, s['id'], m)

            i += 1

    def test_error_log(self):
        util.print_frame()

        cluster = self.cluster

        # Start test-fiall
        p = util.exec_proc_async('%s/.obj%d' % (constant.ARCCI_DIR, self.arch),
                "./test-fiall -z localhost:2181 -c %s -s 10" % cluster['cluster_name'],
                subprocess.PIPE, subprocess.PIPE, subprocess.PIPE);

        # Set up arguments
        server = cluster['servers'][0]
        gw_id = server['id']
        gw_port = server['gateway_port']

        # Check load
        for i in range(20):
            ok = True
            for s in cluster['servers']:
                tps = util.get_tps(s['ip'], s['gateway_port'], 'gw')
                util.log('%s:%d TPS:%d' % (s['ip'], s['gateway_port'], tps))
            time.sleep(1)

        # Check no error
        util.log(' ### BEGIN - ARCCI LOGS ### ')
        (stdout, stderr) = p.communicate()
        for line in stdout.split("\n"):
            util.log(line)
        util.log(' ### END - ARCCI LOGS ### ')

    def test_redis_hang(self):
        util.print_frame()

        server = self.cluster['servers'][0]
        gw = telnetlib.Telnet(server['ip'], server['gateway_port'])
        redis = telnetlib.Telnet(server['ip'], server['redis_port'])

        redis.write("debug sleep 1000\r\n")

        gw.write("dbsize\r\n")
        gw.read_until("\r\n")

        ts = time.time()
        while (time.time() - ts < 6):
            gw.write("dbsize\r\n")
            time.sleep(0.1)

        gw.write("ping\r\n")
        gw.read_until("+PONG\r\n")

        gw.write("info cluster\r\n")
        ret = gw.read_until("\r\n\r\n", 3)
        util.log(ret)

        if "redis_instances_available:5" not in ret:
            self.assertFalse(True, "Disconnection of timed-out redis is not processed in gateway")

    def test_getandplay(self):
        util.print_frame()

        test_limit_mb = 10

        server = self.cluster['servers'][0]
        redis = telnetlib.Telnet(server['ip'], server['redis_port'])

        util.log("Insert large key about 100MB")
        self.insertLargeKey(redis, "test_key")

        # Test getandplay
        start_time = time.time()
        util.log("Start getandplay, start ts:%d" % start_time)

        cmd = "./cluster-util --getandplay %s %d %s %d 0-8191 30000 %d" % (server['ip'], self.getdump_proxy_port,
                                                                        server['ip'], self.playdump_proxy_port, test_limit_mb)
        proc = util.exec_proc_async(util.cluster_util_dir(0), cmd, True, None, subprocess.PIPE, None)

        monitor_file1 = "%s/%s" % (util.cluster_util_dir(0), self.getdump_proxy_log)
        monitor_file2 = "%s/%s" % (util.cluster_util_dir(0), self.playdump_proxy_log)
        self.monitor_filesize_diff(proc, test_limit_mb, monitor_file1, monitor_file2)

        ret = proc.wait()
        self.assertEqual(0, ret)

        elapse_time = time.time() - start_time
        util.log("End getandplay, elapsed:%d" % elapse_time)

        dump_file_size = os.path.getsize("%s/%s" % (util.cluster_util_dir(0), self.getdump_proxy_log))
        play_file_size = os.path.getsize("%s/%s" % (util.cluster_util_dir(0), self.playdump_proxy_log))
        util.log("Dump File Size:%d, Play File Size:%d, elapsed:%d, limit:%dMB/s, actual:%dMB/s"
                 % (dump_file_size, play_file_size, elapse_time, test_limit_mb,
                    (dump_file_size + play_file_size) / elapse_time / 1024 / 1024))
        self.assertTrue((dump_file_size + play_file_size) / (10 * 1024 * 1024) < elapse_time)

    def pgs_add_and_del( self, upgrade_server, type ):
        util.print_frame()

        util.log( '[start] add and del pgs%d. type:%s' % (upgrade_server['id'], type) )
        util.log_server_state( self.cluster )

        # start load generator
        load_gen_list = {}
        for i in range( len(self.cluster['servers']) ):
            server = self.cluster['servers'][i]
            load_gen = load_generator.LoadGenerator(server['id'], server['ip'], server['gateway_port'])
            load_gen.start()
            load_gen_list[i] = load_gen

        # detach pgs from cluster
        cmd = 'pgs_leave %s %d\r\n' % (upgrade_server['cluster_name'], upgrade_server['id'])
        ret = util.cm_command( self.leader_cm['ip'], self.leader_cm['cm_port'], cmd )
        jobj = json.loads(ret)
        self.assertEqual( jobj['msg'], '+OK', 'failed : cmd="%s", reply="%s"' % (cmd[:-2], ret[:-2]) )
        util.log( 'succeeded : cmd="%s", reply="%s"' % (cmd[:-2], ret[:-2]) )

        # set new values
        ip, port = util.get_rand_gateway(self.cluster)
        gw = gateway_mgmt.Gateway( '0' )
        gw.connect( ip, port )
        for i in range( 0, 50 ):
            cmd = 'set %s%d %d\r\n' % (self.key_base, i, i)
            gw.write( cmd )
            res = gw.read_until( '\r\n' )
            self.assertEqual( res, '+OK\r\n', 'failed to set values to gw(%s:%d). cmd:%s, res:%s' % (ip, port, cmd[:-2], res[:-2]) )

        # attach pgs from cluster
        cmd = 'pgs_join %s %d\r\n' % (upgrade_server['cluster_name'], upgrade_server['id'])
        ret = util.cm_command( self.leader_cm['ip'], self.leader_cm['cm_port'], cmd )
        jobj = json.loads(ret)
        self.assertEqual( jobj['msg'], '+OK', 'failed : cmd="%s", reply="%s"' % (cmd[:-2], ret) )
        util.log( 'succeeded : cmd="%s", reply="%s"' % (cmd[:-2], ret[:-2]) )
        time.sleep( 3 )

        # check new values
        redis = redis_mgmt.Redis( upgrade_server['id'] )
        ret = redis.connect( upgrade_server['ip'], upgrade_server['redis_port'] )
        self.assertEquals( ret, 0, 'failed : connect to smr%d(%s:%d)' % (upgrade_server['id'], upgrade_server['ip'], upgrade_server['redis_port']) )

        for i in range( 0, 50 ):
            cmd = 'get %s%d\r\n' % (self.key_base, i)
            redis.write( cmd )
            redis.read_until( '\r\n' )
            res = redis.read_until( '\r\n' )
            self.assertEqual( res, '%d\r\n' % i, 'failed to get values from redis%d. %s != %d' % (upgrade_server['id'], res, i) )
        util.log( 'succeeded : check values with get operations on pgs%d.' % (upgrade_server['id']) )

        # shutdown load generators
        for i in range( len(load_gen_list) ):
            load_gen_list[i].quit()
            load_gen_list[i].join()

        util.log_server_state( self.cluster )

        return 0

    def _test_protocol_error_with_quote(self, server, optional = None):
        util.print_frame()

        self.send_command_and_expect(server
                , 'set "key_without_closing_quote value\r\n'
                , '-ERR Protocol error: unbalanced quotes in request\r\n')
        self.server_reconnect(server)
        if (optional):
            self.send_command_and_expect(optional
                , "set 'key_without_closing_quote value\r\n"
                , '-ERR Protocol error: unbalanced quotes in request\r\n')
            self.server_reconnect(optional)

        self.send_command_and_expect(server
                , "set 'key_without_closing_quote value_without_opening_quote'\r\n"
                , "-ERR wrong number of arguments for 'set' command\r\n")
        if (optional):
            self.send_command_and_expect(optional
                , 'set "key_without_closing_quote value_without_opening_quote"\r\n'
                , "-ERR wrong number of arguments for 'set' command\r\n")

        self.send_command_and_expect(server
                , 'set key value_without_opening_quote"\r\n'
                , '-ERR Protocol error: unbalanced quotes in request\r\n')
        self.server_reconnect(server)
        if (optional):
            self.send_command_and_expect(optional
                , "set key value_without_opening_quote'\r\n"
                , '-ERR Protocol error: unbalanced quotes in request\r\n')
            self.server_reconnect(optional)

    def test_migrate_empty_s3obj(self):
        util.print_frame()
        ip, port = util.get_rand_gateway(self.cluster)
        client = redis_sock.RedisClient(ip, port)

        # Fill some string and empty s3 objects
        keyprefix = 'test_migrate_empty_s3obj'
        for i in range (1000):
            ok, data = client.do_request('set %s_string_%d %d\r\n' % (keyprefix, i, i))
            assert (ok == True)
            ok, data = client.do_request('s3ladd ks %s_s3_%d svc key val 0\r\n' % (keyprefix, i))
            assert (ok == True and data == 1)
            ok, data = client.do_request('s3lrem ks %s_s3_%d svc key val\r\n' % (keyprefix, i))
            assert (ok == True and data == 1)

        ## migration pg0 -> pg1 then pg1 -> pg0
        ret = util.migration(self.cluster, 0, 1, 4096, 8191, 40000)
        self.assertEqual(True, ret, 'Migration Fail')
        ret = util.migration(self.cluster, 1, 0, 4096, 8191, 40000)
        self.assertEqual(True, ret, 'Migration Fail')

        # Check string object
        for i in range (1000):
            ok, data = client.do_request('get %s_string_%d\r\n' % (keyprefix, i))
            assert (ok == True and int(data) == i)

        client.close()

    def save_pages(self, pages):
        '''Write pages to disk

        pages is a list of url, lang tuples
        '''
        parallelpath = ''

        for (url, lang) in pages:
            try:
                (r, tmpname) = self.downloader.download(url)
            except adder.AdderException as e:
                util.print_frame(debug=str(e) + '\n')
            else:
                normalised_name = namechanger.normalise_filename(
                    os.path.basename(tmpname))
                normalised_path = os.path.join(
                    self.corpus_adders[lang].goaldir, normalised_name)

                if not os.path.exists(normalised_path):
                    parallelpath = self.corpus_adders[lang].copy_file_to_corpus(
                        tmpname, url, parallelpath=parallelpath)
                    util.print_frame(
                            debug='adding {}'.format(parallelpath))
                else:
                    parallelpath = normalised_path
        print(file=sys.stderr)

    def links(self):
        links = set()
        for address in self.tree.findall('.//a'):
            href = address.get('href')
            if href is not None:
                if not re.search(
                    'tv.samediggi.no|^#|/rss/feed|switchlanguage|facebook.com|'
                    'Web-tv|user/login|mailto|/Dokumenter|/Dokumeantta|'
                    '/Tjaatsegh|.pdf|.doc|.xls|/images/|/download/|/Biejjielaahkoe|/Kalender|'
                    '/Dahpahusat|javascript|tel:',
                    href):
                    if href.startswith('/'):
                        href = urlparse.urlunparse(
                            (self.parsed_url.scheme,
                             self.parsed_url.netloc,
                             href, '', '', ''))

                    add = False
                    for uff in self.ok_netlocs:
                        if uff in href:
                            add = True
                            links.add(href)

                    if not add:
                        util.print_frame(debug=href.encode('utf8') + '\n')

        return links

    def test_1_consistent_while_slave_is_in_load( self ):
        util.print_frame()
        ip, port = util.get_rand_gateway( self.cluster )
        gw = gateway_mgmt.Gateway( ip )
        gw.connect( ip, port )

        max_key = 5
        key_base = 'load_gen_key'
        for idx in range( max_key ):
            cmd = 'set %s%d 0\r\n' % (key_base, idx)
            gw.write( cmd )
            gw.read_until( '\r\n', 10 )

        try_count = 9999
        for value in range( try_count ):
            for idx in range( max_key ):
                cmd = 'set %s%d %d\r\n' % (key_base, idx, value)
                gw.write( cmd )
                response = gw.read_until( '\r\n', 10 )
                self.assertEquals( response, '+OK\r\n' )

                cmd = 'get %s%d\r\n' % (key_base, idx)
                gw.write( cmd )
                response = gw.read_until( '\r\n', 10 )
                response = gw.read_until( '\r\n', 10 )

                self.assertEquals( response, '%s\r\n' % (value), 'fail! original_value:%d, return_from_slave:%s' % (value, response[1:]) )

    def test_6_from_3_to_6_heartbeat_checkers( self ):
        util.print_frame()

        hbc_svr_list = []
        i = 5000 + len( self.cluster['servers'] )
        for server in self.cluster['servers']:
            i = i + 1
            hbc_svr = {}
            hbc_svr['id'] = i
            hbc_svr['ip'] = server['ip']
            hbc_svr['zk_port'] = server['zk_port']
            hbc_svr_list.append(hbc_svr)

            ret = testbase.setup_cm( i )
            self.assertEquals( 0, ret, 'failed to copy heartbeat checker, server:%d' % hbc_svr['id'] )

            ret = testbase.request_to_start_cm( i, i )
            self.assertEquals( 0, ret,
                               'failed to request_to_start_cm, server:%d' % hbc_svr['id'] )
            self.state_transition()

        # Go back to initial configuration
        for hbc_svr in hbc_svr_list:
            self.assertEqual(0, testbase.request_to_shutdown_cm(hbc_svr),
                    'failed to shutdown confmaster')

    def test_cscandigest(self):
        util.print_frame()

        gateway_list = []
        for server in self.cluster['servers']:
            gateway_list.append(telnetlib.Telnet(server['ip'], server['gateway_port']))

        util.log("run CSCAN test")

        svr = random.choice(gateway_list)
        expect = self.insert_key(svr)

        svr.write('CSCANDIGEST\r\n')
        svr.read_until('\r\n', 3)
        digest1 = svr.read_until('\r\n', 3)

        ret = util.migration(self.cluster, 0, 1, 0, 4095, 40000)
        self.assertEqual(True, ret, 'Migration Fail')

        svr.write('CSCANDIGEST\r\n')
        svr.read_until('\r\n', 3)
        digest2 = svr.read_until('\r\n', 3)

        ret = util.migration(self.cluster, 1, 0, 0, 4095, 40000)
        self.assertEqual(True, ret, 'Migration Fail')

        svr.write('CSCANDIGEST\r\n')
        svr.read_until('\r\n', 3)
        digest3 = svr.read_until('\r\n', 3)

        self.assertEqual(digest1, digest3, "Incompatible Cluster Digest")
        self.assertNotEqual(digest1, digest2, "Incompatible Cluster Digest")

    def test_quorum_policy_of_hanging_master( self ):
        util.print_frame()

        # get master, slave1, slave2
        m, s1, s2 = util.get_mss( self.cluster )
        self.assertNotEqual( m, None, 'master is None.' )
        self.assertNotEqual( s1, None, 'slave1 is None.' )
        self.assertNotEqual( s2, None, 'slave2 is None.' )

        # hang
        smr = smr_mgmt.SMR( m['id'] )
        ret = smr.connect( m['ip'], m['smr_mgmt_port'] )
        self.assertEqual( ret, 0, 'failed to connect to master. %s:%d' % (m['ip'], m['smr_mgmt_port']) )
        smr.write( 'fi delay sleep 1 15000\r\n' )
        time.sleep( 5 )

        # wait for forced master election
        success = False
        new_master = None
        for i in range( 7 ):
            role = util.get_role_of_server( s1 )
            if role == c.ROLE_MASTER:
                success = True
                new_master = s1
                break
            role = util.get_role_of_server( s2 )
            if role == c.ROLE_MASTER:
                success = True
                new_master = s2
                break
            time.sleep( 1 )
        self.assertEqual( success, True, 'failed to forced master election' )

        # shutdown confmaster
        for server in self.cluster['servers']:
            util.shutdown_cm( server['id'] )

        # wait until hanging master wake up
        time.sleep( 5 )

        # check quorum policy
        quorum_of_haning_master = util.get_quorum( m )
        self.assertEqual( 2, quorum_of_haning_master,
                          'invalid quorum of haning master, expected:%d, but:%d' %(2, quorum_of_haning_master) )
        util.log( 'succeeded : quorum of haning master=%d' % quorum_of_haning_master )

        # check quorum policy
        quorum_of_new_master = util.get_quorum( new_master )
        self.assertNotEqual( None, quorum_of_new_master, 'failed : find new master' )
        self.assertEqual( 1, quorum_of_new_master ,
                          'invalid quorum of new master, expected:%d, but:%d' % (1, quorum_of_new_master) )
        util.log( 'succeeded : quorum of new master=%d' % quorum_of_new_master )

        # Go back to initial configuration
        # Recover Confmaster
        self.assertTrue(util.recover_confmaster(self.cluster, [0,1,2], 0), 'failed to recover confmaster')

        return 0

    def test_gateway_lookup( self ):
        util.print_frame()

        # check initial states
        for server in self.cluster['servers']:
            success = False
            for try_cnt in range( 5 ):
                cmd = 'get /RC/NOTIFICATION/CLUSTER/%s/GW/%d' % (self.cluster['cluster_name'], server['id'])
                ret = util.zk_cmd( cmd )
                ret = ret['err']
                if -1 != ret.find('cZxid'):
                    success = True
                    break
                time.sleep( 1 )

            self.assertEqual( success, True, 'failed : cmd="%s", ret="%s"' % (cmd, ret) )
            util.log( 'succeeded : cmd="%s", ret="%s"' % (cmd, ret) )

        # shutdown gateway
        for server in self.cluster['servers']:
            success = False

            ret = util.shutdown_gateway( server['id'], server['gateway_port'] )
            self.assertEqual( ret, 0, 'failed : shutdown gateawy%d' % server['id'] )

            for try_cnt in range( 10 ):
                cmd = 'get /RC/NOTIFICATION/CLUSTER/%s/GW/%d' % (self.cluster['cluster_name'], server['id'])
                ret = util.zk_cmd( cmd )
                ret = ret['err']
                if -1 != ret.find('Node does not exist'):
                    success = True
                    break
                time.sleep( 1 )

            self.assertEqual( success, True, 'failed : cmd="%s", ret="%s"' % (cmd, ret) )
            util.log( 'succeeded : cmd="%s", ret="%s"' % (cmd, ret) )

        # restart gateway
        for server in self.cluster['servers']:
            success = False

            ret = util.start_gateway( server['id'], server['ip'], self.leader_cm['cm_port'], server['cluster_name'], server['gateway_port'])
            self.assertEqual( ret, 0, 'failed : start gateawy%d' % server['id'] )

            for try_cnt in range( 5 ):
                cmd = 'get /RC/NOTIFICATION/CLUSTER/%s/GW/%d' % (self.cluster['cluster_name'], server['id'])
                ret = util.zk_cmd( cmd )
                ret = ret['err']
                if -1 != ret.find('cZxid'):
                    success = True
                    break
                time.sleep( 1 )

            self.assertEqual( success, True, 'failed : cmd="%s", ret="%s"' % (cmd, ret) )
            util.log( 'succeeded : cmd="%s", ret="%s"' % (cmd, ret) )

        return 0

    def _test_multibulk_protocol_error(self, server):
        util.print_frame()

        REDIS_INLINE_MAX_SIZE = 1024*64
        self.send_command_and_expect(server
                , '*%s\r\n' % ('1' * (REDIS_INLINE_MAX_SIZE * 3))
                , '-ERR Protocol error: too big mbulk count string\r\n')
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*%s\r\n' % ('1' * (1024*48))
                , '-ERR Protocol error: invalid multibulk length\r\n')
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*%d\r\n' % (1024*1024 + 1)
                , '-ERR Protocol error: invalid multibulk length\r\n')
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*%d\r\n!\r\n' % (1024*1024)
                , "-ERR Protocol error: expected '$', got '!'\r\n")
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*not_a_number\r\n'
                , '-ERR Protocol error: invalid multibulk length\r\n')
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*-1\r\nset normal command\r\n'
                , '+OK\r\n')

        self.send_command_and_expect(server
                , '*2\r\n$3\r\nget\r\n$%s\r\n' % ('1' * (REDIS_INLINE_MAX_SIZE * 3))
                , '-ERR Protocol error: too big bulk count string\r\n')
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*2\r\n!3\r\nget\r\n$3\r\nkey\r\n'
                , "-ERR Protocol error: expected '$', got '!'\r\n")
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*2\r\n$-1\r\n'
                , '-ERR Protocol error: invalid bulk length\r\n')
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*3\r\n$3\r\nSET\r\n$3\r\nKEY\r\n$%d\r\n' % (512*1024*1024+1)
                , '-ERR Protocol error: invalid bulk length\r\n')
        self.server_reconnect(server)

        self.send_command_and_expect(server
                , '*3\r\n$3\r\nSET\r\n$1\r\na\r\n$0\r\n\r\n'
                , '+OK\r\n')

    def test_basic_op_smr(self):
        util.print_frame()
        f = open("%s/test_basicop_output_smr" % constant.logdir, 'w')
        p = util.exec_proc_async("../redis-2.8.8",
                            "./runtest_smr --accurate",
                            True, None, f, None)

        ret = p.wait()
        f.close()
        self.assertEquals(0, ret)

    def test_basic_op_gateway(self):
        util.print_frame()
        ip, port = util.get_rand_gateway(self.cluster)
        f = open("%s/test_basicop_output_gw" % constant.logdir, 'w')
        p = util.exec_proc_async("../redis-2.8.8",
                            "./runtest_gw --accurate --gw-port "+str(port),
                            True, None, f, None)

        ret = p.wait()
        f.close()
        self.assertEquals(0, ret)

    def test_upgrade_slave_smr( self ):
        util.print_frame()

        # get master, slave1, slave2
        m, s1, s2 = util.get_mss( self.cluster )
        self.assertNotEqual( m, None, 'master is None.' )
        self.assertNotEqual( s1, None, 'slave1 is None.' )
        self.assertNotEqual( s2, None, 'slave2 is None.' )

        ret = util.upgrade_pgs( s1, self.leader_cm, self.cluster )
        self.assertTrue(ret, 'Failed to upgrade slave pgs%d' % s1['id'])

    def test_7_remaining_hbc_connection( self ):
        util.print_frame()

        # check pgs
        for server in self.cluster['servers']:
            before_cnt_redis = util.get_clients_count_of_redis(server['ip'], server['redis_port'])
            before_cnt_smr = util.get_clients_count_of_smr(server['smr_mgmt_port'])

            cmd = 'pgs_leave %s %d forced\r\npgs_del %s %d' % (self.cluster['cluster_name'], server['id'], self.cluster['cluster_name'], server['id'])
            util.cm_command(self.leader_cm['ip'], self.leader_cm['cm_port'], cmd)

        for server in self.cluster['servers']:
            # check redis
            success = False
            for i in range(5):
                after_cnt = util.get_clients_count_of_redis(server['ip'], server['redis_port'])
                if after_cnt <= 2:
                    success = True
                    break
                time.sleep(1)
            self.assertEquals( success, True, 'failed : the number of connections to redis%d(%s:%d) is %d, exptected=n<=2, before=%d' % (server['id'], server['ip'], server['redis_port'], after_cnt, before_cnt_redis) )
            util.log( 'succeeded : the number of connections to redis%d(%s:%d) is %d, exptected=n<=2, before=%d' % (server['id'], server['ip'], server['redis_port'], after_cnt, before_cnt_redis) )

            # check smr
            success = False
            expected = 0
            for i in range(5):
                after_cnt = util.get_clients_count_of_smr(server['smr_mgmt_port'])
                if after_cnt == expected:
                    success = True
                    break
                time.sleep(1)
            self.assertEquals( success, True, 'failed : the number of connections to smr%d(%s:%d) is %d, exptected=%d, before=%d' % (server['id'], server['ip'], server['smr_mgmt_port'], after_cnt, expected, before_cnt_smr) )
            util.log( 'succeeded : the number of connections to smr%d(%s:%d) is %d, exptected=%d, before=%d' % (server['id'], server['ip'], server['smr_mgmt_port'], after_cnt, expected, before_cnt_smr) )

        # check gateway
        for server in self.cluster['servers']:
            before_cnt = util.get_clients_count_of_gw(server['ip'], server['gateway_port'])

            cmd = 'gw_del %s %d' % (self.cluster['cluster_name'], server['id'])
            util.cm_command(self.leader_cm['ip'], self.leader_cm['cm_port'], cmd)

        for server in self.cluster['servers']:
            success = False
            expected = 1
            for i in range(5):
                after_cnt = util.get_clients_count_of_gw(server['ip'], server['gateway_port'])
                if after_cnt == expected:
                    success = True
                    break
                time.sleep(1)

            self.assertEquals( success, True, 'failed : the number of connections to gateway%d(%s:%d) is %d, exptected=%d.' % (server['id'], server['ip'], server['gateway_port'], after_cnt, expected) )
            util.log( 'succeeded : the number of connections to gateway%d(%s:%d) is %d, exptected=%d.' % (server['id'], server['ip'], server['gateway_port'], after_cnt, expected) )