def test_stepdown_triggers_refresh(self):
        c_find_one = MongoReplicaSetClient(
            self.seed, replicaSet=self.name, use_greenlets=use_greenlets)

        # We've started the primary and one secondary
        primary = ha_tools.get_primary()
        secondary = ha_tools.get_secondaries()[0]
        self.assertEqual(
            one(c_find_one.secondaries), _partition_node(secondary))

        ha_tools.stepdown_primary()

        # Make sure the stepdown completes
        sleep(1)

        # Trigger a refresh
        self.assertRaises(AutoReconnect, c_find_one.test.test.find_one)

        # Wait for the immediate refresh to complete - we're not waiting for
        # the periodic refresh, which has been disabled
        sleep(1)

        # We've detected the stepdown
        self.assertTrue(
            not c_find_one.primary
            or _partition_node(primary) != c_find_one.primary)

    def test_passive_and_hidden(self):
        self.c = ReplicaSetConnection(self.seed, replicaSet=self.name, use_greenlets=use_greenlets)
        db = self.c.pymongo_test
        w = len(self.c.secondaries) + 1
        db.test.remove({}, safe=True, w=w)
        db.test.insert({"foo": "bar"}, safe=True, w=w)

        passives = ha_tools.get_passives()
        passives = [_partition_node(member) for member in passives]
        hidden = ha_tools.get_hidden_members()
        hidden = [_partition_node(member) for member in hidden]
        self.assertEqual(self.c.secondaries, set(passives))

        for mode in (ReadPreference.SECONDARY, ReadPreference.SECONDARY_PREFERRED):
            db.read_preference = mode
            for _ in xrange(10):
                cursor = db.test.find()
                cursor.next()
                self.assertTrue(cursor._Cursor__connection_id in passives)
                self.assertTrue(cursor._Cursor__connection_id not in hidden)

        ha_tools.kill_members(ha_tools.get_passives(), 2)
        sleep(2 * MONITOR_INTERVAL)
        db.read_preference = ReadPreference.SECONDARY_PREFERRED

        for _ in xrange(10):
            cursor = db.test.find()
            cursor.next()
            self.assertEqual(cursor._Cursor__connection_id, self.c.primary)

    def test_recovering_member_triggers_refresh(self):
        # To test that find_one() and count() trigger immediate refreshes,
        # we'll create a separate client for each
        self.c_find_one, self.c_count = [
            MongoReplicaSetClient(
                self.seed, replicaSet=self.name, use_greenlets=use_greenlets,
                read_preference=SECONDARY)
            for _ in xrange(2)]

        # We've started the primary and one secondary
        primary = ha_tools.get_primary()
        secondary = ha_tools.get_secondaries()[0]

        # Pre-condition: just make sure they all connected OK
        for c in self.c_find_one, self.c_count:
            self.assertEqual(one(c.secondaries), _partition_node(secondary))

        ha_tools.set_maintenance(secondary, True)

        # Trigger a refresh in various ways
        self.assertRaises(AutoReconnect, self.c_find_one.test.test.find_one)
        self.assertRaises(AutoReconnect, self.c_count.test.test.count)

        # Wait for the immediate refresh to complete - we're not waiting for
        # the periodic refresh, which has been disabled
        sleep(1)

        for c in self.c_find_one, self.c_count:
            self.assertFalse(c.secondaries)
            self.assertEqual(_partition_node(primary), c.primary)

