#
# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
# or more contributor license agreements. Licensed under the Elastic License 2.0;
# you may not use this file except in compliance with the Elastic License 2.0.
#
import asyncio
import functools
import random
import ssl
import time
import timeit
from datetime import datetime, timezone
from unittest.mock import Mock, patch

import pytest
from dateutil.tz import tzutc
from freezegun import freeze_time
from pympler import asizeof

from connectors.utils import (
    ConcurrentTasks,
    ErrorMonitor,
    InvalidIndexNameError,
    MemQueue,
    NonBlockingBoundedSemaphore,
    RetryStrategy,
    TooManyErrors,
    UnknownRetryStrategyError,
    base64url_to_base64,
    decode_base64_value,
    deep_merge_dicts,
    evaluate_timedelta,
    filter_nested_dict_by_keys,
    get_base64_value,
    get_pem_format,
    get_size,
    get_source_klass,
    get_source_klasses,
    has_duplicates,
    html_to_text,
    is_expired,
    iterable_batches_generator,
    next_run,
    parse_datetime_string,
    retryable,
    ssl_context,
    time_to_sleep_between_retries,
    truncate_id,
    url_encode,
    validate_email_address,
    validate_index_name,
)


@pytest.mark.parametrize(
    "cron_statement, now, expected_next_run",
    [
        (
            # Every minute at second=0 - not generated by Kibana
            "0 * * * * *",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 17, 19, 0, tzinfo=timezone.utc),
        ),
        (
            # Every 5 minutes at second 0 - generated by Kibana
            "0 0,5,10,15,20,25,30,35,40,45,50,55 * * * ?",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 17, 20, 0, tzinfo=timezone.utc),
        ),
        (
            # Every 5 minutes at second 0 - not generated by Kibana
            "0 */5 * * * *",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 17, 20, 0, tzinfo=timezone.utc),
        ),
        (
            # Every 10 minutes - generated by Kibana
            "0 0,10,20,30,40,50 * * * ?",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 17, 20, 0, tzinfo=timezone.utc),
        ),
        (
            # Every 15 minutes - generated by Kibana
            "0 0,15,30,45 * * * ?",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 17, 30, 0, tzinfo=timezone.utc),
        ),
        (
            # Every 30 minutes - generated by Kibana
            "0 0,30 * * * ?",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 17, 30, 0, tzinfo=timezone.utc),
        ),
        (
            # Every hour at minute=3 - generated by Kibana; without overflow
            "0 3 * * * ?",
            datetime(2023, 1, 18, 17, 1, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 17, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every hour at minute=3 - generated by Kibana; with overflow
            "0 3 * * * ?",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 18, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every day at 5:03 - generated by Kibana; without overflow
            "0 3 5 * * ?",
            datetime(2023, 1, 18, 5, 1, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 18, 5, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every day at 5:03 - generated by Kibana; with overflow
            "0 3 5 * * ?",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),
            datetime(2023, 1, 19, 5, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every week on Wednesday at 5:03 - generated by Kibana; without overflow
            "0 3 5 ? * 4",
            datetime(2023, 1, 18, 1, 18, 56, 814, tzinfo=timezone.utc),  # Wednesday
            datetime(2023, 1, 18, 5, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every week on Wednesday at 5:03 - generated by Kibana; with overflow
            "0 3 5 ? * 4",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),  # Wednesday
            datetime(2023, 1, 25, 5, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every month on 5th at 5:03 - generated by Kibana; with overflow
            "0 3 5 5 * ?",
            datetime(2023, 1, 18, 17, 18, 56, 814, tzinfo=timezone.utc),  # Wednesday
            datetime(2023, 2, 5, 5, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every month on 5th at 5:03 - generated by Kibana; without overflow
            "0 3 5 5 * ?",
            datetime(2023, 1, 5, 1, 18, 56, 814, tzinfo=timezone.utc),  # Wednesday
            datetime(2023, 1, 5, 5, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every year in January on 5th at 5:03 - generated by Kibana; without overflow
            "0 3 5 5 1 ?",
            datetime(2023, 1, 5, 1, 18, 56, 814, tzinfo=timezone.utc),  # Wednesday
            datetime(2023, 1, 5, 5, 3, 0, tzinfo=timezone.utc),
        ),
        (
            # Every year in January on 5th at 5:03 - generated by Kibana; with overflow
            "0 3 5 5 1 ?",
            datetime(2023, 1, 5, 6, 18, 56, 814, tzinfo=timezone.utc),  # Wednesday
            datetime(2024, 1, 5, 5, 3, 0, tzinfo=timezone.utc),
        ),
    ],
)
def test_next_run(cron_statement, now, expected_next_run):
    # can run within two minutes
    assert next_run(cron_statement, now).isoformat(
        " ", "seconds"
    ) == expected_next_run.isoformat(" ", "seconds")


def test_invalid_names():
    for name in (
        "index?name",
        "index#name",
        "_indexname",
        "-indexname",
        "+indexname",
        "INDEXNAME",
        "..",
        ".",
    ):
        with pytest.raises(InvalidIndexNameError):
            validate_index_name(name)


def test_mem_queue_speed():
    def mem_queue():
        import asyncio

        from connectors.utils import MemQueue

        queue = MemQueue(
            maxmemsize=1024 * 1024, refresh_interval=0.1, refresh_timeout=2
        )

        async def run():
            for _ in range(1000):
                await queue.put("x" * 100)

        asyncio.run(run())

    mem_queue_duration = min(timeit.repeat(mem_queue, number=1, repeat=3))

    def vanilla_queue():
        import asyncio

        queue = asyncio.Queue()

        async def run():
            for _ in range(1000):
                await queue.put("x" * 100)

        asyncio.run(run())

    queue_duration = min(timeit.repeat(vanilla_queue, number=1, repeat=3))

    # mem queue should be 30 times slower at the most
    quotient, _ = divmod(mem_queue_duration, queue_duration)
    assert quotient < 30


@pytest.mark.asyncio
async def test_mem_queue_race():
    item = "small stuff"
    queue = MemQueue(
        maxmemsize=get_size(item) * 2 + 1, refresh_interval=0.01, refresh_timeout=1
    )
    max_size = 0

    async def put_one():
        nonlocal max_size
        await queue.put(item)
        qsize = queue.qmemsize()
        await asyncio.sleep(0)
        if max_size < qsize:
            max_size = qsize

    async def remove_one():
        await queue.get()

    tasks = []
    for _ in range(1000):
        tasks.append(put_one())
        tasks.append(remove_one())

    random.shuffle(tasks)
    tasks = [asyncio.create_task(t) for t in tasks]

    await asyncio.gather(*tasks)
    assert max_size <= get_size(item) * 2


@pytest.mark.asyncio
async def test_mem_queue():
    # Initial timeout is really small so that the test is fast.
    # The part of the test before timeout increase will take at least refresh_timeout
    # seconds to execute, so if timeout is 60 seconds, then it'll take 60+ seconds.
    # Thus we make timeout small and increase it later
    queue = MemQueue(maxmemsize=1024, refresh_interval=0, refresh_timeout=0.15)
    await queue.put("small stuff")

    assert not queue.full()
    assert queue.qmemsize() == asizeof.asizeof("small stuff")

    # let's pile up until it can't accept anymore stuff

    with pytest.raises(asyncio.QueueFull):
        while True:
            await queue.put("x" * 100)

    # We increase the timeout to not be so flaky
    queue.refresh_timeout = 2
    when = []

    async def add_data():
        when.append(time.time())
        # this call gets throttled for at the most 2s before it breaks
        await queue.put("DATA" * 10)
        when.append(time.time())

    async def remove_data():
        await asyncio.sleep(0.1)
        size, item = await queue.get()
        assert (size, item) == (64, "small stuff")
        await asyncio.sleep(0)
        await queue.get()  # removes the 2kb
        assert not queue.full()

    await asyncio.gather(add_data(), remove_data())
    assert when[1] - when[0] < queue.refresh_timeout


@pytest.mark.asyncio
async def test_mem_queue_too_large_item():
    """
    When an item is added to the queue that is larger than the queue capacity then the item is discarded

    Imagine that maximum queue size is 5MB and an item of 6MB is added inside.
    Before the logic was to declare the queue full in this case, but it becomes deadlocked - queue is
    trying to free some space up to fit a 6MB item, but it's not possible.

    After a fix the item is discarded with a log line stating that the document is skipped.
    """
    queue = MemQueue(maxmemsize=1, refresh_interval=0.1, refresh_timeout=0.5)

    # Does not raise an error - it'll be full after the item is added inside
    await queue.put("lala" * 1000)

    with pytest.raises(asyncio.QueueFull) as e:
        await queue.put("x")

    assert e is not None


@pytest.mark.asyncio
async def test_mem_queue_put_nowait():
    queue = MemQueue(
        maxsize=5, maxmemsize=1000, refresh_interval=0.1, refresh_timeout=0.5
    )
    # make queue full by size
    for i in range(5):
        queue.put_nowait(i)

    with pytest.raises(asyncio.QueueFull) as e:
        await queue.put_nowait("x")

    assert e is not None


def test_get_base64_value():
    """This test verify get_base64_value method and convert encoded data into base64"""
    expected_result = get_base64_value("dummy".encode("utf-8"))
    assert expected_result == "ZHVtbXk="


def test_decode_base64_value():
    """This test verify decode_base64_value and decodes base64 encoded data"""
    expected_result = decode_base64_value("ZHVtbXk=".encode("utf-8"))
    assert expected_result == b"dummy"


def test_try_acquire():
    bound_value = 5
    sem = NonBlockingBoundedSemaphore(bound_value)

    for _ in range(bound_value):
        assert sem.try_acquire()

    assert not sem.try_acquire()

    sem.release()
    assert sem.try_acquire()


@pytest.mark.asyncio
async def test_concurrent_runner():
    results = []

    def _results_callback(task):
        results.append(task.result())

    async def coroutine(i):
        await asyncio.sleep(0.1)
        return i

    runner = ConcurrentTasks()
    for i in range(10):
        task = await runner.put(functools.partial(coroutine, i))
        task.add_done_callback(functools.partial(_results_callback))

    await runner.join()
    assert results == list(range(10))


@pytest.mark.asyncio
async def test_concurrent_runner_canceled():
    results = []
    tasks = []

    def _results_callback(task):
        results.append(task.result())

    async def coroutine(i):
        await asyncio.sleep(1)
        return i

    runner = ConcurrentTasks(max_concurrency=10)
    for i in range(10):
        task = await runner.put(functools.partial(coroutine, i))
        task.add_done_callback(functools.partial(_results_callback))
        tasks.append(task)

    runner.cancel()
    await runner.join()
    assert len(tasks) == 10
    for task in tasks:
        assert task.cancelled()
    assert len(results) == 0


@pytest.mark.asyncio
async def test_concurrent_runner_canceled_with_waiting_task():
    results = []

    def _results_callback(task):
        results.append(task.result())

    async def coroutine(i, sleep_time):
        await asyncio.sleep(sleep_time)
        return i

    runner = ConcurrentTasks(max_concurrency=10)
    for i in range(10):
        task = await runner.put(functools.partial(coroutine, i, 1))  # long-running task
        task.add_done_callback(functools.partial(_results_callback))

    # create a task to put a waiting task so that it won't block
    task = asyncio.create_task(runner.put(functools.partial(coroutine, 100, 0.1)))
    task.add_done_callback(functools.partial(_results_callback))
    runner.cancel()
    # wait for the first 10 tasks to be canceled
    await runner.join()
    # wait for the 11th task to complete
    await runner.join()
    assert len(results) == 1
    assert results[0].result() == 100


@pytest.mark.asyncio
async def test_concurrent_runner_fails():
    results = []

    def _results_callback(task):
        results.append(task.result())

    async def coroutine(i):
        await asyncio.sleep(0.1)
        if i == 5:
            msg = "I FAILED"
            raise Exception(msg)
        return i

    runner = ConcurrentTasks()
    for i in range(10):
        task = await runner.put(functools.partial(coroutine, i))
        task.add_done_callback(functools.partial(_results_callback))

    await runner.join()
    assert 5 not in results


@pytest.mark.asyncio
async def test_concurrent_runner_high_concurrency():
    results = []

    def _results_callback(task):
        results.append(task.result())

    async def coroutine(i):
        await asyncio.sleep(0)
        return i

    second_results = []

    def _second_callback(task):
        second_results.append(task.result())

    runner = ConcurrentTasks()
    for i in range(1000):
        task = await runner.put(functools.partial(coroutine, i))
        task.add_done_callback(functools.partial(_results_callback))
        if i == 3:
            task.add_done_callback(functools.partial(_second_callback))

    await runner.join()
    assert results == list(range(1000))
    assert second_results == [3]


@pytest.mark.parametrize(
    "initial_capacity, expected_result",
    [
        (0, True),  # when pool is empty
        (2, True),  # when pool is almost empty
        (5, True),  # when pool is half full
        (8, True),  # when pool is almost full
        (10, False),  # when pool is full
    ],
)
@pytest.mark.asyncio
async def test_concurrent_runner_try_put(initial_capacity, expected_result):
    results = []

    def _results_callback(task):
        results.append(task.result())

    async def coroutine(i):
        await asyncio.sleep(0.1)
        return i

    runner = ConcurrentTasks(10)
    # fill the pool with initial capacity
    for i in range(initial_capacity):
        task = await runner.put(functools.partial(coroutine, i))
        task.add_done_callback(functools.partial(_results_callback))

    # try to put an additional task
    task = runner.try_put(functools.partial(coroutine, 100))
    if task:
        task.add_done_callback(functools.partial(_results_callback))

    await runner.join()

    if expected_result:
        assert task is not None
        assert 100 in results
    else:
        assert task is None
        assert 100 not in results


@pytest.mark.asyncio
async def test_concurrent_runner_join():
    results = []

    def _results_callback(task):
        results.append(task.result())

    async def coroutine(i):
        await asyncio.sleep(0.2)
        return i

    runner = ConcurrentTasks()
    for i in range(3):
        task = await runner.put(functools.partial(coroutine, i))
        task.add_done_callback(functools.partial(_results_callback))

    async def delayed_coroutine():
        await asyncio.sleep(0.1)
        task = await runner.put(functools.partial(coroutine, 3))
        task.add_done_callback(functools.partial(_results_callback))

    # put the 4th task after 0.1 second during the execution of the first 3 tasks
    asyncio.create_task(delayed_coroutine())

    # calling join will wait for the first 3 tasks to finish, the 4th task is not added yet
    await runner.join()
    assert len(results) == 3
    # the fouth task should still in the pool
    assert 3 not in results
    assert len(runner) == 1

    # wait for the 4th task to finish
    await runner.join()
    assert len(results) == 4
    assert 3 in results


@pytest.mark.asyncio
async def test_concurrent_tasks_raise_any_exception():
    async def return_1():
        return 1

    async def raise_error():
        msg = "This task had an error"
        raise Exception(msg)

    async def long_sleep_then_return_2():
        try:
            await asyncio.sleep(100)
            return 2
        except asyncio.CancelledError:
            pass

    runner = ConcurrentTasks()
    await runner.put(functools.partial(return_1))
    await runner.put(functools.partial(raise_error))
    await runner.put(functools.partial(long_sleep_then_return_2))

    await asyncio.sleep(0)
    with pytest.raises(Exception):
        runner.raise_any_exception()


@pytest.mark.asyncio
async def test_concurrent_tasks_join_raise_on_error():
    results = []

    def _results_callback(task):
        results.append(task.result())

    async def return_1():
        return 1

    async def raise_error():
        msg = "This task had an error"
        raise Exception(msg)

    async def long_sleep_then_return_2():
        try:
            await asyncio.sleep(100)
            return 2
        except asyncio.CancelledError:
            pass

    runner = ConcurrentTasks()
    task1 = await runner.put(functools.partial(return_1))
    task1.add_done_callback(functools.partial(_results_callback))
    await runner.put(functools.partial(raise_error))
    task3 = await runner.put(functools.partial(long_sleep_then_return_2))
    task3.add_done_callback(functools.partial(_results_callback))

    with pytest.raises(Exception):
        await runner.join(raise_on_error=True)

    assert len(results) == 1
    assert results[0] == 1


class CustomException(Exception):
    pass


class CustomGeneratorException(Exception):
    pass


@pytest.mark.fail_slow(1)
@pytest.mark.asyncio
async def test_exponential_backoff_retry_async_generator():
    mock_gen = Mock()
    num_retries = 10

    @retryable(
        retries=num_retries,
        interval=0,
        strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
    )
    async def raises_async_generator():
        for _ in range(3):
            mock_gen()
            raise CustomGeneratorException()
            yield 1

    with pytest.raises(CustomGeneratorException):
        async for _ in raises_async_generator():
            pass

    # retried 10 times
    assert mock_gen.call_count == num_retries

    # would lead to roughly ~ 50 seconds of retrying
    @retryable(retries=10, interval=5, strategy=RetryStrategy.LINEAR_BACKOFF)
    async def does_not_raise_async_generator():
        for _ in range(3):
            yield 1

    # would fail, if retried once (retry_interval = 5 seconds). Explicit time boundary for this test: 1 second
    items = []
    async for item in does_not_raise_async_generator():
        items.append(item)

    assert items == [1, 1, 1]


@pytest.mark.fail_slow(1)
def test_exponential_backoff_retry_sync_function():
    mock_func = Mock()
    num_retries = 10

    @retryable(
        retries=num_retries,
        interval=0,
        strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
    )
    def raises_function():
        for _ in range(3):
            mock_func()
            raise CustomException()

    with pytest.raises(CustomException):
        for _ in raises_function():
            pass

    # retried 10 times
    assert mock_func.call_count == num_retries

    # would lead to roughly ~ 50 seconds of retrying
    @retryable(retries=10, interval=5, strategy=RetryStrategy.LINEAR_BACKOFF)
    def does_not_raise_function():
        for _ in range(3):
            yield 1

    # would fail, if retried once (retry_interval = 5 seconds). Explicit time boundary for this test: 1 second
    items = []
    for item in does_not_raise_function():
        items.append(item)

    assert items == [1, 1, 1]


@pytest.mark.fail_slow(1)
@pytest.mark.asyncio
async def test_exponential_backoff_retry_async_function():
    mock_func = Mock()
    num_retries = 10

    @retryable(
        retries=num_retries,
        interval=0,
        strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
    )
    async def raises():
        mock_func()
        raise CustomException()

    with pytest.raises(CustomException):
        await raises()

        # retried 10 times
        assert mock_func.call_count == num_retries

    # would lead to roughly ~ 50 seconds of retrying
    @retryable(retries=10, interval=5, strategy=RetryStrategy.LINEAR_BACKOFF)
    async def does_not_raise():
        pass

    # would fail, if retried once (retry_interval = 5 seconds). Explicit time boundary for this test: 1 second
    await does_not_raise()


@pytest.mark.parametrize(
    "skipped_exceptions",
    [CustomGeneratorException, [CustomGeneratorException, RuntimeError]],
)
@pytest.mark.asyncio
async def test_skipped_exceptions_retry_async_generator(skipped_exceptions):
    mock_gen = Mock()
    num_retries = 10

    @retryable(
        retries=num_retries,
        skipped_exceptions=skipped_exceptions,
    )
    async def raises_async_generator():
        for _ in range(3):
            mock_gen()
            raise CustomGeneratorException()
            yield 1

    with pytest.raises(CustomGeneratorException):
        async for _ in raises_async_generator():
            pass

    assert mock_gen.call_count == 1


@pytest.mark.parametrize(
    "skipped_exceptions", [CustomException, [CustomException, RuntimeError]]
)
@pytest.mark.asyncio
async def test_skipped_exceptions_retry_async_function(skipped_exceptions):
    mock_func = Mock()
    num_retries = 10

    @retryable(
        retries=num_retries,
        skipped_exceptions=skipped_exceptions,
    )
    async def raises():
        mock_func()
        raise CustomException()

    with pytest.raises(CustomException):
        await raises()

        assert mock_func.call_count == 1


@pytest.mark.parametrize(
    "skipped_exceptions", [CustomException, [CustomException, RuntimeError]]
)
@pytest.mark.asyncio
async def test_skipped_exceptions_retry_sync_function(skipped_exceptions):
    mock_func = Mock()
    num_retries = 10

    @retryable(
        retries=num_retries,
        skipped_exceptions=skipped_exceptions,
    )
    def raises():
        mock_func()
        raise CustomException()

    with pytest.raises(CustomException):
        await raises()

        assert mock_func.call_count == 1


def test_retryable_not_implemented_error():
    with pytest.raises(NotImplementedError):

        @retryable()
        class NotSupported:
            pass


class MockSSL:
    """This class contains methods which returns dummy ssl context"""

    def load_verify_locations(self, cadata):
        """This method verify locations"""
        pass


def test_ssl_context():
    """This function test ssl_context with dummy certificate"""
    # Setup
    certificate = "-----BEGIN CERTIFICATE----- Certificate -----END CERTIFICATE-----"

    # Execute
    with patch.object(ssl, "create_default_context", return_value=MockSSL()):
        ssl_context(certificate=certificate)


def test_url_encode():
    """Test the url_encode method by passing a string"""
    # Execute
    encode_response = url_encode("http://ascii.cl?parameter='Click on URL Decode!'")
    # Assert
    assert (
        encode_response
        == "http%3A%2F%2Fascii.cl%3Fparameter%3D'Click%20on%20URL%20Decode%21'"
    )


def test_is_expired():
    """This method checks whether token expires or not"""
    # Execute
    expires_at = datetime.fromisoformat("2023-02-10T09:02:23.629821")
    actual_response = is_expired(expires_at=expires_at)
    # Assert
    assert actual_response is True


@freeze_time("2023-02-18 14:25:26.158843", tz_offset=-4)
def test_evaluate_timedelta():
    """This method tests adding seconds to the current utc time"""
    # Execute
    expected_response = evaluate_timedelta(seconds=86399, time_skew=20)

    # Assert
    assert expected_response == "2023-02-19T14:25:05.158843"


MULTILINE_PEM_CERTIFICATE = """-----BEGIN CERTIFICATE-----
Certificate1
Certificate2
-----END CERTIFICATE-----"""


def test_get_pem_format_with_postfix():
    expected_formatted_pem_key = """-----BEGIN PRIVATE KEY-----
PrivateKey
-----END PRIVATE KEY-----"""
    private_key = "-----BEGIN PRIVATE KEY----- PrivateKey -----END PRIVATE KEY-----"

    formatted_private_key = get_pem_format(
        key=private_key, postfix="-----END PRIVATE KEY-----"
    )
    assert formatted_private_key == expected_formatted_pem_key


def test_get_pem_format_multiline():
    certificate = "-----BEGIN CERTIFICATE----- Certificate1 Certificate2 -----END CERTIFICATE-----"

    assert get_pem_format(key=certificate) == MULTILINE_PEM_CERTIFICATE


def test_get_pem_format_multiple_certificates():
    expected_formatted_multiple_certificates = """-----BEGIN CERTIFICATE-----
Certificate1
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
Certificate2
-----END CERTIFICATE-----
"""
    multiple_certificates = "-----BEGIN CERTIFICATE----- Certificate1 -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- Certificate2 -----END CERTIFICATE-----"

    formatted_multi_certificate = get_pem_format(key=multiple_certificates)
    assert formatted_multi_certificate == expected_formatted_multiple_certificates


def test_get_pem_format_already_multiline_certificate_is_preserved():
    # Copied verbatim from a .pem file: the intact END marker must not be broken.
    assert get_pem_format(key=MULTILINE_PEM_CERTIFICATE) == MULTILINE_PEM_CERTIFICATE


def test_get_pem_format_already_multiline_private_key_is_preserved():
    private_key = """-----BEGIN PRIVATE KEY-----
PrivateKey
-----END PRIVATE KEY-----"""

    assert (
        get_pem_format(key=private_key, postfix="-----END PRIVATE KEY-----")
        == private_key
    )


def test_get_pem_format_multiline_with_surrounding_whitespace_is_normalized():
    certificate = (
        "\n  -----BEGIN CERTIFICATE-----  \n"
        "Certificate1\n"
        "\n"
        "Certificate2\n"
        "  -----END CERTIFICATE-----  \n"
    )

    assert get_pem_format(key=certificate) == MULTILINE_PEM_CERTIFICATE


def test_get_pem_format_single_line_with_trailing_newline_is_reflowed():
    # A trailing newline alone must not be mistaken for multi-line content.
    certificate = (
        "-----BEGIN CERTIFICATE----- Certificate1 Certificate2 "
        "-----END CERTIFICATE-----\n"
    )

    assert get_pem_format(key=certificate) == MULTILINE_PEM_CERTIFICATE


def test_truncate_id():
    long_id = "something-12341361361-21905128510263"
    truncated_id = truncate_id(long_id)

    assert len(truncated_id) < len(long_id)


@pytest.mark.parametrize(
    "_list, should_have_duplicate",
    [([], False), (["abc"], False), (["abc", "def"], False), (["abc", "abc"], True)],
)
def test_has_duplicates(_list, should_have_duplicate):
    assert has_duplicates(_list) == should_have_duplicate


@pytest.mark.parametrize(
    "key_list, source_dict, expected_dict",
    [
        (["one", "two"], {"foo": {"one": 1, "two": 2}}, {}),
        (
            ["one", "two"],
            {"foo": {"one": 1, "two": 2}, "bar": {"one": 1, "three": 3}},
            {"bar": {"one": 1, "three": 3}},
        ),
        (
            ["one", "two"],
            {"foo": {}},
            {"foo": {}},
        ),
        (
            [],
            {"foo": {"one": 1, "two": 2}, "bar": {"one": 1, "three": 3}},
            {},
        ),
    ],
)
def test_filter_nested_dict_by_keys(key_list, source_dict, expected_dict):
    assert filter_nested_dict_by_keys(key_list, source_dict) == expected_dict


@pytest.mark.parametrize(
    "base_dict, new_dict, expected_dict",
    [
        (
            {"foo": {"one": 1, "two": 2}, "bar": {"one": 1, "two": 2}},
            {"foo": {"one": 10}, "bar": {"three": 30}},
            {"foo": {"one": 10, "two": 2}, "bar": {"one": 1, "two": 2, "three": 30}},
        ),
        (
            {"foo": {}, "bar": {}},
            {"foo": {"one": 1}, "bar": {"three": 3}},
            {"foo": {"one": 1}, "bar": {"three": 3}},
        ),
        (
            {
                "foo": {"level2": {"level3": "fafa"}},
                "bar": {"level2": {"level3": "fafa"}},
            },
            {
                "foo": {"level2": {"level3": "foofoo"}},
                "bar": {"level2": {"level4": "haha"}},
            },
            {
                "foo": {"level2": {"level3": "foofoo"}},
                "bar": {"level2": {"level3": "fafa", "level4": "haha"}},
            },
        ),
    ],
)
def test_deep_merge_dicts(base_dict, new_dict, expected_dict):
    assert deep_merge_dicts(base_dict, new_dict) == expected_dict


def test_html_to_text_with_html_with_unclosed_tag():
    invalid_html = "<div>Hello, world!</div><div>Next Line"

    assert html_to_text(invalid_html) == "Hello, world!\nNext Line"


def test_html_to_text_without_html():
    invalid_html = "just text"

    assert html_to_text(invalid_html) == "just text"


def test_html_to_text_with_weird_html():
    invalid_html = "<div/>just</div> text"

    text = html_to_text(invalid_html)

    assert "just" in text
    assert "text" in text


def test_html_to_text_with_none():
    assert html_to_text(None) is None


def test_html_to_text_with_lxml_exception():
    # Here we're just mocking it in such a way, that
    # using BeautifulSoup(html, "lxml") raises an error to emulate the fact
    # that lxml is not available.
    def _init_func(html, parser_type=None, features=None):
        if parser_type == "lxml":
            msg = "LXML not available"
            raise Exception(msg)
        else:
            parser_mock = Mock()
            return parser_mock

    with patch("connectors.utils.BeautifulSoup") as beautiful_soup_patch:
        beautiful_soup_patch.side_effect = _init_func
        html = "lala <br/>"

        # assert it does not throw
        html_to_text(html)

        # and assert is uses default parser
        beautiful_soup_patch.assert_called_with(html, features="html.parser")


def batch_size(value):
    """Used for readability purposes in parametrized tests."""
    return value


@pytest.mark.parametrize(
    "iterable, batch_size_, expected_batches",
    [
        ([1, 2, 3], batch_size(1), [[1], [2], [3]]),
        ([1, 2, 3], batch_size(2), [[1, 2], [3]]),
        (
            [1, 2, 3],
            batch_size(3),
            [
                [1, 2, 3],
            ],
        ),
        (
            [1, 2, 3],
            batch_size(1000),
            [
                [1, 2, 3],
            ],
        ),
        ([], batch_size(20), []),
        ([[]], batch_size(20), [[[]]]),
    ],
)
def test_iterable_batches_generator(iterable, batch_size_, expected_batches):
    actual_batches = []

    for batch in iterable_batches_generator(iterable, batch_size_):
        actual_batches.append(batch)

    assert actual_batches == expected_batches


@pytest.mark.parametrize(
    "base64url_encoded_value, base64_expected_value",
    [("YQ-_", "YQ+/"), ("", ""), (None, None)],
)
def test_base64url_to_base64(base64url_encoded_value, base64_expected_value):
    assert base64url_to_base64(base64url_encoded_value) == base64_expected_value


@pytest.mark.parametrize(
    "email_address, is_valid",
    [
        ("subject@email_address.com", True),
        ("subject", False),
        ("@email_address.com", False),
        ("", False),
        ("subject @email_address.com", False),
        ("subject@email_address", False),
    ],
)
def test_validate_email_address(email_address, is_valid):
    assert validate_email_address(email_address) == is_valid


@pytest.mark.parametrize(
    "strategy, interval, retry, expected_sleep",
    [
        (RetryStrategy.CONSTANT, 10, 2, 10),  # Constant 10 seconds
        (RetryStrategy.LINEAR_BACKOFF, 10, 0, 0),  # 10 * 0 = 0
        (RetryStrategy.LINEAR_BACKOFF, 10, 1, 10),  # 10 * 1 = 10
        (RetryStrategy.LINEAR_BACKOFF, 10, 2, 20),  # 10 * 2 = 20
        (RetryStrategy.EXPONENTIAL_BACKOFF, 10, 0, 1),  # 10 ^ 0 = 1
        (RetryStrategy.EXPONENTIAL_BACKOFF, 10, 1, 10),  # 10 ^ 1 = 10
        (RetryStrategy.EXPONENTIAL_BACKOFF, 10, 2, 100),  # 10 ^ 2 = 100
    ],
)
async def test_time_to_sleep_between_retries(strategy, interval, retry, expected_sleep):
    assert time_to_sleep_between_retries(strategy, interval, retry) == expected_sleep


async def test_time_to_sleep_between_retries_invalid_strategy():
    with pytest.raises(UnknownRetryStrategyError) as e:
        time_to_sleep_between_retries("lalala", 1, 1)

    assert e is not None


@pytest.mark.parametrize(
    "string, parsed_datetime",
    [
        (
            "2024-05-15T12:37:52.429924009Z",
            datetime(2024, 5, 15, 12, 37, 52, 429924, tzutc()),
        ),
        (
            "2024-05-15T12:37:52.429Z",
            datetime(2024, 5, 15, 12, 37, 52, 429000, tzutc()),
        ),
    ],
)
def test_parse_datetime_string_compatibility(string, parsed_datetime):
    assert parse_datetime_string(string) == parsed_datetime


def test_error_monitor_raises_after_too_many_errors_in_window():
    error_monitor = ErrorMonitor(max_error_rate=0.15, error_window_size=100)

    for _ in range(10):
        error_monitor.track_error(Exception("Something went wrong"))

    for _ in range(84):
        error_monitor.track_success()

    for _ in range(5):
        error_monitor.track_error(Exception("Something went wrong"))

    with pytest.raises(TooManyErrors):
        error_monitor.track_error(InvalidIndexNameError("Can't use this name"))


def test_error_monitor_raises_when_errors_were_reported_before():
    # Regression test.
    # Problem fixed was that monitor incorrectly calculates max_error_ratio - it never
    # actually considered either ratio or window size - it was always raising an error if
    # window_size * max_error_ratio errors happened, which is 15 in case of this setup.
    # What it should do is really consider the window and error ratio, e.g:
    # 85 documents were correctly ingested, 15 failed. Window will be: 85 x success, 15 x failure,
    # error_ratio = 0.15, but condition to raise is max_error_ratio < error_ratio - it's false, so
    # no error is raised.
    #
    # Then 90 documents were ingested correctly, window moves and will be: 10 x failure, 90 x success,
    # error_ratio = 0.1; Then 10 errors happen, but error_ratio will stay the same because window will be
    # 90 x success, 10 x failure.

    error_monitor = ErrorMonitor(
        max_error_rate=0.15, error_window_size=100, max_consecutive_errors=100
    )

    # First set things up
    for _ in range(2):
        for _ in range(5):
            error_monitor.track_error(Exception("Something went wrong"))

        for _ in range(40):
            error_monitor.track_success()

    # Then do actual test
    # Before:
    # 5 x failure; 40 x success; 5 x failure; 40 x success, real error_ratio = 0.1
    # After:
    # 40 x success; 5 x failure; 40 x success; 5 x failure, real error_ratio = 0.1
    for _ in range(5):
        error_monitor.track_error(Exception("Something went wrong"))

    # Before:
    # 40 x success; 5 x failure; 40 x success; 5 x failure, real error_ratio = 0.1
    # After:
    # 1 x success; 5x failure; 94 x success, real_error_ratio = 0.05
    for _ in range(94):
        error_monitor.track_success()

    # Before:
    # 1 x success; 5 x failure; 94 x success, real_error_ratio = 0.05
    # After:
    # 85 x success, 15 x failure, real_error_ratio = 0.15.
    # Any error within next 85 documents should trigger the MaxErrorsInWindowExceededError
    for _ in range(15):
        error_monitor.track_error(Exception("Something went wrong again"))

    with pytest.raises(TooManyErrors):
        error_monitor.track_error(InvalidIndexNameError("Can't use this name"))


def test_error_monitor_when_reports_too_many_consecutive_errors():
    error_monitor = ErrorMonitor(max_consecutive_errors=3)

    error_monitor.track_error(Exception("first"))
    error_monitor.track_error(Exception("second"))
    error_monitor.track_error(Exception("third"))

    with pytest.raises(TooManyErrors):
        error_monitor.track_error(Exception("fourth"))


def test_error_monitor_when_reports_too_many_total_errors():
    error_monitor = ErrorMonitor(
        max_total_errors=100, max_consecutive_errors=999, max_error_rate=1
    )

    for _ in range(10):
        error_monitor.track_error(Exception("first"))

    for _ in range(500):
        error_monitor.track_success()

    for _ in range(90):
        error_monitor.track_error(Exception("second"))

    with pytest.raises(TooManyErrors):
        error_monitor.track_error(Exception("third"))


def test_error_monitor_when_reports_too_many_errors_in_window():
    error_monitor = ErrorMonitor(error_window_size=100, max_error_rate=0.05)

    # rate is 0.04
    for _ in range(4):
        error_monitor.track_error(Exception("first"))

    for _ in range(95):
        error_monitor.track_success()

    # rate gets to 0.05 but does not raise yet
    for _ in range(1):
        error_monitor.track_error(Exception("second"))

    # rate stays 0.05 here
    for _ in range(4):
        error_monitor.track_error(Exception("third"))

    with pytest.raises(TooManyErrors):
        error_monitor.track_error(Exception("last"))


def test_error_monitor_when_errors_are_tracked_last_x_errors_are_stored():
    error_monitor = ErrorMonitor(error_queue_size=5)

    for _ in range(5):
        error_monitor.track_error(Exception("first_part"))

    for _ in range(999):
        error_monitor.track_success()

    for _ in range(3):
        error_monitor.track_error(Exception("second_part"))

    errors = error_monitor.error_queue

    assert str(errors[0]) == "first_part"
    assert str(errors[1]) == "first_part"
    assert str(errors[2]) == "second_part"
    assert str(errors[3]) == "second_part"
    assert str(errors[4]) == "second_part"


def test_error_monitor_when_disabled():
    error_monitor = ErrorMonitor(
        enabled=False, max_total_errors=1, max_consecutive_errors=1, max_error_rate=0.01
    )

    for _ in range(9999):
        error_monitor.track_error(Exception("second_part"))


class MyConnector:
    id = "1"  # noqa A003
    service_type = "yea"

    def __init__(self, *args):
        pass


def test_get_source_klass():
    assert get_source_klass("tests.test_utils:MyConnector") is MyConnector


def test_get_source_klasses():
    settings = {
        "sources": {
            "yea": "tests.test_utils:MyConnector",
            "yea2": "tests.test_utils:MyConnector",
        }
    }

    sources = list(get_source_klasses(settings))
    assert sources == [MyConnector, MyConnector]
