diff --git a/tests/common.py b/tests/common.py index 93d70f0e12c..dfe832cc4ce 100644 --- a/tests/common.py +++ b/tests/common.py @@ -23,7 +23,7 @@ from homeassistant.auth import ( providers as auth_providers, ) from homeassistant.auth.permissions import system_policies -from homeassistant.components import mqtt, recorder +from homeassistant.components import recorder from homeassistant.components.device_automation import ( # noqa: F401 _async_get_device_automation_capabilities as async_get_device_automation_capabilities, _async_get_device_automations as async_get_device_automations, @@ -53,13 +53,13 @@ from homeassistant.helpers import ( storage, ) from homeassistant.helpers.json import JSONEncoder -from homeassistant.setup import async_setup_component, setup_component +from homeassistant.setup import setup_component from homeassistant.util.async_ import run_callback_threadsafe import homeassistant.util.dt as date_util from homeassistant.util.unit_system import METRIC_SYSTEM import homeassistant.util.yaml.loader as yaml_loader -from tests.async_mock import AsyncMock, MagicMock, Mock, patch +from tests.async_mock import AsyncMock, Mock, patch _LOGGER = logging.getLogger(__name__) INSTANCES = [] @@ -324,39 +324,6 @@ def mock_state_change_event(hass, new_state, old_state=None): hass.bus.fire(EVENT_STATE_CHANGED, event_data, context=new_state.context) -async def async_mock_mqtt_component(hass, config=None): - """Mock the MQTT component.""" - if config is None: - config = {mqtt.CONF_BROKER: "mock-broker"} - - @ha.callback - def _async_fire_mqtt_message(topic, payload, qos, retain): - async_fire_mqtt_message(hass, topic, payload, qos, retain) - - with patch("paho.mqtt.client.Client") as mock_client: - mock_client = mock_client.return_value - mock_client.connect.return_value = 0 - mock_client.subscribe.return_value = (0, 0) - mock_client.unsubscribe.return_value = (0, 0) - mock_client.publish.side_effect = _async_fire_mqtt_message - - result = await async_setup_component(hass, mqtt.DOMAIN, {mqtt.DOMAIN: config}) - assert result - await hass.async_block_till_done() - - mqtt_component_mock = MagicMock( - spec_set=hass.data["mqtt"], wraps=hass.data["mqtt"] - ) - hass.data["mqtt"].connected = mqtt_component_mock.connected - mqtt_component_mock._mqttc = mock_client - - hass.data["mqtt"] = mqtt_component_mock - return hass.data["mqtt"] - - -mock_mqtt_component = threadsafe_coroutine_factory(async_mock_mqtt_component) - - @ha.callback def mock_component(hass, component): """Mock a component is setup.""" diff --git a/tests/components/alarm_control_panel/common.py b/tests/components/alarm_control_panel/common.py index ce0bde0517c..fa50a1aab41 100644 --- a/tests/components/alarm_control_panel/common.py +++ b/tests/components/alarm_control_panel/common.py @@ -15,7 +15,6 @@ from homeassistant.const import ( SERVICE_ALARM_DISARM, SERVICE_ALARM_TRIGGER, ) -from homeassistant.loader import bind_hass async def async_alarm_disarm(hass, code=None, entity_id=ENTITY_MATCH_ALL): @@ -29,18 +28,6 @@ async def async_alarm_disarm(hass, code=None, entity_id=ENTITY_MATCH_ALL): await hass.services.async_call(DOMAIN, SERVICE_ALARM_DISARM, data, blocking=True) -@bind_hass -def alarm_disarm(hass, code=None, entity_id=ENTITY_MATCH_ALL): - """Send the alarm the command for disarm.""" - data = {} - if code: - data[ATTR_CODE] = code - if entity_id: - data[ATTR_ENTITY_ID] = entity_id - - hass.services.call(DOMAIN, SERVICE_ALARM_DISARM, data) - - async def async_alarm_arm_home(hass, code=None, entity_id=ENTITY_MATCH_ALL): """Send the alarm the command for disarm.""" data = {} @@ -52,18 +39,6 @@ async def async_alarm_arm_home(hass, code=None, entity_id=ENTITY_MATCH_ALL): await hass.services.async_call(DOMAIN, SERVICE_ALARM_ARM_HOME, data, blocking=True) -@bind_hass -def alarm_arm_home(hass, code=None, entity_id=ENTITY_MATCH_ALL): - """Send the alarm the command for arm home.""" - data = {} - if code: - data[ATTR_CODE] = code - if entity_id: - data[ATTR_ENTITY_ID] = entity_id - - hass.services.call(DOMAIN, SERVICE_ALARM_ARM_HOME, data) - - async def async_alarm_arm_away(hass, code=None, entity_id=ENTITY_MATCH_ALL): """Send the alarm the command for disarm.""" data = {} @@ -75,18 +50,6 @@ async def async_alarm_arm_away(hass, code=None, entity_id=ENTITY_MATCH_ALL): await hass.services.async_call(DOMAIN, SERVICE_ALARM_ARM_AWAY, data, blocking=True) -@bind_hass -def alarm_arm_away(hass, code=None, entity_id=ENTITY_MATCH_ALL): - """Send the alarm the command for arm away.""" - data = {} - if code: - data[ATTR_CODE] = code - if entity_id: - data[ATTR_ENTITY_ID] = entity_id - - hass.services.call(DOMAIN, SERVICE_ALARM_ARM_AWAY, data) - - async def async_alarm_arm_night(hass, code=None, entity_id=ENTITY_MATCH_ALL): """Send the alarm the command for disarm.""" data = {} @@ -98,18 +61,6 @@ async def async_alarm_arm_night(hass, code=None, entity_id=ENTITY_MATCH_ALL): await hass.services.async_call(DOMAIN, SERVICE_ALARM_ARM_NIGHT, data, blocking=True) -@bind_hass -def alarm_arm_night(hass, code=None, entity_id=ENTITY_MATCH_ALL): - """Send the alarm the command for arm night.""" - data = {} - if code: - data[ATTR_CODE] = code - if entity_id: - data[ATTR_ENTITY_ID] = entity_id - - hass.services.call(DOMAIN, SERVICE_ALARM_ARM_NIGHT, data) - - async def async_alarm_trigger(hass, code=None, entity_id=ENTITY_MATCH_ALL): """Send the alarm the command for disarm.""" data = {} @@ -121,18 +72,6 @@ async def async_alarm_trigger(hass, code=None, entity_id=ENTITY_MATCH_ALL): await hass.services.async_call(DOMAIN, SERVICE_ALARM_TRIGGER, data, blocking=True) -@bind_hass -def alarm_trigger(hass, code=None, entity_id=ENTITY_MATCH_ALL): - """Send the alarm the command for trigger.""" - data = {} - if code: - data[ATTR_CODE] = code - if entity_id: - data[ATTR_ENTITY_ID] = entity_id - - hass.services.call(DOMAIN, SERVICE_ALARM_TRIGGER, data) - - async def async_alarm_arm_custom_bypass(hass, code=None, entity_id=ENTITY_MATCH_ALL): """Send the alarm the command for disarm.""" data = {} @@ -144,15 +83,3 @@ async def async_alarm_arm_custom_bypass(hass, code=None, entity_id=ENTITY_MATCH_ await hass.services.async_call( DOMAIN, SERVICE_ALARM_ARM_CUSTOM_BYPASS, data, blocking=True ) - - -@bind_hass -def alarm_arm_custom_bypass(hass, code=None, entity_id=ENTITY_MATCH_ALL): - """Send the alarm the command for arm custom bypass.""" - data = {} - if code: - data[ATTR_CODE] = code - if entity_id: - data[ATTR_ENTITY_ID] = entity_id - - hass.services.call(DOMAIN, SERVICE_ALARM_ARM_CUSTOM_BYPASS, data) diff --git a/tests/components/manual_mqtt/test_alarm_control_panel.py b/tests/components/manual_mqtt/test_alarm_control_panel.py index 996811224c6..aa318d61b84 100644 --- a/tests/components/manual_mqtt/test_alarm_control_panel.py +++ b/tests/components/manual_mqtt/test_alarm_control_panel.py @@ -1,6 +1,5 @@ """The tests for the manual_mqtt Alarm Control Panel component.""" from datetime import timedelta -import unittest from homeassistant.components import alarm_control_panel from homeassistant.const import ( @@ -11,1772 +10,1698 @@ from homeassistant.const import ( STATE_ALARM_PENDING, STATE_ALARM_TRIGGERED, ) -from homeassistant.setup import setup_component +from homeassistant.setup import async_setup_component import homeassistant.util.dt as dt_util -from tests.async_mock import Mock, patch +from tests.async_mock import patch from tests.common import ( assert_setup_component, - fire_mqtt_message, - fire_time_changed, - get_test_home_assistant, - mock_mqtt_component, + async_fire_mqtt_message, + async_fire_time_changed, ) from tests.components.alarm_control_panel import common CODE = "HELLO_CODE" -class TestAlarmControlPanelManualMqtt(unittest.TestCase): - """Test the manual_mqtt alarm module.""" - - def setUp(self): # pylint: disable=invalid-name - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() - self.hass.config_entries._async_schedule_save = Mock() - self.mock_publish = mock_mqtt_component(self.hass) - self.addCleanup(self.tear_down_cleanup) - - def tear_down_cleanup(self): - """Stop down everything that was started.""" - self.hass.stop() - - def test_fail_setup_without_state_topic(self): - """Test for failing with no state topic.""" - with assert_setup_component(0) as config: - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - alarm_control_panel.DOMAIN: { - "platform": "mqtt_alarm", - "command_topic": "alarm/command", - } - }, - ) - assert not config[alarm_control_panel.DOMAIN] - - def test_fail_setup_without_command_topic(self): - """Test failing with no command topic.""" - with assert_setup_component(0): - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - alarm_control_panel.DOMAIN: { - "platform": "mqtt_alarm", - "state_topic": "alarm/state", - } - }, - ) - - def test_arm_home_no_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, +async def test_fail_setup_without_state_topic(hass, mqtt_mock): + """Test for failing with no state topic.""" + with assert_setup_component(0) as config: + assert await async_setup_component( + hass, alarm_control_panel.DOMAIN, { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 0, - "disarm_after_trigger": False, + alarm_control_panel.DOMAIN: { + "platform": "mqtt_alarm", "command_topic": "alarm/command", + } + }, + ) + assert not config[alarm_control_panel.DOMAIN] + + +async def test_fail_setup_without_command_topic(hass, mqtt_mock): + """Test failing with no command topic.""" + with assert_setup_component(0): + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + alarm_control_panel.DOMAIN: { + "platform": "mqtt_alarm", "state_topic": "alarm/state", } }, ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_home(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_HOME == self.hass.states.get(entity_id).state - - def test_arm_home_no_pending_when_code_not_req(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "code_arm_required": False, - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_home(self.hass, 0) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_HOME == self.hass.states.get(entity_id).state - - def test_arm_home_with_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 1, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_home(self.hass, CODE, entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - state = self.hass.states.get(entity_id) - assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_HOME - - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_HOME == self.hass.states.get(entity_id).state - - def test_arm_home_with_invalid_code(self): - """Attempt to arm home without a valid code.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 1, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_home(self.hass, f"{CODE}2") - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_arm_away_no_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE, entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - def test_arm_away_no_pending_when_code_not_req(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code_arm_required": False, - "code": CODE, - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, 0, entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - def test_arm_home_with_template_code(self): - """Attempt to arm with a template-based code.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code_template": '{{ "abc" }}', - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_home(self.hass, "abc") - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_ARMED_HOME == state.state - - def test_arm_away_with_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 1, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - state = self.hass.states.get(entity_id) - assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_AWAY - - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - def test_arm_away_with_invalid_code(self): - """Attempt to arm away without a valid code.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 1, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, f"{CODE}2") - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_arm_night_no_pending(self): - """Test arm night method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_night(self.hass, CODE, entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_NIGHT == self.hass.states.get(entity_id).state - - def test_arm_night_no_pending_when_code_not_req(self): - """Test arm night method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code_arm_required": False, - "code": CODE, - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_night(self.hass, 0, entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_NIGHT == self.hass.states.get(entity_id).state - - def test_arm_night_with_pending(self): - """Test arm night method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 1, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_night(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - state = self.hass.states.get(entity_id) - assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_NIGHT - - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_NIGHT == self.hass.states.get(entity_id).state - - # Do not go to the pending state when updating to the same state - common.alarm_arm_night(self.hass, CODE, entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_NIGHT == self.hass.states.get(entity_id).state - - def test_arm_night_with_invalid_code(self): - """Attempt to arm night without a valid code.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 1, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_night(self.hass, f"{CODE}2") - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_trigger_no_pending(self): - """Test triggering when no pending submitted method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "trigger_time": 1, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=60) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - def test_trigger_with_delay(self): - """Test trigger method and switch from pending to triggered.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "delay_time": 1, - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) + + +async def test_arm_home_no_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_home(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_HOME == hass.states.get(entity_id).state + + +async def test_arm_home_no_pending_when_code_not_req(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "code_arm_required": False, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_home(hass, 0) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_HOME == hass.states.get(entity_id).state + + +async def test_arm_home_with_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 1, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_home(hass, CODE, entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + state = hass.states.get(entity_id) + assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_HOME + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_HOME == hass.states.get(entity_id).state + + +async def test_arm_home_with_invalid_code(hass, mqtt_mock): + """Attempt to arm home without a valid code.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 1, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_home(hass, f"{CODE}2") + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_arm_away_no_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE, entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + +async def test_arm_away_no_pending_when_code_not_req(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code_arm_required": False, + "code": CODE, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, 0, entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + +async def test_arm_home_with_template_code(hass, mqtt_mock): + """Attempt to arm with a template-based code.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code_template": '{{ "abc" }}', + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_home(hass, "abc") + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_ARMED_HOME == state.state + + +async def test_arm_away_with_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 1, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + state = hass.states.get(entity_id) + assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_AWAY + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + +async def test_arm_away_with_invalid_code(hass, mqtt_mock): + """Attempt to arm away without a valid code.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 1, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, f"{CODE}2") + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_arm_night_no_pending(hass, mqtt_mock): + """Test arm night method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_night(hass, CODE, entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_NIGHT == hass.states.get(entity_id).state + + +async def test_arm_night_no_pending_when_code_not_req(hass, mqtt_mock): + """Test arm night method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code_arm_required": False, + "code": CODE, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_night(hass, 0, entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_NIGHT == hass.states.get(entity_id).state + + +async def test_arm_night_with_pending(hass, mqtt_mock): + """Test arm night method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 1, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_night(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + state = hass.states.get(entity_id) + assert state.attributes["post_pending_state"] == STATE_ALARM_ARMED_NIGHT + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_NIGHT == hass.states.get(entity_id).state + + # Do not go to the pending state when updating to the same state + await common.async_alarm_arm_night(hass, CODE, entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_NIGHT == hass.states.get(entity_id).state + + +async def test_arm_night_with_invalid_code(hass, mqtt_mock): + """Attempt to arm night without a valid code.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 1, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_night(hass, f"{CODE}2") + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_trigger_no_pending(hass, mqtt_mock): + """Test triggering when no pending submitted method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "trigger_time": 1, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=60) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + +async def test_trigger_with_delay(hass, mqtt_mock): + """Test trigger method and switch from pending to triggered.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "delay_time": 1, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_PENDING == state.state + assert STATE_ALARM_TRIGGERED == state.attributes["post_pending_state"] + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_TRIGGERED == state.state + + +async def test_trigger_zero_trigger_time(hass, mqtt_mock): + """Test disabled trigger.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 0, + "trigger_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_trigger_zero_trigger_time_with_pending(hass, mqtt_mock): + """Test disabled trigger.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 2, + "trigger_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_trigger_with_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 2, + "trigger_time": 3, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + state = hass.states.get(entity_id) + assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED + + future = dt_util.utcnow() + timedelta(seconds=2) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_trigger_with_disarm_after_trigger(hass, mqtt_mock): + """Test disarm after trigger.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "trigger_time": 5, + "pending_time": 0, + "disarm_after_trigger": True, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_trigger_with_zero_specific_trigger_time(hass, mqtt_mock): + """Test trigger method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "trigger_time": 5, + "disarmed": {"trigger_time": 0}, + "pending_time": 0, + "disarm_after_trigger": True, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_trigger_with_unused_zero_specific_trigger_time(hass, mqtt_mock): + """Test disarm after trigger.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "trigger_time": 5, + "armed_home": {"trigger_time": 0}, + "pending_time": 0, + "disarm_after_trigger": True, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_trigger_with_specific_trigger_time(hass, mqtt_mock): + """Test disarm after trigger.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "disarmed": {"trigger_time": 5}, + "pending_time": 0, + "disarm_after_trigger": True, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_back_to_back_trigger_with_no_disarm_after_trigger(hass, mqtt_mock): + """Test no disarm after back to back trigger.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "trigger_time": 5, + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE, entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + +async def test_disarm_while_pending_trigger(hass, mqtt_mock): + """Test disarming while pending state.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "trigger_time": 5, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + await common.async_alarm_disarm(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_disarm_during_trigger_with_invalid_code(hass, mqtt_mock): + """Test disarming while code is invalid.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 5, + "code": f"{CODE}2", + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + await common.async_alarm_disarm(hass, entity_id=entity_id) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + +async def test_trigger_with_unused_specific_delay(hass, mqtt_mock): + """Test trigger method and switch from pending to triggered.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "delay_time": 5, + "pending_time": 0, + "armed_home": {"delay_time": 10}, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_PENDING == state.state + assert STATE_ALARM_TRIGGERED == state.attributes["post_pending_state"] + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + +async def test_trigger_with_specific_delay(hass, mqtt_mock): + """Test trigger method and switch from pending to triggered.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "delay_time": 10, + "pending_time": 0, + "armed_away": {"delay_time": 1}, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_PENDING == state.state + assert STATE_ALARM_TRIGGERED == state.attributes["post_pending_state"] + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + +async def test_trigger_with_pending_and_delay(hass, mqtt_mock): + """Test trigger method and switch from pending to triggered.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "delay_time": 1, + "pending_time": 0, + "triggered": {"pending_time": 1}, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED + + future += timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + +async def test_trigger_with_pending_and_specific_delay(hass, mqtt_mock): + """Test trigger method and switch from pending to triggered.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "delay_time": 10, + "pending_time": 0, + "armed_away": {"delay_time": 1}, + "triggered": {"pending_time": 1}, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_PENDING + assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED + + future += timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert state.state == STATE_ALARM_TRIGGERED + + +async def test_armed_home_with_specific_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 10, + "armed_home": {"pending_time": 2}, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + await common.async_alarm_arm_home(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=2) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_HOME == hass.states.get(entity_id).state + + +async def test_armed_away_with_specific_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 10, + "armed_away": {"pending_time": 2}, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + await common.async_alarm_arm_away(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=2) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + +async def test_armed_night_with_specific_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 10, + "armed_night": {"pending_time": 2}, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + await common.async_alarm_arm_night(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=2) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_NIGHT == hass.states.get(entity_id).state + + +async def test_trigger_with_specific_pending(hass, mqtt_mock): + """Test arm home method.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 10, + "triggered": {"pending_time": 2}, + "trigger_time": 3, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + await common.async_alarm_trigger(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=2) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_TRIGGERED == hass.states.get(entity_id).state + + future = dt_util.utcnow() + timedelta(seconds=5) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_arm_away_after_disabled_disarmed(hass, mqtt_mock): + """Test pending state with and without zero trigger time.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code": CODE, + "pending_time": 0, + "delay_time": 1, + "armed_away": {"pending_time": 1}, + "disarmed": {"trigger_time": 0}, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_away(hass, CODE) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_PENDING == state.state + assert STATE_ALARM_DISARMED == state.attributes["pre_pending_state"] + assert STATE_ALARM_ARMED_AWAY == state.attributes["post_pending_state"] + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_PENDING == state.state + assert STATE_ALARM_DISARMED == state.attributes["pre_pending_state"] + assert STATE_ALARM_ARMED_AWAY == state.attributes["post_pending_state"] + + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_ARMED_AWAY == state.state + + await common.async_alarm_trigger(hass, entity_id=entity_id) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) assert STATE_ALARM_PENDING == state.state + assert STATE_ALARM_ARMED_AWAY == state.attributes["pre_pending_state"] assert STATE_ALARM_TRIGGERED == state.attributes["post_pending_state"] - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_TRIGGERED == state.state - - def test_trigger_zero_trigger_time(self): - """Test disabled trigger.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 0, - "trigger_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_trigger_zero_trigger_time_with_pending(self): - """Test disabled trigger.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 2, - "trigger_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_trigger_with_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 2, - "trigger_time": 3, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - state = self.hass.states.get(entity_id) - assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED - - future = dt_util.utcnow() + timedelta(seconds=2) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_trigger_with_disarm_after_trigger(self): - """Test disarm after trigger.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "trigger_time": 5, - "pending_time": 0, - "disarm_after_trigger": True, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_trigger_with_zero_specific_trigger_time(self): - """Test trigger method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "trigger_time": 5, - "disarmed": {"trigger_time": 0}, - "pending_time": 0, - "disarm_after_trigger": True, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_trigger_with_unused_zero_specific_trigger_time(self): - """Test disarm after trigger.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "trigger_time": 5, - "armed_home": {"trigger_time": 0}, - "pending_time": 0, - "disarm_after_trigger": True, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_trigger_with_specific_trigger_time(self): - """Test disarm after trigger.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "disarmed": {"trigger_time": 5}, - "pending_time": 0, - "disarm_after_trigger": True, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_back_to_back_trigger_with_no_disarm_after_trigger(self): - """Test no disarm after back to back trigger.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "trigger_time": 5, - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE, entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - def test_disarm_while_pending_trigger(self): - """Test disarming while pending state.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "trigger_time": 5, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - common.alarm_disarm(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_disarm_during_trigger_with_invalid_code(self): - """Test disarming while code is invalid.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 5, - "code": f"{CODE}2", - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - common.alarm_disarm(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - def test_trigger_with_unused_specific_delay(self): - """Test trigger method and switch from pending to triggered.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "delay_time": 5, - "pending_time": 0, - "armed_home": {"delay_time": 10}, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_PENDING == state.state - assert STATE_ALARM_TRIGGERED == state.attributes["post_pending_state"] - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_TRIGGERED - - def test_trigger_with_specific_delay(self): - """Test trigger method and switch from pending to triggered.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "delay_time": 10, - "pending_time": 0, - "armed_away": {"delay_time": 1}, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_PENDING == state.state - assert STATE_ALARM_TRIGGERED == state.attributes["post_pending_state"] - - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_TRIGGERED - - def test_trigger_with_pending_and_delay(self): - """Test trigger method and switch from pending to triggered.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "delay_time": 1, - "pending_time": 0, - "triggered": {"pending_time": 1}, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_PENDING - assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED - - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_PENDING - assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED - - future += timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_TRIGGERED - - def test_trigger_with_pending_and_specific_delay(self): - """Test trigger method and switch from pending to triggered.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "delay_time": 10, - "pending_time": 0, - "armed_away": {"delay_time": 1}, - "triggered": {"pending_time": 1}, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_PENDING - assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED - - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_PENDING - assert state.attributes["post_pending_state"] == STATE_ALARM_TRIGGERED - - future += timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert state.state == STATE_ALARM_TRIGGERED - - def test_armed_home_with_specific_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 10, - "armed_home": {"pending_time": 2}, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - common.alarm_arm_home(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=2) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_HOME == self.hass.states.get(entity_id).state - - def test_armed_away_with_specific_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 10, - "armed_away": {"pending_time": 2}, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - common.alarm_arm_away(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=2) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - def test_armed_night_with_specific_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 10, - "armed_night": {"pending_time": 2}, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - common.alarm_arm_night(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=2) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_NIGHT == self.hass.states.get(entity_id).state - - def test_trigger_with_specific_pending(self): - """Test arm home method.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 10, - "triggered": {"pending_time": 2}, - "trigger_time": 3, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - common.alarm_trigger(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=2) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_TRIGGERED == self.hass.states.get(entity_id).state - - future = dt_util.utcnow() + timedelta(seconds=5) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_arm_away_after_disabled_disarmed(self): - """Test pending state with and without zero trigger time.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code": CODE, - "pending_time": 0, - "delay_time": 1, - "armed_away": {"pending_time": 1}, - "disarmed": {"trigger_time": 0}, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_away(self.hass, CODE) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_PENDING == state.state - assert STATE_ALARM_DISARMED == state.attributes["pre_pending_state"] - assert STATE_ALARM_ARMED_AWAY == state.attributes["post_pending_state"] - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_PENDING == state.state - assert STATE_ALARM_DISARMED == state.attributes["pre_pending_state"] - assert STATE_ALARM_ARMED_AWAY == state.attributes["post_pending_state"] - - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_ARMED_AWAY == state.state - - common.alarm_trigger(self.hass, entity_id=entity_id) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_PENDING == state.state - assert STATE_ALARM_ARMED_AWAY == state.attributes["pre_pending_state"] - assert STATE_ALARM_TRIGGERED == state.attributes["post_pending_state"] - - future += timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_TRIGGERED == state.state - - def test_disarm_with_template_code(self): - """Attempt to disarm with a valid or invalid template-based code.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - "alarm_control_panel": { - "platform": "manual_mqtt", - "name": "test", - "code_template": '{{ "" if from_state == "disarmed" else "abc" }}', - "pending_time": 0, - "disarm_after_trigger": False, - "command_topic": "alarm/command", - "state_topic": "alarm/state", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_arm_home(self.hass, "def") - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_ARMED_HOME == state.state - - common.alarm_disarm(self.hass, "def") - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_ARMED_HOME == state.state - - common.alarm_disarm(self.hass, "abc") - self.hass.block_till_done() - - state = self.hass.states.get(entity_id) - assert STATE_ALARM_DISARMED == state.state - - def test_arm_home_via_command_topic(self): - """Test arming home via command topic.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - alarm_control_panel.DOMAIN: { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 1, - "state_topic": "alarm/state", - "command_topic": "alarm/command", - "payload_arm_home": "ARM_HOME", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - # Fire the arm command via MQTT; ensure state changes to pending - fire_mqtt_message(self.hass, "alarm/command", "ARM_HOME") - self.hass.block_till_done() - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - # Fast-forward a little bit - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_HOME == self.hass.states.get(entity_id).state - - def test_arm_away_via_command_topic(self): - """Test arming away via command topic.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - alarm_control_panel.DOMAIN: { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 1, - "state_topic": "alarm/state", - "command_topic": "alarm/command", - "payload_arm_away": "ARM_AWAY", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - # Fire the arm command via MQTT; ensure state changes to pending - fire_mqtt_message(self.hass, "alarm/command", "ARM_AWAY") - self.hass.block_till_done() - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - # Fast-forward a little bit - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_AWAY == self.hass.states.get(entity_id).state - - def test_arm_night_via_command_topic(self): - """Test arming night via command topic.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - alarm_control_panel.DOMAIN: { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 1, - "state_topic": "alarm/state", - "command_topic": "alarm/command", - "payload_arm_night": "ARM_NIGHT", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - # Fire the arm command via MQTT; ensure state changes to pending - fire_mqtt_message(self.hass, "alarm/command", "ARM_NIGHT") - self.hass.block_till_done() - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - # Fast-forward a little bit - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - - assert STATE_ALARM_ARMED_NIGHT == self.hass.states.get(entity_id).state - - def test_disarm_pending_via_command_topic(self): - """Test disarming pending alarm via command topic.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - alarm_control_panel.DOMAIN: { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 1, - "state_topic": "alarm/state", - "command_topic": "alarm/command", - "payload_disarm": "DISARM", - } - }, - ) - self.hass.block_till_done() - - entity_id = "alarm_control_panel.test" - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - common.alarm_trigger(self.hass) - self.hass.block_till_done() - - assert STATE_ALARM_PENDING == self.hass.states.get(entity_id).state - - # Now that we're pending, receive a command to disarm - fire_mqtt_message(self.hass, "alarm/command", "DISARM") - self.hass.block_till_done() - - assert STATE_ALARM_DISARMED == self.hass.states.get(entity_id).state - - def test_state_changes_are_published_to_mqtt(self): - """Test publishing of MQTT messages when state changes.""" - assert setup_component( - self.hass, - alarm_control_panel.DOMAIN, - { - alarm_control_panel.DOMAIN: { - "platform": "manual_mqtt", - "name": "test", - "pending_time": 1, - "trigger_time": 1, - "state_topic": "alarm/state", - "command_topic": "alarm/command", - } - }, - ) - self.hass.block_till_done() - - # Component should send disarmed alarm state on startup - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_DISARMED, 0, True - ) - self.mock_publish.async_publish.reset_mock() - - # Arm in home mode - common.alarm_arm_home(self.hass) - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_PENDING, 0, True - ) - self.mock_publish.async_publish.reset_mock() - # Fast-forward a little bit - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_ARMED_HOME, 0, True - ) - self.mock_publish.async_publish.reset_mock() - - # Arm in away mode - common.alarm_arm_away(self.hass) - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_PENDING, 0, True - ) - self.mock_publish.async_publish.reset_mock() - # Fast-forward a little bit - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_ARMED_AWAY, 0, True - ) - self.mock_publish.async_publish.reset_mock() - - # Arm in night mode - common.alarm_arm_night(self.hass) - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_PENDING, 0, True - ) - self.mock_publish.async_publish.reset_mock() - # Fast-forward a little bit - future = dt_util.utcnow() + timedelta(seconds=1) - with patch( - ( - "homeassistant.components.manual_mqtt.alarm_control_panel." - "dt_util.utcnow" - ), - return_value=future, - ): - fire_time_changed(self.hass, future) - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_ARMED_NIGHT, 0, True - ) - self.mock_publish.async_publish.reset_mock() - - # Disarm - common.alarm_disarm(self.hass) - self.hass.block_till_done() - self.mock_publish.async_publish.assert_called_once_with( - "alarm/state", STATE_ALARM_DISARMED, 0, True - ) + future += timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_TRIGGERED == state.state + + +async def test_disarm_with_template_code(hass, mqtt_mock): + """Attempt to disarm with a valid or invalid template-based code.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + "alarm_control_panel": { + "platform": "manual_mqtt", + "name": "test", + "code_template": '{{ "" if from_state == "disarmed" else "abc" }}', + "pending_time": 0, + "disarm_after_trigger": False, + "command_topic": "alarm/command", + "state_topic": "alarm/state", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_arm_home(hass, "def") + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_ARMED_HOME == state.state + + await common.async_alarm_disarm(hass, "def") + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_ARMED_HOME == state.state + + await common.async_alarm_disarm(hass, "abc") + await hass.async_block_till_done() + + state = hass.states.get(entity_id) + assert STATE_ALARM_DISARMED == state.state + + +async def test_arm_home_via_command_topic(hass, mqtt_mock): + """Test arming home via command topic.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + alarm_control_panel.DOMAIN: { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 1, + "state_topic": "alarm/state", + "command_topic": "alarm/command", + "payload_arm_home": "ARM_HOME", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + # Fire the arm command via MQTT; ensure state changes to pending + async_fire_mqtt_message(hass, "alarm/command", "ARM_HOME") + await hass.async_block_till_done() + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + # Fast-forward a little bit + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_HOME == hass.states.get(entity_id).state + + +async def test_arm_away_via_command_topic(hass, mqtt_mock): + """Test arming away via command topic.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + alarm_control_panel.DOMAIN: { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 1, + "state_topic": "alarm/state", + "command_topic": "alarm/command", + "payload_arm_away": "ARM_AWAY", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + # Fire the arm command via MQTT; ensure state changes to pending + async_fire_mqtt_message(hass, "alarm/command", "ARM_AWAY") + await hass.async_block_till_done() + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + # Fast-forward a little bit + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_AWAY == hass.states.get(entity_id).state + + +async def test_arm_night_via_command_topic(hass, mqtt_mock): + """Test arming night via command topic.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + alarm_control_panel.DOMAIN: { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 1, + "state_topic": "alarm/state", + "command_topic": "alarm/command", + "payload_arm_night": "ARM_NIGHT", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + # Fire the arm command via MQTT; ensure state changes to pending + async_fire_mqtt_message(hass, "alarm/command", "ARM_NIGHT") + await hass.async_block_till_done() + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + # Fast-forward a little bit + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + + assert STATE_ALARM_ARMED_NIGHT == hass.states.get(entity_id).state + + +async def test_disarm_pending_via_command_topic(hass, mqtt_mock): + """Test disarming pending alarm via command topic.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + alarm_control_panel.DOMAIN: { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 1, + "state_topic": "alarm/state", + "command_topic": "alarm/command", + "payload_disarm": "DISARM", + } + }, + ) + await hass.async_block_till_done() + + entity_id = "alarm_control_panel.test" + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + await common.async_alarm_trigger(hass) + await hass.async_block_till_done() + + assert STATE_ALARM_PENDING == hass.states.get(entity_id).state + + # Now that we're pending, receive a command to disarm + async_fire_mqtt_message(hass, "alarm/command", "DISARM") + await hass.async_block_till_done() + + assert STATE_ALARM_DISARMED == hass.states.get(entity_id).state + + +async def test_state_changes_are_published_to_mqtt(hass, mqtt_mock): + """Test publishing of MQTT messages when state changes.""" + assert await async_setup_component( + hass, + alarm_control_panel.DOMAIN, + { + alarm_control_panel.DOMAIN: { + "platform": "manual_mqtt", + "name": "test", + "pending_time": 1, + "trigger_time": 1, + "state_topic": "alarm/state", + "command_topic": "alarm/command", + } + }, + ) + await hass.async_block_till_done() + + # Component should send disarmed alarm state on startup + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_DISARMED, 0, True + ) + mqtt_mock.async_publish.reset_mock() + + # Arm in home mode + await common.async_alarm_arm_home(hass) + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_PENDING, 0, True + ) + mqtt_mock.async_publish.reset_mock() + # Fast-forward a little bit + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_ARMED_HOME, 0, True + ) + mqtt_mock.async_publish.reset_mock() + + # Arm in away mode + await common.async_alarm_arm_away(hass) + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_PENDING, 0, True + ) + mqtt_mock.async_publish.reset_mock() + # Fast-forward a little bit + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_ARMED_AWAY, 0, True + ) + mqtt_mock.async_publish.reset_mock() + + # Arm in night mode + await common.async_alarm_arm_night(hass) + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_PENDING, 0, True + ) + mqtt_mock.async_publish.reset_mock() + # Fast-forward a little bit + future = dt_util.utcnow() + timedelta(seconds=1) + with patch( + ("homeassistant.components.manual_mqtt.alarm_control_panel." "dt_util.utcnow"), + return_value=future, + ): + async_fire_time_changed(hass, future) + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_ARMED_NIGHT, 0, True + ) + mqtt_mock.async_publish.reset_mock() + + # Disarm + await common.async_alarm_disarm(hass) + await hass.async_block_till_done() + mqtt_mock.async_publish.assert_called_once_with( + "alarm/state", STATE_ALARM_DISARMED, 0, True + ) diff --git a/tests/components/mqtt/test_alarm_control_panel.py b/tests/components/mqtt/test_alarm_control_panel.py index 1aa26cf0334..aa6452fd9c8 100644 --- a/tests/components/mqtt/test_alarm_control_panel.py +++ b/tests/components/mqtt/test_alarm_control_panel.py @@ -17,6 +17,7 @@ from homeassistant.const import ( STATE_ALARM_TRIGGERED, STATE_UNKNOWN, ) +from homeassistant.setup import async_setup_component from .test_common import ( help_test_availability_when_connection_lost, @@ -41,11 +42,7 @@ from .test_common import ( help_test_update_with_json_attrs_not_dict, ) -from tests.common import ( - assert_setup_component, - async_fire_mqtt_message, - async_setup_component, -) +from tests.common import assert_setup_component, async_fire_mqtt_message from tests.components.alarm_control_panel import common CODE_NUMBER = "1234" diff --git a/tests/components/mqtt/test_climate.py b/tests/components/mqtt/test_climate.py index 6ccb38eda2d..6a7bdf0b7e6 100644 --- a/tests/components/mqtt/test_climate.py +++ b/tests/components/mqtt/test_climate.py @@ -23,6 +23,7 @@ from homeassistant.components.climate.const import ( SUPPORT_TARGET_TEMPERATURE_RANGE, ) from homeassistant.const import STATE_OFF +from homeassistant.setup import async_setup_component from .test_common import ( help_test_availability_when_connection_lost, @@ -48,7 +49,7 @@ from .test_common import ( ) from tests.async_mock import call -from tests.common import async_fire_mqtt_message, async_setup_component +from tests.common import async_fire_mqtt_message from tests.components.climate import common ENTITY_CLIMATE = "climate.test" diff --git a/tests/components/mqtt/test_common.py b/tests/components/mqtt/test_common.py index 7da0a09f024..4275cc36e13 100644 --- a/tests/components/mqtt/test_common.py +++ b/tests/components/mqtt/test_common.py @@ -10,9 +10,10 @@ from homeassistant.components.mqtt.const import MQTT_DISCONNECTED from homeassistant.components.mqtt.discovery import async_start from homeassistant.const import ATTR_ASSUMED_STATE, STATE_UNAVAILABLE from homeassistant.helpers.dispatcher import async_dispatcher_send +from homeassistant.setup import async_setup_component from tests.async_mock import ANY -from tests.common import async_fire_mqtt_message, async_setup_component, mock_registry +from tests.common import async_fire_mqtt_message, mock_registry DEFAULT_CONFIG_DEVICE_INFO_ID = { "identifiers": ["helloworld"], diff --git a/tests/components/mqtt_eventstream/test_init.py b/tests/components/mqtt_eventstream/test_init.py index 8050535eed4..ecdedf904d4 100644 --- a/tests/components/mqtt_eventstream/test_init.py +++ b/tests/components/mqtt_eventstream/test_init.py @@ -7,15 +7,13 @@ import homeassistant.components.mqtt_eventstream as eventstream from homeassistant.const import EVENT_STATE_CHANGED from homeassistant.core import State, callback from homeassistant.helpers.json import JSONEncoder -from homeassistant.setup import setup_component +from homeassistant.setup import async_setup_component import homeassistant.util.dt as dt_util from tests.async_mock import ANY, patch from tests.common import ( - fire_mqtt_message, - fire_time_changed, - get_test_home_assistant, - mock_mqtt_component, + async_fire_mqtt_message, + async_fire_time_changed, mock_state_change_event, ) @@ -25,177 +23,170 @@ def mock_storage(hass_storage): """Autouse hass_storage for the TestCase tests.""" -class TestMqttEventStream: - """Test the MQTT eventstream module.""" +async def add_eventstream(hass, sub_topic=None, pub_topic=None, ignore_event=None): + """Add a mqtt_eventstream component.""" + config = {} + if sub_topic: + config["subscribe_topic"] = sub_topic + if pub_topic: + config["publish_topic"] = pub_topic + if ignore_event: + config["ignore_event"] = ignore_event + return await async_setup_component( + hass, eventstream.DOMAIN, {eventstream.DOMAIN: config} + ) - def setup_method(self): - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() - self.mock_mqtt = mock_mqtt_component(self.hass) - def teardown_method(self): - """Stop everything that was started.""" - self.hass.stop() +async def test_setup_succeeds(hass, mqtt_mock): + """Test the success of the setup.""" + assert await add_eventstream(hass) - def add_eventstream(self, sub_topic=None, pub_topic=None, ignore_event=None): - """Add a mqtt_eventstream component.""" - config = {} - if sub_topic: - config["subscribe_topic"] = sub_topic - if pub_topic: - config["publish_topic"] = pub_topic - if ignore_event: - config["ignore_event"] = ignore_event - return setup_component( - self.hass, eventstream.DOMAIN, {eventstream.DOMAIN: config} - ) - def test_setup_succeeds(self): - """Test the success of the setup.""" - assert self.add_eventstream() +async def test_setup_with_pub(hass, mqtt_mock): + """Test the setup with subscription.""" + # Should start off with no listeners for all events + assert hass.bus.async_listeners().get("*") is None - def test_setup_with_pub(self): - """Test the setup with subscription.""" - # Should start off with no listeners for all events - assert self.hass.bus.listeners.get("*") is None + assert await add_eventstream(hass, pub_topic="bar") + await hass.async_block_till_done() - assert self.add_eventstream(pub_topic="bar") - self.hass.block_till_done() + # Verify that the event handler has been added as a listener + assert hass.bus.async_listeners().get("*") == 1 - # Verify that the event handler has been added as a listener - assert self.hass.bus.listeners.get("*") == 1 - @patch("homeassistant.components.mqtt.async_subscribe") - def test_subscribe(self, mock_sub): - """Test the subscription.""" - sub_topic = "foo" - assert self.add_eventstream(sub_topic=sub_topic) - self.hass.block_till_done() +async def test_subscribe(hass, mqtt_mock): + """Test the subscription.""" + sub_topic = "foo" + assert await add_eventstream(hass, sub_topic=sub_topic) + await hass.async_block_till_done() - # Verify that the this entity was subscribed to the topic - mock_sub.assert_called_with(self.hass, sub_topic, ANY) + # Verify that the this entity was subscribed to the topic + mqtt_mock.async_subscribe.assert_called_with(sub_topic, ANY, 0, ANY) - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_sends_message(self, mock_utcnow, mock_pub): - """Test the sending of a new message if event changed.""" - now = dt_util.as_utc(dt_util.now()) - e_id = "fake.entity" - pub_topic = "bar" - mock_utcnow.return_value = now +async def test_state_changed_event_sends_message(hass, mqtt_mock): + """Test the sending of a new message if event changed.""" + now = dt_util.as_utc(dt_util.now()) + e_id = "fake.entity" + pub_topic = "bar" + with patch( + ("homeassistant.core.dt_util.utcnow"), return_value=now, + ): # Add the eventstream component for publishing events - assert self.add_eventstream(pub_topic=pub_topic) - self.hass.block_till_done() + assert await add_eventstream(hass, pub_topic=pub_topic) + await hass.async_block_till_done() # Reset the mock because it will have already gotten calls for the # mqtt_eventstream state change on initialization, etc. - mock_pub.reset_mock() + mqtt_mock.async_publish.reset_mock() # Set a state of an entity - mock_state_change_event(self.hass, State(e_id, "on")) - self.hass.block_till_done() + mock_state_change_event(hass, State(e_id, "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() - # The order of the JSON is indeterminate, - # so first just check that publish was called - mock_pub.assert_called_with(self.hass, pub_topic, ANY) - assert mock_pub.called + # The order of the JSON is indeterminate, + # so first just check that publish was called + mqtt_mock.async_publish.assert_called_with(pub_topic, ANY, 0, False) + assert mqtt_mock.async_publish.called - # Get the actual call to publish and make sure it was the one - # we were looking for - msg = mock_pub.call_args[0][2] - event = {} - event["event_type"] = EVENT_STATE_CHANGED - new_state = { - "last_updated": now.isoformat(), - "state": "on", - "entity_id": e_id, - "attributes": {}, - "last_changed": now.isoformat(), - } - event["event_data"] = {"new_state": new_state, "entity_id": e_id} + # Get the actual call to publish and make sure it was the one + # we were looking for + msg = mqtt_mock.async_publish.call_args[0][1] + event = {} + event["event_type"] = EVENT_STATE_CHANGED + new_state = { + "last_updated": now.isoformat(), + "state": "on", + "entity_id": e_id, + "attributes": {}, + "last_changed": now.isoformat(), + } + event["event_data"] = {"new_state": new_state, "entity_id": e_id} - # Verify that the message received was that expected - result = json.loads(msg) - result["event_data"]["new_state"].pop("context") - assert result == event + # Verify that the message received was that expected + result = json.loads(msg) + result["event_data"]["new_state"].pop("context") + assert result == event - @patch("homeassistant.components.mqtt.async_publish") - def test_time_event_does_not_send_message(self, mock_pub): - """Test the sending of a new message if time event.""" - assert self.add_eventstream(pub_topic="bar") - self.hass.block_till_done() - # Reset the mock because it will have already gotten calls for the - # mqtt_eventstream state change on initialization, etc. - mock_pub.reset_mock() +async def test_time_event_does_not_send_message(hass, mqtt_mock): + """Test the sending of a new message if time event.""" + assert await add_eventstream(hass, pub_topic="bar") + await hass.async_block_till_done() - fire_time_changed(self.hass, dt_util.utcnow()) - assert not mock_pub.called + # Reset the mock because it will have already gotten calls for the + # mqtt_eventstream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() - def test_receiving_remote_event_fires_hass_event(self): - """Test the receiving of the remotely fired event.""" - sub_topic = "foo" - assert self.add_eventstream(sub_topic=sub_topic) - self.hass.block_till_done() + async_fire_time_changed(hass, dt_util.utcnow()) + assert not mqtt_mock.async_publish.called - calls = [] - @callback - def listener(_): - calls.append(1) +async def test_receiving_remote_event_fires_hass_event(hass, mqtt_mock): + """Test the receiving of the remotely fired event.""" + sub_topic = "foo" + assert await add_eventstream(hass, sub_topic=sub_topic) + await hass.async_block_till_done() - self.hass.bus.listen_once("test_event", listener) - self.hass.block_till_done() + calls = [] - payload = json.dumps( - {"event_type": "test_event", "event_data": {}}, cls=JSONEncoder - ) - fire_mqtt_message(self.hass, sub_topic, payload) - self.hass.block_till_done() + @callback + def listener(_): + calls.append(1) - assert 1 == len(calls) + hass.bus.async_listen_once("test_event", listener) + await hass.async_block_till_done() - @patch("homeassistant.components.mqtt.async_publish") - def test_ignored_event_doesnt_send_over_stream(self, mock_pub): - """Test the ignoring of sending events if defined.""" - assert self.add_eventstream(pub_topic="bar", ignore_event=["state_changed"]) - self.hass.block_till_done() + payload = json.dumps( + {"event_type": "test_event", "event_data": {}}, cls=JSONEncoder + ) + async_fire_mqtt_message(hass, sub_topic, payload) + await hass.async_block_till_done() - e_id = "entity.test_id" - event = {} - event["event_type"] = EVENT_STATE_CHANGED - new_state = {"state": "on", "entity_id": e_id, "attributes": {}} - event["event_data"] = {"new_state": new_state, "entity_id": e_id} + assert 1 == len(calls) - # Reset the mock because it will have already gotten calls for the - # mqtt_eventstream state change on initialization, etc. - mock_pub.reset_mock() - # Set a state of an entity - mock_state_change_event(self.hass, State(e_id, "on")) - self.hass.block_till_done() +async def test_ignored_event_doesnt_send_over_stream(hass, mqtt_mock): + """Test the ignoring of sending events if defined.""" + assert await add_eventstream(hass, pub_topic="bar", ignore_event=["state_changed"]) + await hass.async_block_till_done() - assert not mock_pub.called + e_id = "entity.test_id" + event = {} + event["event_type"] = EVENT_STATE_CHANGED + new_state = {"state": "on", "entity_id": e_id, "attributes": {}} + event["event_data"] = {"new_state": new_state, "entity_id": e_id} - @patch("homeassistant.components.mqtt.async_publish") - def test_wrong_ignored_event_sends_over_stream(self, mock_pub): - """Test the ignoring of sending events if defined.""" - assert self.add_eventstream(pub_topic="bar", ignore_event=["statee_changed"]) - self.hass.block_till_done() + # Reset the mock because it will have already gotten calls for the + # mqtt_eventstream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() - e_id = "entity.test_id" - event = {} - event["event_type"] = EVENT_STATE_CHANGED - new_state = {"state": "on", "entity_id": e_id, "attributes": {}} - event["event_data"] = {"new_state": new_state, "entity_id": e_id} + # Set a state of an entity + mock_state_change_event(hass, State(e_id, "on")) + await hass.async_block_till_done() - # Reset the mock because it will have already gotten calls for the - # mqtt_eventstream state change on initialization, etc. - mock_pub.reset_mock() + assert not mqtt_mock.async_publish.called - # Set a state of an entity - mock_state_change_event(self.hass, State(e_id, "on")) - self.hass.block_till_done() - assert mock_pub.called +async def test_wrong_ignored_event_sends_over_stream(hass, mqtt_mock): + """Test the ignoring of sending events if defined.""" + assert await add_eventstream(hass, pub_topic="bar", ignore_event=["statee_changed"]) + await hass.async_block_till_done() + + e_id = "entity.test_id" + event = {} + event["event_type"] = EVENT_STATE_CHANGED + new_state = {"state": "on", "entity_id": e_id, "attributes": {}} + event["event_data"] = {"new_state": new_state, "entity_id": e_id} + + # Reset the mock because it will have already gotten calls for the + # mqtt_eventstream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State(e_id, "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + assert mqtt_mock.async_publish.called diff --git a/tests/components/mqtt_statestream/test_init.py b/tests/components/mqtt_statestream/test_init.py index aa9ef0d5de8..c0264060aff 100644 --- a/tests/components/mqtt_statestream/test_init.py +++ b/tests/components/mqtt_statestream/test_init.py @@ -3,14 +3,10 @@ import pytest import homeassistant.components.mqtt_statestream as statestream from homeassistant.core import State -from homeassistant.setup import setup_component +from homeassistant.setup import async_setup_component -from tests.async_mock import ANY, call, patch -from tests.common import ( - get_test_home_assistant, - mock_mqtt_component, - mock_state_change_event, -) +from tests.async_mock import ANY, call +from tests.common import mock_state_change_event @pytest.fixture(autouse=True) @@ -18,360 +14,343 @@ def mock_storage(hass_storage): """Autouse hass_storage for the TestCase tests.""" -class TestMqttStateStream: - """Test the MQTT statestream module.""" - - def setup_method(self): - """Set up things to be run when tests are started.""" - self.hass = get_test_home_assistant() - self.mock_mqtt = mock_mqtt_component(self.hass) - - def teardown_method(self): - """Stop everything that was started.""" - self.hass.stop() - - def add_statestream( - self, - base_topic=None, - publish_attributes=None, - publish_timestamps=None, - publish_include=None, - publish_exclude=None, - ): - """Add a mqtt_statestream component.""" - config = {} - if base_topic: - config["base_topic"] = base_topic - if publish_attributes: - config["publish_attributes"] = publish_attributes - if publish_timestamps: - config["publish_timestamps"] = publish_timestamps - if publish_include: - config["include"] = publish_include - if publish_exclude: - config["exclude"] = publish_exclude - return setup_component( - self.hass, statestream.DOMAIN, {statestream.DOMAIN: config} - ) - - def test_fails_with_no_base(self): - """Setup should fail if no base_topic is set.""" - assert self.add_statestream() is False - - def test_setup_succeeds_without_attributes(self): - """Test the success of the setup with a valid base_topic.""" - assert self.add_statestream(base_topic="pub") - - def test_setup_succeeds_with_attributes(self): - """Test setup with a valid base_topic and publish_attributes.""" - assert self.add_statestream(base_topic="pub", publish_attributes=True) - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_sends_message(self, mock_utcnow, mock_pub): - """Test the sending of a new message if event changed.""" - e_id = "fake.entity" - base_topic = "pub" - - # Add the statestream component for publishing state updates - assert self.add_statestream(base_topic=base_topic) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State(e_id, "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - mock_pub.assert_called_with(self.hass, "pub/fake/entity/state", "on", 1, True) - assert mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_sends_message_and_timestamp( - self, mock_utcnow, mock_pub - ): - """Test the sending of a message and timestamps if event changed.""" - e_id = "another.entity" - base_topic = "pub" - - # Add the statestream component for publishing state updates - assert self.add_statestream( - base_topic=base_topic, publish_attributes=None, publish_timestamps=True - ) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State(e_id, "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - calls = [ - call.async_publish(self.hass, "pub/another/entity/state", "on", 1, True), - call.async_publish( - self.hass, "pub/another/entity/last_changed", ANY, 1, True - ), - call.async_publish( - self.hass, "pub/another/entity/last_updated", ANY, 1, True - ), - ] - - mock_pub.assert_has_calls(calls, any_order=True) - assert mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_attr_sends_message(self, mock_utcnow, mock_pub): - """Test the sending of a new message if attribute changed.""" - e_id = "fake.entity" - base_topic = "pub" - - # Add the statestream component for publishing state updates - assert self.add_statestream(base_topic=base_topic, publish_attributes=True) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - test_attributes = {"testing": "YES", "list": ["a", "b", "c"], "bool": False} - - # Set a state of an entity - mock_state_change_event( - self.hass, State(e_id, "off", attributes=test_attributes) - ) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - calls = [ - call.async_publish(self.hass, "pub/fake/entity/state", "off", 1, True), - call.async_publish(self.hass, "pub/fake/entity/testing", '"YES"', 1, True), - call.async_publish( - self.hass, "pub/fake/entity/list", '["a", "b", "c"]', 1, True - ), - call.async_publish(self.hass, "pub/fake/entity/bool", "false", 1, True), - ] - - mock_pub.assert_has_calls(calls, any_order=True) - assert mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_include_domain(self, mock_utcnow, mock_pub): - """Test that filtering on included domain works as expected.""" - base_topic = "pub" - - incl = {"domains": ["fake"]} - excl = {} - - # Add the statestream component for publishing state updates - # Set the filter to allow fake.* items - assert self.add_statestream( - base_topic=base_topic, publish_include=incl, publish_exclude=excl - ) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State("fake.entity", "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - mock_pub.assert_called_with(self.hass, "pub/fake/entity/state", "on", 1, True) - assert mock_pub.called - - mock_pub.reset_mock() - # Set a state of an entity that shouldn't be included - mock_state_change_event(self.hass, State("fake2.entity", "on")) - self.hass.block_till_done() - - assert not mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_include_entity(self, mock_utcnow, mock_pub): - """Test that filtering on included entity works as expected.""" - base_topic = "pub" - - incl = {"entities": ["fake.entity"]} - excl = {} - - # Add the statestream component for publishing state updates - # Set the filter to allow fake.* items - assert self.add_statestream( - base_topic=base_topic, publish_include=incl, publish_exclude=excl - ) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State("fake.entity", "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - mock_pub.assert_called_with(self.hass, "pub/fake/entity/state", "on", 1, True) - assert mock_pub.called - - mock_pub.reset_mock() - # Set a state of an entity that shouldn't be included - mock_state_change_event(self.hass, State("fake.entity2", "on")) - self.hass.block_till_done() - - assert not mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_exclude_domain(self, mock_utcnow, mock_pub): - """Test that filtering on excluded domain works as expected.""" - base_topic = "pub" - - incl = {} - excl = {"domains": ["fake2"]} - - # Add the statestream component for publishing state updates - # Set the filter to allow fake.* items - assert self.add_statestream( - base_topic=base_topic, publish_include=incl, publish_exclude=excl - ) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State("fake.entity", "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - mock_pub.assert_called_with(self.hass, "pub/fake/entity/state", "on", 1, True) - assert mock_pub.called - - mock_pub.reset_mock() - # Set a state of an entity that shouldn't be included - mock_state_change_event(self.hass, State("fake2.entity", "on")) - self.hass.block_till_done() - - assert not mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_exclude_entity(self, mock_utcnow, mock_pub): - """Test that filtering on excluded entity works as expected.""" - base_topic = "pub" - - incl = {} - excl = {"entities": ["fake.entity2"]} - - # Add the statestream component for publishing state updates - # Set the filter to allow fake.* items - assert self.add_statestream( - base_topic=base_topic, publish_include=incl, publish_exclude=excl - ) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State("fake.entity", "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - mock_pub.assert_called_with(self.hass, "pub/fake/entity/state", "on", 1, True) - assert mock_pub.called - - mock_pub.reset_mock() - # Set a state of an entity that shouldn't be included - mock_state_change_event(self.hass, State("fake.entity2", "on")) - self.hass.block_till_done() - - assert not mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_exclude_domain_include_entity( - self, mock_utcnow, mock_pub - ): - """Test filtering with excluded domain and included entity.""" - base_topic = "pub" - - incl = {"entities": ["fake.entity"]} - excl = {"domains": ["fake"]} - - # Add the statestream component for publishing state updates - # Set the filter to allow fake.* items - assert self.add_statestream( - base_topic=base_topic, publish_include=incl, publish_exclude=excl - ) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State("fake.entity", "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - mock_pub.assert_called_with(self.hass, "pub/fake/entity/state", "on", 1, True) - assert mock_pub.called - - mock_pub.reset_mock() - # Set a state of an entity that shouldn't be included - mock_state_change_event(self.hass, State("fake.entity2", "on")) - self.hass.block_till_done() - - assert not mock_pub.called - - @patch("homeassistant.components.mqtt.async_publish") - @patch("homeassistant.core.dt_util.utcnow") - def test_state_changed_event_include_domain_exclude_entity( - self, mock_utcnow, mock_pub - ): - """Test filtering with included domain and excluded entity.""" - base_topic = "pub" - - incl = {"domains": ["fake"]} - excl = {"entities": ["fake.entity2"]} - - # Add the statestream component for publishing state updates - # Set the filter to allow fake.* items - assert self.add_statestream( - base_topic=base_topic, publish_include=incl, publish_exclude=excl - ) - self.hass.block_till_done() - - # Reset the mock because it will have already gotten calls for the - # mqtt_statestream state change on initialization, etc. - mock_pub.reset_mock() - - # Set a state of an entity - mock_state_change_event(self.hass, State("fake.entity", "on")) - self.hass.block_till_done() - - # Make sure 'on' was published to pub/fake/entity/state - mock_pub.assert_called_with(self.hass, "pub/fake/entity/state", "on", 1, True) - assert mock_pub.called - - mock_pub.reset_mock() - # Set a state of an entity that shouldn't be included - mock_state_change_event(self.hass, State("fake.entity2", "on")) - self.hass.block_till_done() - - assert not mock_pub.called +async def add_statestream( + hass, + base_topic=None, + publish_attributes=None, + publish_timestamps=None, + publish_include=None, + publish_exclude=None, +): + """Add a mqtt_statestream component.""" + config = {} + if base_topic: + config["base_topic"] = base_topic + if publish_attributes: + config["publish_attributes"] = publish_attributes + if publish_timestamps: + config["publish_timestamps"] = publish_timestamps + if publish_include: + config["include"] = publish_include + if publish_exclude: + config["exclude"] = publish_exclude + return await async_setup_component( + hass, statestream.DOMAIN, {statestream.DOMAIN: config} + ) + + +async def test_fails_with_no_base(hass, mqtt_mock): + """Setup should fail if no base_topic is set.""" + assert await add_statestream(hass) is False + + +async def test_setup_succeeds_without_attributes(hass, mqtt_mock): + """Test the success of the setup with a valid base_topic.""" + assert await add_statestream(hass, base_topic="pub") + + +async def test_setup_succeeds_with_attributes(hass, mqtt_mock): + """Test setup with a valid base_topic and publish_attributes.""" + assert await add_statestream(hass, base_topic="pub", publish_attributes=True) + + +async def test_state_changed_event_sends_message(hass, mqtt_mock): + """Test the sending of a new message if event changed.""" + e_id = "fake.entity" + base_topic = "pub" + + # Add the statestream component for publishing state updates + assert await add_statestream(hass, base_topic=base_topic) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State(e_id, "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + mqtt_mock.async_publish.assert_called_with("pub/fake/entity/state", "on", 1, True) + assert mqtt_mock.async_publish.called + + +async def test_state_changed_event_sends_message_and_timestamp(hass, mqtt_mock): + """Test the sending of a message and timestamps if event changed.""" + e_id = "another.entity" + base_topic = "pub" + + # Add the statestream component for publishing state updates + assert await add_statestream( + hass, base_topic=base_topic, publish_attributes=None, publish_timestamps=True + ) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State(e_id, "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + calls = [ + call.async_publish("pub/another/entity/state", "on", 1, True), + call.async_publish("pub/another/entity/last_changed", ANY, 1, True), + call.async_publish("pub/another/entity/last_updated", ANY, 1, True), + ] + + mqtt_mock.async_publish.assert_has_calls(calls, any_order=True) + assert mqtt_mock.async_publish.called + + +async def test_state_changed_attr_sends_message(hass, mqtt_mock): + """Test the sending of a new message if attribute changed.""" + e_id = "fake.entity" + base_topic = "pub" + + # Add the statestream component for publishing state updates + assert await add_statestream(hass, base_topic=base_topic, publish_attributes=True) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + test_attributes = {"testing": "YES", "list": ["a", "b", "c"], "bool": False} + + # Set a state of an entity + mock_state_change_event(hass, State(e_id, "off", attributes=test_attributes)) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + calls = [ + call.async_publish("pub/fake/entity/state", "off", 1, True), + call.async_publish("pub/fake/entity/testing", '"YES"', 1, True), + call.async_publish("pub/fake/entity/list", '["a", "b", "c"]', 1, True), + call.async_publish("pub/fake/entity/bool", "false", 1, True), + ] + + mqtt_mock.async_publish.assert_has_calls(calls, any_order=True) + assert mqtt_mock.async_publish.called + + +async def test_state_changed_event_include_domain(hass, mqtt_mock): + """Test that filtering on included domain works as expected.""" + base_topic = "pub" + + incl = {"domains": ["fake"]} + excl = {} + + # Add the statestream component for publishing state updates + # Set the filter to allow fake.* items + assert await add_statestream( + hass, base_topic=base_topic, publish_include=incl, publish_exclude=excl + ) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State("fake.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + mqtt_mock.async_publish.assert_called_with("pub/fake/entity/state", "on", 1, True) + assert mqtt_mock.async_publish.called + + mqtt_mock.async_publish.reset_mock() + # Set a state of an entity that shouldn't be included + mock_state_change_event(hass, State("fake2.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + assert not mqtt_mock.async_publish.called + + +async def test_state_changed_event_include_entity(hass, mqtt_mock): + """Test that filtering on included entity works as expected.""" + base_topic = "pub" + + incl = {"entities": ["fake.entity"]} + excl = {} + + # Add the statestream component for publishing state updates + # Set the filter to allow fake.* items + assert await add_statestream( + hass, base_topic=base_topic, publish_include=incl, publish_exclude=excl + ) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State("fake.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + mqtt_mock.async_publish.assert_called_with("pub/fake/entity/state", "on", 1, True) + assert mqtt_mock.async_publish.called + + mqtt_mock.async_publish.reset_mock() + # Set a state of an entity that shouldn't be included + mock_state_change_event(hass, State("fake.entity2", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + assert not mqtt_mock.async_publish.called + + +async def test_state_changed_event_exclude_domain(hass, mqtt_mock): + """Test that filtering on excluded domain works as expected.""" + base_topic = "pub" + + incl = {} + excl = {"domains": ["fake2"]} + + # Add the statestream component for publishing state updates + # Set the filter to allow fake.* items + assert await add_statestream( + hass, base_topic=base_topic, publish_include=incl, publish_exclude=excl + ) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State("fake.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + mqtt_mock.async_publish.assert_called_with("pub/fake/entity/state", "on", 1, True) + assert mqtt_mock.async_publish.called + + mqtt_mock.async_publish.reset_mock() + # Set a state of an entity that shouldn't be included + mock_state_change_event(hass, State("fake2.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + assert not mqtt_mock.async_publish.called + + +async def test_state_changed_event_exclude_entity(hass, mqtt_mock): + """Test that filtering on excluded entity works as expected.""" + base_topic = "pub" + + incl = {} + excl = {"entities": ["fake.entity2"]} + + # Add the statestream component for publishing state updates + # Set the filter to allow fake.* items + assert await add_statestream( + hass, base_topic=base_topic, publish_include=incl, publish_exclude=excl + ) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State("fake.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + mqtt_mock.async_publish.assert_called_with("pub/fake/entity/state", "on", 1, True) + assert mqtt_mock.async_publish.called + + mqtt_mock.async_publish.reset_mock() + # Set a state of an entity that shouldn't be included + mock_state_change_event(hass, State("fake.entity2", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + assert not mqtt_mock.async_publish.called + + +async def test_state_changed_event_exclude_domain_include_entity(hass, mqtt_mock): + """Test filtering with excluded domain and included entity.""" + base_topic = "pub" + + incl = {"entities": ["fake.entity"]} + excl = {"domains": ["fake"]} + + # Add the statestream component for publishing state updates + # Set the filter to allow fake.* items + assert await add_statestream( + hass, base_topic=base_topic, publish_include=incl, publish_exclude=excl + ) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State("fake.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + mqtt_mock.async_publish.assert_called_with("pub/fake/entity/state", "on", 1, True) + assert mqtt_mock.async_publish.called + + mqtt_mock.async_publish.reset_mock() + # Set a state of an entity that shouldn't be included + mock_state_change_event(hass, State("fake.entity2", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + assert not mqtt_mock.async_publish.called + + +async def test_state_changed_event_include_domain_exclude_entity(hass, mqtt_mock): + """Test filtering with included domain and excluded entity.""" + base_topic = "pub" + + incl = {"domains": ["fake"]} + excl = {"entities": ["fake.entity2"]} + + # Add the statestream component for publishing state updates + # Set the filter to allow fake.* items + assert await add_statestream( + hass, base_topic=base_topic, publish_include=incl, publish_exclude=excl + ) + await hass.async_block_till_done() + + # Reset the mock because it will have already gotten calls for the + # mqtt_statestream state change on initialization, etc. + mqtt_mock.async_publish.reset_mock() + + # Set a state of an entity + mock_state_change_event(hass, State("fake.entity", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + # Make sure 'on' was published to pub/fake/entity/state + mqtt_mock.async_publish.assert_called_with("pub/fake/entity/state", "on", 1, True) + assert mqtt_mock.async_publish.called + + mqtt_mock.async_publish.reset_mock() + # Set a state of an entity that shouldn't be included + mock_state_change_event(hass, State("fake.entity2", "on")) + await hass.async_block_till_done() + await hass.async_block_till_done() + + assert not mqtt_mock.async_publish.called diff --git a/tests/components/nextbus/test_sensor.py b/tests/components/nextbus/test_sensor.py index 74ea6cce127..4e7f0af8526 100644 --- a/tests/components/nextbus/test_sensor.py +++ b/tests/components/nextbus/test_sensor.py @@ -5,9 +5,10 @@ import pytest import homeassistant.components.nextbus.sensor as nextbus import homeassistant.components.sensor as sensor +from homeassistant.setup import async_setup_component from tests.async_mock import patch -from tests.common import assert_setup_component, async_setup_component +from tests.common import assert_setup_component VALID_AGENCY = "sf-muni" VALID_ROUTE = "F" diff --git a/tests/components/pi_hole/test_init.py b/tests/components/pi_hole/test_init.py index d6cee176775..088b56d75b9 100644 --- a/tests/components/pi_hole/test_init.py +++ b/tests/components/pi_hole/test_init.py @@ -2,12 +2,13 @@ from homeassistant.components import pi_hole from homeassistant.components.pi_hole.const import MIN_TIME_BETWEEN_UPDATES +from homeassistant.setup import async_setup_component from homeassistant.util import dt as dt_util from . import _create_mocked_hole, _patch_config_flow_hole from tests.async_mock import patch -from tests.common import async_fire_time_changed, async_setup_component +from tests.common import async_fire_time_changed def _patch_init_hole(mocked_hole): diff --git a/tests/components/pvpc_hourly_pricing/test_sensor.py b/tests/components/pvpc_hourly_pricing/test_sensor.py index 7ef50113de5..781ce35b99f 100644 --- a/tests/components/pvpc_hourly_pricing/test_sensor.py +++ b/tests/components/pvpc_hourly_pricing/test_sensor.py @@ -7,11 +7,12 @@ from pytz import timezone from homeassistant.components.pvpc_hourly_pricing import ATTR_TARIFF, DOMAIN from homeassistant.const import CONF_NAME from homeassistant.core import ATTR_NOW, EVENT_TIME_CHANGED +from homeassistant.setup import async_setup_component from .conftest import check_valid_state from tests.async_mock import patch -from tests.common import async_setup_component, date_util +from tests.common import date_util from tests.test_util.aiohttp import AiohttpClientMocker diff --git a/tests/components/withings/test_init.py b/tests/components/withings/test_init.py index 565a2d7e921..4f4a85585bf 100644 --- a/tests/components/withings/test_init.py +++ b/tests/components/withings/test_init.py @@ -25,7 +25,8 @@ from .common import ( new_profile_config, ) -from tests.common import MagicMock, MockConfigEntry, patch +from tests.async_mock import MagicMock, patch +from tests.common import MockConfigEntry def config_schema_validate(withings_config) -> dict: diff --git a/tests/components/yandex_transport/test_yandex_transport_sensor.py b/tests/components/yandex_transport/test_yandex_transport_sensor.py index 3583dfa0bdf..e5b6f31990b 100644 --- a/tests/components/yandex_transport/test_yandex_transport_sensor.py +++ b/tests/components/yandex_transport/test_yandex_transport_sensor.py @@ -6,10 +6,11 @@ import pytest import homeassistant.components.sensor as sensor from homeassistant.const import CONF_NAME +from homeassistant.setup import async_setup_component import homeassistant.util.dt as dt_util from tests.async_mock import patch -from tests.common import assert_setup_component, async_setup_component, load_fixture +from tests.common import assert_setup_component, load_fixture REPLY = json.loads(load_fixture("yandex_transport_reply.json"))