Migrate samsungtv tests to use freezegun (#105286)

This commit is contained in:
Jan-Philipp Benecke 2023-12-08 14:37:02 +01:00 committed by GitHub
parent 8721327176
commit 6b7dace8d4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -11,6 +11,7 @@ from async_upnp_client.exceptions import (
UpnpError,
UpnpResponseError,
)
from freezegun.api import FrozenDateTimeFactory
import pytest
from samsungctl import exceptions
from samsungtvws.async_remote import SamsungTVWSAsyncRemote
@ -165,7 +166,9 @@ async def test_setup_websocket(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("rest_api")
async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> None:
async def test_setup_websocket_2(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
) -> None:
"""Test setup of platform from config entry."""
entity_id = f"{DOMAIN}.fake"
@ -194,9 +197,9 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non
assert config_entries[0].data[CONF_MAC] == "aa:bb:ww:ii:ff:ii"
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
@ -205,7 +208,7 @@ async def test_setup_websocket_2(hass: HomeAssistant, mock_now: datetime) -> Non
@pytest.mark.usefixtures("rest_api")
async def test_setup_encrypted_websocket(
hass: HomeAssistant, mock_now: datetime
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
) -> None:
"""Test setup of platform from config entry."""
with patch(
@ -219,9 +222,9 @@ async def test_setup_encrypted_websocket(
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state
@ -229,21 +232,25 @@ async def test_setup_encrypted_websocket(
@pytest.mark.usefixtures("remote")
async def test_update_on(hass: HomeAssistant, mock_now: datetime) -> None:
async def test_update_on(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
) -> None:
"""Testing update tv on."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
@pytest.mark.usefixtures("remote")
async def test_update_off(hass: HomeAssistant, mock_now: datetime) -> None:
async def test_update_off(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
) -> None:
"""Testing update tv off."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
@ -252,16 +259,20 @@ async def test_update_off(hass: HomeAssistant, mock_now: datetime) -> None:
side_effect=[OSError("Boom"), DEFAULT_MOCK],
):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_UNAVAILABLE
async def test_update_off_ws_no_power_state(
hass: HomeAssistant, remotews: Mock, rest_api: Mock, mock_now: datetime
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
remotews: Mock,
rest_api: Mock,
mock_now: datetime,
) -> None:
"""Testing update tv off."""
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
@ -276,9 +287,9 @@ async def test_update_off_ws_no_power_state(
remotews.is_alive.return_value = False
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
@ -287,7 +298,11 @@ async def test_update_off_ws_no_power_state(
@pytest.mark.usefixtures("remotews")
async def test_update_off_ws_with_power_state(
hass: HomeAssistant, remotews: Mock, rest_api: Mock, mock_now: datetime
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
remotews: Mock,
rest_api: Mock,
mock_now: datetime,
) -> None:
"""Testing update tv off."""
with patch.object(
@ -308,9 +323,9 @@ async def test_update_off_ws_with_power_state(
device_info["device"]["PowerState"] = "on"
rest_api.rest_device_info.return_value = device_info
next_update = mock_now + timedelta(minutes=1)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
remotews.start_listening.assert_called_once()
rest_api.rest_device_info.assert_called_once()
@ -324,9 +339,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)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
rest_api.rest_device_info.assert_called_once()
@ -337,9 +352,9 @@ async def test_update_off_ws_with_power_state(
rest_api.rest_device_info.reset_mock()
device_info["device"]["PowerState"] = "off"
next_update = mock_now + timedelta(minutes=3)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
rest_api.rest_device_info.assert_called_once()
@ -350,7 +365,11 @@ async def test_update_off_ws_with_power_state(
async def test_update_off_encryptedws(
hass: HomeAssistant, remoteencws: Mock, rest_api: Mock, mock_now: datetime
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
remoteencws: Mock,
rest_api: Mock,
mock_now: datetime,
) -> None:
"""Testing update tv off."""
await setup_samsungtv_entry(hass, MOCK_ENTRYDATA_ENCRYPTED_WS)
@ -364,9 +383,9 @@ async def test_update_off_encryptedws(
remoteencws.is_alive.return_value = False
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
@ -374,7 +393,9 @@ async def test_update_off_encryptedws(
@pytest.mark.usefixtures("remote")
async def test_update_access_denied(hass: HomeAssistant, mock_now: datetime) -> None:
async def test_update_access_denied(
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
) -> None:
"""Testing update tv access denied exception."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
@ -383,13 +404,14 @@ async def test_update_access_denied(hass: HomeAssistant, mock_now: datetime) ->
side_effect=exceptions.AccessDenied("Boom"),
):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
next_update = mock_now + timedelta(minutes=10)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert [
flow
@ -403,6 +425,7 @@ async def test_update_access_denied(hass: HomeAssistant, mock_now: datetime) ->
@pytest.mark.usefixtures("rest_api")
async def test_update_ws_connection_failure(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_now: datetime,
remotews: Mock,
caplog: pytest.LogCaptureFixture,
@ -416,8 +439,8 @@ async def test_update_ws_connection_failure(
side_effect=ConnectionFailure('{"event": "ms.voiceApp.hide"}'),
), patch.object(remotews, "is_alive", return_value=False):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert (
@ -432,7 +455,10 @@ async def test_update_ws_connection_failure(
@pytest.mark.usefixtures("rest_api")
async def test_update_ws_connection_closed(
hass: HomeAssistant, mock_now: datetime, remotews: Mock
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_now: datetime,
remotews: Mock,
) -> None:
"""Testing update tv connection failure exception."""
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
@ -441,8 +467,8 @@ async def test_update_ws_connection_closed(
remotews, "start_listening", side_effect=ConnectionClosedError(None, None)
), patch.object(remotews, "is_alive", return_value=False):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
@ -451,7 +477,10 @@ async def test_update_ws_connection_closed(
@pytest.mark.usefixtures("rest_api")
async def test_update_ws_unauthorized_error(
hass: HomeAssistant, mock_now: datetime, remotews: Mock
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
mock_now: datetime,
remotews: Mock,
) -> None:
"""Testing update tv unauthorized failure exception."""
await setup_samsungtv_entry(hass, MOCK_CONFIGWS)
@ -460,8 +489,8 @@ async def test_update_ws_unauthorized_error(
remotews, "start_listening", side_effect=UnauthorizedError
), patch.object(remotews, "is_alive", return_value=False):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
assert [
@ -475,7 +504,7 @@ async def test_update_ws_unauthorized_error(
@pytest.mark.usefixtures("remote")
async def test_update_unhandled_response(
hass: HomeAssistant, mock_now: datetime
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
) -> None:
"""Testing update tv unhandled response exception."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
@ -485,9 +514,9 @@ async def test_update_unhandled_response(
side_effect=[exceptions.UnhandledResponse("Boom"), DEFAULT_MOCK],
):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
@ -495,7 +524,7 @@ async def test_update_unhandled_response(
@pytest.mark.usefixtures("remote")
async def test_connection_closed_during_update_can_recover(
hass: HomeAssistant, mock_now: datetime
hass: HomeAssistant, freezer: FrozenDateTimeFactory, mock_now: datetime
) -> None:
"""Testing update tv connection closed exception can recover."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
@ -505,17 +534,17 @@ async def test_connection_closed_during_update_can_recover(
side_effect=[exceptions.ConnectionClosed(), DEFAULT_MOCK],
):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_UNAVAILABLE
next_update = mock_now + timedelta(minutes=10)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
@ -653,7 +682,7 @@ async def test_name(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("remote")
async def test_state(hass: HomeAssistant) -> None:
async def test_state(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
"""Test for state property."""
await setup_samsungtv_entry(hass, MOCK_CONFIG)
await hass.services.async_call(
@ -672,7 +701,8 @@ async def test_state(hass: HomeAssistant) -> None:
with patch(
"homeassistant.components.samsungtv.bridge.Remote",
side_effect=OSError,
), patch("homeassistant.util.dt.utcnow", return_value=next_update):
):
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
@ -1393,7 +1423,11 @@ async def test_upnp_subscribe_events_upnpresponseerror(
@pytest.mark.usefixtures("rest_api", "upnp_notify_server")
async def test_upnp_re_subscribe_events(
hass: HomeAssistant, remotews: Mock, dmr_device: Mock, mock_now: datetime
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
remotews: Mock,
dmr_device: Mock,
mock_now: datetime,
) -> None:
"""Test for Upnp event feedback."""
await setup_samsungtv_entry(hass, MOCK_ENTRY_WS)
@ -1407,9 +1441,9 @@ async def test_upnp_re_subscribe_events(
remotews, "start_listening", side_effect=WebSocketException("Boom")
), patch.object(remotews, "is_alive", return_value=False):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
@ -1417,9 +1451,9 @@ async def test_upnp_re_subscribe_events(
assert dmr_device.async_unsubscribe_services.call_count == 1
next_update = mock_now + timedelta(minutes=10)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
@ -1434,6 +1468,7 @@ async def test_upnp_re_subscribe_events(
)
async def test_upnp_failed_re_subscribe_events(
hass: HomeAssistant,
freezer: FrozenDateTimeFactory,
remotews: Mock,
dmr_device: Mock,
mock_now: datetime,
@ -1452,9 +1487,9 @@ async def test_upnp_failed_re_subscribe_events(
remotews, "start_listening", side_effect=WebSocketException("Boom")
), patch.object(remotews, "is_alive", return_value=False):
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
@ -1462,9 +1497,8 @@ async def test_upnp_failed_re_subscribe_events(
assert dmr_device.async_unsubscribe_services.call_count == 1
next_update = mock_now + timedelta(minutes=10)
with patch("homeassistant.util.dt.utcnow", return_value=next_update), patch.object(
dmr_device, "async_subscribe_services", side_effect=error
):
with patch.object(dmr_device, "async_subscribe_services", side_effect=error):
freezer.move_to(next_update)
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()