33ff6b5b6e
* Avoid creating tasks for checking integrations platforms This is a followup to #110743 to avoid creating a task to check if the integration platform exists. We created tasks because we needed to await async_get_integrations but since its always called from EVENT_COMPONENT_LOADED firing, we can use the async_get_loaded_integration version which does not need to be awaited. This eliminates one task for every loaded component * there is no more race risk * reduce * coro or callback * reduce * tweak * race safe * fix type * fixes * use built-in helper to make it smaller * use built-in helper to make it smaller * use built-in helper to make it smaller * add coverage to ensure exceptions are logged * improve readability a bit * platforms
150 lines
4.5 KiB
Python
150 lines
4.5 KiB
Python
"""Test integration platform helpers."""
|
|
from collections.abc import Callable
|
|
from types import ModuleType
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers.integration_platform import (
|
|
async_process_integration_platforms,
|
|
)
|
|
from homeassistant.setup import ATTR_COMPONENT, EVENT_COMPONENT_LOADED
|
|
|
|
from tests.common import mock_platform
|
|
|
|
|
|
async def test_process_integration_platforms(hass: HomeAssistant) -> None:
|
|
"""Test processing integrations."""
|
|
loaded_platform = Mock()
|
|
mock_platform(hass, "loaded.platform_to_check", loaded_platform)
|
|
hass.config.components.add("loaded")
|
|
|
|
event_platform = Mock()
|
|
mock_platform(hass, "event.platform_to_check", event_platform)
|
|
|
|
processed = []
|
|
|
|
async def _process_platform(hass, domain, platform):
|
|
"""Process platform."""
|
|
processed.append((domain, platform))
|
|
|
|
await async_process_integration_platforms(
|
|
hass, "platform_to_check", _process_platform
|
|
)
|
|
|
|
assert len(processed) == 1
|
|
assert processed[0][0] == "loaded"
|
|
assert processed[0][1] == loaded_platform
|
|
|
|
hass.bus.async_fire(EVENT_COMPONENT_LOADED, {ATTR_COMPONENT: "event"})
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(processed) == 2
|
|
assert processed[1][0] == "event"
|
|
assert processed[1][1] == event_platform
|
|
|
|
hass.bus.async_fire(EVENT_COMPONENT_LOADED, {ATTR_COMPONENT: "event"})
|
|
await hass.async_block_till_done()
|
|
|
|
# Firing again should not check again
|
|
assert len(processed) == 2
|
|
|
|
|
|
@callback
|
|
def _process_platform_callback(
|
|
hass: HomeAssistant, domain: str, platform: ModuleType
|
|
) -> None:
|
|
"""Process platform."""
|
|
raise HomeAssistantError("Non-compliant platform")
|
|
|
|
|
|
async def _process_platform_coro(
|
|
hass: HomeAssistant, domain: str, platform: ModuleType
|
|
) -> None:
|
|
"""Process platform."""
|
|
raise HomeAssistantError("Non-compliant platform")
|
|
|
|
|
|
@pytest.mark.no_fail_on_log_exception
|
|
@pytest.mark.parametrize(
|
|
"process_platform", (_process_platform_callback, _process_platform_coro)
|
|
)
|
|
async def test_process_integration_platforms_non_compliant(
|
|
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, process_platform: Callable
|
|
) -> None:
|
|
"""Test processing integrations using with a non-compliant platform."""
|
|
loaded_platform = Mock()
|
|
mock_platform(hass, "loaded_unique_880.platform_to_check", loaded_platform)
|
|
hass.config.components.add("loaded_unique_880")
|
|
|
|
event_platform = Mock()
|
|
mock_platform(hass, "event_unique_990.platform_to_check", event_platform)
|
|
|
|
processed = []
|
|
|
|
await async_process_integration_platforms(
|
|
hass, "platform_to_check", process_platform
|
|
)
|
|
|
|
assert len(processed) == 0
|
|
assert "Exception in " in caplog.text
|
|
assert "platform_to_check" in caplog.text
|
|
assert "Non-compliant platform" in caplog.text
|
|
assert "loaded_unique_880" in caplog.text
|
|
caplog.clear()
|
|
|
|
hass.bus.async_fire(EVENT_COMPONENT_LOADED, {ATTR_COMPONENT: "event_unique_990"})
|
|
await hass.async_block_till_done()
|
|
|
|
assert "Exception in " in caplog.text
|
|
assert "platform_to_check" in caplog.text
|
|
assert "Non-compliant platform" in caplog.text
|
|
assert "event_unique_990" in caplog.text
|
|
|
|
assert len(processed) == 0
|
|
|
|
|
|
async def test_broken_integration(
|
|
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
"""Test handling an integration with a broken or missing manifest."""
|
|
Mock()
|
|
hass.config.components.add("loaded")
|
|
|
|
event_platform = Mock()
|
|
mock_platform(hass, "event.platform_to_check", event_platform)
|
|
|
|
processed = []
|
|
|
|
async def _process_platform(hass, domain, platform):
|
|
"""Process platform."""
|
|
processed.append((domain, platform))
|
|
|
|
await async_process_integration_platforms(
|
|
hass, "platform_to_check", _process_platform
|
|
)
|
|
|
|
assert len(processed) == 0
|
|
assert "Error importing integration loaded for platform_to_check" in caplog.text
|
|
|
|
|
|
async def test_process_integration_platforms_no_integrations(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test processing integrations when no integrations are loaded."""
|
|
event_platform = Mock()
|
|
mock_platform(hass, "event.platform_to_check", event_platform)
|
|
|
|
processed = []
|
|
|
|
async def _process_platform(hass, domain, platform):
|
|
"""Process platform."""
|
|
processed.append((domain, platform))
|
|
|
|
await async_process_integration_platforms(
|
|
hass, "platform_to_check", _process_platform
|
|
)
|
|
|
|
assert len(processed) == 0
|