starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-27 01:46:11 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add sensor platform to INDI Allsky integration (#182326)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Norbert Rittel <norbert@rittel.de>
This commit is contained in:
Hamish
2026-09-21 19:10:17 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI Norbert Rittel
parent 6ca1b628c6
commit 1699ecac0a
12 changed files with 748 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/indi_allsky/__init__.py
+7 -1
View File
@@ -5,7 +5,7 @@ from homeassistant.core import HomeAssistant
from .coordinator import IndiAllSkyConfigEntry, IndiAllSkyDataUpdateCoordinator
_PLATFORMS: list[Platform] = [Platform.CAMERA]
_PLATFORMS: list[Platform] = [Platform.CAMERA, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: IndiAllSkyConfigEntry) -> bool:
@@ -15,6 +15,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: IndiAllSkyConfigEntry) -
entry.runtime_data = coordinator
entry.async_create_background_task(
hass,
coordinator.client.listen(auto_reconnect=True),
"indi_allsky_ws_events",
)
await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
return True
homeassistant/components/indi_allsky/coordinator.py
+27 -3
View File
@@ -1,9 +1,10 @@
"""DataUpdateCoordinator for INDI Allsky integration."""
from dataclasses import dataclass
import logging
from typing import override
from aioindiallsky import IndiAllSkyClient, IndiAllSkyError
from aioindiallsky import ExposureData, IndiAllSkyClient, IndiAllSkyError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_SSL, CONF_VERIFY_SSL
@@ -19,7 +20,14 @@ _LOGGER = logging.getLogger(__name__)
type IndiAllSkyConfigEntry = ConfigEntry[IndiAllSkyDataUpdateCoordinator]
class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[None]):
@dataclass
class IndiAllSkyData:
"""Data model for INDI Allsky coordinator data."""
exposure: ExposureData | None = None
class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[IndiAllSkyData]):
"""Class to manage fetching INDI Allsky data from the API."""
def __init__(self, hass: HomeAssistant, entry: IndiAllSkyConfigEntry) -> None:
@@ -33,6 +41,13 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[None]):
),
session=async_get_clientsession(hass),
)
self.latest_exposure: ExposureData | None = None
unsub = self.client.register_callback(
"exposure_complete", self._handle_exposure_complete
)
entry.async_on_unload(unsub)
entry.async_on_unload(self.client.disconnect)
super().__init__(
hass,
@@ -42,13 +57,22 @@ class IndiAllSkyDataUpdateCoordinator(DataUpdateCoordinator[None]):
update_interval=None,
)
def _handle_exposure_complete(self, exposure: ExposureData) -> None:
"""Handle new exposure_complete event from WebSocket stream."""
self.latest_exposure = exposure
self.async_set_updated_data(IndiAllSkyData(exposure=exposure))
@override
async def _async_update_data(self) -> None:
async def _async_update_data(self) -> IndiAllSkyData:
"""Fetch INDI Allsky metadata and verify connection."""
try:
await self.client.fetch_image("latestimage")
if not self.client.is_connected:
await self.client.connect()
except IndiAllSkyError as err:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_failed",
) from err
return IndiAllSkyData(exposure=self.latest_exposure)
homeassistant/components/indi_allsky/manifest.json
+1 -1
View File
@@ -5,7 +5,7 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/indi_allsky",
"integration_type": "service",
"iot_class": "local_polling",
"iot_class": "local_push",
"loggers": ["aioindiallsky"],
"quality_scale": "bronze",
"requirements": ["aioindiallsky==0.1.2"]
homeassistant/components/indi_allsky/sensor.py
+120
View File
@@ -0,0 +1,120 @@
"""Support for INDI Allsky sensors."""
from collections.abc import Callable
from dataclasses import dataclass
from typing import override
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import EntityCategory, UnitOfTemperature, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from .coordinator import (
IndiAllSkyConfigEntry,
IndiAllSkyData,
IndiAllSkyDataUpdateCoordinator,
)
from .entity import IndiAllSkyEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class IndiAllSkySensorEntityDescription(SensorEntityDescription):
"""Class describing INDI Allsky sensor entities."""
value_fn: Callable[[IndiAllSkyData], StateType]
SENSOR_DESCRIPTIONS: tuple[IndiAllSkySensorEntityDescription, ...] = (
IndiAllSkySensorEntityDescription(
key="binmode",
translation_key="binmode",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: data.exposure.binmode if data.exposure else None,
),
IndiAllSkySensorEntityDescription(
key="exposure",
translation_key="exposure",
device_class=SensorDeviceClass.DURATION,
native_unit_of_measurement=UnitOfTime.SECONDS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.exposure.exposure if data.exposure else None,
),
IndiAllSkySensorEntityDescription(
key="filename",
translation_key="filename",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: data.exposure.filename if data.exposure else None,
),
IndiAllSkySensorEntityDescription(
key="gain",
translation_key="gain",
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda data: data.exposure.gain if data.exposure else None,
),
IndiAllSkySensorEntityDescription(
key="sqm",
translation_key="sqm",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.exposure.sqm if data.exposure else None,
),
IndiAllSkySensorEntityDescription(
key="stars",
translation_key="stars",
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.exposure.stars if data.exposure else None,
),
IndiAllSkySensorEntityDescription(
key="temperature",
device_class=SensorDeviceClass.TEMPERATURE,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
state_class=SensorStateClass.MEASUREMENT,
value_fn=lambda data: data.exposure.temp if data.exposure else None,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: IndiAllSkyConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up INDI Allsky sensors based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
IndiAllSkySensor(coordinator, entry, description)
for description in SENSOR_DESCRIPTIONS
)
class IndiAllSkySensor(IndiAllSkyEntity, SensorEntity):
"""Representation of an INDI Allsky sensor."""
entity_description: IndiAllSkySensorEntityDescription
def __init__(
self,
coordinator: IndiAllSkyDataUpdateCoordinator,
entry: IndiAllSkyConfigEntry,
description: IndiAllSkySensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator, entry)
self.entity_description = description
self._attr_unique_id = f"{entry.entry_id}_{description.key}"
@property
@override
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data)
homeassistant/components/indi_allsky/strings.json
+22
View File
@@ -26,6 +26,28 @@
}
}
},
"entity": {
"sensor": {
"binmode": {
"name": "Binning mode"
},
"exposure": {
"name": "Exposure time"
},
"filename": {
"name": "Filename"
},
"gain": {
"name": "Gain"
},
"sqm": {
"name": "Sky quality"
},
"stars": {
"name": "Stars"
}
}
},
"exceptions": {
"update_failed": {
"message": "Error communicating with INDI Allsky API"
homeassistant/generated/integrations.json
+1 -1
View File
@@ -3377,7 +3377,7 @@
"name": "INDI Allsky",
"integration_type": "service",
"config_flow": true,
"iot_class": "local_polling"
"iot_class": "local_push"
},
"indianamichiganpower": {
"name": "Indiana Michigan Power",
tests/components/indi_allsky/conftest.py
+26 -3
View File
@@ -1,14 +1,16 @@
"""Common fixtures for the INDI Allsky tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from collections.abc import Callable, Generator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from aioindiallsky import ExposureData
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
from tests.common import MockConfigEntry, load_json_object_fixture
@pytest.fixture(autouse=True)
@@ -30,6 +32,14 @@ def mock_setup_entry() -> Generator[AsyncMock]:
@pytest.fixture
def mock_indi_allsky_client() -> Generator[AsyncMock]:
"""Mock the third-party aioindiallsky client globally across coordinator and config flow."""
callbacks: dict[str, list[Callable[..., Any]]] = {}
def register_callback(
event_type: str, callback: Callable[..., Any]
) -> Callable[[], None]:
callbacks.setdefault(event_type, []).append(callback)
return lambda: callbacks[event_type].remove(callback)
with (
patch(
"homeassistant.components.indi_allsky.coordinator.IndiAllSkyClient",
@@ -44,9 +54,22 @@ def mock_indi_allsky_client() -> Generator[AsyncMock]:
client_instance.fetch_image = AsyncMock(
return_value=b"\xff\xd8\xff\xe0fake_jpeg_data"
)
client_instance.connect = AsyncMock()
client_instance.listen = AsyncMock()
client_instance.disconnect = AsyncMock()
client_instance.is_connected = False
client_instance.register_callback = MagicMock(side_effect=register_callback)
client_instance.callbacks = callbacks
yield client_instance
@pytest.fixture
def mock_exposure_data() -> ExposureData:
"""Fixture to provide sample ExposureData from fixture JSON."""
raw_data = load_json_object_fixture("exposure_complete.json", DOMAIN)
return ExposureData.from_dict(raw_data)
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Fixture to cleanly create an INDI Allsky configuration entry."""
tests/components/indi_allsky/fixtures/exposure_complete.json
+13
View File
@@ -0,0 +1,13 @@
{
"filename": "test.jpg",
"createDate": "2026-08-13 22:53:41",
"exposure": 0.185,
"gain": 0.0,
"binmode": 1,
"night": false,
"camera_id": 1,
"id": 4718,
"temp": -273.15,
"sqm": 32928.83,
"stars": 0
}
tests/components/indi_allsky/snapshots/test_sensor.ambr
+373
View File
@@ -0,0 +1,373 @@
# serializer version: 1
# name: test_sensor_setup_and_states[sensor.indi_allsky_binning_mode-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': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.indi_allsky_binning_mode',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Binning mode',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Binning mode',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'binmode',
'unique_id': '1234567890abcdef1234567890abcdef_binmode',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_binning_mode-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Binning mode',
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_binning_mode',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_exposure_time-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_exposure_time',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Exposure time',
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Exposure time',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'exposure',
'unique_id': '1234567890abcdef1234567890abcdef_exposure',
'unit_of_measurement': <UnitOfTime.SECONDS: 's'>,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_exposure_time-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'duration',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Exposure time',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTime.SECONDS: 's'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_exposure_time',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_filename-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': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.indi_allsky_filename',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Filename',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Filename',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'filename',
'unique_id': '1234567890abcdef1234567890abcdef_filename',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_filename-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Filename',
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_filename',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_gain-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': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.indi_allsky_gain',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Gain',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Gain',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'gain',
'unique_id': '1234567890abcdef1234567890abcdef_gain',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_gain-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Gain',
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_gain',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_sky_quality-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_sky_quality',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Sky quality',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Sky quality',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'sqm',
'unique_id': '1234567890abcdef1234567890abcdef_sqm',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_sky_quality-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Sky quality',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_sky_quality',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_stars-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_stars',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Stars',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Stars',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'stars',
'unique_id': '1234567890abcdef1234567890abcdef_stars',
'unit_of_measurement': None,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_stars-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Stars',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_stars',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.indi_allsky_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Temperature',
'platform': 'indi_allsky',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '1234567890abcdef1234567890abcdef_temperature',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensor_setup_and_states[sensor.indi_allsky_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'temperature',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'INDI Allsky Temperature',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.indi_allsky_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
tests/components/indi_allsky/test_camera.py
+7 -3
View File
@@ -1,12 +1,13 @@
"""Tests for the INDI Allsky camera platform."""
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, patch
from aioindiallsky import IndiAllSkyError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.camera import async_get_image
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
@@ -24,8 +25,11 @@ async def test_camera_setup_and_states(
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)
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.CAMERA]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(
hass, entity_registry, snapshot, mock_config_entry.entry_id
)
@pytest.mark.parametrize(
tests/components/indi_allsky/test_init.py
+13 -5
View File
@@ -1,8 +1,9 @@
"""Test initialization of INDI Allsky integration."""
"""Tests for the INDI Allsky integration."""
from unittest.mock import AsyncMock
from aioindiallsky import IndiAllSkyConnectionError
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
@@ -14,29 +15,36 @@ from tests.common import MockConfigEntry
async def test_setup_and_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful setup and unload of entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_indi_allsky_client.listen.assert_called_once_with(auto_reconnect=True)
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
mock_indi_allsky_client.disconnect.assert_awaited_once()
@pytest.mark.parametrize(
"method_name",
["fetch_image", "connect"],
)
async def test_setup_failure_retry(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
method_name: str,
) -> 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"
)
getattr(
mock_indi_allsky_client, method_name
).side_effect = IndiAllSkyConnectionError("Cannot connect to INDI Allsky server")
await setup_integration(hass, mock_config_entry)
tests/components/indi_allsky/test_sensor.py
+138
View File
@@ -0,0 +1,138 @@
"""Tests for the INDI Allsky sensor platform."""
from dataclasses import replace
from unittest.mock import AsyncMock, patch
from aioindiallsky import ExposureData
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.const import STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.usefixtures(
"entity_registry_enabled_by_default", "mock_indi_allsky_client"
)
async def test_sensor_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."""
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
await snapshot_platform(
hass, entity_registry, snapshot, mock_config_entry.entry_id
)
@pytest.mark.usefixtures("mock_indi_allsky_client")
async def test_disabled_sensors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test that disabled-by-default sensors are registered as disabled."""
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
for entity_id in (
"sensor.indi_allsky_binning_mode",
"sensor.indi_allsky_filename",
"sensor.indi_allsky_gain",
):
entry = entity_registry.async_get(entity_id)
assert entry is not None
assert entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION
for entity_id in (
"sensor.indi_allsky_exposure_time",
"sensor.indi_allsky_sky_quality",
"sensor.indi_allsky_stars",
"sensor.indi_allsky_temperature",
):
entry = entity_registry.async_get(entity_id)
assert entry is not None
assert entry.disabled_by is None
async def test_sensor_updates(
hass: HomeAssistant,
mock_indi_allsky_client: AsyncMock,
mock_config_entry: MockConfigEntry,
mock_exposure_data: ExposureData,
entity_registry: er.EntityRegistry,
) -> None:
"""Test sensor state values update on exposure_complete event."""
# Enable disabled sensors for testing
entity_registry.async_get_or_create(
domain="sensor",
platform="indi_allsky",
unique_id=f"{mock_config_entry.entry_id}_binmode",
suggested_object_id="indi_allsky_binning_mode",
disabled_by=None,
)
entity_registry.async_get_or_create(
domain="sensor",
platform="indi_allsky",
unique_id=f"{mock_config_entry.entry_id}_filename",
suggested_object_id="indi_allsky_filename",
disabled_by=None,
)
entity_registry.async_get_or_create(
domain="sensor",
platform="indi_allsky",
unique_id=f"{mock_config_entry.entry_id}_gain",
suggested_object_id="indi_allsky_gain",
disabled_by=None,
)
with patch("homeassistant.components.indi_allsky._PLATFORMS", [Platform.SENSOR]):
await setup_integration(hass, mock_config_entry)
for callback in mock_indi_allsky_client.callbacks.get("exposure_complete", []):
callback(mock_exposure_data)
await hass.async_block_till_done()
state = hass.states.get("sensor.indi_allsky_exposure_time")
assert state is not None
assert state.state == "0.185"
state = hass.states.get("sensor.indi_allsky_temperature")
assert state is not None
assert state.state == STATE_UNKNOWN
state = hass.states.get("sensor.indi_allsky_sky_quality")
assert state is not None
assert state.state == "32928.83"
state = hass.states.get("sensor.indi_allsky_stars")
assert state is not None
assert state.state == "0"
state = hass.states.get("sensor.indi_allsky_binning_mode")
assert state is not None
assert state.state == "1"
state = hass.states.get("sensor.indi_allsky_filename")
assert state is not None
assert state.state == "test.jpg"
state = hass.states.get("sensor.indi_allsky_gain")
assert state is not None
assert state.state == "0.0"
for callback in mock_indi_allsky_client.callbacks.get("exposure_complete", []):
callback(replace(mock_exposure_data, temp=12.5))
await hass.async_block_till_done()
state = hass.states.get("sensor.indi_allsky_temperature")
assert state is not None
assert state.state == "12.5"