1
0
mirror of https://github.com/home-assistant/core.git synced 2025-09-05 11:16:21 +00:00
Files
.devcontainer
.github
.vscode
docs
homeassistant
script
tests
auth
components
abode
adguard
air_quality
airly
alarm_control_panel
alert
alexa
almond
ambiclimate
ambient_station
androidtv
api
api_streams
apns
apprise
aprs
arcam_fmj
arlo
asuswrt
aurora
auth
automatic
automation
__init__.py
common.py
test_event.py
test_geo_location.py
test_homeassistant.py
test_init.py
test_litejet.py
test_mqtt.py
test_numeric_state.py
test_reproduce_state.py
test_state.py
test_sun.py
test_template.py
test_time.py
test_time_pattern.py
test_webhook.py
test_zone.py
awair
aws
axis
bayesian
binary_sensor
blackbird
bom
broadlink
buienradar
caldav
calendar
camera
canary
cast
cert_expiry
climate
cloud
coinmarketcap
command_line
config
configurator
conversation
coolmaster
counter
cover
daikin
darksky
datadog
deconz
default_config
demo
device_automation
device_sun_light_trigger
device_tracker
dialogflow
directv
discovery
dsmr
dte_energy_bridge
duckdns
dyson
ecobee
ee_brightbox
efergy
emulated_hue
emulated_roku
esphome
everlights
facebook
facebox
fail2ban
fan
feedreader
ffmpeg
fido
file
filesize
filter
flux
folder
folder_watcher
foobot
freedns
fritzbox
frontend
generic
generic_thermostat
geo_json_events
geo_location
geo_rss_events
geofency
geonetnz_quakes
glances
google
google_assistant
google_domains
google_pubsub
google_translate
google_wifi
gpslogger
graphite
group
hangouts
hassio
hddtemp
heos
here_travel_time
history
history_graph
history_stats
homeassistant
homekit
homekit_controller
homematic
homematicip_cloud
honeywell
html5
http
huawei_lte
hue
iaqualink
ifttt
ign_sismologia
image_processing
imap_email_content
influxdb
input_boolean
input_datetime
input_number
input_select
input_text
integration
intent_script
ios
ipma
iqvia
islamic_prayer_times
izone
jewish_calendar
kira
light
linky
litejet
local_file
locative
lock
logbook
logentries
logger
logi_circle
london_air
lovelace
luftdaten
mailbox
mailgun
manual
manual_mqtt
marytts
media_player
melissa
meraki
met
mfi
mhz19
microsoft_face
microsoft_face_detect
microsoft_face_identify
min_max
minio
mobile_app
mochad
modbus
mold_indicator
monoprice
moon
mqtt
mqtt_eventstream
mqtt_json
mqtt_room
mqtt_statestream
mythicbeastsdns
namecheapdns
neato
ness_alarm
nest
nextbus
no_ip
notify
notion
nsw_fuel_station
nsw_rural_fire_service_feed
nuheat
nws
nx584
onboarding
openalpr_cloud
openalpr_local
openhardwaremonitor
opentherm_gw
openuv
owntracks
panel_custom
panel_iframe
persistent_notification
person
pi_hole
pilight
plant
plex
point
prometheus
proximity
ps4
ptvsd
push
pushbullet
python_script
qld_bushfire
qwikswitch
radarr
rainmachine
random
recorder
reddit
remember_the_milk
remote
rest
rest_command
rflink
rfxtrx
ring
rmvtransport
rss_feed_template
samsungtv
scene
script
season
sensor
seventeentrack
shell_command
shopping_list
sigfox
simplisafe
simulated
sleepiq
sma
smartthings
smhi
smtp
snips
solaredge
solarlog
soma
somfy
sonarr
sonos
soundtouch
spaceapi
spc
splunk
sql
ssdp
startca
statistics
statsd
stream
sun
switch
switcher_kis
system_health
system_log
tcp
teksavvy
tellduslive
template
threshold
time_date
timer
tod
tomato
toon
tplink
traccar
tradfri
transmission
transport_nsw
trend
tts
twentemilieu
twilio
uk_transport
unifi
unifi_direct
universal
updater
upnp
uptime
usgs_earthquakes_feed
utility_meter
uvc
vacuum
velbus
verisure
version
vesync
voicerss
vultr
wake_on_lan
water_heater
weather
webhook
weblink
webostv
websocket_api
withings
workday
worldclock
wsdot
wunderground
wwlln
xiaomi
xiaomi_miio
yamaha
yandex_transport
yandextts
yessssms
yr
yweather
zeroconf
zha
zone
zwave
__init__.py
conftest.py
fixtures
helpers
mock
resources
scripts
test_util
testing_config
util
__init__.py
common.py
conftest.py
test_bootstrap.py
test_config.py
test_config_entries.py
test_core.py
test_data_entry_flow.py
test_loader.py
test_main.py
test_requirements.py
test_setup.py
.codecov.yml
.coveragerc
.dockerignore
.gitattributes
.gitignore
.hound.yml
.ignore
.pre-commit-config.yaml
.readthedocs.yml
.travis.yml
CLA.md
CODEOWNERS
CODE_OF_CONDUCT.md
CONTRIBUTING.md
Dockerfile.dev
LICENSE.md
MANIFEST.in
README.rst
azure-pipelines-ci.yml
azure-pipelines-release.yml
azure-pipelines-translation.yml
azure-pipelines-wheels.yml
pylintrc
pyproject.toml
requirements_all.txt
requirements_docs.txt
requirements_test.txt
requirements_test_all.txt
setup.cfg
setup.py
tox.ini
core/tests/components/automation/test_event.py
Paulus Schoutsen 4de97abc3a Black
2019-07-31 12:25:30 -07:00

