home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Move Rainbird to async client library (#84417)

Browse Source 
* Bump pyrainbird to 0.7.0 and move to async library

* Share updates across sensors

* Fix test version and delete dead code

* Add test coverage for yaml configuration

* Address PR feedback
This commit is contained in:
Allen Porter
2022-12-22 13:00:17 -08:00
committed by GitHub
parent 5874b4cdcf
commit 490d2cfb71
15 changed files with 791 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -1033,7 +1033,6 @@ omit =
homeassistant/components/radiotherm/entity.py
homeassistant/components/radiotherm/switch.py
homeassistant/components/radiotherm/util.py
homeassistant/components/rainbird/*
homeassistant/components/raincloud/*
homeassistant/components/rainmachine/__init__.py
homeassistant/components/rainmachine/binary_sensor.py

1
CODEOWNERS
View File

@ -926,6 +926,7 @@ build.json @home-assistant/supervisor
/homeassistant/components/radiotherm/ @bdraco @vinnyfuria
/tests/components/radiotherm/ @bdraco @vinnyfuria
/homeassistant/components/rainbird/ @konikvranik @allenporter
/tests/components/rainbird/ @konikvranik @allenporter
/homeassistant/components/raincloud/ @vanstinator
/homeassistant/components/rainforest_eagle/ @gtdiehl @jcalbert @hastarin
/tests/components/rainforest_eagle/ @gtdiehl @jcalbert @hastarin

79
homeassistant/components/rainbird/__init__.py
View File

@ -1,13 +1,16 @@
"""Support for Rain Bird Irrigation system LNK WiFi Module."""
from __future__ import annotations
import asyncio
import logging
from pyrainbird import RainbirdController
from pyrainbird.async_client import (
AsyncRainbirdClient,
AsyncRainbirdController,
RainbirdApiException,
)
import voluptuous as vol
from homeassistant.components.binary_sensor import BinarySensorEntityDescription
from homeassistant.components.sensor import SensorEntityDescription
from homeassistant.const import (
CONF_FRIENDLY_NAME,
CONF_HOST,
@ -17,49 +20,25 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import discovery
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
CONF_ZONES = "zones"
from .const import (
CONF_ZONES,
RAINBIRD_CONTROLLER,
SENSOR_TYPE_RAINDELAY,
SENSOR_TYPE_RAINSENSOR,
)
from .coordinator import RainbirdUpdateCoordinator
PLATFORMS = [Platform.SWITCH, Platform.SENSOR, Platform.BINARY_SENSOR]
_LOGGER = logging.getLogger(__name__)
RAINBIRD_CONTROLLER = "controller"
DATA_RAINBIRD = "rainbird"
DOMAIN = "rainbird"
SENSOR_TYPE_RAINDELAY = "raindelay"
SENSOR_TYPE_RAINSENSOR = "rainsensor"
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key=SENSOR_TYPE_RAINSENSOR,
name="Rainsensor",
icon="mdi:water",
),
SensorEntityDescription(
key=SENSOR_TYPE_RAINDELAY,
name="Raindelay",
icon="mdi:water-off",
),
)
BINARY_SENSOR_TYPES: tuple[BinarySensorEntityDescription, ...] = (
BinarySensorEntityDescription(
key=SENSOR_TYPE_RAINSENSOR,
name="Rainsensor",
icon="mdi:water",
),
BinarySensorEntityDescription(
key=SENSOR_TYPE_RAINDELAY,
name="Raindelay",
icon="mdi:water-off",
),
)
TRIGGER_TIME_SCHEMA = vol.All(
cv.time_period, cv.positive_timedelta, lambda td: (td.total_seconds() // 60)
)
@ -84,36 +63,46 @@ CONFIG_SCHEMA = vol.Schema(
)
def setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Rain Bird component."""
hass.data[DATA_RAINBIRD] = []
success = False
tasks = []
for controller_config in config[DOMAIN]:
success = success or _setup_controller(hass, controller_config, config)
return success
tasks.append(_setup_controller(hass, controller_config, config))
return all(await asyncio.gather(*tasks))
def _setup_controller(hass, controller_config, config):
async def _setup_controller(hass, controller_config, config):
"""Set up a controller."""
server = controller_config[CONF_HOST]
password = controller_config[CONF_PASSWORD]
controller = RainbirdController(server, password)
client = AsyncRainbirdClient(async_get_clientsession(hass), server, password)
controller = AsyncRainbirdController(client)
position = len(hass.data[DATA_RAINBIRD])
try:
controller.get_serial_number()
except Exception as exc: # pylint: disable=broad-except
await controller.get_serial_number()
except RainbirdApiException as exc:
_LOGGER.error("Unable to setup controller: %s", exc)
return False
hass.data[DATA_RAINBIRD].append(controller)
rain_coordinator = RainbirdUpdateCoordinator(hass, controller.get_rain_sensor_state)
delay_coordinator = RainbirdUpdateCoordinator(hass, controller.get_rain_delay)
_LOGGER.debug("Rain Bird Controller %d set to: %s", position, server)
for platform in PLATFORMS:
discovery.load_platform(
hass,
platform,
DOMAIN,
{RAINBIRD_CONTROLLER: position, **controller_config},
{
RAINBIRD_CONTROLLER: controller,
SENSOR_TYPE_RAINSENSOR: rain_coordinator,
SENSOR_TYPE_RAINDELAY: delay_coordinator,
**controller_config,
},
config,
)
return True

56
homeassistant/components/rainbird/binary_sensor.py
View File

@ -3,8 +3,6 @@ from __future__ import annotations
import logging
from pyrainbird import RainbirdController
from homeassistant.components.binary_sensor import (
BinarySensorEntity,
BinarySensorEntityDescription,
@ -12,56 +10,62 @@ from homeassistant.components.binary_sensor import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import (
BINARY_SENSOR_TYPES,
DATA_RAINBIRD,
RAINBIRD_CONTROLLER,
SENSOR_TYPE_RAINDELAY,
SENSOR_TYPE_RAINSENSOR,
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import SENSOR_TYPE_RAINDELAY, SENSOR_TYPE_RAINSENSOR
_LOGGER = logging.getLogger(__name__)
def setup_platform(
BINARY_SENSOR_TYPES: tuple[BinarySensorEntityDescription, ...] = (
BinarySensorEntityDescription(
key=SENSOR_TYPE_RAINSENSOR,
name="Rainsensor",
icon="mdi:water",
),
BinarySensorEntityDescription(
key=SENSOR_TYPE_RAINDELAY,
name="Raindelay",
icon="mdi:water-off",
),
)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up a Rain Bird sensor."""
if discovery_info is None:
return
controller = hass.data[DATA_RAINBIRD][discovery_info[RAINBIRD_CONTROLLER]]
add_entities(
async_add_entities(
[
RainBirdSensor(controller, description)
RainBirdSensor(discovery_info[description.key], description)
for description in BINARY_SENSOR_TYPES
],
True,
)
class RainBirdSensor(BinarySensorEntity):
class RainBirdSensor(CoordinatorEntity, BinarySensorEntity):
"""A sensor implementation for Rain Bird device."""
def __init__(
self,
controller: RainbirdController,
coordinator: DataUpdateCoordinator,
description: BinarySensorEntityDescription,
) -> None:
"""Initialize the Rain Bird sensor."""
super().__init__(coordinator)
self.entity_description = description
self._controller = controller
def update(self) -> None:
"""Get the latest data and updates the states."""
_LOGGER.debug("Updating sensor: %s", self.name)
state = None
if self.entity_description.key == SENSOR_TYPE_RAINSENSOR:
state = self._controller.get_rain_sensor_state()
elif self.entity_description.key == SENSOR_TYPE_RAINDELAY:
state = self._controller.get_rain_delay()
self._attr_is_on = None if state is None else bool(state)
@property
def is_on(self) -> bool | None:
"""Return True if entity is on."""
return None if self.coordinator.data is None else bool(self.coordinator.data)

10
homeassistant/components/rainbird/const.py Normal file
View File

@ -0,0 +1,10 @@
"""Constants for rainbird."""
DOMAIN = "rainbird"
SENSOR_TYPE_RAINDELAY = "raindelay"
SENSOR_TYPE_RAINSENSOR = "rainsensor"
RAINBIRD_CONTROLLER = "controller"
CONF_ZONES = "zones"

47
homeassistant/components/rainbird/coordinator.py Normal file
View File

@ -0,0 +1,47 @@
"""Update coordinators for rainbird."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
import datetime
import logging
from typing import TypeVar
import async_timeout
from pyrainbird.async_client import RainbirdApiException
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
TIMEOUT_SECONDS = 20
UPDATE_INTERVAL = datetime.timedelta(minutes=1)
_LOGGER = logging.getLogger(__name__)
_T = TypeVar("_T")
class RainbirdUpdateCoordinator(DataUpdateCoordinator[_T]):
"""Coordinator for rainbird API calls."""
def __init__(
self,
hass: HomeAssistant,
update_method: Callable[[], Awaitable[_T]],
) -> None:
"""Initialize ZoneStateUpdateCoordinator."""
super().__init__(
hass,
_LOGGER,
name="Rainbird Zones",
update_method=update_method,
update_interval=UPDATE_INTERVAL,
)
async def _async_update_data(self) -> _T:
"""Fetch data from API endpoint."""
try:
async with async_timeout.timeout(TIMEOUT_SECONDS):
return await self.update_method() # type: ignore[misc]
except RainbirdApiException as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err

59
homeassistant/components/rainbird/sensor.py
View File

@ -3,28 +3,38 @@ from __future__ import annotations
import logging
from pyrainbird import RainbirdController
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import (
DATA_RAINBIRD,
RAINBIRD_CONTROLLER,
SENSOR_TYPE_RAINDELAY,
SENSOR_TYPE_RAINSENSOR,
SENSOR_TYPES,
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType, StateType
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import SENSOR_TYPE_RAINDELAY, SENSOR_TYPE_RAINSENSOR
_LOGGER = logging.getLogger(__name__)
def setup_platform(
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key=SENSOR_TYPE_RAINSENSOR,
name="Rainsensor",
icon="mdi:water",
),
SensorEntityDescription(
key=SENSOR_TYPE_RAINDELAY,
name="Raindelay",
icon="mdi:water-off",
),
)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up a Rain Bird sensor."""
@ -32,29 +42,28 @@ def setup_platform(
if discovery_info is None:
return
controller = hass.data[DATA_RAINBIRD][discovery_info[RAINBIRD_CONTROLLER]]
add_entities(
[RainBirdSensor(controller, description) for description in SENSOR_TYPES],
async_add_entities(
[
RainBirdSensor(discovery_info[description.key], description)
for description in SENSOR_TYPES
],
True,
)
class RainBirdSensor(SensorEntity):
class RainBirdSensor(CoordinatorEntity, SensorEntity):
"""A sensor implementation for Rain Bird device."""
def __init__(
self,
controller: RainbirdController,
coordinator: DataUpdateCoordinator,
description: SensorEntityDescription,
) -> None:
"""Initialize the Rain Bird sensor."""
super().__init__(coordinator)
self.entity_description = description
self._controller = controller
def update(self) -> None:
"""Get the latest data and updates the states."""
_LOGGER.debug("Updating sensor: %s", self.name)
if self.entity_description.key == SENSOR_TYPE_RAINSENSOR:
self._attr_native_value = self._controller.get_rain_sensor_state()
elif self.entity_description.key == SENSOR_TYPE_RAINDELAY:
self._attr_native_value = self._controller.get_rain_delay()
@property
def native_value(self) -> StateType:
"""Return the value reported by the sensor."""
return self.coordinator.data

83
homeassistant/components/rainbird/switch.py
View File

@ -1,17 +1,26 @@
"""Support for Rain Bird Irrigation system LNK WiFi Module."""
from __future__ import annotations
from pyrainbird import AvailableStations, RainbirdController
import logging
from pyrainbird import AvailableStations
from pyrainbird.async_client import AsyncRainbirdController, RainbirdApiException
from pyrainbird.data import States
import voluptuous as vol
from homeassistant.components.switch import SwitchEntity
from homeassistant.const import ATTR_ENTITY_ID, CONF_FRIENDLY_NAME, CONF_TRIGGER_TIME
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import ConfigEntryNotReady, PlatformNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import CONF_ZONES, DATA_RAINBIRD, DOMAIN, RAINBIRD_CONTROLLER
from .const import CONF_ZONES, DOMAIN, RAINBIRD_CONTROLLER
from .coordinator import RainbirdUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
ATTR_DURATION = "duration"
@ -32,10 +41,10 @@ SERVICE_SCHEMA_RAIN_DELAY = vol.Schema(
)
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up Rain Bird switches over a Rain Bird controller."""
@ -43,12 +52,16 @@ def setup_platform(
if discovery_info is None:
return
controller: RainbirdController = hass.data[DATA_RAINBIRD][
discovery_info[RAINBIRD_CONTROLLER]
]
available_stations: AvailableStations = controller.get_available_stations()
controller: AsyncRainbirdController = discovery_info[RAINBIRD_CONTROLLER]
try:
available_stations: AvailableStations = (
await controller.get_available_stations()
)
except RainbirdApiException as err:
raise PlatformNotReady(f"Failed to get stations: {str(err)}") from err
if not (available_stations and available_stations.stations):
return
coordinator = RainbirdUpdateCoordinator(hass, controller.get_zone_states)
devices = []
for zone in range(1, available_stations.stations.count + 1):
if available_stations.stations.active(zone):
@ -57,6 +70,7 @@ def setup_platform(
name = zone_config.get(CONF_FRIENDLY_NAME)
devices.append(
RainBirdSwitch(
coordinator,
controller,
zone,
time,
@ -64,29 +78,34 @@ def setup_platform(
)
)
add_entities(devices, True)
try:
await coordinator.async_config_entry_first_refresh()
except ConfigEntryNotReady as err:
raise PlatformNotReady(f"Failed to load zone state: {str(err)}") from err
def start_irrigation(service: ServiceCall) -> None:
async_add_entities(devices)
async def start_irrigation(service: ServiceCall) -> None:
entity_id = service.data[ATTR_ENTITY_ID]
duration = service.data[ATTR_DURATION]
for device in devices:
if device.entity_id == entity_id:
device.turn_on(duration=duration)
await device.async_turn_on(duration=duration)
hass.services.register(
hass.services.async_register(
DOMAIN,
SERVICE_START_IRRIGATION,
start_irrigation,
schema=SERVICE_SCHEMA_IRRIGATION,
)
def set_rain_delay(service: ServiceCall) -> None:
async def set_rain_delay(service: ServiceCall) -> None:
duration = service.data[ATTR_DURATION]
controller.set_rain_delay(duration)
await controller.set_rain_delay(duration)
hass.services.register(
hass.services.async_register(
DOMAIN,
SERVICE_SET_RAIN_DELAY,
set_rain_delay,
@ -94,12 +113,20 @@ def setup_platform(
)
class RainBirdSwitch(SwitchEntity):
class RainBirdSwitch(CoordinatorEntity, SwitchEntity):
"""Representation of a Rain Bird switch."""
def __init__(self, controller: RainbirdController, zone, time, name):
def __init__(
self,
coordinator: RainbirdUpdateCoordinator[States],
rainbird: AsyncRainbirdController,
zone: int,
time: int,
name: str,
) -> None:
"""Initialize a Rain Bird Switch Device."""
self._rainbird = controller
super().__init__(coordinator)
self._rainbird = rainbird
self._zone = zone
self._name = name
self._state = None
@ -116,24 +143,20 @@ class RainBirdSwitch(SwitchEntity):
"""Get the name of the switch."""
return self._name
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs):
"""Turn the switch on."""
if self._rainbird.irrigate_zone(
await self._rainbird.irrigate_zone(
int(self._zone),
int(kwargs[ATTR_DURATION] if ATTR_DURATION in kwargs else self._duration),
):
self._state = True
)
await self.coordinator.async_request_refresh()
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs):
"""Turn the switch off."""
if self._rainbird.stop_irrigation():
self._state = False
def update(self):
"""Update switch status."""
self._state = self._rainbird.get_zone_state(self._zone)
await self._rainbird.stop_irrigation()
await self.coordinator.async_request_refresh()
@property
def is_on(self):
"""Return true if switch is on."""
return self._state
return self.coordinator.data.active(self._zone)

3
requirements_test_all.txt
View File

@ -1329,6 +1329,9 @@ pyps4-2ndscreen==1.3.1
# homeassistant.components.qwikswitch
pyqwikswitch==0.93
# homeassistant.components.rainbird
pyrainbird==0.7.1
# homeassistant.components.risco
pyrisco==0.5.7

1
tests/components/rainbird/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the rainbird integration."""

116
tests/components/rainbird/conftest.py Normal file
View File

@ -0,0 +1,116 @@
"""Test fixtures for rainbird."""
from __future__ import annotations
from collections.abc import Awaitable, Callable, Generator
from typing import Any
from unittest.mock import patch
from pyrainbird import encryption
import pytest
from homeassistant.components.rainbird import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.test_util.aiohttp import AiohttpClientMocker, AiohttpClientMockResponse
ComponentSetup = Callable[[], Awaitable[bool]]
HOST = "example.com"
URL = "http://example.com/stick"
PASSWORD = "password"
#
# Response payloads below come from pyrainbird test cases.
#
# Get serial number Command 0x85. Serial is 0x12635436566
SERIAL_RESPONSE = "850000012635436566"
# Get available stations command 0x83
AVAILABLE_STATIONS_RESPONSE = "83017F000000" # Mask for 7 zones
EMPTY_STATIONS_RESPONSE = "830000000000"
# Get zone state command 0xBF.
ZONE_3_ON_RESPONSE = "BF0004000000" # Zone 3 is on
ZONE_5_ON_RESPONSE = "BF0010000000" # Zone 5 is on
ZONE_OFF_RESPONSE = "BF0000000000" # All zones off
ZONE_STATE_OFF_RESPONSE = "BF0000000000"
# Get rain sensor state command 0XBE
RAIN_SENSOR_OFF = "BE00"
RAIN_SENSOR_ON = "BE01"
# Get rain delay command 0xB6
RAIN_DELAY = "B60010" # 0x10 is 16
RAIN_DELAY_OFF = "B60000"
# ACK command 0x10, Echo 0x06
ACK_ECHO = "0106"
CONFIG = {
DOMAIN: {
"host": HOST,
"password": PASSWORD,
"trigger_time": 360,
}
}
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return []
@pytest.fixture
def yaml_config() -> dict[str, Any]:
"""Fixture for configuration.yaml."""
return CONFIG
@pytest.fixture
async def setup_integration(
hass: HomeAssistant,
platforms: list[str],
yaml_config: dict[str, Any],
) -> Generator[ComponentSetup, None, None]:
"""Fixture for setting up the component."""
with patch(f"homeassistant.components.{DOMAIN}.PLATFORMS", platforms):
async def func() -> bool:
result = await async_setup_component(hass, DOMAIN, yaml_config)
await hass.async_block_till_done()
return result
yield func
def rainbird_response(data: str) -> bytes:
"""Create a fake API response."""
return encryption.encrypt(
'{"jsonrpc": "2.0", "result": {"data":"%s"}, "id": 1} ' % data,
PASSWORD,
)
def mock_response(data: str) -> AiohttpClientMockResponse:
"""Create a fake AiohttpClientMockResponse."""
return AiohttpClientMockResponse("POST", URL, response=rainbird_response(data))
@pytest.fixture(name="responses")
def mock_responses() -> list[AiohttpClientMockResponse]:
"""Fixture to set up a list of fake API responsees for tests to extend."""
return [mock_response(SERIAL_RESPONSE)]
@pytest.fixture(autouse=True)
def handle_responses(
aioclient_mock: AiohttpClientMocker,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Fixture for command mocking for fake responses to the API url."""
async def handle(method, url, data) -> AiohttpClientMockResponse:
return responses.pop(0)
aioclient_mock.post(URL, side_effect=handle)

78
tests/components/rainbird/test_binary_sensor.py Normal file
View File

@ -0,0 +1,78 @@
"""Tests for rainbird sensor platform."""
import pytest
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .conftest import (
RAIN_DELAY,
RAIN_DELAY_OFF,
RAIN_SENSOR_OFF,
RAIN_SENSOR_ON,
ComponentSetup,
mock_response,
)
from tests.test_util.aiohttp import AiohttpClientMockResponse
@pytest.fixture
def platforms() -> list[Platform]:
"""Fixture to specify platforms to test."""
return [Platform.BINARY_SENSOR]
@pytest.mark.parametrize(
"sensor_payload,expected_state",
[(RAIN_SENSOR_OFF, "off"), (RAIN_SENSOR_ON, "on")],
)
async def test_rainsensor(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
sensor_payload: str,
expected_state: bool,
) -> None:
"""Test rainsensor binary sensor."""
responses.extend(
[
mock_response(sensor_payload),
mock_response(RAIN_DELAY),
]
)
assert await setup_integration()
rainsensor = hass.states.get("binary_sensor.rainsensor")
assert rainsensor is not None
assert rainsensor.state == expected_state
@pytest.mark.parametrize(
"sensor_payload,expected_state",
[(RAIN_DELAY_OFF, "off"), (RAIN_DELAY, "on")],
)
async def test_raindelay(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
sensor_payload: str,
expected_state: bool,
) -> None:
"""Test raindelay binary sensor."""
responses.extend(
[
mock_response(RAIN_SENSOR_OFF),
mock_response(sensor_payload),
]
)
assert await setup_integration()
raindelay = hass.states.get("binary_sensor.raindelay")
assert raindelay is not None
assert raindelay.state == expected_state

34
tests/components/rainbird/test_init.py Normal file
View File

@ -0,0 +1,34 @@
"""Tests for rainbird initialization."""
from http import HTTPStatus
from homeassistant.core import HomeAssistant
from .conftest import URL, ComponentSetup
from tests.test_util.aiohttp import AiohttpClientMocker, AiohttpClientMockResponse
async def test_setup_success(
hass: HomeAssistant,
setup_integration: ComponentSetup,
) -> None:
"""Test successful setup and unload."""
assert await setup_integration()
async def test_setup_communication_failure(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test unable to talk to server on startup, which permanently fails setup."""
responses.clear()
responses.append(
AiohttpClientMockResponse("POST", URL, status=HTTPStatus.SERVICE_UNAVAILABLE)
)
assert not await setup_integration()

49
tests/components/rainbird/test_sensor.py Normal file
View File

@ -0,0 +1,49 @@
"""Tests for rainbird sensor platform."""
import pytest
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .conftest import (
RAIN_DELAY,
RAIN_SENSOR_OFF,
RAIN_SENSOR_ON,
ComponentSetup,
mock_response,
)
from tests.test_util.aiohttp import AiohttpClientMockResponse
@pytest.fixture
def platforms() -> list[str]:
"""Fixture to specify platforms to test."""
return [Platform.SENSOR]
@pytest.mark.parametrize(
"sensor_payload,expected_state",
[(RAIN_SENSOR_OFF, "False"), (RAIN_SENSOR_ON, "True")],
)
async def test_sensors(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
sensor_payload: str,
expected_state: bool,
) -> None:
"""Test sensor platform."""
responses.extend([mock_response(sensor_payload), mock_response(RAIN_DELAY)])
assert await setup_integration()
rainsensor = hass.states.get("sensor.rainsensor")
assert rainsensor is not None
assert rainsensor.state == expected_state
raindelay = hass.states.get("sensor.raindelay")
assert raindelay is not None
assert raindelay.state == "16"

301
tests/components/rainbird/test_switch.py Normal file
View File

@ -0,0 +1,301 @@
"""Tests for rainbird sensor platform."""
from http import HTTPStatus
import logging
import pytest
from homeassistant.components.rainbird import DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from .conftest import (
ACK_ECHO,
AVAILABLE_STATIONS_RESPONSE,
EMPTY_STATIONS_RESPONSE,
HOST,
PASSWORD,
URL,
ZONE_3_ON_RESPONSE,
ZONE_5_ON_RESPONSE,
ZONE_OFF_RESPONSE,
ComponentSetup,
mock_response,
)
from tests.components.switch import common as switch_common
from tests.test_util.aiohttp import AiohttpClientMocker, AiohttpClientMockResponse
@pytest.fixture
def platforms() -> list[str]:
"""Fixture to specify platforms to test."""
return [Platform.SWITCH]
async def test_no_zones(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Test case where listing stations returns no stations."""
responses.append(mock_response(EMPTY_STATIONS_RESPONSE))
assert await setup_integration()
zone = hass.states.get("switch.sprinkler_1")
assert zone is None
async def test_zones(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Test switch platform with fake data that creates 7 zones with one enabled."""
responses.extend(
[mock_response(AVAILABLE_STATIONS_RESPONSE), mock_response(ZONE_5_ON_RESPONSE)]
)
assert await setup_integration()
zone = hass.states.get("switch.sprinkler_1")
assert zone is not None
assert zone.state == "off"
zone = hass.states.get("switch.sprinkler_2")
assert zone is not None
assert zone.state == "off"
zone = hass.states.get("switch.sprinkler_3")
assert zone is not None
assert zone.state == "off"
zone = hass.states.get("switch.sprinkler_4")
assert zone is not None
assert zone.state == "off"
zone = hass.states.get("switch.sprinkler_5")
assert zone is not None
assert zone.state == "on"
zone = hass.states.get("switch.sprinkler_6")
assert zone is not None
assert zone.state == "off"
zone = hass.states.get("switch.sprinkler_7")
assert zone is not None
assert zone.state == "off"
assert not hass.states.get("switch.sprinkler_8")
async def test_switch_on(
hass: HomeAssistant,
setup_integration: ComponentSetup,
aioclient_mock: AiohttpClientMocker,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Test turning on irrigation switch."""
responses.extend(
[mock_response(AVAILABLE_STATIONS_RESPONSE), mock_response(ZONE_OFF_RESPONSE)]
)
assert await setup_integration()
# Initially all zones are off. Pick zone3 as an arbitrary to assert
# state, then update below as a switch.
zone = hass.states.get("switch.sprinkler_3")
assert zone is not None
assert zone.state == "off"
aioclient_mock.mock_calls.clear()
responses.extend(
[
mock_response(ACK_ECHO), # Switch on response
mock_response(ZONE_3_ON_RESPONSE), # Updated zone state
]
)
await switch_common.async_turn_on(hass, "switch.sprinkler_3")
await hass.async_block_till_done()
assert len(aioclient_mock.mock_calls) == 2
aioclient_mock.mock_calls.clear()
# Verify switch state is updated
zone = hass.states.get("switch.sprinkler_3")
assert zone is not None
assert zone.state == "on"
async def test_switch_off(
hass: HomeAssistant,
setup_integration: ComponentSetup,
aioclient_mock: AiohttpClientMocker,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Test turning off irrigation switch."""
responses.extend(
[mock_response(AVAILABLE_STATIONS_RESPONSE), mock_response(ZONE_3_ON_RESPONSE)]
)
assert await setup_integration()
# Initially the test zone is on
zone = hass.states.get("switch.sprinkler_3")
assert zone is not None
assert zone.state == "on"
aioclient_mock.mock_calls.clear()
responses.extend(
[
mock_response(ACK_ECHO), # Switch off response
mock_response(ZONE_OFF_RESPONSE), # Updated zone state
]
)
await switch_common.async_turn_off(hass, "switch.sprinkler_3")
await hass.async_block_till_done()
# One call to change the service and one to refresh state
assert len(aioclient_mock.mock_calls) == 2
# Verify switch state is updated
zone = hass.states.get("switch.sprinkler_3")
assert zone is not None
assert zone.state == "off"
async def test_irrigation_service(
hass: HomeAssistant,
setup_integration: ComponentSetup,
aioclient_mock: AiohttpClientMocker,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Test calling the irrigation service."""
responses.extend(
[mock_response(AVAILABLE_STATIONS_RESPONSE), mock_response(ZONE_3_ON_RESPONSE)]
)
assert await setup_integration()
aioclient_mock.mock_calls.clear()
responses.extend([mock_response(ACK_ECHO), mock_response(ZONE_OFF_RESPONSE)])
await hass.services.async_call(
DOMAIN,
"start_irrigation",
{ATTR_ENTITY_ID: "switch.sprinkler_5", "duration": 30},
blocking=True,
)
# One call to change the service and one to refresh state
assert len(aioclient_mock.mock_calls) == 2
async def test_rain_delay_service(
hass: HomeAssistant,
setup_integration: ComponentSetup,
aioclient_mock: AiohttpClientMocker,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Test calling the rain delay service."""
responses.extend(
[mock_response(AVAILABLE_STATIONS_RESPONSE), mock_response(ZONE_3_ON_RESPONSE)]
)
assert await setup_integration()
aioclient_mock.mock_calls.clear()
responses.extend(
[
mock_response(ACK_ECHO),
]
)
await hass.services.async_call(
DOMAIN, "set_rain_delay", {"duration": 30}, blocking=True
)
assert len(aioclient_mock.mock_calls) == 1
async def test_platform_unavailable(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test failure while listing the stations when setting up the platform."""
responses.append(
AiohttpClientMockResponse("POST", URL, status=HTTPStatus.SERVICE_UNAVAILABLE)
)
with caplog.at_level(logging.WARNING):
assert await setup_integration()
assert "Failed to get stations" in caplog.text
async def test_coordinator_unavailable(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test failure to refresh the update coordinator."""
responses.extend(
[
mock_response(AVAILABLE_STATIONS_RESPONSE),
AiohttpClientMockResponse(
"POST", URL, status=HTTPStatus.SERVICE_UNAVAILABLE
),
],
)
with caplog.at_level(logging.WARNING):
assert await setup_integration()
assert "Failed to load zone state" in caplog.text
@pytest.mark.parametrize(
"yaml_config",
[
{
DOMAIN: {
"host": HOST,
"password": PASSWORD,
"trigger_time": 360,
"zones": {
1: {
"friendly_name": "Garden Sprinkler",
},
2: {
"friendly_name": "Back Yard",
},
},
}
},
],
)
async def test_yaml_config(
hass: HomeAssistant,
setup_integration: ComponentSetup,
responses: list[AiohttpClientMockResponse],
) -> None:
"""Test switch platform with fake data that creates 7 zones with one enabled."""
responses.extend(
[mock_response(AVAILABLE_STATIONS_RESPONSE), mock_response(ZONE_5_ON_RESPONSE)]
)
assert await setup_integration()
assert hass.states.get("switch.garden_sprinkler")
assert not hass.states.get("switch.sprinkler_1")
assert hass.states.get("switch.back_yard")
assert not hass.states.get("switch.sprinkler_2")
assert hass.states.get("switch.sprinkler_3")