"""The tests for the sun automation.""" from contextlib import AbstractContextManager, nullcontext from datetime import datetime, timedelta from typing import Any import astral from astral.sun import elevation as astral_elevation from freezegun import freeze_time import pytest import voluptuous as vol from homeassistant.components import automation, sun from homeassistant.components.sun.trigger import _next_polar_transition from homeassistant.const import ( ATTR_ENTITY_ID, ENTITY_MATCH_ALL, SERVICE_TURN_OFF, SERVICE_TURN_ON, SUN_EVENT_SUNRISE, SUN_EVENT_SUNSET, ) from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.helpers.sun import ( get_astral_event_next, get_astral_observer, get_observer_astral_event_next, ) from homeassistant.helpers.trigger import async_validate_trigger_config from homeassistant.setup import async_setup_component from homeassistant.util import dt as dt_util from tests.common import async_fire_time_changed, mock_component from tests.components.common import assert_trigger_options_supported _TEST_DATETIME = datetime(2015, 9, 15, 1, tzinfo=dt_util.UTC) _SUN_ENTITY_ID = "sun.sun" # Next dawn/dusk after _TEST_DATETIME at the default test location (San Diego), # precomputed to serve as an independent oracle for the trigger's scheduler. _DAWN_DUSK = { ("dawn", "civil"): datetime(2015, 9, 15, 13, 7, 26, 84892, tzinfo=dt_util.UTC), ("dawn", "nautical"): datetime(2015, 9, 15, 12, 38, 34, 55317, tzinfo=dt_util.UTC), ("dawn", "astronomical"): datetime( 2015, 9, 15, 12, 9, 9, 742065, tzinfo=dt_util.UTC ), ("dusk", "civil"): datetime(2015, 9, 15, 2, 21, 39, 56429, tzinfo=dt_util.UTC), ("dusk", "nautical"): datetime(2015, 9, 15, 2, 50, 29, 596023, tzinfo=dt_util.UTC), ("dusk", "astronomical"): datetime( 2015, 9, 15, 3, 19, 52, 663966, tzinfo=dt_util.UTC ), } @pytest.fixture(autouse=True) async def setup_comp(hass: HomeAssistant) -> None: """Initialize components.""" mock_component(hass, "group") await async_setup_component(hass, sun.DOMAIN, {sun.DOMAIN: {}}) async def _arm_automation( hass: HomeAssistant, trigger: dict[str, Any], extra_data: dict[str, Any] ) -> None: """Set up an automation with the given trigger config.""" await async_setup_component( hass, automation.DOMAIN, { automation.DOMAIN: { "trigger": trigger, "action": { "service": "test.automation", "data_template": extra_data, }, } }, ) # --- Legacy ``platform: sun`` backwards compatibility ------------------------ async def test_sunset_trigger( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the legacy sunset trigger.""" now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC) trigger_time = datetime(2015, 9, 16, 2, tzinfo=dt_util.UTC) with freeze_time(now): await async_setup_component( hass, automation.DOMAIN, { automation.DOMAIN: { "trigger": {"platform": "sun", "event": SUN_EVENT_SUNSET}, "action": { "service": "test.automation", "data_template": {"id": "{{ trigger.id}}"}, }, } }, ) await hass.services.async_call( automation.DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_MATCH_ALL}, blocking=True, ) assert len(service_calls) == 1 async_fire_time_changed(hass, trigger_time) await hass.async_block_till_done() assert len(service_calls) == 1 with freeze_time(now): await hass.services.async_call( automation.DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_MATCH_ALL}, blocking=True, ) assert len(service_calls) == 2 async_fire_time_changed(hass, trigger_time) await hass.async_block_till_done() assert len(service_calls) == 3 assert service_calls[2].data["id"] == 0 async def test_sunrise_trigger( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the legacy sunrise trigger.""" now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC) trigger_time = datetime(2015, 9, 16, 14, tzinfo=dt_util.UTC) with freeze_time(now): await async_setup_component( hass, automation.DOMAIN, { automation.DOMAIN: { "trigger": {"platform": "sun", "event": SUN_EVENT_SUNRISE}, "action": {"service": "test.automation"}, } }, ) async_fire_time_changed(hass, trigger_time) await hass.async_block_till_done() assert len(service_calls) == 1 async def test_sunset_trigger_with_offset( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the legacy sunset trigger with offset.""" now = datetime(2015, 9, 15, 23, tzinfo=dt_util.UTC) trigger_time = datetime(2015, 9, 16, 2, 30, tzinfo=dt_util.UTC) with freeze_time(now): await async_setup_component( hass, automation.DOMAIN, { automation.DOMAIN: { "trigger": { "platform": "sun", "event": SUN_EVENT_SUNSET, "offset": "0:30:00", }, "action": { "service": "test.automation", "data_template": { "some": ( "{{ trigger.platform }}" " - {{ trigger.event }}" " - {{ trigger.offset }}" ) }, }, } }, ) async_fire_time_changed(hass, trigger_time) await hass.async_block_till_done() assert len(service_calls) == 1 assert service_calls[0].data["some"] == "sun - sunset - 0:30:00" async def test_sunrise_trigger_with_offset( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the legacy sunrise trigger with offset.""" now = datetime(2015, 9, 13, 23, tzinfo=dt_util.UTC) trigger_time = datetime(2015, 9, 16, 13, 30, tzinfo=dt_util.UTC) with freeze_time(now): await async_setup_component( hass, automation.DOMAIN, { automation.DOMAIN: { "trigger": { "platform": "sun", "event": SUN_EVENT_SUNRISE, "offset": "-0:30:00", }, "action": {"service": "test.automation"}, } }, ) async_fire_time_changed(hass, trigger_time) await hass.async_block_till_done() assert len(service_calls) == 1 # --- Modern solar event triggers --------------------------------------------- @pytest.mark.parametrize( ("trigger_key", "astral_event"), [ ("sun.sunrise", SUN_EVENT_SUNRISE), ("sun.sunset", SUN_EVENT_SUNSET), ("sun.solar_noon", "noon"), ("sun.solar_midnight", "midnight"), ], ) async def test_event_trigger_fires( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, astral_event: str, ) -> None: """Test the modern solar event triggers fire at the event time.""" with freeze_time(_TEST_DATETIME): await _arm_automation(hass, {"platform": trigger_key}, {}) expected = get_astral_event_next(hass, astral_event, _TEST_DATETIME) async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 async def test_event_trigger_reschedules( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test that a solar event trigger reschedules for the following day.""" with freeze_time(_TEST_DATETIME) as freezer: await _arm_automation(hass, {"platform": "sun.sunrise"}, {}) first = get_astral_event_next(hass, SUN_EVENT_SUNRISE, _TEST_DATETIME) freezer.move_to(first + timedelta(seconds=1)) async_fire_time_changed(hass, first + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 # The reschedule computes the next event from the now-advanced clock. second = get_astral_event_next(hass, SUN_EVENT_SUNRISE, first) assert second > first freezer.move_to(second + timedelta(seconds=1)) async_fire_time_changed(hass, second + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 2 async def test_event_trigger_reschedules_on_location_change( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test a solar event trigger reschedules when the location changes.""" with freeze_time(_TEST_DATETIME): await _arm_automation(hass, {"platform": "sun.sunrise"}, {}) first = get_astral_event_next(hass, SUN_EVENT_SUNRISE, _TEST_DATETIME) # Move far east so the next sunrise occurs earlier (in UTC) than before. await hass.config.async_update(latitude=51.5, longitude=0.0) await hass.async_block_till_done() rescheduled = get_astral_event_next(hass, SUN_EVENT_SUNRISE, _TEST_DATETIME) assert rescheduled < first # Firing at the new, earlier event time proves the schedule moved with it: # the original schedule pointed at the later sunrise and would not fire yet. async_fire_time_changed(hass, rescheduled + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 @pytest.mark.parametrize("trigger_key", ["sun.dawn", "sun.dusk"]) @pytest.mark.parametrize("twilight", ["civil", "nautical", "astronomical"]) async def test_dawn_dusk_trigger_fires( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, twilight: str, ) -> None: """Test the dawn and dusk triggers fire for each twilight phase.""" event = trigger_key.split(".")[1] with freeze_time(_TEST_DATETIME): await _arm_automation( hass, {"platform": trigger_key, "options": {"type": twilight}}, {"type": "{{ trigger.type }}"}, ) expected = _DAWN_DUSK[event, twilight] async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 assert service_calls[0].data["type"] == twilight async def test_dawn_defaults_to_civil( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the dawn trigger defaults to civil twilight.""" with freeze_time(_TEST_DATETIME): await _arm_automation( hass, {"platform": "sun.dawn"}, {"type": "{{ trigger.type }}"} ) # Nautical dawn happens earlier than civil dawn and must not fire. civil = _DAWN_DUSK["dawn", "civil"] nautical = _DAWN_DUSK["dawn", "nautical"] assert nautical < civil async_fire_time_changed(hass, nautical + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 0 async_fire_time_changed(hass, civil + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 assert service_calls[0].data["type"] == "civil" @pytest.mark.parametrize( ("trigger_key", "astral_event"), [ ("sun.sunrise", SUN_EVENT_SUNRISE), ("sun.sunset", SUN_EVENT_SUNSET), ("sun.solar_noon", "noon"), ("sun.solar_midnight", "midnight"), ], ) @pytest.mark.parametrize( ("offset_type", "sign"), [("before", -1), ("after", 1)], ids=["before", "after"], ) async def test_event_trigger_offset( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, astral_event: str, offset_type: str, sign: int, ) -> None: """Test the solar event triggers apply a before/after time offset.""" offset = timedelta(hours=1) with freeze_time(_TEST_DATETIME): await _arm_automation( hass, { "platform": trigger_key, "options": {"offset": {"hours": 1}, "offset_type": offset_type}, }, {}, ) expected = get_astral_event_next( hass, astral_event, _TEST_DATETIME, sign * offset ) # The offset shifts the fire time away from the bare event time. assert expected != get_astral_event_next(hass, astral_event, _TEST_DATETIME) async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 @pytest.mark.parametrize("trigger_key", ["sun.dawn", "sun.dusk"]) @pytest.mark.parametrize( ("offset_type", "sign"), [("before", -1), ("after", 1)], ids=["before", "after"], ) async def test_dawn_dusk_trigger_offset( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, offset_type: str, sign: int, ) -> None: """Test the dawn and dusk triggers apply a before/after time offset.""" event = trigger_key.split(".")[1] offset = timedelta(hours=1) with freeze_time(_TEST_DATETIME): await _arm_automation( hass, { "platform": trigger_key, "options": {"offset": {"hours": 1}, "offset_type": offset_type}, }, {}, ) expected = _DAWN_DUSK[event, "civil"] + sign * offset async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 # --- Edge cases: no matching solar event on the following day ---------------- # Longyearbyen, Svalbard (deep polar latitude) and Kotzebue, Alaska (above the # Arctic Circle and far west of its time-zone meridian, i.e. heavily skewed # local time). Both have stretches of weeks without a given solar event. _SVALBARD = (78.22, 15.65, "Europe/Oslo") _KOTZEBUE = (66.8983, -162.5966, "America/Anchorage") # A two-sunrise day is the mirror of Kotzebue's two-sunset day: it needs solar # noon (not midnight) near local midnight, which no real location has because # time zones keep solar noon near local noon. This synthetic location forces it # with a polar latitude on a deliberately ~12 h-offset time zone. _TWO_SUNRISE_LOCATION = (66.5, -32.5, "America/Anchorage") @pytest.mark.parametrize( ("location", "now", "trigger_key", "astral_event", "options", "depression"), [ # Polar night: no sunrise for ~2 months. ( _SVALBARD, datetime(2015, 12, 15, 12, tzinfo=dt_util.UTC), "sun.sunrise", "sunrise", {}, None, ), # Midnight sun (skewed time zone): no sunset for ~6 weeks. ( _KOTZEBUE, datetime(2015, 6, 15, 12, tzinfo=dt_util.UTC), "sun.sunset", "sunset", {}, None, ), # No civil dawn at Svalbard during polar night. ( _SVALBARD, datetime(2015, 12, 15, 12, tzinfo=dt_util.UTC), "sun.dawn", "dawn", {"type": "civil"}, 6.0, ), # No civil dusk at Kotzebue around the summer solstice (skewed tz). ( _KOTZEBUE, datetime(2015, 6, 15, 12, tzinfo=dt_util.UTC), "sun.dusk", "dusk", {"type": "civil"}, 6.0, ), ], ) async def test_event_trigger_no_event_next_day( hass: HomeAssistant, service_calls: list[ServiceCall], location: tuple[float, float, str], now: datetime, trigger_key: str, astral_event: str, options: dict[str, Any], depression: float | None, ) -> None: """Test event triggers scan forward to the next day that has the event.""" latitude, longitude, time_zone = location await hass.config.async_set_time_zone(time_zone) await hass.config.async_update(latitude=latitude, longitude=longitude, elevation=0) with freeze_time(now): await _arm_automation(hass, {"platform": trigger_key, "options": options}, {}) expected = get_observer_astral_event_next( get_astral_observer(hass), astral_event, now, depression=depression ) # The event does not occur on the following day at this latitude/season. assert expected > now + timedelta(days=1) async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 async def test_two_sunsets_on_one_day_at_kotzebue( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test both sunsets fire on a calendar day that has two. Kotzebue's far-west longitude on Alaska time puts solar midnight near local midnight in early August, so 2015-08-07 (local) has two sunsets ~24 h apart - one just after midnight and one just before - and the scheduler must fire for both rather than skipping the second. """ latitude, longitude, time_zone = _KOTZEBUE await hass.config.async_set_time_zone(time_zone) await hass.config.async_update(latitude=latitude, longitude=longitude, elevation=0) # 2015-08-07 08:03:42 UTC (00:03 local) and 2015-08-08 07:59:25 UTC (23:59 local) now = datetime(2015, 8, 7, 7, tzinfo=dt_util.UTC) with freeze_time(now) as freezer: await _arm_automation(hass, {"platform": "sun.sunset"}, {}) first = get_astral_event_next(hass, "sunset", now) second = get_astral_event_next(hass, "sunset", first) # Two sunsets ~24 h apart that share one local calendar day. assert dt_util.as_local(first).date() == dt_util.as_local(second).date() assert timedelta(hours=23) < second - first < timedelta(hours=25) freezer.move_to(first + timedelta(seconds=1)) async_fire_time_changed(hass, first + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 freezer.move_to(second + timedelta(seconds=1)) async_fire_time_changed(hass, second + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 2 async def test_two_sunrises_on_one_day( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test both sunrises fire on a calendar day that has two. The mirror of the two-sunset case: a synthetic polar location on a deliberately offset time zone puts solar noon near local midnight, so 2015-03-07 (local) has two sunrises ~24 h apart - one just after midnight and one just before - and the scheduler must fire for both. """ latitude, longitude, time_zone = _TWO_SUNRISE_LOCATION await hass.config.async_set_time_zone(time_zone) await hass.config.async_update(latitude=latitude, longitude=longitude, elevation=0) # 2015-03-07 09:02:36 UTC (00:02 local) and 2015-03-08 08:58:43 UTC (23:58 local) now = datetime(2015, 3, 7, 9, tzinfo=dt_util.UTC) with freeze_time(now) as freezer: await _arm_automation(hass, {"platform": "sun.sunrise"}, {}) first = get_astral_event_next(hass, "sunrise", now) second = get_astral_event_next(hass, "sunrise", first) # Two sunrises ~24 h apart that share one local calendar day. assert dt_util.as_local(first).date() == dt_util.as_local(second).date() assert timedelta(hours=23) < second - first < timedelta(hours=25) freezer.move_to(first + timedelta(seconds=1)) async_fire_time_changed(hass, first + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 freezer.move_to(second + timedelta(seconds=1)) async_fire_time_changed(hass, second + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 2 @pytest.mark.parametrize( ("trigger_key", "astral_event", "now", "above_horizon"), [ # Midnight sun at Svalbard: solar midnight still occurs, with the sun's # lowest point above the horizon. ( "sun.solar_midnight", "midnight", datetime(2015, 6, 15, tzinfo=dt_util.UTC), True, ), # Polar night at Svalbard: solar noon still occurs, with the sun's # highest point below the horizon. ( "sun.solar_noon", "noon", datetime(2015, 12, 15, tzinfo=dt_util.UTC), False, ), ], ) async def test_solar_noon_midnight_in_polar_regions( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, astral_event: str, now: datetime, above_horizon: bool, ) -> None: """Test solar noon/midnight fire even when the sun never crosses the horizon. Solar noon and solar midnight are the extremes of the sun's daily arc, so they occur every day regardless of the horizon: solar midnight happens during midnight sun (sun stays up) and solar noon happens during polar night (sun stays down). """ latitude, longitude, time_zone = _SVALBARD await hass.config.async_set_time_zone(time_zone) await hass.config.async_update(latitude=latitude, longitude=longitude, elevation=0) with freeze_time(now): await _arm_automation(hass, {"platform": trigger_key}, {}) expected = get_astral_event_next(hass, astral_event, now) # The defining property: at the sun's lowest point it is still up # (midnight sun), or at its highest point it is still down (polar night). elevation = astral_elevation(get_astral_observer(hass), expected) assert (elevation > -0.833) is above_horizon async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 # --- Golden and blue hour triggers ------------------------------------------- # For each golden/blue hour trigger, the (astral event, depression) of its # morning (rising, found as a "dawn") and evening (descending, a "dusk") boundary # crossing. depression is the negated boundary elevation astral expects. _GOLDEN_BLUE_CROSSINGS = { ("sun.golden_hour_started", "morning"): ("dawn", 4), ("sun.golden_hour_started", "evening"): ("dusk", -6), ("sun.golden_hour_ended", "morning"): ("dawn", -6), ("sun.golden_hour_ended", "evening"): ("dusk", 4), ("sun.blue_hour_started", "morning"): ("dawn", 6), ("sun.blue_hour_started", "evening"): ("dusk", 4), ("sun.blue_hour_ended", "morning"): ("dawn", 4), ("sun.blue_hour_ended", "evening"): ("dusk", 6), } @pytest.mark.parametrize(("trigger_key", "period"), list(_GOLDEN_BLUE_CROSSINGS)) async def test_golden_blue_hour_trigger_fires( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, period: str, ) -> None: """Test the golden/blue hour triggers fire at the boundary for each period.""" event, depression = _GOLDEN_BLUE_CROSSINGS[trigger_key, period] with freeze_time(_TEST_DATETIME): await _arm_automation( hass, {"platform": trigger_key, "options": {"period": period}}, {"period": "{{ trigger.period }}"}, ) expected = get_observer_astral_event_next( get_astral_observer(hass), event, _TEST_DATETIME, depression=depression ) async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 assert service_calls[0].data["period"] == period async def test_golden_blue_hour_trigger_any_fires_at_both_crossings( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test that with period ``any`` the trigger fires at both daily crossings.""" with freeze_time(_TEST_DATETIME) as freezer: await _arm_automation( hass, {"platform": "sun.golden_hour_started", "options": {"period": "any"}}, {}, ) observer = get_astral_observer(hass) # The morning (rising to -4°) and evening (descending to 6°) crossings. morning = get_observer_astral_event_next( observer, "dawn", _TEST_DATETIME, depression=4 ) evening = get_observer_astral_event_next( observer, "dusk", _TEST_DATETIME, depression=-6 ) first, second = sorted((morning, evening)) freezer.move_to(first + timedelta(seconds=1)) async_fire_time_changed(hass, first + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 freezer.move_to(second + timedelta(seconds=1)) async_fire_time_changed(hass, second + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 2 async def test_golden_hour_defaults_to_any( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the golden hour trigger defaults to the ``any`` period.""" with freeze_time(_TEST_DATETIME): await _arm_automation( hass, {"platform": "sun.golden_hour_started"}, {"period": "{{ trigger.period }}"}, ) # The evening crossing (descending to 6°) is the next one after the anchor. evening = get_observer_astral_event_next( get_astral_observer(hass), "dusk", _TEST_DATETIME, depression=-6 ) async_fire_time_changed(hass, evening + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 assert service_calls[0].data["period"] == "any" @pytest.mark.parametrize( ("trigger_key", "period", "event", "depression"), [ ("sun.golden_hour_started", "morning", "dawn", 4), ("sun.golden_hour_ended", "evening", "dusk", 4), ], ) @pytest.mark.parametrize( ("offset_type", "sign"), [("before", -1), ("after", 1)], ids=["before", "after"], ) async def test_golden_blue_hour_trigger_offset( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, period: str, event: str, depression: int, offset_type: str, sign: int, ) -> None: """Test the golden/blue hour triggers apply a before/after time offset.""" offset = timedelta(hours=1) with freeze_time(_TEST_DATETIME): await _arm_automation( hass, { "platform": trigger_key, "options": { "period": period, "offset": {"hours": 1}, "offset_type": offset_type, }, }, {}, ) observer = get_astral_observer(hass) expected = get_observer_astral_event_next( observer, event, _TEST_DATETIME, sign * offset, depression=depression ) # The offset shifts the fire time away from the bare crossing time. assert expected != get_observer_astral_event_next( observer, event, _TEST_DATETIME, depression=depression ) async_fire_time_changed(hass, expected + timedelta(seconds=1)) await hass.async_block_till_done() assert len(service_calls) == 1 # --- Midnight sun and polar night triggers ----------------------------------- # Precomputed solar-midnight/noon horizon crossings at Svalbard, serving as an # independent oracle for the crossing scheduler (truncated to whole seconds). _POLAR_TRIGGER_CASES = [ ( "sun.midnight_sun_started", datetime(2015, 4, 1, 12, tzinfo=dt_util.UTC), datetime(2015, 4, 18, 22, 56, 36, tzinfo=dt_util.UTC), ), ( "sun.midnight_sun_ended", datetime(2015, 8, 1, 12, tzinfo=dt_util.UTC), datetime(2015, 8, 25, 22, 59, 19, tzinfo=dt_util.UTC), ), ( "sun.polar_night_started", datetime(2015, 10, 1, 12, tzinfo=dt_util.UTC), datetime(2015, 10, 28, 10, 41, 13, tzinfo=dt_util.UTC), ), ( "sun.polar_night_ended", datetime(2015, 2, 1, 12, tzinfo=dt_util.UTC), datetime(2015, 2, 15, 11, 11, 34, tzinfo=dt_util.UTC), ), ] @pytest.mark.parametrize(("trigger_key", "now", "expected"), _POLAR_TRIGGER_CASES) async def test_midnight_sun_polar_night_trigger_fires( hass: HomeAssistant, service_calls: list[ServiceCall], trigger_key: str, now: datetime, expected: datetime, ) -> None: """Test the midnight sun/polar night triggers fire at the horizon crossing.""" latitude, longitude, time_zone = _SVALBARD await hass.config.async_set_time_zone(time_zone) await hass.config.async_update(latitude=latitude, longitude=longitude, elevation=0) with freeze_time(now): await _arm_automation(hass, {"platform": trigger_key}, {}) # Two seconds' slack absorbs the sub-second part dropped from the oracle. async_fire_time_changed(hass, expected + timedelta(seconds=2)) await hass.async_block_till_done() assert len(service_calls) == 1 async def test_midnight_sun_trigger_stays_unscheduled_outside_polar( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test a midnight sun trigger arms without error and never fires off the poles. The default test location (San Diego) never has a midnight sun, so the trigger is left unscheduled rather than raising when the automation is set up. """ with freeze_time(_TEST_DATETIME): await _arm_automation(hass, {"platform": "sun.midnight_sun_started"}, {}) async_fire_time_changed(hass, _TEST_DATETIME + timedelta(days=400)) await hass.async_block_till_done() assert len(service_calls) == 0 async def test_midnight_sun_trigger_offset_catches_pending_crossing( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test a multi-day ``after`` offset still fires for an already-passed crossing. With an ``after`` offset the fire time trails the crossing, so when the trigger is (re)armed in the window between the two, the pending crossing must still be found and scheduled rather than skipped to the next year. """ latitude, longitude, time_zone = _SVALBARD await hass.config.async_set_time_zone(time_zone) await hass.config.async_update(latitude=latitude, longitude=longitude, elevation=0) # Midnight sun starts at the 2015-04-18 22:56 solar midnight; a 3-day "after" # offset pushes the fire time to 2015-04-21. now sits between the two. now = datetime(2015, 4, 20, 12, tzinfo=dt_util.UTC) expected = datetime(2015, 4, 21, 22, 56, 36, tzinfo=dt_util.UTC) with freeze_time(now): await _arm_automation( hass, { "platform": "sun.midnight_sun_started", "options": {"offset": {"days": 3}, "offset_type": "after"}, }, {}, ) async_fire_time_changed(hass, expected + timedelta(seconds=2)) await hass.async_block_till_done() assert len(service_calls) == 1 @pytest.mark.parametrize( ("latitude", "has_crossing"), [ (32.87, False), # San Diego, well below the polar circle (60.0, False), # below the polar circle (65.0, False), # at the short-circuit threshold, still no polar period (78.22, True), # Svalbard, deep polar ], ) def test_next_polar_transition_short_circuits_non_polar( latitude: float, has_crossing: bool ) -> None: """Test the crossing scan is skipped where no polar period can occur. Below the polar circles the scan can never find a crossing, so it must short-circuit rather than run its ~year-long loop in the event loop. """ observer = astral.Observer(latitude, 15.0, 0) result = _next_polar_transition( observer, "midnight", _TEST_DATETIME, target_above=True, offset=timedelta(0) ) assert (result is not None) is has_crossing def test_next_polar_transition_large_before_offset() -> None: """Test a large ``before`` offset still reaches the following annual crossing. A negative (before) offset advances the fire time, so the nearest crossing's fire can already be in the past. Anchoring the scan on ``now - offset`` (not a fixed lookback) must then carry it forward to next year's crossing instead of returning None and leaving the trigger unscheduled. """ observer = astral.Observer(*_SVALBARD[:2], 0) now = datetime(2015, 2, 27, 12, tzinfo=dt_util.UTC) # The 2015-04-18 crossing's fire (60 days earlier) is already past, so the # next fire derives from the 2016-04-17 22:56:39 solar-midnight crossing. expected = datetime(2016, 2, 17, 22, 56, 39, tzinfo=dt_util.UTC) result = _next_polar_transition( observer, "midnight", now, target_above=True, offset=timedelta(days=-60) ) assert result == expected # --- Sun elevation triggers -------------------------------------------------- async def test_elevation_changed_trigger( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the elevation changed trigger fires on any elevation change.""" await _arm_automation( hass, { "platform": "sun.elevation_changed", "options": {"threshold": {"type": "any"}}, }, {}, ) hass.states.async_set(_SUN_ENTITY_ID, "above_horizon", {"elevation": 5.0}) await hass.async_block_till_done() calls_before = len(service_calls) hass.states.async_set(_SUN_ENTITY_ID, "above_horizon", {"elevation": 6.0}) await hass.async_block_till_done() assert len(service_calls) == calls_before + 1 async def test_elevation_crossed_threshold_trigger( hass: HomeAssistant, service_calls: list[ServiceCall] ) -> None: """Test the elevation crossed threshold trigger fires only on crossing.""" await _arm_automation( hass, { "platform": "sun.elevation_crossed_threshold", "options": {"threshold": {"type": "above", "value": {"number": 10}}}, }, {}, ) hass.states.async_set(_SUN_ENTITY_ID, "above_horizon", {"elevation": 5.0}) await hass.async_block_till_done() calls_before = len(service_calls) # Crossing from below to above the threshold fires once. hass.states.async_set(_SUN_ENTITY_ID, "above_horizon", {"elevation": 15.0}) await hass.async_block_till_done() assert len(service_calls) == calls_before + 1 # Staying above the threshold does not fire again. hass.states.async_set(_SUN_ENTITY_ID, "above_horizon", {"elevation": 20.0}) await hass.async_block_till_done() assert len(service_calls) == calls_before + 1 # --- Validation -------------------------------------------------------------- @pytest.mark.parametrize( "trigger_key", [ "sun.sunrise", "sun.sunset", "sun.solar_noon", "sun.solar_midnight", "sun.dawn", "sun.dusk", "sun.golden_hour_started", "sun.golden_hour_ended", "sun.blue_hour_started", "sun.blue_hour_ended", "sun.midnight_sun_started", "sun.midnight_sun_ended", "sun.polar_night_started", "sun.polar_night_ended", ], ) async def test_event_trigger_options(hass: HomeAssistant, trigger_key: str) -> None: """Test the solar event triggers reject behavior and duration options.""" await assert_trigger_options_supported( hass, trigger_key, None, supports_behavior=False, supports_duration=False, supports_target=False, ) @pytest.mark.parametrize( ("trigger_key", "base_options", "supports_duration"), [ ("sun.elevation_changed", {"threshold": {"type": "any"}}, False), ( "sun.elevation_crossed_threshold", {"threshold": {"type": "above", "value": {"number": 10}}}, True, ), ], ) async def test_elevation_trigger_options( hass: HomeAssistant, trigger_key: str, base_options: dict[str, Any], supports_duration: bool, ) -> None: """Test the elevation triggers support the expected options without a target.""" await assert_trigger_options_supported( hass, trigger_key, base_options, supports_behavior=False, supports_duration=supports_duration, supports_target=False, ) @pytest.mark.parametrize( ("twilight", "valid"), [ ("civil", True), ("nautical", True), ("astronomical", True), ("invalid", False), ], ) async def test_dawn_dusk_twilight_validation( hass: HomeAssistant, twilight: str, valid: bool ) -> None: """Test the dawn/dusk triggers validate the twilight option.""" config = {"platform": "sun.dawn", "options": {"type": twilight}} if valid: await async_validate_trigger_config(hass, [config]) else: with pytest.raises(vol.Invalid): await async_validate_trigger_config(hass, [config]) @pytest.mark.parametrize( ("trigger_key", "period", "expectation"), [ ("sun.golden_hour_started", "any", nullcontext()), ("sun.golden_hour_started", "morning", nullcontext()), ("sun.golden_hour_started", "evening", nullcontext()), ("sun.golden_hour_started", "invalid", pytest.raises(vol.Invalid)), ("sun.blue_hour_ended", "any", nullcontext()), ("sun.blue_hour_ended", "morning", nullcontext()), ("sun.blue_hour_ended", "evening", nullcontext()), ("sun.blue_hour_ended", "invalid", pytest.raises(vol.Invalid)), ], ) async def test_golden_blue_hour_period_validation( hass: HomeAssistant, trigger_key: str, period: str, expectation: AbstractContextManager, ) -> None: """Test the golden/blue hour triggers validate the period option.""" config = {"platform": trigger_key, "options": {"period": period}} with expectation: await async_validate_trigger_config(hass, [config])