jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 01:08:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate emulated_roku to use runtime_data to fix flakey tests (#141795)

Browse Source
This commit is contained in:
J. Nick Koston 2025-03-29 23:55:58 -10:00 committed by GitHub
parent beb92a7f9c
commit 65261de7cc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 67 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
homeassistant/components/emulated_roku/__init__.py
View File

@ -46,6 +46,8 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
type EmulatedRokuConfigEntry = ConfigEntry[EmulatedRoku]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the emulated roku component."""
@ -65,22 +67,21 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
async def async_setup_entry(
hass: HomeAssistant, entry: EmulatedRokuConfigEntry
) -> bool:
"""Set up an emulated roku server from a config entry."""
config = config_entry.data
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
name = config[CONF_NAME]
listen_port = config[CONF_LISTEN_PORT]
host_ip = config.get(CONF_HOST_IP) or await async_get_source_ip(hass)
advertise_ip = config.get(CONF_ADVERTISE_IP)
advertise_port = config.get(CONF_ADVERTISE_PORT)
upnp_bind_multicast = config.get(CONF_UPNP_BIND_MULTICAST)
config = entry.data
name: str = config[CONF_NAME]
listen_port: int = config[CONF_LISTEN_PORT]
host_ip: str = config.get(CONF_HOST_IP) or await async_get_source_ip(hass)
advertise_ip: str | None = config.get(CONF_ADVERTISE_IP)
advertise_port: int | None = config.get(CONF_ADVERTISE_PORT)
upnp_bind_multicast: bool | None = config.get(CONF_UPNP_BIND_MULTICAST)
server = EmulatedRoku(
hass,
entry.entry_id,
name,
host_ip,
listen_port,
@ -88,14 +89,12 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
advertise_port,
upnp_bind_multicast,
)
hass.data[DOMAIN][name] = server
entry.runtime_data = server
return await server.setup()
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(
hass: HomeAssistant, entry: EmulatedRokuConfigEntry
) -> bool:
"""Unload a config entry."""
name = entry.data[CONF_NAME]
server = hass.data[DOMAIN].pop(name)
return await server.unload()
return await entry.runtime_data.unload()

53
homeassistant/components/emulated_roku/binding.py
View File

@ -5,7 +5,13 @@ import logging
from emulated_roku import EmulatedRokuCommandHandler, EmulatedRokuServer
from homeassistant.const import EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import CoreState, EventOrigin
from homeassistant.core import (
CALLBACK_TYPE,
CoreState,
Event,
EventOrigin,
HomeAssistant,
)
LOGGER = logging.getLogger(__package__)
@ -27,16 +33,18 @@ class EmulatedRoku:
def __init__(
self,
hass,
name,
host_ip,
listen_port,
advertise_ip,
advertise_port,
upnp_bind_multicast,
):
hass: HomeAssistant,
entry_id: str,
name: str,
host_ip: str,
listen_port: int,
advertise_ip: str | None,
advertise_port: int | None,
upnp_bind_multicast: bool | None,
) -> None:
"""Initialize the properties."""
self.hass = hass
self.entry_id = entry_id
self.roku_usn = name
self.host_ip = host_ip
@ -47,21 +55,21 @@ class EmulatedRoku:
self.bind_multicast = upnp_bind_multicast
self._api_server = None
self._api_server: EmulatedRokuServer | None = None
self._unsub_start_listener = None
self._unsub_stop_listener = None
self._unsub_start_listener: CALLBACK_TYPE | None = None
self._unsub_stop_listener: CALLBACK_TYPE | None = None
async def setup(self):
async def setup(self) -> bool:
"""Start the emulated_roku server."""
class EventCommandHandler(EmulatedRokuCommandHandler):
"""emulated_roku command handler to turn commands into events."""
def __init__(self, hass):
def __init__(self, hass: HomeAssistant) -> None:
self.hass = hass
def on_keydown(self, roku_usn, key):
def on_keydown(self, roku_usn: str, key: str) -> None:
"""Handle keydown event."""
self.hass.bus.async_fire(
EVENT_ROKU_COMMAND,
@ -73,7 +81,7 @@ class EmulatedRoku:
EventOrigin.local,
)
def on_keyup(self, roku_usn, key):
def on_keyup(self, roku_usn: str, key: str) -> None:
"""Handle keyup event."""
self.hass.bus.async_fire(
EVENT_ROKU_COMMAND,
@ -85,7 +93,7 @@ class EmulatedRoku:
EventOrigin.local,
)
def on_keypress(self, roku_usn, key):
def on_keypress(self, roku_usn: str, key: str) -> None:
"""Handle keypress event."""
self.hass.bus.async_fire(
EVENT_ROKU_COMMAND,
@ -97,7 +105,7 @@ class EmulatedRoku:
EventOrigin.local,
)
def launch(self, roku_usn, app_id):
def launch(self, roku_usn: str, app_id: str) -> None:
"""Handle launch event."""
self.hass.bus.async_fire(
EVENT_ROKU_COMMAND,
@ -129,17 +137,19 @@ class EmulatedRoku:
bind_multicast=self.bind_multicast,
)
async def emulated_roku_stop(event):
async def emulated_roku_stop(event: Event | None) -> None:
"""Wrap the call to emulated_roku.close."""
LOGGER.debug("Stopping emulated_roku %s", self.roku_usn)
self._unsub_stop_listener = None
assert self._api_server is not None
await self._api_server.close()
async def emulated_roku_start(event):
async def emulated_roku_start(event: Event | None) -> None:
"""Wrap the call to emulated_roku.start."""
try:
LOGGER.debug("Starting emulated_roku %s", self.roku_usn)
self._unsub_start_listener = None
assert self._api_server is not None
await self._api_server.start()
except OSError:
LOGGER.exception(
@ -165,7 +175,7 @@ class EmulatedRoku:
return True
async def unload(self):
async def unload(self) -> bool:
"""Unload the emulated_roku server."""
LOGGER.debug("Unloading emulated_roku %s", self.roku_usn)
@ -177,6 +187,7 @@ class EmulatedRoku:
self._unsub_stop_listener()
self._unsub_stop_listener = None
assert self._api_server is not None
await self._api_server.close()
return True

31
tests/components/emulated_roku/test_binding.py
View File

@ -1,6 +1,7 @@
"""Tests for emulated_roku library bindings."""
from unittest.mock import AsyncMock, Mock, patch
from uuid import uuid4
from homeassistant.components.emulated_roku.binding import (
ATTR_APP_ID,
@ -14,14 +15,15 @@ from homeassistant.components.emulated_roku.binding import (
ROKU_COMMAND_LAUNCH,
EmulatedRoku,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import Event, HomeAssistant
async def test_events_fired_properly(hass: HomeAssistant) -> None:
"""Test that events are fired correctly."""
binding = EmulatedRoku(
hass, "Test Emulated Roku", "1.2.3.4", 8060, None, None, None
)
random_name = uuid4().hex
# Note that this test is accessing the internal EmulatedRoku class
# and should be refactored in the future not to do so.
binding = EmulatedRoku(hass, "x", random_name, "1.2.3.4", 8060, None, None, None)
events = []
roku_event_handler = None
@ -41,8 +43,9 @@ async def test_events_fired_properly(hass: HomeAssistant) -> None:
return Mock(start=AsyncMock(), close=AsyncMock())
def listener(event):
events.append(event)
def listener(event: Event) -> None:
if event.data[ATTR_SOURCE_NAME] == random_name:
events.append(event)
with patch(
"homeassistant.components.emulated_roku.binding.EmulatedRokuServer", instantiate
@ -53,10 +56,10 @@ async def test_events_fired_properly(hass: HomeAssistant) -> None:
assert roku_event_handler is not None
roku_event_handler.on_keydown("Test Emulated Roku", "A")
roku_event_handler.on_keyup("Test Emulated Roku", "A")
roku_event_handler.on_keypress("Test Emulated Roku", "C")
roku_event_handler.launch("Test Emulated Roku", "1")
roku_event_handler.on_keydown(random_name, "A")
roku_event_handler.on_keyup(random_name, "A")
roku_event_handler.on_keypress(random_name, "C")
roku_event_handler.launch(random_name, "1")
await hass.async_block_till_done()
@ -64,20 +67,20 @@ async def test_events_fired_properly(hass: HomeAssistant) -> None:
assert events[0].event_type == EVENT_ROKU_COMMAND
assert events[0].data[ATTR_COMMAND_TYPE] == ROKU_COMMAND_KEYDOWN
assert events[0].data[ATTR_SOURCE_NAME] == "Test Emulated Roku"
assert events[0].data[ATTR_SOURCE_NAME] == random_name
assert events[0].data[ATTR_KEY] == "A"
assert events[1].event_type == EVENT_ROKU_COMMAND
assert events[1].data[ATTR_COMMAND_TYPE] == ROKU_COMMAND_KEYUP
assert events[1].data[ATTR_SOURCE_NAME] == "Test Emulated Roku"
assert events[1].data[ATTR_SOURCE_NAME] == random_name
assert events[1].data[ATTR_KEY] == "A"
assert events[2].event_type == EVENT_ROKU_COMMAND
assert events[2].data[ATTR_COMMAND_TYPE] == ROKU_COMMAND_KEYPRESS
assert events[2].data[ATTR_SOURCE_NAME] == "Test Emulated Roku"
assert events[2].data[ATTR_SOURCE_NAME] == random_name
assert events[2].data[ATTR_KEY] == "C"
assert events[3].event_type == EVENT_ROKU_COMMAND
assert events[3].data[ATTR_COMMAND_TYPE] == ROKU_COMMAND_LAUNCH
assert events[3].data[ATTR_SOURCE_NAME] == "Test Emulated Roku"
assert events[3].data[ATTR_SOURCE_NAME] == random_name
assert events[3].data[ATTR_APP_ID] == "1"

14
tests/components/emulated_roku/test_init.py
View File

@ -86,16 +86,6 @@ async def test_setup_entry_successful(hass: HomeAssistant) -> None:
assert await emulated_roku.async_setup_entry(hass, entry) is True
assert len(instantiate.mock_calls) == 1
assert hass.data[emulated_roku.DOMAIN]
roku_instance = hass.data[emulated_roku.DOMAIN]["Emulated Roku Test"]
assert roku_instance.roku_usn == "Emulated Roku Test"
assert roku_instance.host_ip == "1.2.3.5"
assert roku_instance.listen_port == 8060
assert roku_instance.advertise_ip == "1.2.3.4"
assert roku_instance.advertise_port == 8071
assert roku_instance.bind_multicast is False
async def test_unload_entry(hass: HomeAssistant) -> None:
@ -113,10 +103,6 @@ async def test_unload_entry(hass: HomeAssistant) -> None:
):
assert await emulated_roku.async_setup_entry(hass, entry) is True
assert emulated_roku.DOMAIN in hass.data
await hass.async_block_till_done()
assert await emulated_roku.async_unload_entry(hass, entry)
assert len(hass.data[emulated_roku.DOMAIN]) == 0