mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 12:47:08 +00:00
Little cleanup of Alert tests (#80333)
This commit is contained in:
parent
7e1a7bed69
commit
6aa47e871d
@ -28,9 +28,11 @@ from homeassistant.const import (
|
|||||||
STATE_OFF,
|
STATE_OFF,
|
||||||
STATE_ON,
|
STATE_ON,
|
||||||
)
|
)
|
||||||
from homeassistant.core import HomeAssistant, callback
|
from homeassistant.core import HomeAssistant, ServiceCall
|
||||||
from homeassistant.setup import async_setup_component
|
from homeassistant.setup import async_setup_component
|
||||||
|
|
||||||
|
from tests.common import async_mock_service
|
||||||
|
|
||||||
NAME = "alert_test"
|
NAME = "alert_test"
|
||||||
DONE_MESSAGE = "alert_gone"
|
DONE_MESSAGE = "alert_gone"
|
||||||
NOTIFIER = "test"
|
NOTIFIER = "test"
|
||||||
@ -71,49 +73,10 @@ TEST_NOACK = [
|
|||||||
ENTITY_ID = f"{DOMAIN}.{NAME}"
|
ENTITY_ID = f"{DOMAIN}.{NAME}"
|
||||||
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def async_turn_on(hass, entity_id):
|
|
||||||
"""Async reset the alert.
|
|
||||||
|
|
||||||
This is a legacy helper method. Do not use it for new tests.
|
|
||||||
"""
|
|
||||||
data = {ATTR_ENTITY_ID: entity_id}
|
|
||||||
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data))
|
|
||||||
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def async_turn_off(hass, entity_id):
|
|
||||||
"""Async acknowledge the alert.
|
|
||||||
|
|
||||||
This is a legacy helper method. Do not use it for new tests.
|
|
||||||
"""
|
|
||||||
data = {ATTR_ENTITY_ID: entity_id}
|
|
||||||
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data))
|
|
||||||
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def async_toggle(hass, entity_id):
|
|
||||||
"""Async toggle acknowledgment of alert.
|
|
||||||
|
|
||||||
This is a legacy helper method. Do not use it for new tests.
|
|
||||||
"""
|
|
||||||
data = {ATTR_ENTITY_ID: entity_id}
|
|
||||||
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TOGGLE, data))
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def mock_notifier(hass):
|
def mock_notifier(hass: HomeAssistant) -> list[ServiceCall]:
|
||||||
"""Mock for notifier."""
|
"""Mock for notifier."""
|
||||||
events = []
|
return async_mock_service(hass, notify.DOMAIN, NOTIFIER)
|
||||||
|
|
||||||
@callback
|
|
||||||
def record_event(event):
|
|
||||||
"""Add recorded event to set."""
|
|
||||||
events.append(event)
|
|
||||||
|
|
||||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
|
||||||
|
|
||||||
return events
|
|
||||||
|
|
||||||
|
|
||||||
async def test_setup(hass):
|
async def test_setup(hass):
|
||||||
@ -135,8 +98,13 @@ async def test_silence(hass, mock_notifier):
|
|||||||
assert await async_setup_component(hass, DOMAIN, TEST_CONFIG)
|
assert await async_setup_component(hass, DOMAIN, TEST_CONFIG)
|
||||||
hass.states.async_set("sensor.test", STATE_ON)
|
hass.states.async_set("sensor.test", STATE_ON)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
async_turn_off(hass, ENTITY_ID)
|
|
||||||
await hass.async_block_till_done()
|
await hass.services.async_call(
|
||||||
|
DOMAIN,
|
||||||
|
SERVICE_TURN_OFF,
|
||||||
|
{ATTR_ENTITY_ID: ENTITY_ID},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
||||||
|
|
||||||
# alert should not be silenced on next fire
|
# alert should not be silenced on next fire
|
||||||
@ -153,11 +121,22 @@ async def test_reset(hass, mock_notifier):
|
|||||||
assert await async_setup_component(hass, DOMAIN, TEST_CONFIG)
|
assert await async_setup_component(hass, DOMAIN, TEST_CONFIG)
|
||||||
hass.states.async_set("sensor.test", STATE_ON)
|
hass.states.async_set("sensor.test", STATE_ON)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
async_turn_off(hass, ENTITY_ID)
|
|
||||||
await hass.async_block_till_done()
|
await hass.services.async_call(
|
||||||
|
DOMAIN,
|
||||||
|
SERVICE_TURN_OFF,
|
||||||
|
{ATTR_ENTITY_ID: ENTITY_ID},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
|
|
||||||
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
||||||
async_turn_on(hass, ENTITY_ID)
|
|
||||||
await hass.async_block_till_done()
|
await hass.services.async_call(
|
||||||
|
DOMAIN,
|
||||||
|
SERVICE_TURN_ON,
|
||||||
|
{ATTR_ENTITY_ID: ENTITY_ID},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
||||||
|
|
||||||
|
|
||||||
@ -167,73 +146,63 @@ async def test_toggle(hass, mock_notifier):
|
|||||||
hass.states.async_set("sensor.test", STATE_ON)
|
hass.states.async_set("sensor.test", STATE_ON)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
||||||
async_toggle(hass, ENTITY_ID)
|
|
||||||
await hass.async_block_till_done()
|
await hass.services.async_call(
|
||||||
|
DOMAIN,
|
||||||
|
SERVICE_TOGGLE,
|
||||||
|
{ATTR_ENTITY_ID: ENTITY_ID},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
assert hass.states.get(ENTITY_ID).state == STATE_OFF
|
||||||
async_toggle(hass, ENTITY_ID)
|
|
||||||
await hass.async_block_till_done()
|
await hass.services.async_call(
|
||||||
|
DOMAIN,
|
||||||
|
SERVICE_TOGGLE,
|
||||||
|
{ATTR_ENTITY_ID: ENTITY_ID},
|
||||||
|
blocking=True,
|
||||||
|
)
|
||||||
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
assert hass.states.get(ENTITY_ID).state == STATE_ON
|
||||||
|
|
||||||
|
|
||||||
async def test_notification_no_done_message(hass):
|
async def test_notification_no_done_message(
|
||||||
|
hass: HomeAssistant, mock_notifier: list[ServiceCall]
|
||||||
|
) -> None:
|
||||||
"""Test notifications."""
|
"""Test notifications."""
|
||||||
events = []
|
|
||||||
config = deepcopy(TEST_CONFIG)
|
config = deepcopy(TEST_CONFIG)
|
||||||
del config[DOMAIN][NAME][CONF_DONE_MESSAGE]
|
del config[DOMAIN][NAME][CONF_DONE_MESSAGE]
|
||||||
|
|
||||||
@callback
|
|
||||||
def record_event(event):
|
|
||||||
"""Add recorded event to set."""
|
|
||||||
events.append(event)
|
|
||||||
|
|
||||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
|
||||||
|
|
||||||
assert await async_setup_component(hass, DOMAIN, config)
|
assert await async_setup_component(hass, DOMAIN, config)
|
||||||
assert len(events) == 0
|
assert len(mock_notifier) == 0
|
||||||
|
|
||||||
hass.states.async_set("sensor.test", STATE_ON)
|
hass.states.async_set("sensor.test", STATE_ON)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert len(events) == 1
|
assert len(mock_notifier) == 1
|
||||||
|
|
||||||
hass.states.async_set("sensor.test", STATE_OFF)
|
hass.states.async_set("sensor.test", STATE_OFF)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert len(events) == 1
|
assert len(mock_notifier) == 1
|
||||||
|
|
||||||
|
|
||||||
async def test_notification(hass):
|
async def test_notification(
|
||||||
|
hass: HomeAssistant, mock_notifier: list[ServiceCall]
|
||||||
|
) -> None:
|
||||||
"""Test notifications."""
|
"""Test notifications."""
|
||||||
events = []
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def record_event(event):
|
|
||||||
"""Add recorded event to set."""
|
|
||||||
events.append(event)
|
|
||||||
|
|
||||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
|
||||||
|
|
||||||
assert await async_setup_component(hass, DOMAIN, TEST_CONFIG)
|
assert await async_setup_component(hass, DOMAIN, TEST_CONFIG)
|
||||||
assert len(events) == 0
|
assert len(mock_notifier) == 0
|
||||||
|
|
||||||
hass.states.async_set("sensor.test", STATE_ON)
|
hass.states.async_set("sensor.test", STATE_ON)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert len(events) == 1
|
assert len(mock_notifier) == 1
|
||||||
|
|
||||||
hass.states.async_set("sensor.test", STATE_OFF)
|
hass.states.async_set("sensor.test", STATE_OFF)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert len(events) == 2
|
assert len(mock_notifier) == 2
|
||||||
|
|
||||||
|
|
||||||
async def test_no_notifiers(hass: HomeAssistant) -> None:
|
async def test_no_notifiers(
|
||||||
|
hass: HomeAssistant, mock_notifier: list[ServiceCall]
|
||||||
|
) -> None:
|
||||||
"""Test we send no notifications when there are not no."""
|
"""Test we send no notifications when there are not no."""
|
||||||
events = []
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def record_event(event):
|
|
||||||
"""Add recorded event to set."""
|
|
||||||
events.append(event)
|
|
||||||
|
|
||||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
|
||||||
|
|
||||||
assert await async_setup_component(
|
assert await async_setup_component(
|
||||||
hass,
|
hass,
|
||||||
DOMAIN,
|
DOMAIN,
|
||||||
@ -248,15 +217,15 @@ async def test_no_notifiers(hass: HomeAssistant) -> None:
|
|||||||
}
|
}
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
assert len(events) == 0
|
assert len(mock_notifier) == 0
|
||||||
|
|
||||||
hass.states.async_set("sensor.test", STATE_ON)
|
hass.states.async_set("sensor.test", STATE_ON)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert len(events) == 0
|
assert len(mock_notifier) == 0
|
||||||
|
|
||||||
hass.states.async_set("sensor.test", STATE_OFF)
|
hass.states.async_set("sensor.test", STATE_OFF)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert len(events) == 0
|
assert len(mock_notifier) == 0
|
||||||
|
|
||||||
|
|
||||||
async def test_sending_non_templated_notification(hass, mock_notifier):
|
async def test_sending_non_templated_notification(hass, mock_notifier):
|
||||||
@ -324,25 +293,16 @@ async def test_sending_data_notification(hass, mock_notifier):
|
|||||||
assert last_event.data[notify.ATTR_DATA] == TEST_DATA
|
assert last_event.data[notify.ATTR_DATA] == TEST_DATA
|
||||||
|
|
||||||
|
|
||||||
async def test_skipfirst(hass):
|
async def test_skipfirst(hass: HomeAssistant, mock_notifier: list[ServiceCall]) -> None:
|
||||||
"""Test skipping first notification."""
|
"""Test skipping first notification."""
|
||||||
config = deepcopy(TEST_CONFIG)
|
config = deepcopy(TEST_CONFIG)
|
||||||
config[DOMAIN][NAME][CONF_SKIP_FIRST] = True
|
config[DOMAIN][NAME][CONF_SKIP_FIRST] = True
|
||||||
events = []
|
|
||||||
|
|
||||||
@callback
|
|
||||||
def record_event(event):
|
|
||||||
"""Add recorded event to set."""
|
|
||||||
events.append(event)
|
|
||||||
|
|
||||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
|
||||||
|
|
||||||
assert await async_setup_component(hass, DOMAIN, config)
|
assert await async_setup_component(hass, DOMAIN, config)
|
||||||
assert len(events) == 0
|
assert len(mock_notifier) == 0
|
||||||
|
|
||||||
hass.states.async_set("sensor.test", STATE_ON)
|
hass.states.async_set("sensor.test", STATE_ON)
|
||||||
await hass.async_block_till_done()
|
await hass.async_block_till_done()
|
||||||
assert len(events) == 0
|
assert len(mock_notifier) == 0
|
||||||
|
|
||||||
|
|
||||||
async def test_done_message_state_tracker_reset_on_cancel(hass):
|
async def test_done_message_state_tracker_reset_on_cancel(hass):
|
||||||
|
Loading…
x
Reference in New Issue
Block a user