"""Integration tests for lambda function."""
import json
import time
import uuid

import pytest
import requests
from confluent_kafka import Consumer
from confluent_kafka import KafkaException
from secrets_manager.lambda_ext import LambdaSecretsManager

import config  # noqa


@pytest.mark.parametrize(
    'vendor', ('smf', 'tunespeak')  # There are allowed vendors in the tests defined by AWS WAF (see terraform-infra)
)
def test_fan_response_api(vendor):
    """Test the fan response API successful request."""
    request_payload = {'email_address': f'{uuid.uuid4()}@mail.com'}
    response = send_fan_response_request(vendor, request_payload)
    assert response.status_code == 200
    topic = config.VENDOR_TOPICS[vendor]
    assert is_message_in_topic(topic, request_payload) is True


@pytest.mark.parametrize(
    'vendor', ('wyng', 'mtl')  # There are NOT allowed vendors in the tests defined by AWS WAF (see terraform-infra)
)
def test_fan_response_api_denied(vendor):
    """Test the fan response API failure."""
    request_payload = {'email_address': f'{uuid.uuid4()}@mail.com'}
    response = send_fan_response_request(vendor, request_payload)
    assert response.status_code == 403


# Utility functions

def send_fan_response_request(vendor, payload):
    """Send a request to the fan response API."""
    url = f'https://qa-fan-response-api.theorchard.io/{vendor}/response'
    token = get_token_for_vendor(vendor)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json'
    }
    return requests.post(url, headers=headers, json=payload)


def get_token_for_vendor(vendor):
    """Get the Auth0 token for the vendor."""
    secrets_manager_client = LambdaSecretsManager(environment='qa', service_name='fan-response-api')
    auth0_config = secrets_manager_client.get_cred(f'{vendor.upper()}_AUTH0_CONFIG')
    headers = {'content-type': 'application/json'}
    res = requests.post('https://qa-orchard.auth0.com/oauth/token', headers=headers, json=auth0_config)
    data = res.json()
    return data['access_token']


def is_message_in_topic(topic, message, timeout=60):
    """Check if the message is in the Kafka topic."""
    conf = {
        'bootstrap.servers': config.KAFKA_BOOTSTRAP_SERVERS,
        'security.protocol': config.KAFKA_SECURITY_PROTOCOL,
        'group.id': f'{config.LAMBDA_NAME}-integration-test-{uuid.uuid4()}',
        'auto.offset.reset': 'earliest',
    }
    consumer = Consumer(conf)
    consumer.subscribe([topic])
    try:
        start_time = time.time()
        while time.time() - start_time < timeout:
            msg = consumer.poll(1.0)  # timeout in seconds
            if msg is None:
                continue
            if msg.error():
                raise KafkaException(msg.error())
            topic_message = msg.value().decode('utf-8')
            if json.loads(topic_message) == message:
                return True
        raise TimeoutError(f'Message not found in topic "{topic}" after {timeout} seconds.')
    finally:
        consumer.close()