jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Switcher move _async_call_api to entity.py (#132877)

Browse Source 
* Switcher move _async_call_api to entity.py

* fix based on requested changes

* fix based on requested changes
This commit is contained in:
YogevBokobza 2024-12-14 21:06:36 +02:00 committed by GitHub
parent 9e2a3ea0e5
commit ff1df757b1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 91 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
homeassistant/components/switcher_kis/cover.py
View File

@ -2,10 +2,8 @@
from __future__ import annotations
import logging
from typing import Any, cast
from aioswitcher.api import SwitcherApi, SwitcherBaseResponse
from aioswitcher.device import DeviceCategory, ShutterDirection, SwitcherShutter
from homeassistant.components.cover import (
@ -16,7 +14,6 @@ from homeassistant.components.cover import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -24,8 +21,6 @@ from .const import SIGNAL_DEVICE_ADD
from .coordinator import SwitcherDataUpdateCoordinator
from .entity import SwitcherEntity
_LOGGER = logging.getLogger(__name__)
API_SET_POSITON = "set_position"
API_STOP = "stop_shutter"
@ -92,32 +87,6 @@ class SwitcherBaseCoverEntity(SwitcherEntity, CoverEntity):
data.direction[self._cover_id] == ShutterDirection.SHUTTER_UP
)
async def _async_call_api(self, api: str, *args: Any) -> None:
"""Call Switcher API."""
_LOGGER.debug("Calling api for %s, api: '%s', args: %s", self.name, api, args)
response: SwitcherBaseResponse | None = None
error = None
try:
async with SwitcherApi(
self.coordinator.data.device_type,
self.coordinator.data.ip_address,
self.coordinator.data.device_id,
self.coordinator.data.device_key,
self.coordinator.token,
) as swapi:
response = await getattr(swapi, api)(*args)
except (TimeoutError, OSError, RuntimeError) as err:
error = repr(err)
if error or not response or not response.successful:
self.coordinator.last_update_success = False
self.async_write_ha_state()
raise HomeAssistantError(
f"Call api for {self.name} failed, api: '{api}', "
f"args: {args}, response/error: {response or error}"
)
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close cover."""
await self._async_call_api(API_SET_POSITON, 0, self._cover_id)

34
homeassistant/components/switcher_kis/entity.py
View File

@ -1,11 +1,19 @@
"""Base class for Switcher entities."""
import logging
from typing import Any
from aioswitcher.api import SwitcherApi, SwitcherBaseResponse
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import SwitcherDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
class SwitcherEntity(CoordinatorEntity[SwitcherDataUpdateCoordinator]):
"""Base class for Switcher entities."""
@ -18,3 +26,29 @@ class SwitcherEntity(CoordinatorEntity[SwitcherDataUpdateCoordinator]):
self._attr_device_info = DeviceInfo(
connections={(dr.CONNECTION_NETWORK_MAC, coordinator.mac_address)}
)
async def _async_call_api(self, api: str, *args: Any) -> None:
"""Call Switcher API."""
_LOGGER.debug("Calling api for %s, api: '%s', args: %s", self.name, api, args)
response: SwitcherBaseResponse | None = None
error = None
try:
async with SwitcherApi(
self.coordinator.data.device_type,
self.coordinator.data.ip_address,
self.coordinator.data.device_id,
self.coordinator.data.device_key,
self.coordinator.token,
) as swapi:
response = await getattr(swapi, api)(*args)
except (TimeoutError, OSError, RuntimeError) as err:
error = repr(err)
if error or not response or not response.successful:
self.coordinator.last_update_success = False
self.async_write_ha_state()
raise HomeAssistantError(
f"Call api for {self.name} failed, api: '{api}', "
f"args: {args}, response/error: {response or error}"
)

31
homeassistant/components/switcher_kis/light.py
View File

@ -2,16 +2,13 @@
from __future__ import annotations
import logging
from typing import Any, cast
from aioswitcher.api import SwitcherApi, SwitcherBaseResponse
from aioswitcher.device import DeviceCategory, DeviceState, SwitcherLight
from homeassistant.components.light import ColorMode, LightEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -19,8 +16,6 @@ from .const import SIGNAL_DEVICE_ADD
from .coordinator import SwitcherDataUpdateCoordinator
from .entity import SwitcherEntity
_LOGGER = logging.getLogger(__name__)
API_SET_LIGHT = "set_light"
@ -79,32 +74,6 @@ class SwitcherBaseLightEntity(SwitcherEntity, LightEntity):
data = cast(SwitcherLight, self.coordinator.data)
return bool(data.light[self._light_id] == DeviceState.ON)
async def _async_call_api(self, api: str, *args: Any) -> None:
"""Call Switcher API."""
_LOGGER.debug("Calling api for %s, api: '%s', args: %s", self.name, api, args)
response: SwitcherBaseResponse | None = None
error = None
try:
async with SwitcherApi(
self.coordinator.data.device_type,
self.coordinator.data.ip_address,
self.coordinator.data.device_id,
self.coordinator.data.device_key,
self.coordinator.token,
) as swapi:
response = await getattr(swapi, api)(*args)
except (TimeoutError, OSError, RuntimeError) as err:
error = repr(err)
if error or not response or not response.successful:
self.coordinator.last_update_success = False
self.async_write_ha_state()
raise HomeAssistantError(
f"Call api for {self.name} failed, api: '{api}', "
f"args: {args}, response/error: {response or error}"
)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the light on."""
await self._async_call_api(API_SET_LIGHT, DeviceState.ON, self._light_id)

31
homeassistant/components/switcher_kis/switch.py
View File

@ -6,7 +6,7 @@ from datetime import timedelta
import logging
from typing import Any
from aioswitcher.api import Command, SwitcherApi, SwitcherBaseResponse
from aioswitcher.api import Command
from aioswitcher.device import DeviceCategory, DeviceState
import voluptuous as vol
@ -96,35 +96,6 @@ class SwitcherBaseSwitchEntity(SwitcherEntity, SwitchEntity):
self.control_result = None
self.async_write_ha_state()
async def _async_call_api(self, api: str, *args: Any) -> None:
"""Call Switcher API."""
_LOGGER.debug(
"Calling api for %s, api: '%s', args: %s", self.coordinator.name, api, args
)
response: SwitcherBaseResponse | None = None
error = None
try:
async with SwitcherApi(
self.coordinator.data.device_type,
self.coordinator.data.ip_address,
self.coordinator.data.device_id,
self.coordinator.data.device_key,
) as swapi:
response = await getattr(swapi, api)(*args)
except (TimeoutError, OSError, RuntimeError) as err:
error = repr(err)
if error or not response or not response.successful:
_LOGGER.error(
"Call api for %s failed, api: '%s', args: %s, response/error: %s",
self.coordinator.name,
api,
args,
response or error,
)
self.coordinator.last_update_success = False
@property
def is_on(self) -> bool:
"""Return True if entity is on."""

12
tests/components/switcher_kis/conftest.py
View File

@ -60,19 +60,11 @@ def mock_api():
patchers = [
patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.connect",
"homeassistant.components.switcher_kis.entity.SwitcherApi.connect",
new=api_mock,
),
patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.disconnect",
new=api_mock,
),
patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.connect",
new=api_mock,
),
patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.disconnect",
"homeassistant.components.switcher_kis.entity.SwitcherApi.disconnect",
new=api_mock,
),
]

