core/tests/components/conftest.py

"""Fixtures for component testing."""

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator, Callable, Coroutine, Generator, Mapping
from functools import lru_cache
from importlib.util import find_spec
import inspect
from ipaddress import IPv4Address, IPv4Network
from pathlib import Path
import re
import string
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock, patch

from aiohasupervisor import SupervisorClient, SupervisorNotFoundError
from aiohasupervisor.addons import AddonsClient
from aiohasupervisor.backups import BackupsClient
from aiohasupervisor.discovery import DiscoveryClient
from aiohasupervisor.homeassistant import HomeAssistantClient
from aiohasupervisor.host import HostClient
from aiohasupervisor.ingress import IngressClient
from aiohasupervisor.jobs import JobsClient
from aiohasupervisor.models import (
    AddonStage,
    AddonState,
    Discovery,
    DockerNetwork,
    GreenInfo,
    HomeAssistantInfo,
    HomeAssistantStats,
    HostInfo,
    InstalledAddon,
    JobsInfo,
    LogLevel,
    MountsInfo,
    NetworkInfo,
    OSInfo,
    Repository,
    ResolutionInfo,
    RootInfo,
    StoreAddon,
    StoreInfo,
    SupervisorInfo,
    SupervisorState,
    SupervisorStats,
    UpdateChannel,
    YellowInfo,
)
from aiohasupervisor.mounts import MountsClient
from aiohasupervisor.network import NetworkClient
from aiohasupervisor.os import OSClient
from aiohasupervisor.resolution import ResolutionClient
from aiohasupervisor.store import StoreClient
from aiohasupervisor.supervisor import SupervisorManagementClient
import pytest
import voluptuous as vol