def setup_package():
    global mongod_started_with_ssl
    global sync_cx
    global sync_db
    global sync_collection
    global is_replica_set
    global rs_name
    global w
    global hosts
    global arbiters
    global primary
    global secondaries

    connectTimeoutMS = socketTimeoutMS = 30 * 1000

    # Store a regular synchronous pymongo MongoClient for convenience while
    # testing. Try over SSL first.
    try:
        sync_cx = pymongo.MongoClient(
            host, port,
            connectTimeoutMS=connectTimeoutMS,
            socketTimeoutMS=socketTimeoutMS,
            ssl=True)

        mongod_started_with_ssl = True
    except pymongo.errors.ConnectionFailure:
        sync_cx = pymongo.MongoClient(
            host, port,
            connectTimeoutMS=connectTimeoutMS,
            socketTimeoutMS=socketTimeoutMS,
            ssl=False)

    sync_db = sync_cx.motor_test
    sync_collection = sync_db.test_collection

    is_replica_set = False
    response = sync_cx.admin.command('ismaster')
    if 'setName' in response:
        is_replica_set = True
        rs_name = str(response['setName'])
        w = len(response['hosts'])
        hosts = set([_partition_node(h) for h in response["hosts"]])
        arbiters = set([
            _partition_node(h) for h in response.get("arbiters", [])])

        repl_set_status = sync_cx.admin.command('replSetGetStatus')
        primary_info = [
            m for m in repl_set_status['members']
            if m['stateStr'] == 'PRIMARY'][0]

        primary = _partition_node(primary_info['name'])
        secondaries = [
            _partition_node(m['name']) for m in repl_set_status['members']
            if m['stateStr'] == 'SECONDARY']

    def setUp(self):
        super(MotorTest, self).setUp()

        # Store a regular synchronous pymongo MongoClient for convenience while
        # testing. Set a timeout so we don't hang a test because, say, Mongo
        # isn't up or is hung by a long-running $where clause.
        connectTimeoutMS = socketTimeoutMS = 30 * 1000
        if self.ssl:
            if not HAVE_SSL:
                raise SkipTest("Python compiled without SSL")
            try:
                self.sync_cx = pymongo.MongoClient(
                    host, port, connectTimeoutMS=connectTimeoutMS,
                    socketTimeoutMS=socketTimeoutMS,
                    ssl=True)
            except pymongo.errors.ConnectionFailure:
                raise SkipTest("mongod doesn't support SSL, or is down")
        else:
            self.sync_cx = pymongo.MongoClient(
                host, port, connectTimeoutMS=connectTimeoutMS,
                socketTimeoutMS=socketTimeoutMS,
                ssl=False)

        self.is_replica_set = False
        response = self.sync_cx.admin.command('ismaster')
        if 'setName' in response:
            self.is_replica_set = True
            self.name = str(response['setName'])
            self.w = len(response['hosts'])
            self.hosts = set([_partition_node(h) for h in response["hosts"]])
            self.arbiters = set([
                _partition_node(h) for h in response.get("arbiters", [])])

            repl_set_status = self.sync_cx.admin.command('replSetGetStatus')
            primary_info = [
                m for m in repl_set_status['members']
                if m['stateStr'] == 'PRIMARY'][0]

            self.primary = _partition_node(primary_info['name'])
            self.secondaries = [
                _partition_node(m['name']) for m in repl_set_status['members']
                if m['stateStr'] == 'SECONDARY']
            
        self.sync_db = self.sync_cx.pymongo_test
        self.sync_coll = self.sync_db.test_collection
        self.sync_coll.drop()

        # Make some test data
        self.sync_coll.ensure_index([('s', pymongo.ASCENDING)], unique=True)
        self.sync_coll.insert(
            [{'_id': i, 's': hex(i)} for i in range(200)])

        self.cx = self.motor_client_sync()

    def setUp(self):
        members = [
            # primary
            {'tags': {'dc': 'ny', 'name': 'primary'}},

            # secondary
            {'tags': {'dc': 'la', 'name': 'secondary'}, 'priority': 0},

            # other_secondary
            {'tags': {'dc': 'ny', 'name': 'other_secondary'}, 'priority': 0},
        ]

        res = ha_tools.start_replica_set(members)
        self.seed, self.name = res

        primary = ha_tools.get_primary()
        self.primary = _partition_node(primary)
        self.primary_tags = ha_tools.get_tags(primary)
        # Make sure priority worked
        self.assertEqual('primary', self.primary_tags['name'])

        self.primary_dc = {'dc': self.primary_tags['dc']}

        secondaries = ha_tools.get_secondaries()

        (secondary, ) = [
            s for s in secondaries
            if ha_tools.get_tags(s)['name'] == 'secondary']

        self.secondary = _partition_node(secondary)
        self.secondary_tags = ha_tools.get_tags(secondary)
        self.secondary_dc = {'dc': self.secondary_tags['dc']}

        (other_secondary, ) = [
            s for s in secondaries
            if ha_tools.get_tags(s)['name'] == 'other_secondary']

        self.other_secondary = _partition_node(other_secondary)
        self.other_secondary_tags = ha_tools.get_tags(other_secondary)
        self.other_secondary_dc = {'dc': self.other_secondary_tags['dc']}

        self.c = MongoReplicaSetClient(
            self.seed, replicaSet=self.name, use_greenlets=use_greenlets)
        self.db = self.c.pymongo_test
        self.w = len(self.c.secondaries) + 1
        self.db.test.remove({}, w=self.w)
        self.db.test.insert(
            [{'foo': i} for i in xrange(10)], w=self.w)

        self.clear_ping_times()

    def setUp(self):
        super(TestBulkWriteConcern, self).setUp()
        client = get_client()
        ismaster = client.test.command('ismaster')
        self.is_repl = bool(ismaster.get('setName'))
        self.w = len(ismaster.get("hosts", []))

        self.secondary = None
        if self.w > 1:
            for member in ismaster['hosts']:
                if member != ismaster['primary']:
                    host, port = _partition_node(member)
                    self.secondary = MongoClient(host, port)
                    break

        self.client = client
        self.coll = client.pymongo_test.test
        self.coll.remove()

        # We tested wtimeout errors by specifying a write concern greater than
        # the number of members, but in MongoDB 2.7.8+ this causes a different
        # sort of error, "Not enough data-bearing nodes". In recent servers we
        # use a failpoint to pause replication on a secondary.
        self.need_replication_stopped = version.at_least(self.client,
                                                         (2, 7, 8))

        self.test_commands_enabled = ("enableTestCommands=1"
                                      in get_command_line(self.client)["argv"])

    def test_request_during_failover(self):
        primary = _partition_node(ha_tools.get_primary())
        secondary = _partition_node(ha_tools.get_random_secondary())

        self.assertTrue(self.c.auto_start_request)
        self.assertTrue(self.c.in_request())

        rs_state = self.c._MongoReplicaSetClient__rs_state
        primary_pool = rs_state.get(primary).pool
        secondary_pool = rs_state.get(secondary).pool

        # Trigger start_request on primary pool
        utils.assertReadFrom(self, self.c, primary, PRIMARY)
        self.assertTrue(primary_pool.in_request())

        # Fail over
        ha_tools.kill_primary()
        sleep(5)

        patience_seconds = 60
        for _ in range(patience_seconds):
            try:
                if ha_tools.ha_tools_debug:
                    print 'Waiting for failover'
                if ha_tools.get_primary():
                    # We have a new primary
                    break
            except ConnectionFailure:
                pass

            sleep(1)
        else:
            self.fail("Problem with test: No new primary after %s seconds"
                % patience_seconds)

        try:
            # Trigger start_request on secondary_pool, which is becoming new
            # primary
            self.c.test.test.find_one()
        except AutoReconnect:
            # We've noticed the failover now
            pass

        # The old secondary is now primary
        utils.assertReadFrom(self, self.c, secondary, PRIMARY)
        self.assertTrue(self.c.in_request())
        self.assertTrue(secondary_pool.in_request())

    def test_monitor_removes_recovering_member(self):
        self.c = MongoReplicaSetClient(
            self.seed, replicaSet=self.name, use_greenlets=use_greenlets)

        secondaries = ha_tools.get_secondaries()

        for mode in SECONDARY, SECONDARY_PREFERRED:
            partitioned_secondaries = [_partition_node(s) for s in secondaries]
            utils.assertReadFromAll(self, self.c, partitioned_secondaries, mode)

        secondary, recovering_secondary = secondaries
        ha_tools.set_maintenance(recovering_secondary, True)
        sleep(2 * MONITOR_INTERVAL)

        for mode in SECONDARY, SECONDARY_PREFERRED:
            # Don't read from recovering member
            utils.assertReadFrom(self, self.c, _partition_node(secondary), mode)

    def setup_rs(self):
        """Determine server's replica set config."""
        response = self.sync_cx.admin.command("ismaster")
        if "setName" in response:
            self.is_replica_set = True
            self.rs_name = str(response["setName"])
            self.rs_uri = self.uri + "?replicaSet=" + self.rs_name
            self.w = len(response["hosts"])
            self.hosts = set([_partition_node(h) for h in response["hosts"]])
            self.arbiters = set([_partition_node(h) for h in response.get("arbiters", [])])

            repl_set_status = self.sync_cx.admin.command("replSetGetStatus")
            primary_info = [m for m in repl_set_status["members"] if m["stateStr"] == "PRIMARY"][0]

            self.primary = _partition_node(primary_info["name"])
            self.secondaries = [
                _partition_node(m["name"]) for m in repl_set_status["members"] if m["stateStr"] == "SECONDARY"
            ]

    def test_recovering(self):
        self.c = ReplicaSetConnection(
            self.seed, replicaSet=self.name, use_greenlets=use_greenlets,
            auto_start_request=False)

        secondaries = ha_tools.get_secondaries()

        for mode in SECONDARY, SECONDARY_PREFERRED:
            partitioned_secondaries = [_partition_node(s) for s in secondaries]
            utils.assertReadFromAll(self, self.c, partitioned_secondaries, mode)

        secondary, recovering_secondary = secondaries
        ha_tools.set_maintenance(recovering_secondary, True)
        sleep(2 * MONITOR_INTERVAL)

        for mode in SECONDARY, SECONDARY_PREFERRED:
            # Don't read from recovering member
            utils.assertReadFrom(self, self.c, _partition_node(secondary), mode)

    def setUp(self):
        members = [
            # primary
            {"tags": {"dc": "ny", "name": "primary"}},
            # secondary
            {"tags": {"dc": "la", "name": "secondary"}, "priority": 0},
            # other_secondary
            {"tags": {"dc": "ny", "name": "other_secondary"}, "priority": 0},
        ]

        res = ha_tools.start_replica_set(members)
        self.seed, self.name = res

        primary = ha_tools.get_primary()
        self.primary = _partition_node(primary)
        self.primary_tags = ha_tools.get_tags(primary)
        # Make sure priority worked
        self.assertEqual("primary", self.primary_tags["name"])

        self.primary_dc = {"dc": self.primary_tags["dc"]}

        secondaries = ha_tools.get_secondaries()

        (secondary,) = [s for s in secondaries if ha_tools.get_tags(s)["name"] == "secondary"]

        self.secondary = _partition_node(secondary)
        self.secondary_tags = ha_tools.get_tags(secondary)
        self.secondary_dc = {"dc": self.secondary_tags["dc"]}

        (other_secondary,) = [s for s in secondaries if ha_tools.get_tags(s)["name"] == "other_secondary"]

        self.other_secondary = _partition_node(other_secondary)
        self.other_secondary_tags = ha_tools.get_tags(other_secondary)
        self.other_secondary_dc = {"dc": self.other_secondary_tags["dc"]}

        self.c = ReplicaSetConnection(self.seed, replicaSet=self.name, use_greenlets=use_greenlets)
        self.db = self.c.pymongo_test
        self.w = len(self.c.secondaries) + 1
        self.db.test.remove({}, safe=True, w=self.w)
        self.db.test.insert([{"foo": i} for i in xrange(10)], safe=True, w=self.w)

        self.clear_ping_times()

 def setup_rs(self):
     """Determine server's replica set config."""
     response = self.sync_cx.admin.command('ismaster')
     if 'setName' in response:
         self.is_replica_set = True
         self.rs_name = str(response['setName'])
         self.rs_uri = self.uri + '?replicaSet=' + self.rs_name
         self.w = len(response['hosts'])
         self.hosts = set([_partition_node(h) for h in response["hosts"]])
         self.arbiters = set([
             _partition_node(h) for h in response.get("arbiters", [])])
 
         repl_set_status = self.sync_cx.admin.command('replSetGetStatus')
         primary_info = [
             m for m in repl_set_status['members']
             if m['stateStr'] == 'PRIMARY'][0]
 
         self.primary = _partition_node(primary_info['name'])
         self.secondaries = [
             _partition_node(m['name']) for m in repl_set_status['members']
             if m['stateStr'] == 'SECONDARY']

    def test_passive_and_hidden(self):
        self.c = ReplicaSetConnection(
            self.seed, replicaSet=self.name, use_greenlets=use_greenlets)

        passives = ha_tools.get_passives()
        passives = [_partition_node(member) for member in passives]
        self.assertEqual(self.c.secondaries, set(passives))

        for mode in SECONDARY, SECONDARY_PREFERRED:
            utils.assertReadFromAll(self, self.c, passives, mode)

        ha_tools.kill_members(ha_tools.get_passives(), 2)
        sleep(2 * MONITOR_INTERVAL)
        utils.assertReadFrom(self, self.c, self.c.primary, SECONDARY_PREFERRED)

    def test_primary_stepdown(self):
        c = MongoReplicaSetClient(
            self.seed, replicaSet=self.name, use_greenlets=use_greenlets)
        self.assertTrue(bool(len(c.secondaries)))
        primary = c.primary
        ha_tools.stepdown_primary()

        # Wait for new primary
        patience_seconds = 30
        for _ in xrange(patience_seconds):
            sleep(1)
            rs_state = c._MongoReplicaSetClient__rs_state
            if rs_state.writer and rs_state.writer != primary:
                # New primary stepped up
                new_primary = _partition_node(ha_tools.get_primary())
                self.assertEqual(new_primary, rs_state.writer)
                new_secondaries = partition_nodes(ha_tools.get_secondaries())
                self.assertEqual(set(new_secondaries), rs_state.secondaries)
                break
        else:
            self.fail(
                "No new primary after %s seconds. Old primary was %s, current"
                " is %s" % (patience_seconds, primary, ha_tools.get_primary()))

def partition_nodes(nodes):
    """Translate from ['host:port', ...] to [(host, port), ...]"""
    return [_partition_node(node) for node in nodes]