<s> """<STR_LIT>""" <EOL> import json <EOL> import unittest <EOL> from unittest . mock import ANY , patch <EOL> import homeassistant . components . mqtt_eventstream as eventstream <EOL> from homeassistant . const import EVENT_STATE_CHANGED <EOL> from homeassistant . core import State <EOL> from homeassistant . remote import JSONEncoder <EOL> import homeassistant . util . dt as dt_util <EOL> from tests . common import ( <EOL> get_test_home_assistant , <EOL> mock_mqtt_component , <EOL> fire_mqtt_message , <EOL> mock_state_change_event , <EOL> fire_time_changed <EOL> ) <EOL> class TestMqttEventStream ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( TestMqttEventStream , self ) . setUp ( ) <EOL> self . hass = get_test_home_assistant ( ) <EOL> self . mock_mqtt = mock_mqtt_component ( self . hass ) <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> self . hass . stop ( ) <EOL> def add_eventstream ( self , sub_topic = None , pub_topic = None ) : <EOL> """<STR_LIT>""" <EOL> config = { } <EOL> if sub_topic : <EOL> config [ '<STR_LIT>' ] = sub_topic <EOL> if pub_topic : <EOL> config [ '<STR_LIT>' ] = pub_topic <EOL> return eventstream . setup ( self . hass , { eventstream . DOMAIN : config } ) <EOL> def test_setup_succeeds ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( self . add_eventstream ( ) ) <EOL> def test_setup_with_pub ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . hass . bus . listeners . get ( '<STR_LIT:*>' ) , None ) <EOL> self . assertTrue ( self . add_eventstream ( pub_topic = '<STR_LIT:bar>' ) ) <EOL> self . hass . pool . block_till_done ( ) <EOL> self . assertEqual ( self . hass . bus . listeners . get ( '<STR_LIT:*>' ) , <NUM_LIT:1> ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_subscribe ( self , mock_sub ) : <EOL> """<STR_LIT>""" <EOL> sub_topic = '<STR_LIT:foo>' <EOL> self . assertTrue ( self . add_eventstream ( sub_topic = sub_topic ) ) <EOL> self . hass . pool . block_till_done ( ) <EOL> mock_sub . assert_called_with ( self . hass , sub_topic , ANY ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_state_changed_event_sends_message ( self , mock_utcnow , mock_pub ) : <EOL> """<STR_LIT>""" <EOL> now = dt_util . as_utc ( dt_util . now ( ) ) <EOL> e_id = '<STR_LIT>' <EOL> pub_topic = '<STR_LIT:bar>' <EOL> mock_utcnow . return_value = now <EOL> self . assertTrue ( self . add_eventstream ( pub_topic = pub_topic ) ) <EOL> self . hass . pool . block_till_done ( ) <EOL> mock_pub . reset_mock ( ) <EOL> mock_state_change_event ( self . hass , State ( e_id , '<STR_LIT>' ) ) <EOL> self . hass . pool . block_till_done ( ) <EOL> mock_pub . assert_called_with ( self . hass , pub_topic , ANY ) <EOL> self . assertTrue ( mock_pub . called ) <EOL> msg = mock_pub . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] <EOL> event = { } <EOL> event [ '<STR_LIT>' ] = EVENT_STATE_CHANGED <EOL> new_state = { <EOL> "<STR_LIT>" : now . isoformat ( ) , <EOL> "<STR_LIT:state>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : e_id , <EOL> "<STR_LIT>" : { } , <EOL> "<STR_LIT>" : now . isoformat ( ) <EOL> } <EOL> event [ '<STR_LIT>' ] = { "<STR_LIT>" : new_state , "<STR_LIT>" : e_id } <EOL> self . assertEqual ( json . loads ( msg ) , event ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_time_event_does_not_send_message ( self , mock_pub ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( self . add_eventstream ( pub_topic = '<STR_LIT:bar>' ) ) <EOL> self . hass . pool . block_till_done ( ) <EOL> mock_pub . reset_mock ( ) <EOL> fire_time_changed ( self . hass , dt_util . utcnow ( ) ) <EOL> self . assertFalse ( mock_pub . called ) <EOL> def test_receiving_remote_event_fires_hass_event ( self ) : <EOL> """<STR_LIT>""" <EOL> sub_topic = '<STR_LIT:foo>' <EOL> self . assertTrue ( self . add_eventstream ( sub_topic = sub_topic ) ) <EOL> self . hass . pool . block_till_done ( ) <EOL> calls = [ ] <EOL> self . hass . bus . listen_once ( '<STR_LIT>' , lambda _ : calls . append ( <NUM_LIT:1> ) ) <EOL> self . hass . pool . block_till_done ( ) <EOL> payload = json . dumps ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : { } } , <EOL> cls = JSONEncoder <EOL> ) <EOL> fire_mqtt_message ( self . hass , sub_topic , payload ) <EOL> self . hass . pool . block_till_done ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( calls ) ) </s>