170 lines
4.8 KiB
Python

"""The tests for the Event automation."""
import pytest
from homeassistant.core import Context
from homeassistant.setup import async_setup_component
import homeassistant.components.automation as automation
from tests.common import mock_component
from tests.components.automation import common
from tests.common import async_mock_service
@pytest.fixture
def calls(hass):
"""Track calls to a mock serivce."""
return async_mock_service(hass, "test", "automation")
@pytest.fixture(autouse=True)
def setup_comp(hass):
"""Initialize components."""
mock_component(hass, "group")
async def test_if_fires_on_event(hass, calls):
"""Test the firing of events."""
context = Context()
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {"platform": "event", "event_type": "test_event"},
"action": {"service": "test.automation"},
}
},
)
hass.bus.async_fire("test_event", context=context)
await hass.async_block_till_done()
assert 1 == len(calls)
assert calls[0].context.parent_id == context.id
await common.async_turn_off(hass)
await hass.async_block_till_done()
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_event_extra_data(hass, calls):
"""Test the firing of events still matches with event data."""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {"platform": "event", "event_type": "test_event"},
"action": {"service": "test.automation"},
}
},
)
hass.bus.async_fire("test_event", {"extra_key": "extra_data"})
await hass.async_block_till_done()
assert 1 == len(calls)
await common.async_turn_off(hass)
await hass.async_block_till_done()
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_event_with_data(hass, calls):
"""Test the firing of events with data."""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
"platform": "event",
"event_type": "test_event",
"event_data": {"some_attr": "some_value"},
},
"action": {"service": "test.automation"},
}
},
)
hass.bus.async_fire("test_event", {"some_attr": "some_value", "another": "value"})
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_event_with_empty_data_config(hass, calls):
"""Test the firing of events with empty data config.
The frontend automation editor can produce configurations with an
empty dict for event_data instead of no key.
"""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
"platform": "event",
"event_type": "test_event",
"event_data": {},
},
"action": {"service": "test.automation"},
}
},
)
hass.bus.async_fire("test_event", {"some_attr": "some_value", "another": "value"})
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_fires_on_event_with_nested_data(hass, calls):
"""Test the firing of events with nested data."""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
"platform": "event",
"event_type": "test_event",
"event_data": {"parent_attr": {"some_attr": "some_value"}},
},
"action": {"service": "test.automation"},
}
},
)
hass.bus.async_fire(
"test_event", {"parent_attr": {"some_attr": "some_value", "another": "value"}}
)
await hass.async_block_till_done()
assert 1 == len(calls)
async def test_if_not_fires_if_event_data_not_matches(hass, calls):
"""Test firing of event if no match."""
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {
"platform": "event",
"event_type": "test_event",
"event_data": {"some_attr": "some_value"},
},
"action": {"service": "test.automation"},
}
},
)
hass.bus.async_fire("test_event", {"some_attr": "some_other_value"})
await hass.async_block_till_done()
assert 0 == len(calls)