mirror of
https://github.com/home-assistant/core.git
synced 2026-09-26 17:31:15 -04:00
New integration: INDI Allsky (#179123)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Co-authored-by: Joostlek <joostlek@outlook.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Hamish
co-authored by
Copilot Autofix powered by AI
Joostlek
Claude Opus 4.8
Copilot Autofix powered by AI
Joostlek
Claude Opus 4.8
parent
61a63b82d0
commit
362fa725e6
@@ -312,6 +312,7 @@ homeassistant.components.imgw_pib.*
|
||||
homeassistant.components.immich.*
|
||||
homeassistant.components.incomfort.*
|
||||
homeassistant.components.indevolt.*
|
||||
homeassistant.components.indi_allsky.*
|
||||
homeassistant.components.inels.*
|
||||
homeassistant.components.infrared.*
|
||||
homeassistant.components.input_button.*
|
||||
|
||||
Generated
+2
@@ -887,6 +887,8 @@ CLAUDE.md @home-assistant/core
|
||||
/tests/components/incomfort/ @jbouwh
|
||||
/homeassistant/components/indevolt/ @xirt
|
||||
/tests/components/indevolt/ @xirt
|
||||
/homeassistant/components/indi_allsky/ @TN-1
|
||||
/tests/components/indi_allsky/ @TN-1
|
||||
/homeassistant/components/inels/ @epdevlab
|
||||
/tests/components/inels/ @epdevlab
|
||||
/homeassistant/components/influxdb/ @mdegat01 @Robbie1221
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
"""The INDI Allsky integration."""
|
||||
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .coordinator import IndiAllSkyConfigEntry, IndiAllSkyDataUpdateCoordinator
|
||||
|
||||
_PLATFORMS: list[Platform] = [Platform.CAMERA]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: IndiAllSkyConfigEntry) -> bool:
|
||||
"""Set up INDI Allsky from a config entry."""
|
||||
coordinator = IndiAllSkyDataUpdateCoordinator(hass, entry)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
entry.runtime_data = coordinator
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: IndiAllSkyConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)
|
||||
@@ -0,0 +1,55 @@
|
||||
"""Support for INDI Allsky camera."""
|
||||
|
||||
from typing import override
|
||||
|
||||
from aioindiallsky import IndiAllSkyError
|
||||
|
||||
from homeassistant.components.camera import Camera
|
||||
from homeassistant.components.image import infer_image_type
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from .coordinator import IndiAllSkyConfigEntry, IndiAllSkyDataUpdateCoordinator
|
||||
from .entity import IndiAllSkyEntity
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: IndiAllSkyConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the INDI Allsky camera platform."""
|
||||
coordinator = entry.runtime_data
|
||||
async_add_entities([IndiAllSkyCamera(coordinator, entry)])
|
||||
|
||||
|
||||
class IndiAllSkyCamera(IndiAllSkyEntity, Camera):
|
||||
"""Representation of an INDI Allsky camera."""
|
||||
|
||||
_attr_name = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: IndiAllSkyDataUpdateCoordinator,
|
||||
entry: IndiAllSkyConfigEntry,
|
||||
) -> None:
|
||||
"""Initialize the camera."""
|
||||
super().__init__(coordinator, entry)
|
||||
Camera.__init__(self)
|
||||
self._attr_unique_id = entry.entry_id
|
||||
|
||||
@override
|
||||
async def async_camera_image(
|
||||
self, width: int | None = None, height: int | None = None
|
||||
) -> bytes | None:
|
||||
"""Return bytes of current camera image."""
|
||||
try:
|
||||
image: bytes = await self.coordinator.client.fetch_image("latestimage")
|
||||
except IndiAllSkyError:
|
||||
return None
|
||||
else:
|
||||
if content_type := infer_image_type(image):
|
||||
self.content_type = content_type
|
||||
return image
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Config flow for INDI Allsky integration."""
|
||||
|
||||
import logging
|
||||
from typing import Any, override
|
||||
|
||||
from aioindiallsky import (
|
||||
IndiAllSkyAuthError,
|
||||
IndiAllSkyClient,
|
||||
IndiAllSkyConnectionError,
|
||||
IndiAllSkyTimeoutError,
|
||||
)
|
||||
import probatio
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.selector import (
|
||||
BooleanSelector,
|
||||
NumberSelector,
|
||||
NumberSelectorConfig,
|
||||
NumberSelectorMode,
|
||||
TextSelector,
|
||||
TextSelectorConfig,
|
||||
TextSelectorType,
|
||||
)
|
||||
|
||||
from .const import DOMAIN
|
||||
from .util import get_ssl_context, normalize_host
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
STEP_USER_DATA_SCHEMA = probatio.Schema(
|
||||
{
|
||||
probatio.Required(CONF_HOST): TextSelector(
|
||||
TextSelectorConfig(
|
||||
type=TextSelectorType.TEXT,
|
||||
autocomplete="host",
|
||||
),
|
||||
),
|
||||
probatio.Required(CONF_PORT, default=443): NumberSelector(
|
||||
NumberSelectorConfig(
|
||||
min=1,
|
||||
max=65535,
|
||||
mode=NumberSelectorMode.BOX,
|
||||
),
|
||||
),
|
||||
probatio.Optional(CONF_SSL, default=True): BooleanSelector(),
|
||||
probatio.Optional(CONF_VERIFY_SSL, default=True): BooleanSelector(),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
|
||||
"""Validate that the user input allows us to connect to INDI Allsky."""
|
||||
client = IndiAllSkyClient(
|
||||
host=data[CONF_HOST],
|
||||
port=int(data[CONF_PORT]),
|
||||
ssl=get_ssl_context(
|
||||
data.get(CONF_SSL, True),
|
||||
data.get(CONF_VERIFY_SSL, True),
|
||||
),
|
||||
session=async_get_clientsession(hass),
|
||||
)
|
||||
|
||||
try:
|
||||
await client.fetch_image("latestimage")
|
||||
except IndiAllSkyAuthError as err:
|
||||
_LOGGER.error(
|
||||
"Authentication failed for INDI Allsky at %s:%s: %s",
|
||||
data[CONF_HOST],
|
||||
data[CONF_PORT],
|
||||
err,
|
||||
)
|
||||
raise InvalidAuth from err
|
||||
except (IndiAllSkyConnectionError, IndiAllSkyTimeoutError) as err:
|
||||
_LOGGER.error(
|
||||
"Cannot connect to INDI Allsky instance at %s:%s: %s",
|
||||
data[CONF_HOST],
|
||||
data[CONF_PORT],
|
||||
err,
|
||||
)
|
||||
raise CannotConnect from err
|
||||
except Exception as err:
|
||||
raise Unknown from err
|
||||
|
||||
|
||||
class IndiAllSkyConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for INDI Allsky."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
@override
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the initial step."""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
if user_input is not None:
|
||||
user_input[CONF_HOST] = normalize_host(user_input[CONF_HOST])
|
||||
user_input[CONF_PORT] = int(user_input[CONF_PORT])
|
||||
self._async_abort_entries_match(
|
||||
{
|
||||
CONF_HOST: user_input[CONF_HOST],
|
||||
CONF_PORT: user_input[CONF_PORT],
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
await validate_input(self.hass, user_input)
|
||||
except CannotConnect:
|
||||
errors["base"] = "cannot_connect"
|
||||
except InvalidAuth:
|
||||
errors["base"] = "invalid_auth"
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected exception")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
port = user_input[CONF_PORT]
|
||||
default_port = 443 if user_input.get(CONF_SSL, True) else 80
|
||||
host_str = (
|
||||
f"{user_input[CONF_HOST]}:{port}"
|
||||
if port != default_port
|
||||
else user_input[CONF_HOST]
|
||||
)
|
||||
return self.async_create_entry(
|
||||
title=f"INDI Allsky ({host_str})",
|
||||
data=user_input,
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=self.add_suggested_values_to_schema(
|
||||
STEP_USER_DATA_SCHEMA, user_input
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
|
||||
class CannotConnect(HomeAssistantError):
|
||||
"""Error to indicate we cannot connect."""
|
||||
|
||||
|
||||
class InvalidAuth(HomeAssistantError):
|
||||
"""Error to indicate there is invalid auth."""
|
||||
|
||||
|
||||
class Unknown(HomeAssistantError):
|
||||
"""Unexpected error."""
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Constants for the INDI Allsky integration."""
|
||||
|
||||
DOMAIN = "indi_allsky"
|
||||
@@ -0,0 +1,54 @@
|
||||
"""DataUpdateCoordinator for INDI Allsky integration."""
|
||||
|
||||
import logging
|
||||
from typing import override
|
||||
|
||||
from aioindiallsky import IndiAllSkyClient, IndiAllSkyError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN
|
||||
from .util import get_ssl_context
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
type IndiAllSkyConfigEntry = ConfigEntry[IndiAllSkyDataUpdateCoordinator]
|
||||
|
||||
|
||||
class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[None]):
|
||||
"""Class to manage fetching INDI Allsky data from the API."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, entry: IndiAllSkyConfigEntry) -> None:
|
||||
"""Initialize the coordinator."""
|
||||
self.client = IndiAllSkyClient(
|
||||
host=entry.data[CONF_HOST],
|
||||
port=int(entry.data[CONF_PORT]),
|
||||
ssl=get_ssl_context(
|
||||
entry.data[CONF_SSL],
|
||||
entry.data[CONF_VERIFY_SSL],
|
||||
),
|
||||
session=async_get_clientsession(hass),
|
||||
)
|
||||
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
config_entry=entry,
|
||||
name=DOMAIN,
|
||||
update_interval=None,
|
||||
)
|
||||
|
||||
@override
|
||||
async def _async_update_data(self) -> None:
|
||||
"""Fetch INDI Allsky metadata and verify connection."""
|
||||
try:
|
||||
await self.client.fetch_image("latestimage")
|
||||
except IndiAllSkyError as err:
|
||||
raise UpdateFailed(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="update_failed",
|
||||
) from err
|
||||
@@ -0,0 +1,26 @@
|
||||
"""Base entity for the INDI Allsky integration."""
|
||||
|
||||
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN
|
||||
from .coordinator import IndiAllSkyConfigEntry, IndiAllSkyDataUpdateCoordinator
|
||||
|
||||
|
||||
class IndiAllSkyEntity(CoordinatorEntity[IndiAllSkyDataUpdateCoordinator]):
|
||||
"""Base class for INDI Allsky entities."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: IndiAllSkyDataUpdateCoordinator,
|
||||
entry: IndiAllSkyConfigEntry,
|
||||
) -> None:
|
||||
"""Initialize the entity."""
|
||||
super().__init__(coordinator)
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, entry.entry_id)},
|
||||
name=entry.title,
|
||||
entry_type=DeviceEntryType.SERVICE,
|
||||
)
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"domain": "indi_allsky",
|
||||
"name": "INDI Allsky",
|
||||
"codeowners": ["@TN-1"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/indi_allsky",
|
||||
"integration_type": "service",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["aioindiallsky"],
|
||||
"quality_scale": "bronze",
|
||||
"requirements": ["aioindiallsky==0.1.1"]
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
rules:
|
||||
# Bronze
|
||||
action-setup:
|
||||
status: exempt
|
||||
comment: Integration does not register custom actions.
|
||||
appropriate-polling: done
|
||||
brands: done
|
||||
common-modules: done
|
||||
config-flow-test-coverage: done
|
||||
config-flow: done
|
||||
dependency-transparency: done
|
||||
docs-actions:
|
||||
status: exempt
|
||||
comment: Integration does not register custom actions.
|
||||
docs-conditions:
|
||||
status: exempt
|
||||
comment: Integration does not register custom conditions.
|
||||
docs-high-level-description: done
|
||||
docs-installation-instructions: done
|
||||
docs-removal-instructions: done
|
||||
docs-triggers:
|
||||
status: exempt
|
||||
comment: Integration does not register custom triggers.
|
||||
entity-event-setup:
|
||||
status: exempt
|
||||
comment: Integration does not register custom events.
|
||||
entity-unique-id: done
|
||||
has-entity-name: done
|
||||
runtime-data: done
|
||||
test-before-configure: done
|
||||
test-before-setup: done
|
||||
unique-config-entry: done
|
||||
|
||||
# Silver
|
||||
action-exceptions: todo
|
||||
config-entry-unloading: done
|
||||
docs-configuration-parameters: todo
|
||||
docs-installation-parameters: todo
|
||||
entity-unavailable: todo
|
||||
integration-owner: done
|
||||
log-when-unavailable: todo
|
||||
parallel-updates: done
|
||||
reauthentication-flow:
|
||||
status: exempt
|
||||
comment: Integration does not require reauthentication.
|
||||
test-coverage: done
|
||||
|
||||
# Gold
|
||||
devices: done
|
||||
diagnostics: todo
|
||||
discovery-update-info: todo
|
||||
discovery: todo
|
||||
docs-data-update: todo
|
||||
docs-examples: todo
|
||||
docs-known-limitations: todo
|
||||
docs-supported-devices: todo
|
||||
docs-supported-functions: todo
|
||||
docs-troubleshooting: todo
|
||||
docs-use-cases: todo
|
||||
dynamic-devices: todo
|
||||
entity-category: done
|
||||
entity-device-class: done
|
||||
entity-disabled-by-default: done
|
||||
entity-translations: done
|
||||
exception-translations: done
|
||||
icon-translations:
|
||||
status: exempt
|
||||
comment: Entities define no custom icons.
|
||||
reconfiguration-flow: todo
|
||||
repair-issues:
|
||||
status: exempt
|
||||
comment: Integration does not require user intervention repairs.
|
||||
stale-devices: todo
|
||||
|
||||
# Platinum
|
||||
async-dependency: done
|
||||
inject-websession: done
|
||||
strict-typing: done
|
||||
@@ -0,0 +1,34 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"port": "[%key:common::config_flow::data::port%]",
|
||||
"ssl": "[%key:common::config_flow::data::ssl%]",
|
||||
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
|
||||
},
|
||||
"data_description": {
|
||||
"host": "The hostname or IP address of your INDI Allsky server.",
|
||||
"port": "The port number of your INDI Allsky web server.",
|
||||
"ssl": "Connect using SSL/HTTPS.",
|
||||
"verify_ssl": "Verify the SSL certificate of the INDI Allsky server."
|
||||
},
|
||||
"description": "Enter the connection details for your INDI Allsky instance."
|
||||
}
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
"update_failed": {
|
||||
"message": "Error communicating with INDI Allsky API"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Utilities for the INDI Allsky integration."""
|
||||
|
||||
import ipaddress
|
||||
import ssl
|
||||
|
||||
from homeassistant.util.ssl import get_default_context, get_default_no_verify_context
|
||||
|
||||
|
||||
def normalize_host(host: str) -> str:
|
||||
"""Normalize hostname or IP address into canonical form."""
|
||||
host_clean = host.strip(" []")
|
||||
try:
|
||||
return str(ipaddress.ip_address(host_clean))
|
||||
except ValueError:
|
||||
return host_clean.lower().removesuffix(".")
|
||||
|
||||
|
||||
def get_ssl_context(ssl_enabled: bool, verify_ssl: bool) -> bool | ssl.SSLContext:
|
||||
"""Return SSL configuration for IndiAllSkyClient."""
|
||||
if not ssl_enabled:
|
||||
return False
|
||||
if not verify_ssl:
|
||||
return get_default_no_verify_context()
|
||||
return get_default_context()
|
||||
Generated
+1
@@ -374,6 +374,7 @@ FLOWS = {
|
||||
"improv_ble",
|
||||
"incomfort",
|
||||
"indevolt",
|
||||
"indi_allsky",
|
||||
"inels",
|
||||
"influxdb",
|
||||
"inkbird",
|
||||
|
||||
@@ -3356,6 +3356,12 @@
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"indi_allsky": {
|
||||
"name": "INDI Allsky",
|
||||
"integration_type": "service",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"indianamichiganpower": {
|
||||
"name": "Indiana Michigan Power",
|
||||
"integration_type": "virtual",
|
||||
|
||||
@@ -2878,6 +2878,16 @@ disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.indi_allsky.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.inels.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
||||
Generated
+3
@@ -311,6 +311,9 @@ aioimaplib==2.0.1
|
||||
# homeassistant.components.immich
|
||||
aioimmich==0.17.0
|
||||
|
||||
# homeassistant.components.indi_allsky
|
||||
aioindiallsky==0.1.1
|
||||
|
||||
# homeassistant.components.ipp
|
||||
aioipp==0.19.0
|
||||
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
"""Tests for the INDI Allsky integration."""
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def setup_integration(
|
||||
hass: HomeAssistant,
|
||||
config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Set up the INDI Allsky integration for testing."""
|
||||
config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
@@ -0,0 +1,63 @@
|
||||
"""Common fixtures for the INDI Allsky tests."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.indi_allsky.const import DOMAIN
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_system_random() -> Generator[None]:
|
||||
"""Mock random.SystemRandom.getrandbits to produce deterministic camera access tokens."""
|
||||
with patch("random.SystemRandom.getrandbits", return_value=123123123123):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.indi_allsky.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_indi_allsky_client() -> Generator[AsyncMock]:
|
||||
"""Mock the third-party aioindiallsky client globally across coordinator and config flow."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.indi_allsky.coordinator.IndiAllSkyClient",
|
||||
autospec=True,
|
||||
) as mock_client,
|
||||
patch(
|
||||
"homeassistant.components.indi_allsky.config_flow.IndiAllSkyClient",
|
||||
new=mock_client,
|
||||
),
|
||||
):
|
||||
client_instance = mock_client.return_value
|
||||
client_instance.fetch_image = AsyncMock(
|
||||
return_value=b"\xff\xd8\xff\xe0fake_jpeg_data"
|
||||
)
|
||||
yield client_instance
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Fixture to cleanly create an INDI Allsky configuration entry."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="INDI Allsky",
|
||||
data={
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_PORT: 443,
|
||||
CONF_SSL: True,
|
||||
CONF_VERIFY_SSL: True,
|
||||
},
|
||||
entry_id="1234567890abcdef1234567890abcdef",
|
||||
)
|
||||
@@ -0,0 +1,54 @@
|
||||
# serializer version: 1
|
||||
# name: test_camera_setup_and_states[camera.indi_allsky-entry]
|
||||
EntityRegistryEntrySnapshot({
|
||||
'aliases': list([
|
||||
None,
|
||||
]),
|
||||
'area_id': None,
|
||||
'capabilities': None,
|
||||
'config_entry_id': <ANY>,
|
||||
'config_subentry_id': <ANY>,
|
||||
'device_class': None,
|
||||
'device_id': <ANY>,
|
||||
'disabled_by': None,
|
||||
'domain': 'camera',
|
||||
'entity_category': None,
|
||||
'entity_id': 'camera.indi_allsky',
|
||||
'has_entity_name': True,
|
||||
'hidden_by': None,
|
||||
'icon': None,
|
||||
'id': <ANY>,
|
||||
'labels': set({
|
||||
}),
|
||||
'name': None,
|
||||
'object_id_base': None,
|
||||
'options': dict({
|
||||
}),
|
||||
'original_device_class': None,
|
||||
'original_icon': None,
|
||||
'original_name': None,
|
||||
'platform': 'indi_allsky',
|
||||
'previous_unique_id': None,
|
||||
'suggested_object_id': None,
|
||||
'supported_features': 0,
|
||||
'translation_key': None,
|
||||
'unique_id': '1234567890abcdef1234567890abcdef',
|
||||
'unit_of_measurement': None,
|
||||
})
|
||||
# ---
|
||||
# name: test_camera_setup_and_states[camera.indi_allsky-state]
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
<CameraEntityStateAttribute.ACCESS_TOKEN: 'access_token'>: '1caab5c3b3',
|
||||
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: '/api/camera_proxy/camera.indi_allsky?token=1caab5c3b3',
|
||||
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky',
|
||||
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <CameraEntityFeature: 0>,
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'camera.indi_allsky',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': 'idle',
|
||||
})
|
||||
# ---
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Tests for the INDI Allsky camera platform."""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from aioindiallsky import IndiAllSkyError
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.camera import async_get_image
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry, snapshot_platform
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_indi_allsky_client")
|
||||
async def test_camera_setup_and_states(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
snapshot: SnapshotAssertion,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test standard successful setup and entity snapshots using snapshot_platform."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("image_bytes", "expected_content_type"),
|
||||
[
|
||||
pytest.param(b"\xff\xd8\xff\xe0fake_jpeg_data", "image/jpeg", id="jpeg"),
|
||||
pytest.param(b"\x89PNG\r\n\x1a\nfake_png_data", "image/png", id="png"),
|
||||
],
|
||||
)
|
||||
async def test_camera_image_and_update(
|
||||
hass: HomeAssistant,
|
||||
mock_indi_allsky_client: AsyncMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
image_bytes: bytes,
|
||||
expected_content_type: str,
|
||||
) -> None:
|
||||
"""Test camera image fetching and content type inference."""
|
||||
mock_indi_allsky_client.fetch_image.return_value = image_bytes
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
image = await async_get_image(hass, "camera.indi_allsky")
|
||||
assert image.content == image_bytes
|
||||
assert image.content_type == expected_content_type
|
||||
|
||||
|
||||
async def test_camera_image_fetch_failure(
|
||||
hass: HomeAssistant,
|
||||
mock_indi_allsky_client: AsyncMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test camera image fetching failure handling."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
mock_indi_allsky_client.fetch_image.side_effect = IndiAllSkyError("Fetch error")
|
||||
|
||||
with pytest.raises(HomeAssistantError, match="Unable to get image"):
|
||||
await async_get_image(hass, "camera.indi_allsky")
|
||||
@@ -0,0 +1,243 @@
|
||||
"""Test the INDI Allsky Config flow."""
|
||||
|
||||
import ssl
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from aioindiallsky import IndiAllSkyAuthError, IndiAllSkyConnectionError
|
||||
import pytest
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.indi_allsky.const import DOMAIN
|
||||
from homeassistant.components.indi_allsky.util import get_ssl_context, normalize_host
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("host", "port", "ssl_enabled", "verify_ssl", "expected_title"),
|
||||
[
|
||||
pytest.param(
|
||||
"127.0.0.1",
|
||||
443,
|
||||
True,
|
||||
False,
|
||||
"INDI Allsky (127.0.0.1)",
|
||||
id="ipv4_default_port",
|
||||
),
|
||||
pytest.param(
|
||||
"127.0.0.1",
|
||||
8443,
|
||||
True,
|
||||
True,
|
||||
"INDI Allsky (127.0.0.1:8443)",
|
||||
id="ipv4_custom_port",
|
||||
),
|
||||
pytest.param(
|
||||
"2001:db8::1",
|
||||
443,
|
||||
True,
|
||||
True,
|
||||
"INDI Allsky (2001:db8::1)",
|
||||
id="ipv6_default_port",
|
||||
),
|
||||
pytest.param(
|
||||
"2001:db8::1",
|
||||
8080,
|
||||
False,
|
||||
True,
|
||||
"INDI Allsky (2001:db8::1:8080)",
|
||||
id="ipv6_custom_port",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_form_success(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: AsyncMock,
|
||||
mock_indi_allsky_client: AsyncMock,
|
||||
host: str,
|
||||
port: int,
|
||||
ssl_enabled: bool,
|
||||
verify_ssl: bool,
|
||||
expected_title: str,
|
||||
) -> None:
|
||||
"""Test we get the form, validate the client, and create a successful entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: host,
|
||||
CONF_PORT: port,
|
||||
CONF_SSL: ssl_enabled,
|
||||
CONF_VERIFY_SSL: verify_ssl,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == expected_title
|
||||
assert result["data"] == {
|
||||
CONF_HOST: host,
|
||||
CONF_PORT: port,
|
||||
CONF_SSL: ssl_enabled,
|
||||
CONF_VERIFY_SSL: verify_ssl,
|
||||
}
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("side_effect", "error_key"),
|
||||
[
|
||||
(IndiAllSkyConnectionError("Cannot connect"), "cannot_connect"),
|
||||
(IndiAllSkyAuthError("Invalid key"), "invalid_auth"),
|
||||
(Exception("Unexpected error"), "unknown"),
|
||||
],
|
||||
)
|
||||
async def test_form_failures_and_recovery(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: AsyncMock,
|
||||
mock_indi_allsky_client: AsyncMock,
|
||||
side_effect: Exception,
|
||||
error_key: str,
|
||||
) -> None:
|
||||
"""Test handling validation failures and ensuring the flow can recover."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
mock_indi_allsky_client.fetch_image.side_effect = side_effect
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_PORT: 443,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": error_key}
|
||||
|
||||
mock_indi_allsky_client.fetch_image.side_effect = None
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: "127.0.0.1",
|
||||
CONF_PORT: 443,
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"duplicate_host",
|
||||
[
|
||||
"127.0.0.1",
|
||||
" 127.0.0.1 ",
|
||||
],
|
||||
)
|
||||
async def test_form_already_configured(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
duplicate_host: str,
|
||||
) -> None:
|
||||
"""Test duplicate host/port configurations abort early."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: duplicate_host,
|
||||
CONF_PORT: 443,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"duplicate_host",
|
||||
[
|
||||
"2001:db8::1",
|
||||
"[2001:db8::1]",
|
||||
"2001:0db8:0000:0000:0000:0000:0000:0001",
|
||||
"2001:DB8::1",
|
||||
" [2001:db8::1] ",
|
||||
],
|
||||
)
|
||||
async def test_form_already_configured_ipv6(
|
||||
hass: HomeAssistant,
|
||||
duplicate_host: str,
|
||||
) -> None:
|
||||
"""Test duplicate IPv6 configurations abort regardless of formatting variation."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title="INDI Allsky (2001:db8::1)",
|
||||
data={
|
||||
CONF_HOST: "2001:db8::1",
|
||||
CONF_PORT: 443,
|
||||
},
|
||||
entry_id="ipv6_entry",
|
||||
)
|
||||
entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: duplicate_host,
|
||||
CONF_PORT: 443,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("input_host", "expected_host"),
|
||||
[
|
||||
("127.0.0.1", "127.0.0.1"),
|
||||
(" 127.0.0.1 ", "127.0.0.1"),
|
||||
("allsky.local", "allsky.local"),
|
||||
("2001:db8::1", "2001:db8::1"),
|
||||
("[2001:db8::1]", "2001:db8::1"),
|
||||
("2001:0db8:0000:0000:0000:0000:0000:0001", "2001:db8::1"),
|
||||
("2001:DB8::1", "2001:db8::1"),
|
||||
(" [2001:db8::1] ", "2001:db8::1"),
|
||||
],
|
||||
)
|
||||
def test_normalize_host(input_host: str, expected_host: str) -> None:
|
||||
"""Test host normalization for IPv4, IPv6, and hostnames."""
|
||||
assert normalize_host(input_host) == expected_host
|
||||
|
||||
|
||||
def test_get_ssl_context() -> None:
|
||||
"""Test get_ssl_context return values for various SSL setting combinations."""
|
||||
assert get_ssl_context(ssl_enabled=False, verify_ssl=True) is False
|
||||
assert get_ssl_context(ssl_enabled=False, verify_ssl=False) is False
|
||||
|
||||
ctx_verified = get_ssl_context(ssl_enabled=True, verify_ssl=True)
|
||||
assert isinstance(ctx_verified, ssl.SSLContext)
|
||||
assert ctx_verified.verify_mode != ssl.CERT_NONE
|
||||
|
||||
ctx_no_verify = get_ssl_context(ssl_enabled=True, verify_ssl=False)
|
||||
assert isinstance(ctx_no_verify, ssl.SSLContext)
|
||||
assert ctx_no_verify.verify_mode == ssl.CERT_NONE
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Test initialization of INDI Allsky integration."""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from aioindiallsky import IndiAllSkyConnectionError
|
||||
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import setup_integration
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_setup_and_unload_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_indi_allsky_client: AsyncMock,
|
||||
) -> None:
|
||||
"""Test successful setup and unload of entry."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_setup_failure_retry(
|
||||
hass: HomeAssistant,
|
||||
mock_indi_allsky_client: AsyncMock,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test that an API connection failure during initial setup places entry in retry state."""
|
||||
mock_indi_allsky_client.fetch_image.side_effect = IndiAllSkyConnectionError(
|
||||
"Cannot connect to INDI Allsky server"
|
||||
)
|
||||
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
Reference in New Issue
Block a user