Rewrite demo unittest tests to pytest style test functions (#41845)

This commit is contained in:
CurrentThread 2020-10-19 23:29:44 +02:00 committed by GitHub
parent 7e3aa6ddce
commit e129ea4db7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 258 additions and 261 deletions

View file

@ -1,5 +1,4 @@
"""The tests for the demo platform."""
import unittest
import pytest
@ -14,15 +13,11 @@ from homeassistant.const import (
ATTR_UNIT_OF_MEASUREMENT,
LENGTH_KILOMETERS,
)
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 patch
from tests.common import (
assert_setup_component,
fire_time_changed,
get_test_home_assistant,
)
from tests.common import assert_setup_component, async_fire_time_changed
CONFIG = {geo_location.DOMAIN: [{"platform": "demo"}]}
@ -33,54 +28,40 @@ def mock_legacy_time(legacy_patchable_time):
yield
class TestDemoPlatform(unittest.TestCase):
"""Test the demo platform."""
async def test_setup_platform(hass):
"""Test setup of demo platform via configuration."""
utcnow = dt_util.utcnow()
# Patching 'utcnow' to gain more control over the timed update.
with patch("homeassistant.util.dt.utcnow", return_value=utcnow):
with assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(hass, geo_location.DOMAIN, CONFIG)
await hass.async_block_till_done()
def setUp(self):
"""Initialize values for this testcase class."""
self.hass = get_test_home_assistant()
self.addCleanup(self.hass.stop)
# In this test, one zone and geolocation entities have been
# generated.
all_states = [
hass.states.get(entity_id)
for entity_id in hass.states.async_entity_ids(geo_location.DOMAIN)
]
assert len(all_states) == NUMBER_OF_DEMO_DEVICES
def test_setup_platform(self):
"""Test setup of demo platform via configuration."""
utcnow = dt_util.utcnow()
# Patching 'utcnow' to gain more control over the timed update.
with patch("homeassistant.util.dt.utcnow", return_value=utcnow):
with assert_setup_component(1, geo_location.DOMAIN):
assert setup_component(self.hass, geo_location.DOMAIN, CONFIG)
self.hass.block_till_done()
for state in all_states:
# Check a single device's attributes.
if state.domain != geo_location.DOMAIN:
# ignore home zone state
continue
assert abs(state.attributes[ATTR_LATITUDE] - hass.config.latitude) < 1.0
assert abs(state.attributes[ATTR_LONGITUDE] - hass.config.longitude) < 1.0
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == LENGTH_KILOMETERS
# In this test, one zone and geolocation entities have been
# generated.
all_states = [
self.hass.states.get(entity_id)
for entity_id in self.hass.states.entity_ids(geo_location.DOMAIN)
]
assert len(all_states) == NUMBER_OF_DEMO_DEVICES
for state in all_states:
# Check a single device's attributes.
if state.domain != geo_location.DOMAIN:
# ignore home zone state
continue
assert (
abs(state.attributes[ATTR_LATITUDE] - self.hass.config.latitude)
< 1.0
)
assert (
abs(state.attributes[ATTR_LONGITUDE] - self.hass.config.longitude)
< 1.0
)
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == LENGTH_KILOMETERS
# Update (replaces 1 device).
fire_time_changed(self.hass, utcnow + DEFAULT_UPDATE_INTERVAL)
self.hass.block_till_done()
# Get all states again, ensure that the number of states is still
# the same, but the lists are different.
all_states_updated = [
self.hass.states.get(entity_id)
for entity_id in self.hass.states.entity_ids(geo_location.DOMAIN)
]
assert len(all_states_updated) == NUMBER_OF_DEMO_DEVICES
assert all_states != all_states_updated
# Update (replaces 1 device).
async_fire_time_changed(hass, utcnow + DEFAULT_UPDATE_INTERVAL)
await hass.async_block_till_done()
# Get all states again, ensure that the number of states is still
# the same, but the lists are different.
all_states_updated = [
hass.states.get(entity_id)
for entity_id in hass.states.async_entity_ids(geo_location.DOMAIN)
]
assert len(all_states_updated) == NUMBER_OF_DEMO_DEVICES
assert all_states != all_states_updated

View file

@ -1,5 +1,6 @@
"""The tests for the notify demo platform."""
import unittest
import logging
import pytest
import voluptuous as vol
@ -8,189 +9,198 @@ import homeassistant.components.demo.notify as demo
import homeassistant.components.notify as notify
from homeassistant.core import callback
from homeassistant.helpers import discovery
from homeassistant.setup import setup_component
from homeassistant.setup import async_setup_component
from tests.async_mock import patch
from tests.common import assert_setup_component, get_test_home_assistant
from tests.components.notify import common
from tests.common import assert_setup_component
CONFIG = {notify.DOMAIN: {"platform": "demo"}}
class TestNotifyDemo(unittest.TestCase):
"""Test the demo notify."""
@pytest.fixture
def events(hass):
"""Fixture that catches notify events."""
events = []
hass.bus.async_listen(demo.EVENT_NOTIFY, callback(lambda e: events.append(e)))
yield events
def setUp(self): # pylint: disable=invalid-name
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.events = []
self.calls = []
@callback
def record_event(event):
"""Record event to send notification."""
self.events.append(event)
@pytest.fixture
def calls():
"""Fixture to calls."""
return []
self.hass.bus.listen(demo.EVENT_NOTIFY, record_event)
self.addCleanup(self.tear_down_cleanup)
def tear_down_cleanup(self):
"""Stop down everything that was started."""
self.hass.stop()
def _setup_notify(self):
with assert_setup_component(1, notify.DOMAIN) as config:
assert setup_component(self.hass, notify.DOMAIN, CONFIG)
assert config[notify.DOMAIN]
self.hass.block_till_done()
def test_setup(self):
"""Test setup."""
self._setup_notify()
@patch("homeassistant.components.demo.notify.get_service", autospec=True)
def test_no_notify_service(self, mock_demo_get_service):
"""Test missing platform notify service instance."""
mock_demo_get_service.return_value = None
with self.assertLogs(
"homeassistant.components.notify", level="ERROR"
) as log_handle:
self._setup_notify()
self.hass.block_till_done()
assert mock_demo_get_service.called
assert log_handle.output == [
"ERROR:homeassistant.components.notify:"
"Failed to initialize notification service demo"
]
@patch("homeassistant.components.demo.notify.get_service", autospec=True)
def test_discover_notify(self, mock_demo_get_service):
"""Test discovery of notify demo platform."""
assert notify.DOMAIN not in self.hass.config.components
discovery.load_platform(
self.hass, "notify", "demo", {"test_key": "test_val"}, {"notify": {}}
)
self.hass.block_till_done()
assert notify.DOMAIN in self.hass.config.components
assert mock_demo_get_service.called
assert mock_demo_get_service.mock_calls[0][1] == (
self.hass,
{},
{"test_key": "test_val"},
)
@pytest.fixture
def record_calls(calls):
"""Fixture to record calls."""
@callback
def record_calls(self, *args):
def record_calls(*args):
"""Record calls."""
self.calls.append(args)
calls.append(args)
def test_sending_none_message(self):
"""Test send with None as message."""
self._setup_notify()
with pytest.raises(vol.Invalid):
common.send_message(self.hass, None)
self.hass.block_till_done()
assert len(self.events) == 0
return record_calls
def test_sending_templated_message(self):
"""Send a templated message."""
self._setup_notify()
self.hass.states.set("sensor.temperature", 10)
common.send_message(
self.hass,
"{{ states.sensor.temperature.state }}",
"{{ states.sensor.temperature.name }}",
@pytest.fixture(name="mock_demo_notify")
def mock_demo_notify_fixture():
"""Mock demo notify service."""
with patch("homeassistant.components.demo.notify.get_service", autospec=True) as ns:
yield ns
async def setup_notify(hass):
"""Test setup."""
with assert_setup_component(1, notify.DOMAIN) as config:
assert await async_setup_component(hass, notify.DOMAIN, CONFIG)
assert config[notify.DOMAIN]
await hass.async_block_till_done()
async def test_no_notify_service(hass, mock_demo_notify, caplog):
"""Test missing platform notify service instance."""
caplog.set_level(logging.ERROR)
mock_demo_notify.return_value = None
await setup_notify(hass)
await hass.async_block_till_done()
assert mock_demo_notify.called
assert "Failed to initialize notification service demo" in caplog.text
async def test_discover_notify(hass, mock_demo_notify):
"""Test discovery of notify demo platform."""
assert notify.DOMAIN not in hass.config.components
mock_demo_notify.return_value = None
discovery.load_platform(
hass, "notify", "demo", {"test_key": "test_val"}, {"notify": {}}
)
await hass.async_block_till_done()
assert notify.DOMAIN in hass.config.components
assert mock_demo_notify.called
assert mock_demo_notify.mock_calls[0][1] == (
hass,
{},
{"test_key": "test_val"},
)
async def test_sending_none_message(hass, events):
"""Test send with None as message."""
await setup_notify(hass)
with pytest.raises(vol.Invalid):
await hass.services.async_call(
notify.DOMAIN, notify.SERVICE_NOTIFY, {notify.ATTR_MESSAGE: None}
)
self.hass.block_till_done()
last_event = self.events[-1]
assert last_event.data[notify.ATTR_TITLE] == "temperature"
assert last_event.data[notify.ATTR_MESSAGE] == 10
await hass.async_block_till_done()
assert len(events) == 0
def test_method_forwards_correct_data(self):
"""Test that all data from the service gets forwarded to service."""
self._setup_notify()
common.send_message(self.hass, "my message", "my title", {"hello": "world"})
self.hass.block_till_done()
assert len(self.events) == 1
data = self.events[0].data
assert {
"message": "my message",
"title": "my title",
"data": {"hello": "world"},
} == data
def test_calling_notify_from_script_loaded_from_yaml_without_title(self):
"""Test if we can call a notify from a script."""
self._setup_notify()
step = {
"service": "notify.notify",
"data": {
"data": {
"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}
}
},
"data_template": {"message": "Test 123 {{ 2 + 2 }}\n"},
}
setup_component(self.hass, "script", {"script": {"test": {"sequence": step}}})
self.hass.services.call("script", "test")
self.hass.block_till_done()
assert len(self.events) == 1
assert {
"message": "Test 123 4",
"data": {
"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}
},
} == self.events[0].data
async def test_sending_templated_message(hass, events):
"""Send a templated message."""
await setup_notify(hass)
hass.states.async_set("sensor.temperature", 10)
data = {
notify.ATTR_MESSAGE: "{{states.sensor.temperature.state}}",
notify.ATTR_TITLE: "{{ states.sensor.temperature.name }}",
}
await hass.services.async_call(notify.DOMAIN, notify.SERVICE_NOTIFY, data)
await hass.async_block_till_done()
last_event = events[-1]
assert last_event.data[notify.ATTR_TITLE] == "temperature"
assert last_event.data[notify.ATTR_MESSAGE] == 10
def test_calling_notify_from_script_loaded_from_yaml_with_title(self):
"""Test if we can call a notify from a script."""
self._setup_notify()
step = {
"service": "notify.notify",
"data": {
"data": {
"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}
}
},
"data_template": {"message": "Test 123 {{ 2 + 2 }}\n", "title": "Test"},
}
setup_component(self.hass, "script", {"script": {"test": {"sequence": step}}})
self.hass.services.call("script", "test")
self.hass.block_till_done()
assert len(self.events) == 1
assert {
"message": "Test 123 4",
"title": "Test",
"data": {
"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}
},
} == self.events[0].data
def test_targets_are_services(self):
"""Test that all targets are exposed as individual services."""
self._setup_notify()
assert self.hass.services.has_service("notify", "demo") is not None
service = "demo_test_target_name"
assert self.hass.services.has_service("notify", service) is not None
async def test_method_forwards_correct_data(hass, events):
"""Test that all data from the service gets forwarded to service."""
await setup_notify(hass)
data = {
notify.ATTR_MESSAGE: "my message",
notify.ATTR_TITLE: "my title",
notify.ATTR_DATA: {"hello": "world"},
}
await hass.services.async_call(notify.DOMAIN, notify.SERVICE_NOTIFY, data)
await hass.async_block_till_done()
assert len(events) == 1
data = events[0].data
assert {
"message": "my message",
"title": "my title",
"data": {"hello": "world"},
} == data
def test_messages_to_targets_route(self):
"""Test message routing to specific target services."""
self._setup_notify()
self.hass.bus.listen_once("notify", self.record_calls)
self.hass.services.call(
"notify",
"demo_test_target_name",
{"message": "my message", "title": "my title", "data": {"hello": "world"}},
)
async def test_calling_notify_from_script_loaded_from_yaml_without_title(hass, events):
"""Test if we can call a notify from a script."""
await setup_notify(hass)
step = {
"service": "notify.notify",
"data": {
"data": {"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}}
},
"data_template": {"message": "Test 123 {{ 2 + 2 }}\n"},
}
await async_setup_component(
hass, "script", {"script": {"test": {"sequence": step}}}
)
await hass.services.async_call("script", "test")
await hass.async_block_till_done()
assert len(events) == 1
assert {
"message": "Test 123 4",
"data": {"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}},
} == events[0].data
self.hass.block_till_done()
data = self.calls[0][0].data
async def test_calling_notify_from_script_loaded_from_yaml_with_title(hass, events):
"""Test if we can call a notify from a script."""
await setup_notify(hass)
step = {
"service": "notify.notify",
"data": {
"data": {"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}}
},
"data_template": {"message": "Test 123 {{ 2 + 2 }}\n", "title": "Test"},
}
await async_setup_component(
hass, "script", {"script": {"test": {"sequence": step}}}
)
await hass.services.async_call("script", "test")
await hass.async_block_till_done()
assert len(events) == 1
assert {
"message": "Test 123 4",
"title": "Test",
"data": {"push": {"sound": "US-EN-Morgan-Freeman-Roommate-Is-Arriving.wav"}},
} == events[0].data
assert {
"message": "my message",
"target": ["test target id"],
"title": "my title",
"data": {"hello": "world"},
} == data
async def test_targets_are_services(hass):
"""Test that all targets are exposed as individual services."""
await setup_notify(hass)
assert hass.services.has_service("notify", "demo") is not None
service = "demo_test_target_name"
assert hass.services.has_service("notify", service) is not None
async def test_messages_to_targets_route(hass, calls, record_calls):
"""Test message routing to specific target services."""
await setup_notify(hass)
hass.bus.async_listen_once("notify", record_calls)
await hass.services.async_call(
"notify",
"demo_test_target_name",
{"message": "my message", "title": "my title", "data": {"hello": "world"}},
)
await hass.async_block_till_done()
data = calls[0][0].data
assert {
"message": "my message",
"target": ["test target id"],
"title": "my title",
"data": {"hello": "world"},
} == data

View file

@ -1,57 +1,63 @@
"""The tests for the demo remote component."""
# pylint: disable=protected-access
import unittest
import pytest
import homeassistant.components.remote as remote
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.setup import setup_component
from tests.common import get_test_home_assistant
from tests.components.remote import common
from homeassistant.components.remote import ATTR_COMMAND
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.setup import async_setup_component
ENTITY_ID = "remote.remote_one"
SERVICE_SEND_COMMAND = "send_command"
class TestDemoRemote(unittest.TestCase):
"""Test the demo remote."""
@pytest.fixture(autouse=True)
async def setup_component(hass):
"""Initialize components."""
assert await async_setup_component(
hass, remote.DOMAIN, {"remote": {"platform": "demo"}}
)
await hass.async_block_till_done()
# pylint: disable=invalid-name
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
assert setup_component(
self.hass, remote.DOMAIN, {"remote": {"platform": "demo"}}
)
self.hass.block_till_done()
self.addCleanup(self.tear_down_cleanup)
async def test_methods(hass):
"""Test if services call the entity methods as expected."""
await hass.services.async_call(
remote.DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
def tear_down_cleanup(self):
"""Stop down everything that was started."""
self.hass.stop()
await hass.services.async_call(
remote.DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
def test_methods(self):
"""Test if services call the entity methods as expected."""
common.turn_on(self.hass, entity_id=ENTITY_ID)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
await hass.services.async_call(
remote.DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}
)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
common.turn_off(self.hass, entity_id=ENTITY_ID)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_ID)
assert state.state == STATE_OFF
data = {
ATTR_ENTITY_ID: ENTITY_ID,
ATTR_COMMAND: ["test"],
}
common.turn_on(self.hass, entity_id=ENTITY_ID)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_ID)
assert state.state == STATE_ON
common.send_command(self.hass, "test", entity_id=ENTITY_ID)
self.hass.block_till_done()
state = self.hass.states.get(ENTITY_ID)
assert state.attributes == {
"friendly_name": "Remote One",
"last_command_sent": "test",
"supported_features": 0,
}
await hass.services.async_call(remote.DOMAIN, SERVICE_SEND_COMMAND, data)
await hass.async_block_till_done()
state = hass.states.get(ENTITY_ID)
assert state.attributes == {
"friendly_name": "Remote One",
"last_command_sent": "test",
"supported_features": 0,
}