"""Test MQTT utils."""

import asyncio
from collections.abc import Callable
from datetime import timedelta
from pathlib import Path
import shutil
import tempfile
from unittest.mock import MagicMock, patch

import pytest

from homeassistant.components import mqtt
from homeassistant.components.mqtt.const import DOMAIN
from homeassistant.components.mqtt.models import MessageCallbackType
from homeassistant.components.mqtt.util import EnsureJobAfterCooldown
from homeassistant.config_entries import ConfigEntryDisabler, ConfigEntryState
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import CoreState, HomeAssistant
from homeassistant.util.dt import utcnow

from tests.common import MockConfigEntry, async_fire_time_changed
from tests.typing import MqttMockHAClient, MqttMockPahoClient


async def test_canceling_debouncer_on_shutdown(
    hass: HomeAssistant,
    record_calls: MessageCallbackType,
    mock_debouncer: asyncio.Event,
    setup_with_birth_msg_client_mock: MqttMockPahoClient,
) -> None:
    """Test canceling the debouncer when HA shuts down."""
    mqtt_client_mock = setup_with_birth_msg_client_mock
    # Mock we are past initial setup
    await mock_debouncer.wait()
    with patch("homeassistant.components.mqtt.client.SUBSCRIBE_COOLDOWN", 2):
        mock_debouncer.clear()
        await mqtt.async_subscribe(hass, "test/state1", record_calls)
        async_fire_time_changed(hass, utcnow() + timedelta(seconds=0.1))
        # Stop HA so the scheduled debouncer task will be canceled
        mqtt_client_mock.subscribe.reset_mock()
        hass.bus.fire(EVENT_HOMEASSISTANT_STOP)
        await mqtt.async_subscribe(hass, "test/state2", record_calls)
        async_fire_time_changed(hass, utcnow() + timedelta(seconds=0.1))
        await mqtt.async_subscribe(hass, "test/state3", record_calls)
        async_fire_time_changed(hass, utcnow() + timedelta(seconds=0.1))
        await mqtt.async_subscribe(hass, "test/state4", record_calls)
        async_fire_time_changed(hass, utcnow() + timedelta(seconds=0.1))
        await mqtt.async_subscribe(hass, "test/state5", record_calls)
        async_fire_time_changed(hass, utcnow() + timedelta(seconds=5))
        await hass.async_block_till_done(wait_background_tasks=True)
        # Assert the debouncer subscribe job was not executed
        assert not mock_debouncer.is_set()
        mqtt_client_mock.subscribe.assert_not_called()

        # Note that the broker connection will not be disconnected gracefully
        await hass.async_block_till_done()
        async_fire_time_changed(hass, utcnow() + timedelta(seconds=5))
        await asyncio.sleep(0)
        await hass.async_block_till_done(wait_background_tasks=True)
        mqtt_client_mock.subscribe.assert_not_called()
        mqtt_client_mock.disconnect.assert_not_called()


