mirror of
https://github.com/home-assistant/core.git
synced 2025-07-18 18:57:06 +00:00
Use freezer.tick in SamsungTV tests (#143954)
This commit is contained in:
epenet
GitHub
parent
857db679ae
commit
8b9c4dadd0
@ -3,7 +3,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Awaitable, Callable, Generator
|
||||
from datetime import datetime
|
||||
from socket import AddressFamily # pylint: disable=no-name-in-module
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
@ -21,7 +20,6 @@ from samsungtvws.exceptions import ResponseError
|
||||
from samsungtvws.remote import ChannelEmitCommand
|
||||
|
||||
from homeassistant.components.samsungtv.const import WEBSOCKET_SSL_PORT
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from .const import SAMPLE_DEVICE_INFO_UE48JU6400, SAMPLE_DEVICE_INFO_WIFI
|
||||
|
||||
@ -285,12 +283,6 @@ def remoteencws_fixture() -> Generator[Mock]:
|
||||
yield remoteencws
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_now() -> datetime:
|
||||
"""Fixture for dtutil.now."""
|
||||
return dt_util.utcnow()
|
||||
|
||||
|
||||
@pytest.fixture(name="mac_address", autouse=True)
|
||||
def mac_address_fixture() -> Generator[Mock]:
|
||||
"""Patch getmac.get_mac_address."""
|
||||
|
||||
@ -1,7 +1,7 @@
|
||||
"""Tests for samsungtv component."""
|
||||
|
||||
from copy import deepcopy
|
||||
from datetime import datetime, timedelta
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from unittest.mock import DEFAULT as DEFAULT_MOCK, AsyncMock, Mock, call, patch
|
||||
|
||||
@ -78,7 +78,6 @@ from homeassistant.const import (
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ServiceNotSupported
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from . import async_wait_config_entry_reload, setup_samsungtv_entry
|
||||
from .const import (
|
||||
@ -153,7 +152,7 @@ async def test_setup_websocket(hass: HomeAssistant) -> None:
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_setup_websocket_2(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test setup of platform from config entry."""
|
||||
entity_id = f"{MP_DOMAIN}.fake"
|
||||
@ -182,9 +181,8 @@ async def test_setup_websocket_2(
|
||||
|
||||
assert config_entries[0].data[CONF_MAC] == "aa:bb:aa:aa:aa:aa"
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(entity_id)
|
||||
@ -194,7 +192,7 @@ async def test_setup_websocket_2(
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_setup_encrypted_websocket(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Test setup of platform from config entry."""
|
||||
with patch(
|
||||
@ -207,9 +205,8 @@ async def test_setup_encrypted_websocket(
|
||||
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -218,15 +215,12 @@ async def test_setup_encrypted_websocket(
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remote")
|
||||
async def test_update_on(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
|
||||
) -> None:
|
||||
async def test_update_on(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
|
||||
"""Testing update tv on."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIG)
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -234,9 +228,7 @@ async def test_update_on(
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("remote")
|
||||
async def test_update_off(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
|
||||
) -> None:
|
||||
async def test_update_off(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
|
||||
"""Testing update tv off."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIG)
|
||||
|
||||
@ -244,9 +236,8 @@ async def test_update_off(
|
||||
"homeassistant.components.samsungtv.bridge.Remote",
|
||||
side_effect=[OSError("Boom"), DEFAULT_MOCK],
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -254,11 +245,7 @@ async def test_update_off(
|
||||
|
||||
|
||||
async def test_update_off_ws_no_power_state(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remotews: Mock,
|
||||
rest_api: Mock,
|
||||
mock_now: datetime,
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock, rest_api: Mock
|
||||
) -> None:
|
||||
"""Testing update tv off."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
@ -272,9 +259,8 @@ async def test_update_off_ws_no_power_state(
|
||||
remotews.start_listening = Mock(side_effect=WebSocketException("Boom"))
|
||||
remotews.is_alive.return_value = False
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -284,11 +270,7 @@ async def test_update_off_ws_no_power_state(
|
||||
|
||||
@pytest.mark.usefixtures("remotews")
|
||||
async def test_update_off_ws_with_power_state(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remotews: Mock,
|
||||
rest_api: Mock,
|
||||
mock_now: datetime,
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock, rest_api: Mock
|
||||
) -> None:
|
||||
"""Testing update tv off."""
|
||||
with (
|
||||
@ -311,9 +293,9 @@ async def test_update_off_ws_with_power_state(
|
||||
device_info = deepcopy(SAMPLE_DEVICE_INFO_WIFI)
|
||||
device_info["device"]["PowerState"] = "on"
|
||||
rest_api.rest_device_info.return_value = device_info
|
||||
next_update = mock_now + timedelta(minutes=1)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
remotews.start_listening.assert_called_once()
|
||||
@ -327,9 +309,9 @@ async def test_update_off_ws_with_power_state(
|
||||
|
||||
# Second update uses device_info(ON)
|
||||
rest_api.rest_device_info.reset_mock()
|
||||
next_update = mock_now + timedelta(minutes=2)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
rest_api.rest_device_info.assert_called_once()
|
||||
@ -340,9 +322,9 @@ async def test_update_off_ws_with_power_state(
|
||||
# Third update uses device_info (OFF)
|
||||
rest_api.rest_device_info.reset_mock()
|
||||
device_info["device"]["PowerState"] = "off"
|
||||
next_update = mock_now + timedelta(minutes=3)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
|
||||
freezer.tick(timedelta(minutes=1))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
rest_api.rest_device_info.assert_called_once()
|
||||
@ -358,7 +340,6 @@ async def test_update_off_encryptedws(
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remoteencws: Mock,
|
||||
rest_api: Mock,
|
||||
mock_now: datetime,
|
||||
) -> None:
|
||||
"""Testing update tv off."""
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
|
||||
@ -371,9 +352,8 @@ async def test_update_off_encryptedws(
|
||||
remoteencws.start_listening = Mock(side_effect=WebSocketException("Boom"))
|
||||
remoteencws.is_alive.return_value = False
|
||||
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -383,7 +363,7 @@ async def test_update_off_encryptedws(
|
||||
|
||||
@pytest.mark.usefixtures("remote")
|
||||
async def test_update_access_denied(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Testing update tv access denied exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIG)
|
||||
@ -392,14 +372,12 @@ async def test_update_access_denied(
|
||||
"homeassistant.components.samsungtv.bridge.Remote",
|
||||
side_effect=exceptions.AccessDenied("Boom"),
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
next_update = mock_now + timedelta(minutes=10)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
assert [
|
||||
@ -415,7 +393,6 @@ async def test_update_access_denied(
|
||||
async def test_update_ws_connection_failure(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
mock_now: datetime,
|
||||
remotews: Mock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
@ -430,9 +407,8 @@ async def test_update_ws_connection_failure(
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
assert (
|
||||
@ -447,10 +423,7 @@ async def test_update_ws_connection_failure(
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_update_ws_connection_closed(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
mock_now: datetime,
|
||||
remotews: Mock,
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock
|
||||
) -> None:
|
||||
"""Testing update tv connection failure exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
@ -461,9 +434,8 @@ async def test_update_ws_connection_closed(
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -472,10 +444,7 @@ async def test_update_ws_connection_closed(
|
||||
|
||||
@pytest.mark.usefixtures("rest_api")
|
||||
async def test_update_ws_unauthorized_error(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
mock_now: datetime,
|
||||
remotews: Mock,
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, remotews: Mock
|
||||
) -> None:
|
||||
"""Testing update tv unauthorized failure exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
|
||||
@ -484,9 +453,8 @@ async def test_update_ws_unauthorized_error(
|
||||
patch.object(remotews, "start_listening", side_effect=UnauthorizedError),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
assert [
|
||||
@ -500,7 +468,7 @@ async def test_update_ws_unauthorized_error(
|
||||
|
||||
@pytest.mark.usefixtures("remote")
|
||||
async def test_update_unhandled_response(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Testing update tv unhandled response exception."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIG)
|
||||
@ -509,9 +477,8 @@ async def test_update_unhandled_response(
|
||||
"homeassistant.components.samsungtv.bridge.Remote",
|
||||
side_effect=[exceptions.UnhandledResponse("Boom"), DEFAULT_MOCK],
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -520,7 +487,7 @@ async def test_update_unhandled_response(
|
||||
|
||||
@pytest.mark.usefixtures("remote")
|
||||
async def test_connection_closed_during_update_can_recover(
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
|
||||
hass: HomeAssistant, freezer: FrozenDateTimeFactory
|
||||
) -> None:
|
||||
"""Testing update tv connection closed exception can recover."""
|
||||
await setup_samsungtv_entry(hass, MOCK_CONFIG)
|
||||
@ -529,17 +496,15 @@ async def test_connection_closed_during_update_can_recover(
|
||||
"homeassistant.components.samsungtv.bridge.Remote",
|
||||
side_effect=[exceptions.ConnectionClosed(), DEFAULT_MOCK],
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
|
||||
next_update = mock_now + timedelta(minutes=10)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -689,13 +654,12 @@ async def test_state(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> Non
|
||||
# Should be STATE_UNAVAILABLE after the timer expires
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
next_update = dt_util.utcnow() + timedelta(seconds=20)
|
||||
with patch(
|
||||
"homeassistant.components.samsungtv.bridge.Remote",
|
||||
side_effect=OSError,
|
||||
):
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(seconds=20))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -1390,7 +1354,6 @@ async def test_upnp_re_subscribe_events(
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remotews: Mock,
|
||||
dmr_device: Mock,
|
||||
mock_now: datetime,
|
||||
) -> None:
|
||||
"""Test for Upnp event feedback."""
|
||||
await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
|
||||
@ -1406,9 +1369,8 @@ async def test_upnp_re_subscribe_events(
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -1416,9 +1378,8 @@ async def test_upnp_re_subscribe_events(
|
||||
assert dmr_device.async_subscribe_services.call_count == 1
|
||||
assert dmr_device.async_unsubscribe_services.call_count == 1
|
||||
|
||||
next_update = mock_now + timedelta(minutes=10)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -1437,7 +1398,6 @@ async def test_upnp_failed_re_subscribe_events(
|
||||
freezer: FrozenDateTimeFactory,
|
||||
remotews: Mock,
|
||||
dmr_device: Mock,
|
||||
mock_now: datetime,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
error: Exception,
|
||||
) -> None:
|
||||
@ -1455,9 +1415,8 @@ async def test_upnp_failed_re_subscribe_events(
|
||||
),
|
||||
patch.object(remotews, "is_alive", return_value=False),
|
||||
):
|
||||
next_update = mock_now + timedelta(minutes=5)
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
@ -1465,10 +1424,9 @@ async def test_upnp_failed_re_subscribe_events(
|
||||
assert dmr_device.async_subscribe_services.call_count == 1
|
||||
assert dmr_device.async_unsubscribe_services.call_count == 1
|
||||
|
||||
next_update = mock_now + timedelta(minutes=10)
|
||||
with patch.object(dmr_device, "async_subscribe_services", side_effect=error):
|
||||
freezer.move_to(next_update)
|
||||
async_fire_time_changed(hass, next_update)
|
||||
freezer.tick(timedelta(minutes=5))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
|
||||
state = hass.states.get(ENTITY_ID)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user