from homeassistant import components, loader
from homeassistant.components import repairs
from homeassistant.config_entries import (
    DISCOVERY_SOURCES,
    ConfigEntriesFlowManager,
    FlowResult,
    OptionsFlowManager,
)
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import (
    Context,
    EntityServiceResponse,
    HassJobType,
    HomeAssistant,
    ServiceCall,
    ServiceRegistry,
    ServiceResponse,
    SupportsResponse,
    callback,
)
from homeassistant.data_entry_flow import (
    FlowContext,
    FlowHandler,
    FlowManager,
    FlowResultType,
    section,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.translation import async_get_translations
from homeassistant.helpers.typing import VolSchemaType
from homeassistant.util import yaml as yaml_util

from tests.common import QualityScaleStatus, get_quality_scale

if TYPE_CHECKING:
    from homeassistant.components.hassio import AddonManager

    from .conversation import MockAgent
    from .device_tracker.common import MockScanner
    from .light.common import MockLight
    from .sensor.common import MockSensor
    from .switch.common import MockSwitch

pytest.register_assert_rewrite("tests.components.common")

# Regex for accessing the integration name from the test path
RE_REQUEST_DOMAIN = re.compile(r".*tests\/components\/([^/]+)\/.*")


@pytest.fixture(scope="session", autouse=find_spec("zeroconf") is not None)
def patch_zeroconf_multiple_catcher() -> Generator[None]:
    """If installed, patch zeroconf wrapper that detects if multiple instances are used."""
    with patch(
        "homeassistant.components.zeroconf.install_multiple_zeroconf_catcher",
        side_effect=lambda zc: None,
    ):
        yield


@pytest.fixture(scope="session", autouse=True)
def prevent_io() -> Generator[None]:
    """Fixture to prevent certain I/O from happening."""
    with patch(
        "homeassistant.components.http.ban.load_yaml_config_file",
    ):
        yield


@pytest.fixture
def entity_registry_enabled_by_default() -> Generator[None]:
    """Test fixture that ensures all entities are enabled in the registry."""
    with (
        patch(
            "homeassistant.helpers.entity.Entity.entity_registry_enabled_default",
            return_value=True,
        ),
        patch(
            "homeassistant.components.device_tracker.config_entry.ScannerEntity.entity_registry_enabled_default",
            return_value=True,
        ),
    ):
        yield


# TTS test fixtures
@pytest.fixture(name="mock_tts_get_cache_files")
def mock_tts_get_cache_files_fixture() -> Generator[MagicMock]:
    """Mock the list TTS cache function."""
    from .tts.common import mock_tts_get_cache_files_fixture_helper  # noqa: PLC0415

    yield from mock_tts_get_cache_files_fixture_helper()


@pytest.fixture(name="mock_tts_init_cache_dir")
def mock_tts_init_cache_dir_fixture(
    init_tts_cache_dir_side_effect: Any,
) -> Generator[MagicMock]:
    """Mock the TTS cache dir in memory."""
    from .tts.common import mock_tts_init_cache_dir_fixture_helper  # noqa: PLC0415

    yield from mock_tts_init_cache_dir_fixture_helper(init_tts_cache_dir_side_effect)


@pytest.fixture(name="init_tts_cache_dir_side_effect")
def init_tts_cache_dir_side_effect_fixture() -> Any:
    """Return the cache dir."""
    from .tts.common import (  # noqa: PLC0415
        init_tts_cache_dir_side_effect_fixture_helper,
    )

    return init_tts_cache_dir_side_effect_fixture_helper()


@pytest.fixture(name="mock_tts_cache_dir")
def mock_tts_cache_dir_fixture(
    tmp_path: Path,
    mock_tts_init_cache_dir: MagicMock,
    mock_tts_get_cache_files: MagicMock,
    request: pytest.FixtureRequest,
) -> Generator[Path]:
    """Mock the TTS cache dir with empty dir."""
    from .tts.common import mock_tts_cache_dir_fixture_helper  # noqa: PLC0415

    yield from mock_tts_cache_dir_fixture_helper(
        tmp_path, mock_tts_init_cache_dir, mock_tts_get_cache_files, request
    )


@pytest.fixture(name="tts_mutagen_mock")
def tts_mutagen_mock_fixture() -> Generator[MagicMock]:
    """Mock writing tags."""
    from .tts.common import tts_mutagen_mock_fixture_helper  # noqa: PLC0415

    yield from tts_mutagen_mock_fixture_helper()


@pytest.fixture(name="mock_conversation_agent")
def mock_conversation_agent_fixture(hass: HomeAssistant) -> MockAgent:
    """Mock a conversation agent."""
    from .conversation.common import (  # noqa: PLC0415
        mock_conversation_agent_fixture_helper,
    )

    return mock_conversation_agent_fixture_helper(hass)


@pytest.fixture(scope="session", autouse=find_spec("haffmpeg") is not None)
def prevent_ffmpeg_subprocess() -> Generator[None]:
    """If installed, prevent ffmpeg from creating a subprocess."""
    with patch(
        "homeassistant.components.ffmpeg.FFVersion.get_version", return_value="6.0"
    ):
        yield


@pytest.fixture
def mock_light_entities() -> list[MockLight]:
    """Return mocked light entities."""
    from .light.common import MockLight  # noqa: PLC0415

    return [
        MockLight("Ceiling", STATE_ON),
        MockLight("Ceiling", STATE_OFF),
        MockLight(None, STATE_OFF),
    ]


@pytest.fixture
def mock_sensor_entities() -> dict[str, MockSensor]:
    """Return mocked sensor entities."""
    from .sensor.common import get_mock_sensor_entities  # noqa: PLC0415

    return get_mock_sensor_entities()


@pytest.fixture
def mock_switch_entities() -> list[MockSwitch]:
    """Return mocked toggle entities."""
    from .switch.common import get_mock_switch_entities  # noqa: PLC0415

    return get_mock_switch_entities()


@pytest.fixture
def mock_legacy_device_scanner() -> MockScanner:
    """Return mocked legacy device scanner entity."""
    from .device_tracker.common import MockScanner  # noqa: PLC0415

    return MockScanner()


@pytest.fixture
def mock_legacy_device_tracker_setup() -> Callable[[HomeAssistant, MockScanner], None]:
    """Return setup callable for legacy device tracker setup."""
    from .device_tracker.common import mock_legacy_device_tracker_setup  # noqa: PLC0415

    return mock_legacy_device_tracker_setup


@pytest.fixture(name="addon_manager")
def addon_manager_fixture(
    hass: HomeAssistant, supervisor_client: AsyncMock
) -> AddonManager:
    """Return an AddonManager instance."""
    from .hassio.common import mock_addon_manager  # noqa: PLC0415

    return mock_addon_manager(hass)


@pytest.fixture(name="discovery_info")
def discovery_info_fixture() -> list[Discovery]:
    """Return the discovery info from the supervisor."""
    return []


@pytest.fixture(name="discovery_info_side_effect")
def discovery_info_side_effect_fixture() -> Any | None:
    """Return the discovery info from the supervisor."""
    return None


@pytest.fixture(name="get_addon_discovery_info")
def get_addon_discovery_info_fixture(
    supervisor_client: AsyncMock,
    discovery_info: list[Discovery],
    discovery_info_side_effect: Any | None,
) -> AsyncMock:
    """Mock get add-on discovery info."""
    supervisor_client.discovery.list.return_value = discovery_info
    supervisor_client.discovery.list.side_effect = discovery_info_side_effect
    return supervisor_client.discovery.list


@pytest.fixture(name="get_discovery_message_side_effect")
def get_discovery_message_side_effect_fixture() -> Any | None:
    """Side effect for getting a discovery message by uuid."""
    return None


@pytest.fixture(name="get_discovery_message")
def get_discovery_message_fixture(
    supervisor_client: AsyncMock, get_discovery_message_side_effect: Any | None
) -> AsyncMock:
    """Mock getting a discovery message by uuid."""
    supervisor_client.discovery.get.side_effect = get_discovery_message_side_effect
    return supervisor_client.discovery.get


@pytest.fixture(name="addon_store_info_side_effect")
def addon_store_info_side_effect_fixture() -> Any | None:
    """Return the add-on store info side effect."""
    return None


@pytest.fixture(name="addon_store_info")
def addon_store_info_fixture(
    supervisor_client: AsyncMock,
    addon_store_info_side_effect: Any | None,
) -> AsyncMock:
    """Mock Supervisor add-on store info."""
    from .hassio.common import mock_addon_store_info  # noqa: PLC0415

    return mock_addon_store_info(supervisor_client, addon_store_info_side_effect)


@pytest.fixture(name="addon_info_side_effect")
def addon_info_side_effect_fixture() -> Any | None:
    """Return the add-on info side effect."""
    return None


@pytest.fixture(name="addon_info")
def addon_info_fixture(
    supervisor_client: AsyncMock, addon_info_side_effect: Any | None
) -> AsyncMock:
    """Mock Supervisor add-on info."""
    from .hassio.common import mock_addon_info  # noqa: PLC0415

    return mock_addon_info(supervisor_client, addon_info_side_effect)


@pytest.fixture(name="addon_not_installed")
def addon_not_installed_fixture(
    addon_store_info: AsyncMock, addon_info: AsyncMock
) -> AsyncMock:
    """Mock add-on not installed."""
    from .hassio.common import mock_addon_not_installed  # noqa: PLC0415

    addon_info.side_effect = SupervisorNotFoundError
    return mock_addon_not_installed(addon_store_info, addon_info)


@pytest.fixture(name="addon_installed")
def addon_installed_fixture(
    addon_store_info: AsyncMock, addon_info: AsyncMock
) -> AsyncMock:
    """Mock add-on already installed but not running."""
    from .hassio.common import mock_addon_installed  # noqa: PLC0415

    return mock_addon_installed(addon_store_info, addon_info)


@pytest.fixture(name="addon_running")
def addon_running_fixture(
    addon_store_info: AsyncMock, addon_info: AsyncMock
) -> AsyncMock:
    """Mock add-on already running."""
    from .hassio.common import mock_addon_running  # noqa: PLC0415

    return mock_addon_running(addon_store_info, addon_info)


@pytest.fixture(name="install_addon_side_effect")
def install_addon_side_effect_fixture(
    addon_store_info: AsyncMock, addon_info: AsyncMock
) -> Any | None:
    """Return the install add-on side effect."""

    from .hassio.common import mock_install_addon_side_effect  # noqa: PLC0415

    return mock_install_addon_side_effect(addon_store_info, addon_info)


@pytest.fixture(name="install_addon")
def install_addon_fixture(
    supervisor_client: AsyncMock,
    install_addon_side_effect: Any | None,
) -> AsyncMock:
    """Mock install add-on."""
    supervisor_client.store.install_addon.side_effect = install_addon_side_effect
    return supervisor_client.store.install_addon


@pytest.fixture(name="start_addon_side_effect")
def start_addon_side_effect_fixture(
    addon_store_info: AsyncMock, addon_info: AsyncMock
) -> Any | None:
    """Return the start add-on options side effect."""
    from .hassio.common import mock_start_addon_side_effect  # noqa: PLC0415

    return mock_start_addon_side_effect(addon_store_info, addon_info)


@pytest.fixture(name="start_addon")
def start_addon_fixture(
    supervisor_client: AsyncMock, start_addon_side_effect: Any | None
) -> AsyncMock:
    """Mock start add-on."""
    supervisor_client.addons.start_addon.side_effect = start_addon_side_effect
    return supervisor_client.addons.start_addon


@pytest.fixture(name="restart_addon_side_effect")
def restart_addon_side_effect_fixture() -> Any | None:
    """Return the restart add-on options side effect."""
    return None


@pytest.fixture(name="restart_addon")
def restart_addon_fixture(
    supervisor_client: AsyncMock,
    restart_addon_side_effect: Any | None,
) -> AsyncMock:
    """Mock restart add-on."""
    supervisor_client.addons.restart_addon.side_effect = restart_addon_side_effect
    return supervisor_client.addons.restart_addon


@pytest.fixture(name="stop_addon")
def stop_addon_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock stop add-on."""
    return supervisor_client.addons.stop_addon


@pytest.fixture(name="addon_options")
def addon_options_fixture(addon_info: AsyncMock) -> dict[str, Any]:
    """Mock add-on options."""
    return addon_info.return_value.options


@pytest.fixture(name="set_addon_options_side_effect")
def set_addon_options_side_effect_fixture(
    addon_options: dict[str, Any],
) -> Any | None:
    """Return the set add-on options side effect."""
    from .hassio.common import mock_set_addon_options_side_effect  # noqa: PLC0415

    return mock_set_addon_options_side_effect(addon_options)


@pytest.fixture(name="set_addon_options")
def set_addon_options_fixture(
    supervisor_client: AsyncMock,
    set_addon_options_side_effect: Any | None,
) -> AsyncMock:
    """Mock set add-on options."""
    supervisor_client.addons.set_addon_options.side_effect = (
        set_addon_options_side_effect
    )
    return supervisor_client.addons.set_addon_options


@pytest.fixture(name="uninstall_addon")
def uninstall_addon_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock uninstall add-on."""
    return supervisor_client.addons.uninstall_addon


@pytest.fixture(name="create_backup")
def create_backup_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock create backup."""
    return supervisor_client.backups.partial_backup


@pytest.fixture(name="update_addon")
def update_addon_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock update add-on."""
    return supervisor_client.store.update_addon


@pytest.fixture(name="store_addons")
def store_addons_fixture() -> list[StoreAddon]:
    """Mock store addons list."""
    return []


@pytest.fixture(name="store_repositories")
def store_repositories_fixture() -> list[Repository]:
    """Mock store repositories list."""
    return []


@pytest.fixture(name="store_info")
def store_info_fixture(
    supervisor_client: AsyncMock,
    store_addons: list[StoreAddon],
    store_repositories: list[Repository],
) -> AsyncMock:
    """Mock store info."""
    supervisor_client.store.info.return_value = StoreInfo(
        addons=store_addons, repositories=store_repositories
    )
    return supervisor_client.store.info


@pytest.fixture(name="addon_stats")
def addon_stats_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock addon stats info."""
    from .hassio.common import mock_addon_stats  # noqa: PLC0415

    return mock_addon_stats(supervisor_client)


@pytest.fixture(name="addon_changelog")
def addon_changelog_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock addon changelog."""
    supervisor_client.store.addon_changelog.return_value = ""
    return supervisor_client.store.addon_changelog


@pytest.fixture(name="supervisor_is_connected")
def supervisor_is_connected_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock supervisor is connected."""
    supervisor_client.supervisor.ping.return_value = None
    return supervisor_client.supervisor.ping


@pytest.fixture(name="resolution_info")
def resolution_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock resolution info from supervisor."""
    supervisor_client.resolution.info.return_value = ResolutionInfo(
        suggestions=[],
        unsupported=[],
        unhealthy=[],
        issues=[],
        checks=[],
    )
    return supervisor_client.resolution.info


@pytest.fixture(name="resolution_suggestions_for_issue")
def resolution_suggestions_for_issue_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock suggestions by issue from supervisor resolution."""
    supervisor_client.resolution.suggestions_for_issue.return_value = []
    return supervisor_client.resolution.suggestions_for_issue


@pytest.fixture(name="jobs_info")
def jobs_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock jobs info from supervisor."""
    supervisor_client.jobs.info.return_value = JobsInfo(ignore_conditions=[], jobs=[])
    return supervisor_client.jobs.info


@pytest.fixture(name="os_yellow_info")
def os_yellow_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock yellow info API from supervisor OS."""
    supervisor_client.os.yellow_info.return_value = YellowInfo(
        disk_led=True, heartbeat_led=True, power_led=True
    )
    return supervisor_client.os.yellow_info


@pytest.fixture(name="os_green_info")
def os_green_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock green info API from supervisor OS."""
    supervisor_client.os.green_info.return_value = GreenInfo(
        activity_led=True, power_led=True, system_health_led=True
    )
    return supervisor_client.os.green_info


@pytest.fixture(name="supervisor_root_info")
def supervisor_root_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock root info API from supervisor."""
    supervisor_client.info.return_value = RootInfo(
        supervisor="222",
        homeassistant="0.110.0",
        hassos="1.2.3",
        docker="",
        hostname=None,
        operating_system=None,
        features=[],
        machine=None,
        machine_id=None,
        arch="",
        state=SupervisorState.RUNNING,
        supported_arch=[],
        supported=True,
        channel=UpdateChannel.STABLE,
        logging=LogLevel.INFO,
        timezone="Etc/UTC",
    )
    return supervisor_client.info


@pytest.fixture(name="host_info")
def host_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock host info API from supervisor."""
    supervisor_client.host.info.return_value = HostInfo(
        agent_version=None,
        apparmor_version=None,
        chassis="vm",
        virtualization=None,
        cpe=None,
        deployment=None,
        disk_free=1.6,
        disk_total=100.0,
        disk_used=98.4,
        disk_life_time=None,
        features=[],
        hostname=None,
        llmnr_hostname=None,
        kernel="4.19.0-6-amd64",
        operating_system="Debian GNU/Linux 10 (buster)",
        timezone=None,
        dt_utc=None,
        dt_synchronized=None,
        use_ntp=None,
        startup_time=None,
        boot_timestamp=None,
        broadcast_llmnr=None,
        broadcast_mdns=None,
    )
    return supervisor_client.host.info


@pytest.fixture(name="homeassistant_info")
def homeassistant_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock Home Assistant info API from supervisor."""
    supervisor_client.homeassistant.info.return_value = HomeAssistantInfo(
        version="1.0.0",
        version_latest="1.0.0",
        update_available=False,
        machine=None,
        ip_address=IPv4Address("172.30.32.1"),
        arch=None,
        image="homeassistant",
        boot=True,
        port=8123,
        ssl=False,
        watchdog=True,
        audio_input=None,
        audio_output=None,
        backups_exclude_database=False,
        duplicate_log_file=False,
    )
    return supervisor_client.homeassistant.info


@pytest.fixture(name="supervisor_info")
def supervisor_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock supervisor info API from supervisor."""
    supervisor_client.supervisor.info.return_value = SupervisorInfo(
        version="1.0.0",
        version_latest="1.0.0",
        update_available=False,
        channel=UpdateChannel.STABLE,
        arch="",
        supported=True,
        healthy=True,
        ip_address=IPv4Address("172.30.32.2"),
        timezone=None,
        logging=LogLevel.INFO,
        debug=False,
        debug_block=False,
        diagnostics=None,
        auto_update=True,
        country=None,
        detect_blocking_io=False,
    )
    return supervisor_client.supervisor.info


@pytest.fixture(name="addons_list")
def addons_list_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock addons list API from supervisor."""
    supervisor_client.addons.list.return_value = [
        InstalledAddon(
            detached=False,
            advanced=False,
            available=True,
            build=False,
            description="",
            homeassistant=None,
            icon=False,
            logo=False,
            name="test",
            repository="core",
            slug="test",
            stage=AddonStage.STABLE,
            update_available=True,
            url="https://github.com/home-assistant/addons/test",
            version_latest="2.0.1",
            version="2.0.0",
            state=AddonState.STARTED,
        ),
        InstalledAddon(
            detached=False,
            advanced=False,
            available=True,
            build=False,
            description="",
            homeassistant=None,
            icon=False,
            logo=False,
            name="test2",
            repository="core",
            slug="test2",
            stage=AddonStage.STABLE,
            update_available=False,
            url="https://github.com",
            version_latest="3.1.0",
            version="3.1.0",
            state=AddonState.STOPPED,
        ),
    ]
    return supervisor_client.addons.list


@pytest.fixture(name="network_info")
def network_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock network info API from supervisor."""
    supervisor_client.network.info.return_value = NetworkInfo(
        interfaces=[],
        docker=DockerNetwork(
            interface="hassio",
            address=IPv4Network("172.30.32.0/23"),
            gateway=IPv4Address("172.30.32.1"),
            dns=IPv4Address("172.30.32.3"),
        ),
        host_internet=True,
        supervisor_internet=True,
    )
    return supervisor_client.network.info


@pytest.fixture(name="os_info")
def os_info_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock os info API from supervisor."""
    supervisor_client.os.info.return_value = OSInfo(
        version="1.0.0",
        version_latest="1.0.0",
        update_available=False,
        board=None,
        boot=None,
        data_disk=None,
        boot_slots={},
    )
    return supervisor_client.os.info


@pytest.fixture(name="homeassistant_stats")
def homeassistant_stats_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock Home Assistant stats API from supervisor."""
    supervisor_client.homeassistant.stats.return_value = HomeAssistantStats(
        cpu_percent=0.99,
        memory_usage=182611968,
        memory_limit=3977146368,
        memory_percent=4.59,
        network_rx=362570232,
        network_tx=82374138,
        blk_read=46010945536,
        blk_write=15051526144,
    )
    return supervisor_client.homeassistant.stats


@pytest.fixture(name="supervisor_stats")
def supervisor_stats_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock supervisor stats API from supervisor."""
    supervisor_client.supervisor.stats.return_value = SupervisorStats(
        cpu_percent=0.99,
        memory_usage=182611968,
        memory_limit=3977146368,
        memory_percent=4.59,
        network_rx=362570232,
        network_tx=82374138,
        blk_read=46010945536,
        blk_write=15051526144,
    )
    return supervisor_client.supervisor.stats


@pytest.fixture(name="ingress_panels")
def ingress_panels_fixture(supervisor_client: AsyncMock) -> AsyncMock:
    """Mock ingress panels API from supervisor."""
    supervisor_client.ingress.panels.return_value = {}
    return supervisor_client.ingress.panels


@pytest.fixture(name="supervisor_client")
def supervisor_client() -> Generator[AsyncMock]:
    """Mock the supervisor client."""
    supervisor_client = AsyncMock(spec=SupervisorClient)
    supervisor_client.addons = AsyncMock(spec=AddonsClient)
    supervisor_client.backups = AsyncMock(spec=BackupsClient)
    supervisor_client.discovery = AsyncMock(spec=DiscoveryClient)
    supervisor_client.homeassistant = AsyncMock(spec=HomeAssistantClient)
    supervisor_client.host = AsyncMock(spec=HostClient)
    supervisor_client.ingress = AsyncMock(spec=IngressClient)
    supervisor_client.jobs = AsyncMock(spec=JobsClient)
    supervisor_client.jobs.info.return_value = JobsInfo(ignore_conditions=[], jobs=[])
    supervisor_client.mounts = AsyncMock(spec=MountsClient)
    supervisor_client.mounts.info.return_value = MagicMock(
        spec=MountsInfo, default_backup_mount=None, mounts=[]
    )
    supervisor_client.network = AsyncMock(spec=NetworkClient)
    supervisor_client.os = AsyncMock(spec=OSClient)
    supervisor_client.resolution = AsyncMock(spec=ResolutionClient)
    supervisor_client.supervisor = AsyncMock(spec=SupervisorManagementClient)
    supervisor_client.store = AsyncMock(spec=StoreClient)

    with (
        patch(
            "homeassistant.components.hassio.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.handler.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.addon_manager.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.addon_panel.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.backup.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.discovery.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.coordinator.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.issues.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.jobs.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.repairs.get_supervisor_client",
            return_value=supervisor_client,
        ),
        patch(
            "homeassistant.components.hassio.update_helper.get_supervisor_client",
            return_value=supervisor_client,
        ),
    ):
        yield supervisor_client


def _validate_translation_placeholders(
    full_key: str,
    translation: str,
    description_placeholders: dict[str, str] | None,
    translation_errors: dict[str, str],
) -> str | None:
    """Raise if translation exists with missing placeholders."""
    tuples = list(string.Formatter().parse(translation))
    for _, placeholder, _, _ in tuples:
        if placeholder is None:
            continue
        if (
            description_placeholders is None
            or placeholder not in description_placeholders
        ):
            translation_errors[full_key] = (
                f"Description not found for placeholder `{placeholder}` in {full_key}"
            )


async def _validate_translation(
    hass: HomeAssistant,
    translation_errors: dict[str, str],
    ignore_translations_for_mock_domains: set[str],
    category: str,
    component: str,
    key: str,
    description_placeholders: Mapping[str, str] | None,
    *,
    translation_required: bool = True,
) -> None:
    """Raise if translation doesn't exist."""
    full_key = f"component.{component}.{category}.{key}"
    if component in ignore_translations_for_mock_domains:
        try:
            integration = await loader.async_get_integration(hass, component)
        except loader.IntegrationNotFound:
            return
        component_paths = components.__path__
        if not any(
            Path(f"{component_path}/{component}") == integration.file_path
            for component_path in component_paths
        ):
            return
        # If the integration exists, translation errors should be ignored via the
        # ignore_missing_translations fixture instead of the
        # ignore_translations_for_mock_domains fixture.
        translation_errors[full_key] = f"The integration '{component}' exists"
        return

    translations = await async_get_translations(hass, "en", category, [component])

    if full_key.endswith("."):
        for subkey, translation in translations.items():
            if subkey.startswith(full_key):
                _validate_translation_placeholders(
                    subkey, translation, description_placeholders, translation_errors
                )
        return
    if (translation := translations.get(full_key)) is not None:
        _validate_translation_placeholders(
            full_key, translation, description_placeholders, translation_errors
        )
        return

    if not translation_required:
        return

    if full_key not in translation_errors:
        for k in translation_errors:
            if k.endswith(".") and full_key.startswith(k):
                full_key = k
                break
    if translation_errors.get(full_key) in {"used", "unused"}:
        # If the integration does not exist, translation errors should be ignored
        # via the ignore_translations_for_mock_domains fixture instead of the
        # ignore_missing_translations fixture.
        try:
            await loader.async_get_integration(hass, component)
        except loader.IntegrationNotFound:
            translation_errors[full_key] = (
                f"Translation not found for {component}: `{category}.{key}`. "
                f"The integration '{component}' does not exist."
            )
            return

        # This translation key is in the ignore list, mark it as used
        translation_errors[full_key] = "used"
        return

    translation_errors[full_key] = (
        f"Translation not found for {component}: `{category}.{key}`. "
        f"Please add to homeassistant/components/{component}/strings.json"
    )


@pytest.fixture
def ignore_missing_translations() -> str | list[str]:
    """Ignore specific missing translations.

    Override or parametrize this fixture with a fixture that returns
    a list of missing translation that should be ignored.
    """
    return []


@pytest.fixture
def ignore_translations_for_mock_domains() -> str | list[str]:
    """Don't validate translations for specific domains.

    Override or parametrize this fixture with a fixture that returns
    a list of domains for which translations should not be validated.
    This should only be used when testing mocked integrations.
    """
    return []


@lru_cache
def _get_integration_quality_scale(integration: str) -> dict[str, Any]:
    """Get the quality scale for an integration."""
    try:
        return yaml_util.load_yaml_dict(
            f"homeassistant/components/{integration}/quality_scale.yaml"
        ).get("rules", {})
    except FileNotFoundError:
        return {}


def _get_integration_quality_scale_rule(integration: str, rule: str) -> str:
    """Get the quality scale for an integration."""
    quality_scale = _get_integration_quality_scale(integration)
    if not quality_scale or rule not in quality_scale:
        return "todo"
    status = quality_scale[rule]
    return status if isinstance(status, str) else status["status"]


async def _check_step_or_section_translations(
    hass: HomeAssistant,
    translation_errors: dict[str, str],
    category: str,
    integration: str,
    translation_prefix: str,
    description_placeholders: dict[str, str],
    data_schema: vol.Schema | None,
    ignore_translations_for_mock_domains: set[str],
) -> None:
    # neither title nor description are required
    # - title defaults to integration name
    # - description is optional
    for header in ("title", "description"):
        await _validate_translation(
            hass,
            translation_errors,
            ignore_translations_for_mock_domains,
            category,
            integration,
            f"{translation_prefix}.{header}",
            description_placeholders,
            translation_required=False,
        )

    if not data_schema:
        return

    for data_key, data_value in data_schema.schema.items():
        if isinstance(data_value, section):
            # check the nested section
            await _check_step_or_section_translations(
                hass,
                translation_errors,
                category,
                integration,
                f"{translation_prefix}.sections.{data_key}",
                description_placeholders,
                data_value.schema,
                ignore_translations_for_mock_domains,
            )
            continue
        iqs_config_flow = _get_integration_quality_scale_rule(
            integration, "config-flow"
        )
        # data and data_description are compulsory
        for header in ("data", "data_description"):
            await _validate_translation(
                hass,
                translation_errors,
                ignore_translations_for_mock_domains,
                category,
                integration,
                f"{translation_prefix}.{header}.{data_key}",
                description_placeholders,
                translation_required=(iqs_config_flow == "done"),
            )


async def _check_config_flow_result_translations(
    manager: FlowManager,
    flow: FlowHandler,
    result: FlowResult[FlowContext, str],
    translation_errors: dict[str, str],
    ignore_translations_for_mock_domains: set[str],
) -> None:
    if result["type"] is FlowResultType.CREATE_ENTRY:
        # No need to check translations for a completed flow
        return

    key_prefix = ""
    description_placeholders = result.get("description_placeholders")
    if isinstance(manager, ConfigEntriesFlowManager):
        category = "config"
        integration = flow.handler
    elif isinstance(manager, OptionsFlowManager):
        category = "options"
        integration = flow.hass.config_entries.async_get_entry(flow.handler).domain
    elif isinstance(manager, repairs.RepairsFlowManager):
        category = "issues"
        integration = flow.handler
        issue_id = flow.issue_id
        issue = ir.async_get(flow.hass).async_get_issue(integration, issue_id)
        if issue is None:
            # Issue was deleted mid-flow (e.g., config entry removed), skip check
            return
        key_prefix = f"{issue.translation_key}.fix_flow."
        description_placeholders = {
            # Both are used in issue translations, and description_placeholders
            # takes precedence over translation_placeholders
            **(issue.translation_placeholders or {}),
            **(description_placeholders or {}),
        }
    else:
        return

    # Check if this flow has been seen before
    # Gets set to False on first run, and to True on subsequent runs
    setattr(flow, "__flow_seen_before", hasattr(flow, "__flow_seen_before"))

    if result["type"] is FlowResultType.FORM:
        if step_id := result.get("step_id"):
            await _check_step_or_section_translations(
                flow.hass,
                translation_errors,
                category,
                integration,
                f"{key_prefix}step.{step_id}",
                description_placeholders,
                result["data_schema"],
                ignore_translations_for_mock_domains,
            )

        if errors := result.get("errors"):
            for error in errors.values():
                await _validate_translation(
                    flow.hass,
                    translation_errors,
                    ignore_translations_for_mock_domains,
                    category,
                    integration,
                    f"{key_prefix}error.{error}",
                    description_placeholders,
                )
        return

    if result["type"] is FlowResultType.ABORT:
        # We don't need translations for a discovery flow which immediately
        # aborts, since such flows won't be seen by users
        if not flow.__flow_seen_before and flow.source in DISCOVERY_SOURCES:
            return
        await _validate_translation(
            flow.hass,
            translation_errors,
            ignore_translations_for_mock_domains,
            category,
            integration,
            f"{key_prefix}abort.{result['reason']}",
            description_placeholders,
        )


async def _check_create_issue_translations(
    issue_registry: ir.IssueRegistry,
    issue: ir.IssueEntry,
    translation_errors: dict[str, str],
    ignore_translations_for_mock_domains: set[str],
) -> None:
    if issue.translation_key is None:
        # `translation_key` is only None on dismissed issues
        return
    await _validate_translation(
        issue_registry.hass,
        translation_errors,
        ignore_translations_for_mock_domains,
        "issues",
        issue.domain,
        f"{issue.translation_key}.title",
        issue.translation_placeholders,
    )
    if not issue.is_fixable:
        # Description is required for non-fixable issues
        await _validate_translation(
            issue_registry.hass,
            translation_errors,
            ignore_translations_for_mock_domains,
            "issues",
            issue.domain,
            f"{issue.translation_key}.description",
            issue.translation_placeholders,
        )


def _get_request_quality_scale(
    request: pytest.FixtureRequest, rule: str
) -> QualityScaleStatus:
    if not (match := RE_REQUEST_DOMAIN.match(str(request.path))):
        return QualityScaleStatus.TODO
    integration = match.groups(1)[0]
    return get_quality_scale(integration).get(rule, QualityScaleStatus.TODO)


async def _check_exception_translation(
    hass: HomeAssistant,
    exception: HomeAssistantError,
    translation_errors: dict[str, str],
    request: pytest.FixtureRequest,
    ignore_translations_for_mock_domains: set[str],
) -> None:
    if exception.translation_key is None:
        if (
            _get_request_quality_scale(request, "exception-translations")
            is QualityScaleStatus.DONE
        ):
            translation_errors["quality_scale"] = (
                f"Found untranslated {type(exception).__name__} exception: {exception}"
            )
        return
    await _validate_translation(
        hass,
        translation_errors,
        ignore_translations_for_mock_domains,
        "exceptions",
        exception.translation_domain,
        f"{exception.translation_key}.message",
        exception.translation_placeholders,
    )


_DYNAMIC_SERVICE_DOMAINS = {
    "esphome",
    "notify",
    "rest_command",
    "script",
    "shell_command",
    "tts",
}
"""These domains create services dynamically.

name/description translations are not required.
"""


async def _check_service_registration_translation(
    hass: HomeAssistant,
    domain: str,
    service_name: str,
    description_placeholders: Mapping[str, str] | None,
    translation_errors: dict[str, str],
    ignore_translations_for_mock_domains: set[str],
) -> None:
    # Use trailing . to check all subkeys
    # This validates placeholders only, and only if the translation exists
    await _validate_translation(
        hass,
        translation_errors,
        ignore_translations_for_mock_domains,
        "services",
        domain,
        f"{service_name}.",
        description_placeholders,
    )
    # Service `name` and `description` should be compulsory
    # unless for specific domains where the services are dynamically created
    if domain not in _DYNAMIC_SERVICE_DOMAINS:
        for subkey in ("name", "description"):
            await _validate_translation(
                hass,
                translation_errors,
                ignore_translations_for_mock_domains,
                "services",
                domain,
                f"{service_name}.{subkey}",
                description_placeholders,
                translation_required=True,
            )


@pytest.fixture(autouse=True)
async def check_translations(
    ignore_missing_translations: str | list[str],
    ignore_translations_for_mock_domains: str | list[str],
    request: pytest.FixtureRequest,
) -> AsyncGenerator[None]:
    """Check that translation requirements are met.

    Current checks:
    - data entry flow results (ConfigFlow/OptionsFlow/RepairFlow)
    - issue registry entries
    - action (service) exceptions
    """
    if not isinstance(ignore_missing_translations, list):
        ignore_missing_translations = [ignore_missing_translations]

    if not isinstance(ignore_translations_for_mock_domains, list):
        ignored_domains = {ignore_translations_for_mock_domains}
    else:
        ignored_domains = set(ignore_translations_for_mock_domains)

    # Set all ignored translation keys to "unused"
    translation_errors = dict.fromkeys(ignore_missing_translations, "unused")

    translation_coros = set()

    # Keep reference to original functions
    _original_flow_manager_async_handle_step = FlowManager._async_handle_step
    _original_issue_registry_async_create_issue = ir.IssueRegistry.async_get_or_create
    _original_service_registry_async_call = ServiceRegistry.async_call
    _original_service_registry_async_register = ServiceRegistry.async_register

    # Prepare override functions
    async def _flow_manager_async_handle_step(
        self: FlowManager, flow: FlowHandler, *args
    ) -> FlowResult:
        result = await _original_flow_manager_async_handle_step(self, flow, *args)
        await _check_config_flow_result_translations(
            self, flow, result, translation_errors, ignored_domains
        )
        return result

    def _issue_registry_async_create_issue(
        self: ir.IssueRegistry, domain: str, issue_id: str, *args, **kwargs
    ) -> ir.IssueEntry:
        result = _original_issue_registry_async_create_issue(
            self, domain, issue_id, *args, **kwargs
        )
        translation_coros.add(
            _check_create_issue_translations(
                self, result, translation_errors, ignored_domains
            )
        )
        return result

    async def _service_registry_async_call(
        self: ServiceRegistry,
        domain: str,
        service: str,
        service_data: dict[str, Any] | None = None,
        blocking: bool = False,
        context: Context | None = None,
        target: dict[str, Any] | None = None,
        return_response: bool = False,
    ) -> ServiceResponse:
        try:
            return await _original_service_registry_async_call(
                self,
                domain,
                service,
                service_data,
                blocking,
                context,
                target,
                return_response,
            )
        except HomeAssistantError as err:
            translation_coros.add(
                _check_exception_translation(
                    self._hass,
                    err,
                    translation_errors,
                    request,
                    ignored_domains,
                )
            )
            raise

    @callback
    def _service_registry_async_register(
        self: ServiceRegistry,
        domain: str,
        service: str,
        service_func: Callable[
            [ServiceCall],
            Coroutine[Any, Any, ServiceResponse | EntityServiceResponse]
            | ServiceResponse
            | EntityServiceResponse
            | None,
        ],
        schema: VolSchemaType | None = None,
        supports_response: SupportsResponse = SupportsResponse.NONE,
        job_type: HassJobType | None = None,
        *,
        description_placeholders: Mapping[str, str] | None = None,
    ) -> None:
        if (
            (current_frame := inspect.currentframe()) is None
            or (caller := current_frame.f_back) is None
            or (
                # async_mock_service is used in tests to register test services
                caller.f_code.co_name != "async_mock_service"
                # ServiceRegistry.async_register can also be called directly in
                # a test module
                and not caller.f_code.co_filename.startswith(
                    str(Path(__file__).parents[0])
                )
            )
        ):
            translation_coros.add(
                _check_service_registration_translation(
                    self._hass,
                    domain,
                    service,
                    description_placeholders,
                    translation_errors,
                    ignored_domains,
                )
            )

        _original_service_registry_async_register(
            self,
            domain,
            service,
            service_func,
            schema,
            supports_response,
            job_type,
            description_placeholders=description_placeholders,
        )

    # Use override functions
    with (
        patch(
            "homeassistant.data_entry_flow.FlowManager._async_handle_step",
            _flow_manager_async_handle_step,
        ),
        patch(
            "homeassistant.helpers.issue_registry.IssueRegistry.async_get_or_create",
            _issue_registry_async_create_issue,
        ),
        patch(
            "homeassistant.core.ServiceRegistry.async_call",
            _service_registry_async_call,
        ),
        patch(
            "homeassistant.core.ServiceRegistry.async_register",
            _service_registry_async_register,
        ),
    ):
        yield

    await asyncio.gather(*translation_coros)

    # Run final checks
    unused_ignore = [k for k, v in translation_errors.items() if v == "unused"]
    if unused_ignore:
        # Some ignored translations were not used
        pytest.fail(
            f"Unused ignore translations: {', '.join(unused_ignore)}. "
            "Please remove them from the ignore_missing_translations fixture."
        )
    for description in translation_errors.values():
        if description != "used":
            pytest.fail(description)


@pytest.fixture(name="enable_labs_preview_features")
def enable_labs_preview_features() -> Generator[None]:
    """Enable labs preview features."""
    with patch(
        "homeassistant.components.labs.async_is_preview_feature_enabled",
        return_value=True,
    ):
        yield