From f5f4570a2943dab2a75f2bd6141851e791874327 Mon Sep 17 00:00:00 2001 From: Alexander Pitkin Date: Tue, 20 Oct 2020 00:26:40 +0300 Subject: [PATCH] Rewrite alert test to pytest (#41102) * Rewrite alert test to pytest * Rewrite alert test to pytest. Add mock_notifier fixture --- tests/components/alert/test_init.py | 409 +++++++++++++--------------- 1 file changed, 191 insertions(+), 218 deletions(-) diff --git a/tests/components/alert/test_init.py b/tests/components/alert/test_init.py index 60c2bd6b809..3796322a5b5 100644 --- a/tests/components/alert/test_init.py +++ b/tests/components/alert/test_init.py @@ -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