mirror of
https://github.com/home-assistant/core.git
synced 2025-07-16 01:37:08 +00:00
Rewrite alert test to pytest (#41102)
* Rewrite alert test to pytest * Rewrite alert test to pytest. Add mock_notifier fixture
This commit is contained in:
Alexander Pitkin
GitHub
parent
b8417a2ce2
commit
f5f4570a29
@ -1,7 +1,8 @@
|
||||
"""The tests for the Alert component."""
|
||||
# pylint: disable=protected-access
|
||||
from copy import deepcopy
|
||||
import unittest
|
||||
|
||||
import pytest
|
||||
|
||||
import homeassistant.components.alert as alert
|
||||
from homeassistant.components.alert import DOMAIN
|
||||
@ -19,9 +20,7 @@ from homeassistant.const import (
|
||||
STATE_ON,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.setup import setup_component
|
||||
|
||||
from tests.common import get_test_home_assistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
NAME = "alert_test"
|
||||
DONE_MESSAGE = "alert_gone"
|
||||
@ -63,14 +62,6 @@ TEST_NOACK = [
|
||||
ENTITY_ID = f"{alert.DOMAIN}.{NAME}"
|
||||
|
||||
|
||||
def turn_on(hass, entity_id):
|
||||
"""Reset the alert.
|
||||
|
||||
This is a legacy helper method. Do not use it for new tests.
|
||||
"""
|
||||
hass.add_job(async_turn_on, hass, entity_id)
|
||||
|
||||
|
||||
@callback
|
||||
def async_turn_on(hass, entity_id):
|
||||
"""Async reset the alert.
|
||||
@ -81,14 +72,6 @@ def async_turn_on(hass, entity_id):
|
||||
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data))
|
||||
|
||||
|
||||
def turn_off(hass, entity_id):
|
||||
"""Acknowledge alert.
|
||||
|
||||
This is a legacy helper method. Do not use it for new tests.
|
||||
"""
|
||||
hass.add_job(async_turn_off, hass, entity_id)
|
||||
|
||||
|
||||
@callback
|
||||
def async_turn_off(hass, entity_id):
|
||||
"""Async acknowledge the alert.
|
||||
@ -99,14 +82,6 @@ def async_turn_off(hass, entity_id):
|
||||
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data))
|
||||
|
||||
|
||||
def toggle(hass, entity_id):
|
||||
"""Toggle acknowledgment of alert.
|
||||
|
||||
This is a legacy helper method. Do not use it for new tests.
|
||||
"""
|
||||
hass.add_job(async_toggle, hass, entity_id)
|
||||
|
||||
|
||||
@callback
|
||||
def async_toggle(hass, entity_id):
|
||||
"""Async toggle acknowledgment of alert.
|
||||
@ -117,239 +92,237 @@ def async_toggle(hass, entity_id):
|
||||
hass.async_create_task(hass.services.async_call(DOMAIN, SERVICE_TOGGLE, data))
|
||||
|
||||
|
||||
# pylint: disable=invalid-name
|
||||
class TestAlert(unittest.TestCase):
|
||||
"""Test the alert module."""
|
||||
@pytest.fixture
|
||||
def mock_notifier(hass):
|
||||
"""Mock for notifier."""
|
||||
events = []
|
||||
|
||||
def setUp(self):
|
||||
"""Set up things to be run when tests are started."""
|
||||
self.hass = get_test_home_assistant()
|
||||
self._setup_notify()
|
||||
self.addCleanup(self.hass.stop)
|
||||
@callback
|
||||
def record_event(event):
|
||||
"""Add recorded event to set."""
|
||||
events.append(event)
|
||||
|
||||
def _setup_notify(self):
|
||||
events = []
|
||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
||||
|
||||
@callback
|
||||
def record_event(event):
|
||||
"""Add recorded event to set."""
|
||||
events.append(event)
|
||||
return events
|
||||
|
||||
self.hass.services.register(notify.DOMAIN, NOTIFIER, record_event)
|
||||
|
||||
return events
|
||||
async def test_is_on(hass):
|
||||
"""Test is_on method."""
|
||||
hass.states.async_set(ENTITY_ID, STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert alert.is_on(hass, ENTITY_ID)
|
||||
hass.states.async_set(ENTITY_ID, STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert not alert.is_on(hass, ENTITY_ID)
|
||||
|
||||
def test_is_on(self):
|
||||
"""Test is_on method."""
|
||||
self.hass.states.set(ENTITY_ID, STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
assert alert.is_on(self.hass, ENTITY_ID)
|
||||
self.hass.states.set(ENTITY_ID, STATE_OFF)
|
||||
self.hass.block_till_done()
|
||||
assert not alert.is_on(self.hass, ENTITY_ID)
|
||||
|
||||
def test_setup(self):
|
||||
"""Test setup method."""
|
||||
assert setup_component(self.hass, alert.DOMAIN, TEST_CONFIG)
|
||||
assert STATE_IDLE == self.hass.states.get(ENTITY_ID).state
|
||||
async def test_setup(hass):
|
||||
"""Test setup method."""
|
||||
assert await async_setup_component(hass, alert.DOMAIN, TEST_CONFIG)
|
||||
assert STATE_IDLE == hass.states.get(ENTITY_ID).state
|
||||
|
||||
def test_fire(self):
|
||||
"""Test the alert firing."""
|
||||
assert setup_component(self.hass, alert.DOMAIN, TEST_CONFIG)
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_ON == self.hass.states.get(ENTITY_ID).state
|
||||
|
||||
def test_silence(self):
|
||||
"""Test silencing the alert."""
|
||||
assert setup_component(self.hass, alert.DOMAIN, TEST_CONFIG)
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
turn_off(self.hass, ENTITY_ID)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_OFF == self.hass.states.get(ENTITY_ID).state
|
||||
async def test_fire(hass, mock_notifier):
|
||||
"""Test the alert firing."""
|
||||
assert await async_setup_component(hass, alert.DOMAIN, TEST_CONFIG)
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_ON == hass.states.get(ENTITY_ID).state
|
||||
|
||||
# alert should not be silenced on next fire
|
||||
self.hass.states.set("sensor.test", STATE_OFF)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_IDLE == self.hass.states.get(ENTITY_ID).state
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_ON == self.hass.states.get(ENTITY_ID).state
|
||||
|
||||
def test_reset(self):
|
||||
"""Test resetting the alert."""
|
||||
assert setup_component(self.hass, alert.DOMAIN, TEST_CONFIG)
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
turn_off(self.hass, ENTITY_ID)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_OFF == self.hass.states.get(ENTITY_ID).state
|
||||
turn_on(self.hass, ENTITY_ID)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_ON == self.hass.states.get(ENTITY_ID).state
|
||||
async def test_silence(hass, mock_notifier):
|
||||
"""Test silencing the alert."""
|
||||
assert await async_setup_component(hass, alert.DOMAIN, TEST_CONFIG)
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
async_turn_off(hass, ENTITY_ID)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_OFF == hass.states.get(ENTITY_ID).state
|
||||
|
||||
def test_toggle(self):
|
||||
"""Test toggling alert."""
|
||||
assert setup_component(self.hass, alert.DOMAIN, TEST_CONFIG)
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_ON == self.hass.states.get(ENTITY_ID).state
|
||||
toggle(self.hass, ENTITY_ID)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_OFF == self.hass.states.get(ENTITY_ID).state
|
||||
toggle(self.hass, ENTITY_ID)
|
||||
self.hass.block_till_done()
|
||||
assert STATE_ON == self.hass.states.get(ENTITY_ID).state
|
||||
# alert should not be silenced on next fire
|
||||
hass.states.async_set("sensor.test", STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_IDLE == hass.states.get(ENTITY_ID).state
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_ON == hass.states.get(ENTITY_ID).state
|
||||
|
||||
def test_notification_no_done_message(self):
|
||||
"""Test notifications."""
|
||||
events = []
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
del config[alert.DOMAIN][NAME][alert.CONF_DONE_MESSAGE]
|
||||
|
||||
@callback
|
||||
def record_event(event):
|
||||
"""Add recorded event to set."""
|
||||
events.append(event)
|
||||
async def test_reset(hass, mock_notifier):
|
||||
"""Test resetting the alert."""
|
||||
assert await async_setup_component(hass, alert.DOMAIN, TEST_CONFIG)
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
async_turn_off(hass, ENTITY_ID)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_OFF == hass.states.get(ENTITY_ID).state
|
||||
async_turn_on(hass, ENTITY_ID)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_ON == hass.states.get(ENTITY_ID).state
|
||||
|
||||
self.hass.services.register(notify.DOMAIN, NOTIFIER, record_event)
|
||||
|
||||
assert setup_component(self.hass, alert.DOMAIN, config)
|
||||
assert len(events) == 0
|
||||
async def test_toggle(hass, mock_notifier):
|
||||
"""Test toggling alert."""
|
||||
assert await async_setup_component(hass, alert.DOMAIN, TEST_CONFIG)
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_ON == hass.states.get(ENTITY_ID).state
|
||||
async_toggle(hass, ENTITY_ID)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_OFF == hass.states.get(ENTITY_ID).state
|
||||
async_toggle(hass, ENTITY_ID)
|
||||
await hass.async_block_till_done()
|
||||
assert STATE_ON == hass.states.get(ENTITY_ID).state
|
||||
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
assert len(events) == 1
|
||||
|
||||
self.hass.states.set("sensor.test", STATE_OFF)
|
||||
self.hass.block_till_done()
|
||||
assert len(events) == 1
|
||||
async def test_notification_no_done_message(hass):
|
||||
"""Test notifications."""
|
||||
events = []
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
del config[alert.DOMAIN][NAME][alert.CONF_DONE_MESSAGE]
|
||||
|
||||
def test_notification(self):
|
||||
"""Test notifications."""
|
||||
events = []
|
||||
@callback
|
||||
def record_event(event):
|
||||
"""Add recorded event to set."""
|
||||
events.append(event)
|
||||
|
||||
@callback
|
||||
def record_event(event):
|
||||
"""Add recorded event to set."""
|
||||
events.append(event)
|
||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
||||
|
||||
self.hass.services.register(notify.DOMAIN, NOTIFIER, record_event)
|
||||
assert await async_setup_component(hass, alert.DOMAIN, config)
|
||||
assert len(events) == 0
|
||||
|
||||
assert setup_component(self.hass, alert.DOMAIN, TEST_CONFIG)
|
||||
assert len(events) == 0
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert len(events) == 1
|
||||
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
assert len(events) == 1
|
||||
hass.states.async_set("sensor.test", STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert len(events) == 1
|
||||
|
||||
self.hass.states.set("sensor.test", STATE_OFF)
|
||||
self.hass.block_till_done()
|
||||
assert len(events) == 2
|
||||
|
||||
def test_sending_non_templated_notification(self):
|
||||
"""Test notifications."""
|
||||
events = self._setup_notify()
|
||||
async def test_notification(hass):
|
||||
"""Test notifications."""
|
||||
events = []
|
||||
|
||||
assert setup_component(self.hass, alert.DOMAIN, TEST_CONFIG)
|
||||
@callback
|
||||
def record_event(event):
|
||||
"""Add recorded event to set."""
|
||||
events.append(event)
|
||||
|
||||
self.hass.states.set(TEST_ENTITY, STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
self.assertEqual(1, len(events))
|
||||
last_event = events[-1]
|
||||
self.assertEqual(last_event.data[notify.ATTR_MESSAGE], NAME)
|
||||
hass.services.async_register(notify.DOMAIN, NOTIFIER, record_event)
|
||||
|
||||
def test_sending_templated_notification(self):
|
||||
"""Test templated notification."""
|
||||
events = self._setup_notify()
|
||||
assert await async_setup_component(hass, alert.DOMAIN, TEST_CONFIG)
|
||||
assert len(events) == 0
|
||||
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_ALERT_MESSAGE] = TEMPLATE
|
||||
assert setup_component(self.hass, alert.DOMAIN, config)
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert len(events) == 1
|
||||
|
||||
self.hass.states.set(TEST_ENTITY, STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
self.assertEqual(1, len(events))
|
||||
last_event = events[-1]
|
||||
self.assertEqual(last_event.data[notify.ATTR_MESSAGE], TEST_ENTITY)
|
||||
hass.states.async_set("sensor.test", STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert len(events) == 2
|
||||
|
||||
def test_sending_templated_done_notification(self):
|
||||
"""Test templated notification."""
|
||||
events = self._setup_notify()
|
||||
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_DONE_MESSAGE] = TEMPLATE
|
||||
assert setup_component(self.hass, alert.DOMAIN, config)
|
||||
async def test_sending_non_templated_notification(hass, mock_notifier):
|
||||
"""Test notifications."""
|
||||
assert await async_setup_component(hass, alert.DOMAIN, TEST_CONFIG)
|
||||
|
||||
self.hass.states.set(TEST_ENTITY, STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
self.hass.states.set(TEST_ENTITY, STATE_OFF)
|
||||
self.hass.block_till_done()
|
||||
self.assertEqual(2, len(events))
|
||||
last_event = events[-1]
|
||||
self.assertEqual(last_event.data[notify.ATTR_MESSAGE], TEST_ENTITY)
|
||||
hass.states.async_set(TEST_ENTITY, STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_notifier) == 1
|
||||
last_event = mock_notifier[-1]
|
||||
assert last_event.data[notify.ATTR_MESSAGE] == NAME
|
||||
|
||||
def test_sending_titled_notification(self):
|
||||
"""Test notifications."""
|
||||
events = self._setup_notify()
|
||||
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_TITLE] = TITLE
|
||||
assert setup_component(self.hass, alert.DOMAIN, config)
|
||||
async def test_sending_templated_notification(hass, mock_notifier):
|
||||
"""Test templated notification."""
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_ALERT_MESSAGE] = TEMPLATE
|
||||
assert await async_setup_component(hass, alert.DOMAIN, config)
|
||||
|
||||
self.hass.states.set(TEST_ENTITY, STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
self.assertEqual(1, len(events))
|
||||
last_event = events[-1]
|
||||
self.assertEqual(last_event.data[notify.ATTR_TITLE], TEST_TITLE)
|
||||
hass.states.async_set(TEST_ENTITY, STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_notifier) == 1
|
||||
last_event = mock_notifier[-1]
|
||||
assert last_event.data[notify.ATTR_MESSAGE] == TEST_ENTITY
|
||||
|
||||
def test_sending_data_notification(self):
|
||||
"""Test notifications."""
|
||||
events = self._setup_notify()
|
||||
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_DATA] = TEST_DATA
|
||||
assert setup_component(self.hass, alert.DOMAIN, config)
|
||||
async def test_sending_templated_done_notification(hass, mock_notifier):
|
||||
"""Test templated notification."""
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_DONE_MESSAGE] = TEMPLATE
|
||||
assert await async_setup_component(hass, alert.DOMAIN, config)
|
||||
|
||||
self.hass.states.set(TEST_ENTITY, STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
self.assertEqual(1, len(events))
|
||||
last_event = events[-1]
|
||||
self.assertEqual(last_event.data[notify.ATTR_DATA], TEST_DATA)
|
||||
hass.states.async_set(TEST_ENTITY, STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
hass.states.async_set(TEST_ENTITY, STATE_OFF)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_notifier) == 2
|
||||
last_event = mock_notifier[-1]
|
||||
assert last_event.data[notify.ATTR_MESSAGE] == TEST_ENTITY
|
||||
|
||||
def test_skipfirst(self):
|
||||
"""Test skipping first notification."""
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_SKIP_FIRST] = True
|
||||
events = []
|
||||
|
||||
@callback
|
||||
def record_event(event):
|
||||
"""Add recorded event to set."""
|
||||
events.append(event)
|
||||
async def test_sending_titled_notification(hass, mock_notifier):
|
||||
"""Test notifications."""
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_TITLE] = TITLE
|
||||
assert await async_setup_component(hass, alert.DOMAIN, config)
|
||||
|
||||
self.hass.services.register(notify.DOMAIN, NOTIFIER, record_event)
|
||||
hass.states.async_set(TEST_ENTITY, STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_notifier) == 1
|
||||
last_event = mock_notifier[-1]
|
||||
assert last_event.data[notify.ATTR_TITLE] == TEST_TITLE
|
||||
|
||||
assert setup_component(self.hass, alert.DOMAIN, config)
|
||||
assert len(events) == 0
|
||||
|
||||
self.hass.states.set("sensor.test", STATE_ON)
|
||||
self.hass.block_till_done()
|
||||
assert len(events) == 0
|
||||
async def test_sending_data_notification(hass, mock_notifier):
|
||||
"""Test notifications."""
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.CONF_DATA] = TEST_DATA
|
||||
assert await async_setup_component(hass, alert.DOMAIN, config)
|
||||
|
||||
def test_noack(self):
|
||||
"""Test no ack feature."""
|
||||
entity = alert.Alert(self.hass, *TEST_NOACK)
|
||||
self.hass.add_job(entity.begin_alerting)
|
||||
self.hass.block_till_done()
|
||||
hass.states.async_set(TEST_ENTITY, STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert len(mock_notifier) == 1
|
||||
last_event = mock_notifier[-1]
|
||||
assert last_event.data[notify.ATTR_DATA] == TEST_DATA
|
||||
|
||||
def test_done_message_state_tracker_reset_on_cancel(self):
|
||||
"""Test that the done message is reset when canceled."""
|
||||
entity = alert.Alert(self.hass, *TEST_NOACK)
|
||||
entity._cancel = lambda *args: None
|
||||
assert entity._send_done_message is False
|
||||
entity._send_done_message = True
|
||||
self.hass.add_job(entity.end_alerting)
|
||||
self.hass.block_till_done()
|
||||
assert entity._send_done_message is False
|
||||
|
||||
async def test_skipfirst(hass):
|
||||
"""Test skipping first notification."""
|
||||
config = deepcopy(TEST_CONFIG)
|
||||
config[alert.DOMAIN][NAME][alert.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, alert.DOMAIN, config)
|
||||
assert len(events) == 0
|
||||
|
||||
hass.states.async_set("sensor.test", STATE_ON)
|
||||
await hass.async_block_till_done()
|
||||
assert len(events) == 0
|
||||
|
||||
|
||||
async def test_noack(hass):
|
||||
"""Test no ack feature."""
|
||||
entity = alert.Alert(hass, *TEST_NOACK)
|
||||
hass.async_add_job(entity.begin_alerting)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def test_done_message_state_tracker_reset_on_cancel(hass):
|
||||
"""Test that the done message is reset when canceled."""
|
||||
entity = alert.Alert(hass, *TEST_NOACK)
|
||||
entity._cancel = lambda *args: None
|
||||
assert entity._send_done_message is False
|
||||
entity._send_done_message = True
|
||||
hass.async_add_job(entity.end_alerting)
|
||||
await hass.async_block_till_done()
|
||||
assert entity._send_done_message is False
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user