'Get leader for partitions reload metadata if it is not available'

@patch('kafka.SimpleClient._get_conn')
@patch('kafka.client.KafkaProtocol')
def test_get_leader_for_partitions_reloads_metadata(self, protocol, conn):

    mock_conn(conn)
    brokers = [BrokerMetadata(0, 'broker_1', 4567, None), BrokerMetadata(1, 'broker_2', 5678, None)]
    resp0_brokers = list(map(itemgetter(0, 1, 2), brokers))
    topics = [(NO_LEADER, 'topic_no_partitions', [])]
    protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
    client = SimpleClient(hosts=['broker_1:4567'])
    self.assertDictEqual({}, client.topics_to_brokers)
    topics = [(NO_ERROR, 'topic_one_partition', [(NO_ERROR, 0, 0, [0, 1], [0, 1])])]
    protocol.decode_metadata_response.return_value = MetadataResponse[0](resp0_brokers, topics)
    leader = client._get_leader_for_partition('topic_one_partition', 0)
    self.assertEqual(brokers[0], leader)
    self.assertDictEqual({TopicPartition('topic_one_partition', 0): brokers[0]}, client.topics_to_brokers)