8
tests/components/switcher_kis/test_button.py
View File

@ -42,7 +42,7 @@ async def test_assume_button(
assert hass.states.get(SWING_OFF_EID) is None
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
BUTTON_DOMAIN,
@ -79,7 +79,7 @@ async def test_swing_button(
assert hass.states.get(SWING_OFF_EID) is not None
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
BUTTON_DOMAIN,
@ -103,7 +103,7 @@ async def test_control_device_fail(
# Test exception during set hvac mode
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
side_effect=RuntimeError("fake error"),
) as mock_control_device:
with pytest.raises(HomeAssistantError):
@ -130,7 +130,7 @@ async def test_control_device_fail(
# Test error response during turn on
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
return_value=SwitcherBaseResponse(None),
) as mock_control_device:
with pytest.raises(HomeAssistantError):

18
tests/components/switcher_kis/test_climate.py
View File

@ -49,7 +49,7 @@ async def test_climate_hvac_mode(
# Test set hvac mode heat
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
CLIMATE_DOMAIN,
@ -71,7 +71,7 @@ async def test_climate_hvac_mode(
# Test set hvac mode off
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
CLIMATE_DOMAIN,
@ -108,7 +108,7 @@ async def test_climate_temperature(
# Test set target temperature
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
CLIMATE_DOMAIN,
@ -128,7 +128,7 @@ async def test_climate_temperature(
# Test set target temperature - incorrect params
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
with pytest.raises(ServiceValidationError):
await hass.services.async_call(
@ -160,7 +160,7 @@ async def test_climate_fan_level(
# Test set fan level to high
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
CLIMATE_DOMAIN,
@ -195,7 +195,7 @@ async def test_climate_swing(
# Test set swing mode on
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
CLIMATE_DOMAIN,
@ -218,7 +218,7 @@ async def test_climate_swing(
# Test set swing mode off
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
) as mock_control_device:
await hass.services.async_call(
CLIMATE_DOMAIN,
@ -249,7 +249,7 @@ async def test_control_device_fail(hass: HomeAssistant, mock_bridge, mock_api) -
# Test exception during set hvac mode
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
side_effect=RuntimeError("fake error"),
) as mock_control_device:
with pytest.raises(HomeAssistantError):
@ -276,7 +276,7 @@ async def test_control_device_fail(hass: HomeAssistant, mock_bridge, mock_api) -
# Test error response during turn on
with patch(
"homeassistant.components.switcher_kis.climate.SwitcherApi.control_breeze_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_breeze_device",
return_value=SwitcherBaseResponse(None),
) as mock_control_device:
with pytest.raises(HomeAssistantError):

12
tests/components/switcher_kis/test_cover.py
View File

@ -115,7 +115,7 @@ async def test_cover(
# Test set position
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.set_position"
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_position"
) as mock_control_device:
await hass.services.async_call(
COVER_DOMAIN,
@ -136,7 +136,7 @@ async def test_cover(
# Test open
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.set_position"
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_position"
) as mock_control_device:
await hass.services.async_call(
COVER_DOMAIN,
@ -156,7 +156,7 @@ async def test_cover(
# Test close
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.set_position"
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_position"
) as mock_control_device:
await hass.services.async_call(
COVER_DOMAIN,
@ -176,7 +176,7 @@ async def test_cover(
# Test stop
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.stop_shutter"
"homeassistant.components.switcher_kis.entity.SwitcherApi.stop_shutter"
) as mock_control_device:
await hass.services.async_call(
COVER_DOMAIN,
@ -232,7 +232,7 @@ async def test_cover_control_fail(
# Test exception during set position
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.set_position",
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_position",
side_effect=RuntimeError("fake error"),
) as mock_control_device:
with pytest.raises(HomeAssistantError):
@ -257,7 +257,7 @@ async def test_cover_control_fail(
# Test error response during set position
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.set_position",
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_position",
return_value=SwitcherBaseResponse(None),
) as mock_control_device:
with pytest.raises(HomeAssistantError):

8
tests/components/switcher_kis/test_light.py
View File

@ -86,7 +86,7 @@ async def test_light(
# Test turning on light
with patch(
"homeassistant.components.switcher_kis.light.SwitcherApi.set_light",
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_light",
) as mock_set_light:
await hass.services.async_call(
LIGHT_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id}, blocking=True
@ -99,7 +99,7 @@ async def test_light(
# Test turning off light
with patch(
"homeassistant.components.switcher_kis.light.SwitcherApi.set_light"
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_light"
) as mock_set_light:
await hass.services.async_call(
LIGHT_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: entity_id}, blocking=True
@ -153,7 +153,7 @@ async def test_light_control_fail(
# Test exception during turn on
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.set_light",
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_light",
side_effect=RuntimeError("fake error"),
) as mock_control_device:
with pytest.raises(HomeAssistantError):
@ -178,7 +178,7 @@ async def test_light_control_fail(
# Test error response during turn on
with patch(
"homeassistant.components.switcher_kis.cover.SwitcherApi.set_light",
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_light",
return_value=SwitcherBaseResponse(None),
) as mock_control_device:
with pytest.raises(HomeAssistantError):

26
tests/components/switcher_kis/test_services.py
View File

@ -16,6 +16,7 @@ from homeassistant.components.switcher_kis.const import (
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.config_validation import time_period_str
from homeassistant.util import slugify
@ -48,7 +49,7 @@ async def test_turn_on_with_timer_service(
assert state.state == STATE_OFF
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.control_device"
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_device"
) as mock_control_device:
await hass.services.async_call(
DOMAIN,
@ -78,7 +79,7 @@ async def test_set_auto_off_service(hass: HomeAssistant, mock_bridge, mock_api)
entity_id = f"{SWITCH_DOMAIN}.{slugify(device.name)}"
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.set_auto_shutdown"
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_auto_shutdown"
) as mock_set_auto_shutdown:
await hass.services.async_call(
DOMAIN,
@ -95,7 +96,7 @@ async def test_set_auto_off_service(hass: HomeAssistant, mock_bridge, mock_api)
@pytest.mark.parametrize("mock_bridge", [[DUMMY_WATER_HEATER_DEVICE]], indirect=True)
async def test_set_auto_off_service_fail(
hass: HomeAssistant, mock_bridge, mock_api, caplog: pytest.LogCaptureFixture
hass: HomeAssistant, mock_bridge, mock_api
) -> None:
"""Test set auto off service failed."""
await init_integration(hass)
@ -105,24 +106,21 @@ async def test_set_auto_off_service_fail(
entity_id = f"{SWITCH_DOMAIN}.{slugify(device.name)}"
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.set_auto_shutdown",
"homeassistant.components.switcher_kis.entity.SwitcherApi.set_auto_shutdown",
return_value=None,
) as mock_set_auto_shutdown:
await hass.services.async_call(
DOMAIN,
SERVICE_SET_AUTO_OFF_NAME,
{ATTR_ENTITY_ID: entity_id, CONF_AUTO_OFF: DUMMY_AUTO_OFF_SET},
blocking=True,
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_SET_AUTO_OFF_NAME,
{ATTR_ENTITY_ID: entity_id, CONF_AUTO_OFF: DUMMY_AUTO_OFF_SET},
blocking=True,
)
assert mock_api.call_count == 2
mock_set_auto_shutdown.assert_called_once_with(
time_period_str(DUMMY_AUTO_OFF_SET)
)
assert (
f"Call api for {device.name} failed, api: 'set_auto_shutdown'"
in caplog.text
)
state = hass.states.get(entity_id)
assert state.state == STATE_UNAVAILABLE

36
tests/components/switcher_kis/test_switch.py
View File

@ -16,6 +16,7 @@ from homeassistant.const import (
STATE_UNAVAILABLE,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.util import slugify
from . import init_integration
@ -47,7 +48,7 @@ async def test_switch(
# Test turning on
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.control_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_device",
) as mock_control_device:
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id}, blocking=True
@ -60,7 +61,7 @@ async def test_switch(
# Test turning off
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.control_device"
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_device"
) as mock_control_device:
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: entity_id}, blocking=True
@ -78,7 +79,6 @@ async def test_switch_control_fail(
mock_bridge,
mock_api,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test switch control fail."""
await init_integration(hass)
@ -97,18 +97,19 @@ async def test_switch_control_fail(
# Test exception during turn on
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.control_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_device",
side_effect=RuntimeError("fake error"),
) as mock_control_device:
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_api.call_count == 2
mock_control_device.assert_called_once_with(Command.ON)
assert (
f"Call api for {device.name} failed, api: 'control_device'" in caplog.text
)
state = hass.states.get(entity_id)
assert state.state == STATE_UNAVAILABLE
@ -121,17 +122,18 @@ async def test_switch_control_fail(
# Test error response during turn on
with patch(
"homeassistant.components.switcher_kis.switch.SwitcherApi.control_device",
"homeassistant.components.switcher_kis.entity.SwitcherApi.control_device",
return_value=SwitcherBaseResponse(None),
) as mock_control_device:
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id}, blocking=True
)
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
assert mock_api.call_count == 4
mock_control_device.assert_called_once_with(Command.ON)
assert (
f"Call api for {device.name} failed, api: 'control_device'" in caplog.text
)
state = hass.states.get(entity_id)
assert state.state == STATE_UNAVAILABLE