async def test_canceling_debouncer_normal(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test canceling the debouncer before completion."""

    async def _async_myjob() -> None:
        await asyncio.sleep(1.0)

    debouncer = EnsureJobAfterCooldown(0.0, _async_myjob)
    debouncer.async_schedule()
    await asyncio.sleep(0.01)
    assert debouncer._task is not None
    await debouncer.async_cleanup()
    assert debouncer._task is None


async def test_canceling_debouncer_throws(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test canceling the debouncer when HA shuts down."""

    async def _async_myjob() -> None:
        await asyncio.sleep(1.0)

    debouncer = EnsureJobAfterCooldown(0.0, _async_myjob)
    debouncer.async_schedule()
    await asyncio.sleep(0.01)
    assert debouncer._task is not None
    # let debouncer._task fail by mocking it
    with patch.object(debouncer, "_task") as task:
        task.cancel = MagicMock(return_value=True)
        await debouncer.async_cleanup()
        assert "Error cleaning up task" in caplog.text
        await hass.async_block_till_done()
        async_fire_time_changed(hass, utcnow() + timedelta(seconds=5))
        await hass.async_block_till_done()


async def help_create_test_certificate_file(
    hass: HomeAssistant,
    mock_temp_dir: str,
    option: str,
    content: bytes = b"old content",
) -> None:
    """Help creating a certificate test file."""
    temp_dir = Path(tempfile.gettempdir()) / mock_temp_dir

    def _create_file() -> None:
        if not temp_dir.exists():
            temp_dir.mkdir(0o700)
        temp_file = temp_dir / option
        with open(temp_file, "wb") as old_file:
            old_file.write(content)
            old_file.close()

    await hass.async_add_executor_job(_create_file)


@pytest.mark.parametrize(
    ("option", "content"),
    [
        (mqtt.CONF_CERTIFICATE, "### CA CERTIFICATE ###"),
        (mqtt.CONF_CLIENT_CERT, "### CLIENT CERTIFICATE ###"),
        (mqtt.CONF_CLIENT_KEY, "### PRIVATE KEY ###"),
    ],
)
@pytest.mark.parametrize("temp_dir_prefix", ["create-test1"])
async def test_async_create_certificate_temp_files(
    hass: HomeAssistant,
    mock_temp_dir: str,
    option: str,
    content: str,
) -> None:
    """Test creating and reading and recovery certificate files."""
    config = {option: content}

    # Create old file to be able to assert it is replaced and recovered
    await help_create_test_certificate_file(hass, mock_temp_dir, option)
    await mqtt.util.async_create_certificate_temp_files(hass, config)
    file_path = await hass.async_add_executor_job(mqtt.util.get_file_path, option)
    assert file_path is not None
    assert (
        await hass.async_add_executor_job(
            mqtt.util.migrate_certificate_file_to_content, file_path
        )
        == content
    )

    # Make sure old files are removed to test certificate and dir creation
    def _remove_old_files() -> None:
        temp_dir = Path(tempfile.gettempdir()) / mock_temp_dir
        shutil.rmtree(temp_dir)

    await hass.async_add_executor_job(_remove_old_files)

    # Test a new dir and file is created correctly
    await mqtt.util.async_create_certificate_temp_files(hass, config)
    file_path = await hass.async_add_executor_job(mqtt.util.get_file_path, option)
    assert file_path is not None
    assert (
        await hass.async_add_executor_job(
            mqtt.util.migrate_certificate_file_to_content, file_path
        )
        == content
    )


@pytest.mark.parametrize("temp_dir_prefix", ["create-test2"])
async def test_certificate_temp_files_with_auto_mode(
    hass: HomeAssistant,
    mock_temp_dir: str,
) -> None:
    """Test creating and reading and recovery certificate files with auto mode."""
    config = {mqtt.CONF_CERTIFICATE: "auto"}

    # Create old file to be able to assert it is removed with auto option
    await help_create_test_certificate_file(hass, mock_temp_dir, mqtt.CONF_CERTIFICATE)
    await mqtt.util.async_create_certificate_temp_files(hass, config)
    file_path = await hass.async_add_executor_job(mqtt.util.get_file_path, "auto")
    assert file_path is None
    assert (
        await hass.async_add_executor_job(
            mqtt.util.migrate_certificate_file_to_content, "auto"
        )
        == "auto"
    )


async def test_reading_non_exitisting_certificate_file() -> None:
    """Test reading a non existing certificate file."""
    assert (
        mqtt.util.migrate_certificate_file_to_content("/home/file_not_exists") is None
    )


async def test_return_default_get_file_path(
    hass: HomeAssistant, mock_temp_dir: str
) -> None:
    """Test get_file_path returns default."""

    def _get_file_path(file_path: Path) -> bool:
        return (
            not file_path.exists()
            and mqtt.util.get_file_path("some_option", "mydefault") == "mydefault"
        )

    temp_dir = Path(tempfile.gettempdir()) / mock_temp_dir
    assert await hass.async_add_executor_job(_get_file_path, temp_dir)


async def test_waiting_for_client_not_loaded(
    hass: HomeAssistant,
    mqtt_client_mock: MqttMockPahoClient,
) -> None:
    """Test waiting for client while mqtt entry is not yet loaded."""
    hass.set_state(CoreState.starting)
    await hass.async_block_till_done()

    entry = MockConfigEntry(
        domain=DOMAIN,
        data={"broker": "test-broker"},
        state=ConfigEntryState.NOT_LOADED,
        version=mqtt.CONFIG_ENTRY_VERSION,
        minor_version=mqtt.CONFIG_ENTRY_MINOR_VERSION,
    )
    entry.add_to_hass(hass)

    unsubs: list[Callable[[], None]] = []

    async def _async_just_in_time_subscribe() -> None:
        assert await mqtt.async_wait_for_mqtt_client(hass)
        # Awaiting a second time should work too and return True
        assert await mqtt.async_wait_for_mqtt_client(hass)
        unsubs.append(await mqtt.async_subscribe(hass, "test_topic", lambda msg: None))

    # Simulate some integration waiting for the client to become available
    for _ in range(4):
        hass.async_create_task(_async_just_in_time_subscribe())

    assert entry.state is ConfigEntryState.NOT_LOADED
    assert len(unsubs) == 0
    assert await hass.config_entries.async_setup(entry.entry_id)
    await hass.async_block_till_done()
    assert len(unsubs) == 4
    for unsub in unsubs:
        unsub()


async def test_waiting_for_client_loaded(
    hass: HomeAssistant,
    mqtt_mock: MqttMockHAClient,
) -> None:
    """Test waiting for client where mqtt entry is loaded."""
    unsub: Callable[[], None] | None = None

    async def _async_just_in_time_subscribe() -> None:
        nonlocal unsub
        assert await mqtt.async_wait_for_mqtt_client(hass)
        unsub = await mqtt.async_subscribe(hass, "test_topic", lambda msg: None)

    entry = hass.config_entries.async_entries(DOMAIN)[0]
    assert entry.state is ConfigEntryState.LOADED

    await _async_just_in_time_subscribe()

    assert unsub is not None
    unsub()


async def test_waiting_for_client_entry_fails(
    hass: HomeAssistant,
    mqtt_client_mock: MqttMockPahoClient,
) -> None:
    """Test waiting for client where mqtt entry is failing."""
    hass.set_state(CoreState.starting)
    await hass.async_block_till_done()

    entry = MockConfigEntry(
        domain=DOMAIN,
        data={"broker": "test-broker"},
        state=ConfigEntryState.NOT_LOADED,
        version=mqtt.CONFIG_ENTRY_VERSION,
        minor_version=mqtt.CONFIG_ENTRY_MINOR_VERSION,
    )
    entry.add_to_hass(hass)

    async def _async_just_in_time_subscribe() -> None:
        assert not await mqtt.async_wait_for_mqtt_client(hass)

    hass.async_create_task(_async_just_in_time_subscribe())
    assert entry.state is ConfigEntryState.NOT_LOADED
    with patch(
        "homeassistant.components.mqtt.async_setup_entry",
        side_effect=Exception,
    ):
        await hass.config_entries.async_setup(entry.entry_id)
    assert entry.state is ConfigEntryState.SETUP_ERROR  # type:ignore[comparison-overlap]


async def test_waiting_for_client_setup_fails(
    hass: HomeAssistant,
    mqtt_client_mock: MqttMockPahoClient,
) -> None:
    """Test waiting for client where mqtt entry is failing during setup."""
    hass.set_state(CoreState.starting)
    await hass.async_block_till_done()

    entry = MockConfigEntry(
        domain=DOMAIN,
        data={"broker": "test-broker"},
        state=ConfigEntryState.NOT_LOADED,
        version=mqtt.CONFIG_ENTRY_VERSION,
        minor_version=mqtt.CONFIG_ENTRY_MINOR_VERSION,
    )
    entry.add_to_hass(hass)

    async def _async_just_in_time_subscribe() -> None:
        assert not await mqtt.async_wait_for_mqtt_client(hass)

    hass.async_create_task(_async_just_in_time_subscribe())
    assert entry.state is ConfigEntryState.NOT_LOADED

    # Simulate MQTT setup fails before the client would become available
    mqtt_client_mock.connect.side_effect = Exception
    assert not await hass.config_entries.async_setup(entry.entry_id)
    assert entry.state is ConfigEntryState.SETUP_ERROR  # type:ignore[comparison-overlap]


@patch("homeassistant.components.mqtt.util.AVAILABILITY_TIMEOUT", 0.01)
async def test_waiting_for_client_timeout(
    hass: HomeAssistant,
) -> None:
    """Test waiting for client with timeout."""
    hass.set_state(CoreState.starting)
    await hass.async_block_till_done()

    entry = MockConfigEntry(
        domain=DOMAIN,
        data={"broker": "test-broker"},
        state=ConfigEntryState.NOT_LOADED,
        version=mqtt.CONFIG_ENTRY_VERSION,
        minor_version=mqtt.CONFIG_ENTRY_MINOR_VERSION,
    )
    entry.add_to_hass(hass)

    assert entry.state is ConfigEntryState.NOT_LOADED
    # returns False after timeout
    assert not await mqtt.async_wait_for_mqtt_client(hass)


async def test_waiting_for_client_with_disabled_entry(
    hass: HomeAssistant,
) -> None:
    """Test waiting for client with timeout."""
    hass.set_state(CoreState.starting)
    await hass.async_block_till_done()

    entry = MockConfigEntry(
        domain=DOMAIN,
        data={"broker": "test-broker"},
        state=ConfigEntryState.NOT_LOADED,
        version=mqtt.CONFIG_ENTRY_VERSION,
        minor_version=mqtt.CONFIG_ENTRY_MINOR_VERSION,
    )
    entry.add_to_hass(hass)

    # Disable MQTT config entry
    await hass.config_entries.async_set_disabled_by(
        entry.entry_id, ConfigEntryDisabler.USER
    )

    assert entry.state is ConfigEntryState.NOT_LOADED

    # returns False because entry is disabled
    assert not await mqtt.async_wait_for_mqtt_client(hass)
