"""Test for the Schedule integration.""" from collections.abc import Callable, Coroutine from typing import Any from unittest.mock import patch from freezegun.api import FrozenDateTimeFactory import pytest from syrupy.assertion import SnapshotAssertion from homeassistant.components.schedule.const import ( ATTR_NEXT_EVENT, CONF_ALL_DAYS, CONF_DATA, CONF_FRIDAY, CONF_FROM, CONF_MONDAY, CONF_SATURDAY, CONF_SUNDAY, CONF_THURSDAY, CONF_TO, CONF_TUESDAY, CONF_WEDNESDAY, DOMAIN, SERVICE_GET, ) from homeassistant.const import ( ATTR_EDITABLE, ATTR_FRIENDLY_NAME, ATTR_ICON, ATTR_NAME, CONF_ENTITY_ID, CONF_ICON, CONF_NAME, EVENT_STATE_CHANGED, SERVICE_RELOAD, STATE_OFF, STATE_ON, ) from homeassistant.core import Context, HomeAssistant from homeassistant.helpers import entity_registry as er from homeassistant.setup import async_setup_component from tests.common import MockUser, async_capture_events, async_fire_time_changed from tests.typing import WebSocketGenerator @pytest.mark.parametrize("invalid_config", [None, {"name with space": None}]) async def test_invalid_config(hass: HomeAssistant, invalid_config) -> None: """Test invalid configs.""" assert not await async_setup_component(hass, DOMAIN, {DOMAIN: invalid_config}) @pytest.mark.parametrize( ("schedule", "error"), [ ( [ {CONF_FROM: "00:00:00", CONF_TO: "23:59:59"}, {CONF_FROM: "07:00:00", CONF_TO: "08:00:00"}, ], "Overlapping times found in schedule", ), ( [ {CONF_FROM: "07:00:00", CONF_TO: "08:00:00"}, {CONF_FROM: "07:00:00", CONF_TO: "08:00:00"}, ], "Overlapping times found in schedule", ), ( [ {CONF_FROM: "07:59:00", CONF_TO: "09:00:00"}, {CONF_FROM: "07:00:00", CONF_TO: "08:00:00"}, ], "Overlapping times found in schedule", ), ( [ {CONF_FROM: "06:00:00", CONF_TO: "07:00:00"}, {CONF_FROM: "06:59:00", CONF_TO: "08:00:00"}, ], "Overlapping times found in schedule", ), ( [ {CONF_FROM: "06:00:00", CONF_TO: "05:00:00"}, ], "Invalid time range, from 06:00:00 is after 05:00:00", ), ], ) async def test_invalid_schedules( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], caplog: pytest.LogCaptureFixture, schedule: list[dict[str, str]], error: str, ) -> None: """Test overlapping time ranges invalidate.""" assert not await schedule_setup( config={ DOMAIN: { "from_yaml": { CONF_NAME: "from yaml", CONF_ICON: "mdi:party-pooper", CONF_SUNDAY: schedule, } } } ) assert error in caplog.text async def test_events_one_day( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], caplog: pytest.LogCaptureFixture, freezer: FrozenDateTimeFactory, ) -> None: """Test events only during one day of the week.""" freezer.move_to("2022-08-30 13:20:00-07:00") assert await schedule_setup( config={ DOMAIN: { "from_yaml": { CONF_NAME: "from yaml", CONF_ICON: "mdi:party-popper", CONF_SUNDAY: {CONF_FROM: "07:00:00", CONF_TO: "11:00:00"}, } } }, items=[], ) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T07:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T11:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-11T07:00:00-07:00" async def test_adjacent_cross_midnight( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], caplog: pytest.LogCaptureFixture, freezer: FrozenDateTimeFactory, ) -> None: """Test adjacent events don't toggle on->off->on.""" freezer.move_to("2022-08-30 13:20:00-07:00") assert await schedule_setup( config={ DOMAIN: { "from_yaml": { CONF_NAME: "from yaml", CONF_ICON: "mdi:party-popper", CONF_SUNDAY: {CONF_FROM: "23:00:00", CONF_TO: "24:00:00"}, CONF_MONDAY: {CONF_FROM: "00:00:00", CONF_TO: "01:00:00"}, } } }, items=[], ) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T23:00:00-07:00" state_changes = async_capture_events(hass, EVENT_STATE_CHANGED) freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-05T00:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-05T01:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-11T23:00:00-07:00" await hass.async_block_till_done() assert len(state_changes) == 3 for event in state_changes[:-1]: assert event.data["new_state"].state == STATE_ON assert state_changes[2].data["new_state"].state == STATE_OFF async def test_adjacent_within_day( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], caplog: pytest.LogCaptureFixture, freezer: FrozenDateTimeFactory, ) -> None: """Test adjacent events don't toggle on->off->on.""" freezer.move_to("2022-08-30 13:20:00-07:00") assert await schedule_setup( config={ DOMAIN: { "from_yaml": { CONF_NAME: "from yaml", CONF_ICON: "mdi:party-popper", CONF_SUNDAY: [ {CONF_FROM: "22:00:00", CONF_TO: "22:30:00"}, {CONF_FROM: "22:30:00", CONF_TO: "23:00:00"}, ], } } }, items=[], ) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T22:00:00-07:00" state_changes = async_capture_events(hass, EVENT_STATE_CHANGED) freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T22:30:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T23:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-11T22:00:00-07:00" await hass.async_block_till_done() assert len(state_changes) == 3 for event in state_changes[:-1]: assert event.data["new_state"].state == STATE_ON assert state_changes[2].data["new_state"].state == STATE_OFF async def test_non_adjacent_within_day( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], caplog: pytest.LogCaptureFixture, freezer: FrozenDateTimeFactory, ) -> None: """Test adjacent events don't toggle on->off->on.""" freezer.move_to("2022-08-30 13:20:00-07:00") assert await schedule_setup( config={ DOMAIN: { "from_yaml": { CONF_NAME: "from yaml", CONF_ICON: "mdi:party-popper", CONF_SUNDAY: [ {CONF_FROM: "22:00:00", CONF_TO: "22:15:00"}, {CONF_FROM: "22:30:00", CONF_TO: "23:00:00"}, ], } } }, items=[], ) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T22:00:00-07:00" state_changes = async_capture_events(hass, EVENT_STATE_CHANGED) freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T22:15:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T22:30:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T23:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-11T22:00:00-07:00" await hass.async_block_till_done() assert len(state_changes) == 4 assert state_changes[0].data["new_state"].state == STATE_ON assert state_changes[1].data["new_state"].state == STATE_OFF assert state_changes[2].data["new_state"].state == STATE_ON assert state_changes[3].data["new_state"].state == STATE_OFF @pytest.mark.parametrize( "schedule", [ {CONF_FROM: "00:00:00", CONF_TO: "24:00"}, {CONF_FROM: "00:00:00", CONF_TO: "24:00:00"}, ], ) async def test_to_midnight( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], caplog: pytest.LogCaptureFixture, schedule: list[dict[str, str]], freezer: FrozenDateTimeFactory, ) -> None: """Test time range allow to 24:00.""" freezer.move_to("2022-08-30 13:20:00-07:00") assert await schedule_setup( config={ DOMAIN: { "from_yaml": { CONF_NAME: "from yaml", CONF_ICON: "mdi:party-popper", CONF_SUNDAY: schedule, } } }, items=[], ) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-04T00:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-05T00:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-09-11T00:00:00-07:00" async def test_setup_no_config(hass: HomeAssistant, hass_admin_user: MockUser) -> None: """Test component setup with no config.""" count_start = len(hass.states.async_entity_ids()) assert await async_setup_component(hass, DOMAIN, {}) with patch( "homeassistant.config.load_yaml_config_file", autospec=True, return_value={} ): await hass.services.async_call( DOMAIN, SERVICE_RELOAD, blocking=True, context=Context(user_id=hass_admin_user.id), ) await hass.async_block_till_done() assert count_start == len(hass.states.async_entity_ids()) @pytest.mark.freeze_time("2022-08-10 20:10:00-07:00") async def test_load( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], ) -> None: """Test set up from storage and YAML.""" assert await schedule_setup() state = hass.states.get(f"{DOMAIN}.from_storage") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_FRIENDLY_NAME] == "from storage" assert state.attributes[ATTR_EDITABLE] is True assert state.attributes[ATTR_ICON] == "mdi:party-popper" assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-08-12T17:00:00-07:00" state = hass.states.get(f"{DOMAIN}.from_yaml") assert state assert state.state == STATE_ON assert state.attributes[ATTR_FRIENDLY_NAME] == "from yaml" assert state.attributes[ATTR_EDITABLE] is False assert state.attributes[ATTR_ICON] == "mdi:party-pooper" assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-08-10T23:59:59-07:00" async def test_schedule_updates( hass: HomeAssistant, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], freezer: FrozenDateTimeFactory, ) -> None: """Test the schedule updates when time changes.""" freezer.move_to("2022-08-10 20:10:00-07:00") assert await schedule_setup() state = hass.states.get(f"{DOMAIN}.from_storage") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-08-12T17:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get(f"{DOMAIN}.from_storage") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-08-12T23:59:59-07:00" async def test_ws_list( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], ) -> None: """Test listing via WS.""" assert await schedule_setup() client = await hass_ws_client(hass) await client.send_json({"id": 1, "type": f"{DOMAIN}/list"}) resp = await client.receive_json() assert resp["success"] result = {item["id"]: item for item in resp["result"]} assert len(result) == 1 assert result["from_storage"][ATTR_NAME] == "from storage" assert result["from_storage"][CONF_FRIDAY] == [ {CONF_FROM: "17:00:00", CONF_TO: "23:59:59", CONF_DATA: {"party_level": "epic"}} ] assert result["from_storage"][CONF_SATURDAY] == [ {CONF_FROM: "00:00:00", CONF_TO: "23:59:59"} ] assert result["from_storage"][CONF_SUNDAY] == [ {CONF_FROM: "00:00:00", CONF_TO: "24:00:00", CONF_DATA: {"entry": "VIPs only"}} ] assert "from_yaml" not in result async def test_ws_delete( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entity_registry: er.EntityRegistry, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], ) -> None: """Test WS delete cleans up entity registry.""" assert await schedule_setup() state = hass.states.get("schedule.from_storage") assert state is not None assert ( entity_registry.async_get_entity_id(DOMAIN, DOMAIN, "from_storage") is not None ) client = await hass_ws_client(hass) await client.send_json( {"id": 1, "type": f"{DOMAIN}/delete", f"{DOMAIN}_id": "from_storage"} ) resp = await client.receive_json() assert resp["success"] state = hass.states.get("schedule.from_storage") assert state is None assert entity_registry.async_get_entity_id(DOMAIN, DOMAIN, "from_storage") is None @pytest.mark.freeze_time("2022-08-10 20:10:00-07:00") @pytest.mark.parametrize( ("to", "next_event", "saved_to", "icon_dict"), [ ( "23:59:59", "2022-08-10T23:59:59-07:00", "23:59:59", {CONF_ICON: "mdi:party-pooper"}, ), ( "24:00", "2022-08-11T00:00:00-07:00", "24:00:00", {CONF_ICON: "mdi:party-popper"}, ), ( "24:00:00", "2022-08-11T00:00:00-07:00", "24:00:00", {}, ), ], ) async def test_update( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entity_registry: er.EntityRegistry, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], to: str, next_event: str, saved_to: str, icon_dict: dict, ) -> None: """Test updating the schedule.""" assert await schedule_setup() state = hass.states.get("schedule.from_storage") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_FRIENDLY_NAME] == "from storage" assert state.attributes[ATTR_ICON] == "mdi:party-popper" assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-08-12T17:00:00-07:00" assert ( entity_registry.async_get_entity_id(DOMAIN, DOMAIN, "from_storage") is not None ) client = await hass_ws_client(hass) await client.send_json( { "id": 1, "type": f"{DOMAIN}/update", f"{DOMAIN}_id": "from_storage", CONF_NAME: "Party pooper", **icon_dict, CONF_MONDAY: [], CONF_TUESDAY: [], CONF_WEDNESDAY: [{CONF_FROM: "17:00:00", CONF_TO: to}], CONF_THURSDAY: [], CONF_FRIDAY: [], CONF_SATURDAY: [], CONF_SUNDAY: [], } ) resp = await client.receive_json() assert resp["success"] state = hass.states.get("schedule.from_storage") assert state assert state.state == STATE_ON assert state.attributes[ATTR_FRIENDLY_NAME] == "Party pooper" assert state.attributes.get(ATTR_ICON) == icon_dict.get(CONF_ICON) assert state.attributes[ATTR_NEXT_EVENT].isoformat() == next_event await client.send_json({"id": 2, "type": f"{DOMAIN}/list"}) resp = await client.receive_json() assert resp["success"] result = {item["id"]: item for item in resp["result"]} assert len(result) == 1 assert result["from_storage"][CONF_WEDNESDAY] == [ {CONF_FROM: "17:00:00", CONF_TO: saved_to} ] @pytest.mark.freeze_time("2022-08-11 8:52:00-07:00") @pytest.mark.parametrize( ("to", "next_event", "saved_to"), [ ("14:00:00", "2022-08-15T14:00:00-07:00", "14:00:00"), ("24:00", "2022-08-16T00:00:00-07:00", "24:00:00"), ("24:00:00", "2022-08-16T00:00:00-07:00", "24:00:00"), ], ) async def test_ws_create( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, entity_registry: er.EntityRegistry, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], freezer: FrozenDateTimeFactory, to: str, next_event: str, saved_to: str, ) -> None: """Test create WS.""" freezer.move_to("2022-08-11 8:52:00-07:00") assert await schedule_setup(items=[]) state = hass.states.get("schedule.party_mode") assert state is None assert entity_registry.async_get_entity_id(DOMAIN, DOMAIN, "party_mode") is None client = await hass_ws_client(hass) await client.send_json( { "id": 1, "type": f"{DOMAIN}/create", "name": "Party mode", "icon": "mdi:party-popper", "monday": [{"from": "12:00:00", "to": to}], } ) resp = await client.receive_json() assert resp["success"] state = hass.states.get("schedule.party_mode") assert state assert state.state == STATE_OFF assert state.attributes[ATTR_FRIENDLY_NAME] == "Party mode" assert state.attributes[ATTR_EDITABLE] is True assert state.attributes[ATTR_ICON] == "mdi:party-popper" assert state.attributes[ATTR_NEXT_EVENT].isoformat() == "2022-08-15T12:00:00-07:00" freezer.move_to(state.attributes[ATTR_NEXT_EVENT]) async_fire_time_changed(hass) state = hass.states.get("schedule.party_mode") assert state assert state.state == STATE_ON assert state.attributes[ATTR_NEXT_EVENT].isoformat() == next_event await client.send_json({"id": 2, "type": f"{DOMAIN}/list"}) resp = await client.receive_json() assert resp["success"] result = {item["id"]: item for item in resp["result"]} assert len(result) == 1 assert result["party_mode"][CONF_MONDAY] == [ {CONF_FROM: "12:00:00", CONF_TO: saved_to} ] async def test_service_get( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, snapshot: SnapshotAssertion, schedule_setup: Callable[..., Coroutine[Any, Any, bool]], ) -> None: """Test getting a single schedule via service.""" assert await schedule_setup() entity_id = "schedule.from_storage" # Test retrieving a single schedule via service call service_result = await hass.services.async_call( DOMAIN, SERVICE_GET, { CONF_ENTITY_ID: entity_id, }, blocking=True, return_response=True, ) result = service_result.get(entity_id) assert set(result) == CONF_ALL_DAYS assert result == snapshot(name=f"{entity_id}-get") # Now we update the schedule via WS client = await hass_ws_client(hass) await client.send_json( { "id": 1, "type": f"{DOMAIN}/update", f"{DOMAIN}_id": entity_id.rsplit(".", maxsplit=1)[-1], CONF_NAME: "Party pooper", CONF_ICON: "mdi:party-pooper", CONF_MONDAY: [], CONF_TUESDAY: [], CONF_WEDNESDAY: [{CONF_FROM: "17:00:00", CONF_TO: "19:00:00"}], CONF_THURSDAY: [], CONF_FRIDAY: [], CONF_SATURDAY: [], CONF_SUNDAY: [], } ) resp = await client.receive_json() assert resp["success"] # Test retrieving the schedule via service call after WS update service_result = await hass.services.async_call( DOMAIN, SERVICE_GET, { CONF_ENTITY_ID: entity_id, }, blocking=True, return_response=True, ) result = service_result.get(entity_id) assert set(result) == CONF_ALL_DAYS assert result == snapshot(name=f"{entity_id}-get-after-update")