"""Testing the KafkaExecutor."""
from unittest.mock import MagicMock
from unittest.mock import patch

import pytest

from feed_ingestion.util.kafka import executor
from feed_ingestion.util.kafka import structs


@pytest.fixture
def mock_boto3():
    """Mock boto3."""
    boto3_path = ('feed_ingestion.util.kafka.executor.boto3')
    with patch(boto3_path) as boto3:
        boto3.client.return_value = MagicMock()
        cluster_list = {'ClusterInfoList': [
                {
                    'ClusterName': ['test-cluster'],
                    'ClusterArn': 'test-cluster-arn'
                }
            ]
        }
        client = boto3.client()
        client.list_clusters.return_value = cluster_list
        client.get_bootstrap_brokers.return_value = {
            'BootstrapBrokerStringTls':
            'broker-1:1234,broker-2:1234,broker-3:1234'}
        yield dict(boto3=boto3, client=client)


def test_executor_init(mock_boto3):
    """Test executor returns configured consumer."""
    mock_boto3['boto3'].client.return_value = mock_boto3['client']
    mock_executor = executor.KafkaExecutor(cluster_name='test-cluster')
    expected_config = {
        'bootstrap_servers': [
            'broker-1:1234', 'broker-2:1234', 'broker-3:1234'],
        'security_protocol': 'SSL'
    }
    assert isinstance(mock_executor, executor.KafkaExecutor)
    assert mock_executor.basic_config == expected_config


@patch('feed_ingestion.util.kafka.executor.KafkaConsumer')
def test_executor_returns_consumer(mock_consumer, mock_boto3):
    """Test executor returns configured consumer."""
    mock_boto3['boto3'].client.return_value = mock_boto3['client']
    mock_executor = executor.KafkaExecutor(cluster_name='test-cluster')
    consumer = mock_executor.consumer(
        client_id='test-id',
        auto_offset_reset=structs.AutoOffsetReset.LATEST)
    assert consumer is not None


@patch('feed_ingestion.util.kafka.executor.KafkaProducer')
def test_executor_returns_producer(mock_producer, mock_boto3):
    """Test executor returns configured producer."""
    mock_boto3['boto3'].client.return_value = mock_boto3['client']
    mock_executor = executor.KafkaExecutor(cluster_name='test-cluster')
    producer = mock_executor.producer(client_id='test-id')
    assert producer is not None


@patch('feed_ingestion.util.kafka.executor.KafkaAdminClient')
def test_executor_returns_admin_client(mock_admin_client, mock_boto3):
    """Test executor returns configured admin_client."""
    mock_boto3['boto3'].client.return_value = mock_boto3['client']
    mock_executor = executor.KafkaExecutor(cluster_name='test-cluster')
    client = mock_executor.admin_client(client_id='test-id')
    assert client is not None