evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge multiple context managers in tests (#48146)

Browse source
This commit is contained in:
Franck Nijhof 2021-03-27 09:17:15 +01:00 • committed by GitHub
parent 79af18a8ab
commit ad13a9295e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
33 changed files with 590 additions and 627 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
tests/auth/test_init.py
View file

@ -501,9 +501,8 @@ async def test_refresh_token_provider_validation(mock_hass):
with patch(
"homeassistant.auth.providers.insecure_example.ExampleAuthProvider.async_validate_refresh_token",
side_effect=InvalidAuthError("Invalid access"),
) as call:
with pytest.raises(InvalidAuthError):
manager.async_create_access_token(refresh_token, ip)
) as call, pytest.raises(InvalidAuthError):
manager.async_create_access_token(refresh_token, ip)
call.assert_called_with(refresh_token, ip)

34
tests/components/androidtv/test_media_player.py
View file

@ -933,12 +933,11 @@ async def test_update_lock_not_acquired(hass):
with patch(
"androidtv.androidtv.androidtv_async.AndroidTVAsync.update",
side_effect=LockNotAcquiredException,
):
with patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key]:
await hass.helpers.entity_component.async_update_entity(entity_id)
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_OFF
), patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key]:
await hass.helpers.entity_component.async_update_entity(entity_id)
state = hass.states.get(entity_id)
assert state is not None
assert state.state == STATE_OFF
with patchers.patch_shell(SHELL_RESPONSE_STANDBY)[patch_key]:
await hass.helpers.entity_component.async_update_entity(entity_id)
@ -1206,19 +1205,18 @@ async def test_connection_closed_on_ha_stop(hass):
"""Test that the ADB socket connection is closed when HA stops."""
patch_key, entity_id = _setup(CONFIG_ANDROIDTV_ADB_SERVER)
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[patch_key]:
with patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(
hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER
)
await hass.async_block_till_done()
with patchers.PATCH_ADB_DEVICE_TCP, patchers.patch_connect(True)[
patch_key
], patchers.patch_shell(SHELL_RESPONSE_OFF)[patch_key]:
assert await async_setup_component(hass, DOMAIN, CONFIG_ANDROIDTV_ADB_SERVER)
await hass.async_block_till_done()
with patch(
"androidtv.androidtv.androidtv_async.AndroidTVAsync.adb_close"
) as adb_close:
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
assert adb_close.called
with patch(
"androidtv.androidtv.androidtv_async.AndroidTVAsync.adb_close"
) as adb_close:
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
await hass.async_block_till_done()
assert adb_close.called
async def test_exception(hass):

13
tests/components/apprise/test_notify.py
View file

@ -28,13 +28,14 @@ async def test_apprise_config_load_fail02(hass):
BASE_COMPONENT: {"name": "test", "platform": "apprise", "config": "/path/"}
}
with patch("apprise.Apprise.add", return_value=False):
with patch("apprise.AppriseConfig.add", return_value=True):
assert await async_setup_component(hass, BASE_COMPONENT, config)
await hass.async_block_till_done()
with patch("apprise.Apprise.add", return_value=False), patch(
"apprise.AppriseConfig.add", return_value=True
):
assert await async_setup_component(hass, BASE_COMPONENT, config)
await hass.async_block_till_done()
# Test that our service failed to load
assert not hass.services.has_service(BASE_COMPONENT, "test")
# Test that our service failed to load
assert not hass.services.has_service(BASE_COMPONENT, "test")
async def test_apprise_config_load_okay(hass, tmp_path):

31
tests/components/bond/common.py
View file

@ -72,20 +72,23 @@ async def setup_platform(
)
mock_entry.add_to_hass(hass)
with patch("homeassistant.components.bond.PLATFORMS", [platform]):
with patch_bond_version(return_value=bond_version), patch_bond_bridge(
return_value=bridge
), patch_bond_token(return_value=token), patch_bond_device_ids(
return_value=[bond_device_id]
), patch_start_bpup(), patch_bond_device(
return_value=discovered_device
), patch_bond_device_properties(
return_value=props
), patch_bond_device_state(
return_value=state
):
assert await async_setup_component(hass, BOND_DOMAIN, {})
await hass.async_block_till_done()
with patch(
"homeassistant.components.bond.PLATFORMS", [platform]
), patch_bond_version(return_value=bond_version), patch_bond_bridge(
return_value=bridge
), patch_bond_token(
return_value=token
), patch_bond_device_ids(
return_value=[bond_device_id]
), patch_start_bpup(), patch_bond_device(
return_value=discovered_device
), patch_bond_device_properties(
return_value=props
), patch_bond_device_state(
return_value=state
):
assert await async_setup_component(hass, BOND_DOMAIN, {})
await hass.async_block_till_done()
return mock_entry

50
tests/components/device_tracker/test_init.py
View file

@ -237,19 +237,18 @@ async def test_update_stale(hass, mock_device_tracker_conf):
with patch(
"homeassistant.components.device_tracker.legacy.dt_util.utcnow",
return_value=register_time,
):
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(
hass,
device_tracker.DOMAIN,
{
device_tracker.DOMAIN: {
CONF_PLATFORM: "test",
device_tracker.CONF_CONSIDER_HOME: 59,
}
},
)
await hass.async_block_till_done()
), assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(
hass,
device_tracker.DOMAIN,
{
device_tracker.DOMAIN: {
CONF_PLATFORM: "test",
device_tracker.CONF_CONSIDER_HOME: 59,
}
},
)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.dev1").state == STATE_HOME
@ -458,19 +457,18 @@ async def test_see_passive_zone_state(hass, mock_device_tracker_conf):
with patch(
"homeassistant.components.device_tracker.legacy.dt_util.utcnow",
return_value=register_time,
):
with assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(
hass,
device_tracker.DOMAIN,
{
device_tracker.DOMAIN: {
CONF_PLATFORM: "test",
device_tracker.CONF_CONSIDER_HOME: 59,
}
},
)
await hass.async_block_till_done()
), assert_setup_component(1, device_tracker.DOMAIN):
assert await async_setup_component(
hass,
device_tracker.DOMAIN,
{
device_tracker.DOMAIN: {
CONF_PLATFORM: "test",
device_tracker.CONF_CONSIDER_HOME: 59,
}
},
)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.dev1")
attrs = state.attributes

25
tests/components/elgato/test_light.py
View file

@ -93,17 +93,16 @@ async def test_light_unavailable(
with patch(
"homeassistant.components.elgato.light.Elgato.light",
side_effect=ElgatoError,
), patch(
"homeassistant.components.elgato.light.Elgato.state",
side_effect=ElgatoError,
):
with patch(
"homeassistant.components.elgato.light.Elgato.state",
side_effect=ElgatoError,
):
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.frenck"},
blocking=True,
)
await hass.async_block_till_done()
state = hass.states.get("light.frenck")
assert state.state == STATE_UNAVAILABLE
await hass.services.async_call(
LIGHT_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "light.frenck"},
blocking=True,
)
await hass.async_block_till_done()
state = hass.states.get("light.frenck")
assert state.state == STATE_UNAVAILABLE

105
tests/components/emulated_hue/test_init.py
View file

@ -13,29 +13,30 @@ def test_config_google_home_entity_id_to_number():
with patch(
"homeassistant.components.emulated_hue.load_json",
return_value={"1": "light.test2"},
) as json_loader:
with patch("homeassistant.components.emulated_hue.save_json") as json_saver:
number = conf.entity_id_to_number("light.test")
assert number == "2"
) as json_loader, patch(
"homeassistant.components.emulated_hue.save_json"
) as json_saver:
number = conf.entity_id_to_number("light.test")
assert number == "2"
assert json_saver.mock_calls[0][1][1] == {
"1": "light.test2",
"2": "light.test",
}
assert json_saver.mock_calls[0][1][1] == {
"1": "light.test2",
"2": "light.test",
}
assert json_saver.call_count == 1
assert json_loader.call_count == 1
assert json_saver.call_count == 1
assert json_loader.call_count == 1
number = conf.entity_id_to_number("light.test")
assert number == "2"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test")
assert number == "2"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test2")
assert number == "1"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test2")
assert number == "1"
assert json_saver.call_count == 1
entity_id = conf.number_to_entity_id("1")
assert entity_id == "light.test2"
entity_id = conf.number_to_entity_id("1")
assert entity_id == "light.test2"
def test_config_google_home_entity_id_to_number_altered():
@ -47,28 +48,29 @@ def test_config_google_home_entity_id_to_number_altered():
with patch(
"homeassistant.components.emulated_hue.load_json",
return_value={"21": "light.test2"},
) as json_loader:
with patch("homeassistant.components.emulated_hue.save_json") as json_saver:
number = conf.entity_id_to_number("light.test")
assert number == "22"
assert json_saver.call_count == 1
assert json_loader.call_count == 1
) as json_loader, patch(
"homeassistant.components.emulated_hue.save_json"
) as json_saver:
number = conf.entity_id_to_number("light.test")
assert number == "22"
assert json_saver.call_count == 1
assert json_loader.call_count == 1
assert json_saver.mock_calls[0][1][1] == {
"21": "light.test2",
"22": "light.test",
}
assert json_saver.mock_calls[0][1][1] == {
"21": "light.test2",
"22": "light.test",
}
number = conf.entity_id_to_number("light.test")
assert number == "22"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test")
assert number == "22"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test2")
assert number == "21"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test2")
assert number == "21"
assert json_saver.call_count == 1
entity_id = conf.number_to_entity_id("21")
assert entity_id == "light.test2"
entity_id = conf.number_to_entity_id("21")
assert entity_id == "light.test2"
def test_config_google_home_entity_id_to_number_empty():
@ -79,25 +81,26 @@ def test_config_google_home_entity_id_to_number_empty():
with patch(
"homeassistant.components.emulated_hue.load_json", return_value={}
) as json_loader:
with patch("homeassistant.components.emulated_hue.save_json") as json_saver:
number = conf.entity_id_to_number("light.test")
assert number == "1"
assert json_saver.call_count == 1
assert json_loader.call_count == 1
) as json_loader, patch(
"homeassistant.components.emulated_hue.save_json"
) as json_saver:
number = conf.entity_id_to_number("light.test")
assert number == "1"
assert json_saver.call_count == 1
assert json_loader.call_count == 1
assert json_saver.mock_calls[0][1][1] == {"1": "light.test"}
assert json_saver.mock_calls[0][1][1] == {"1": "light.test"}
number = conf.entity_id_to_number("light.test")
assert number == "1"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test")
assert number == "1"
assert json_saver.call_count == 1
number = conf.entity_id_to_number("light.test2")
assert number == "2"
assert json_saver.call_count == 2
number = conf.entity_id_to_number("light.test2")
assert number == "2"
assert json_saver.call_count == 2
entity_id = conf.number_to_entity_id("2")
assert entity_id == "light.test2"
entity_id = conf.number_to_entity_id("2")
assert entity_id == "light.test2"
def test_config_alexa_entity_id_to_number():

30
tests/components/enocean/test_config_flow.py
View file

@ -73,13 +73,14 @@ async def test_detection_flow_with_custom_path(hass):
USER_PROVIDED_PATH = EnOceanFlowHandler.MANUAL_PATH_VALUE
FAKE_DONGLE_PATH = "/fake/dongle"
with patch(DONGLE_VALIDATE_PATH_METHOD, Mock(return_value=True)):
with patch(DONGLE_DETECT_METHOD, Mock(return_value=[FAKE_DONGLE_PATH])):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "detect"},
data={CONF_DEVICE: USER_PROVIDED_PATH},
)
with patch(DONGLE_VALIDATE_PATH_METHOD, Mock(return_value=True)), patch(
DONGLE_DETECT_METHOD, Mock(return_value=[FAKE_DONGLE_PATH])
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "detect"},
data={CONF_DEVICE: USER_PROVIDED_PATH},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "manual"
@ -90,13 +91,14 @@ async def test_detection_flow_with_invalid_path(hass):
USER_PROVIDED_PATH = "/invalid/path"
FAKE_DONGLE_PATH = "/fake/dongle"
with patch(DONGLE_VALIDATE_PATH_METHOD, Mock(return_value=False)):
with patch(DONGLE_DETECT_METHOD, Mock(return_value=[FAKE_DONGLE_PATH])):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "detect"},
data={CONF_DEVICE: USER_PROVIDED_PATH},
)
with patch(DONGLE_VALIDATE_PATH_METHOD, Mock(return_value=False)), patch(
DONGLE_DETECT_METHOD, Mock(return_value=[FAKE_DONGLE_PATH])
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": "detect"},
data={CONF_DEVICE: USER_PROVIDED_PATH},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "detect"

48
tests/components/filter/test_sensor.py
View file

@ -116,24 +116,23 @@ async def test_chain_history(hass, values, missing=False):
with patch(
"homeassistant.components.history.state_changes_during_period",
return_value=fake_states,
), patch(
"homeassistant.components.history.get_last_state_changes",
return_value=fake_states,
):
with patch(
"homeassistant.components.history.get_last_state_changes",
return_value=fake_states,
):
with assert_setup_component(1, "sensor"):
assert await async_setup_component(hass, "sensor", config)
await hass.async_block_till_done()
with assert_setup_component(1, "sensor"):
assert await async_setup_component(hass, "sensor", config)
await hass.async_block_till_done()
for value in values:
hass.states.async_set(config["sensor"]["entity_id"], value.state)
await hass.async_block_till_done()
for value in values:
hass.states.async_set(config["sensor"]["entity_id"], value.state)
await hass.async_block_till_done()
state = hass.states.get("sensor.test")
if missing:
assert state.state == "18.05"
else:
assert state.state == "17.05"
state = hass.states.get("sensor.test")
if missing:
assert state.state == "18.05"
else:
assert state.state == "17.05"
async def test_source_state_none(hass, values):
@ -234,18 +233,17 @@ async def test_history_time(hass):
with patch(
"homeassistant.components.history.state_changes_during_period",
return_value=fake_states,
), patch(
"homeassistant.components.history.get_last_state_changes",
return_value=fake_states,
):
with patch(
"homeassistant.components.history.get_last_state_changes",
return_value=fake_states,
):
with assert_setup_component(1, "sensor"):
assert await async_setup_component(hass, "sensor", config)
await hass.async_block_till_done()
with assert_setup_component(1, "sensor"):
assert await async_setup_component(hass, "sensor", config)
await hass.async_block_till_done()
state = hass.states.get("sensor.test")
assert state.state == "18.0"
await hass.async_block_till_done()
state = hass.states.get("sensor.test")
assert state.state == "18.0"
async def test_setup(hass):

91
tests/components/geo_json_events/test_geo_location.py
View file

@ -198,60 +198,59 @@ async def test_setup_race_condition(hass, legacy_patchable_time):
utcnow = dt_util.utcnow()
with patch("homeassistant.util.dt.utcnow", return_value=utcnow), patch(
"geojson_client.generic_feed.GenericFeed"
) as mock_feed:
with assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(hass, geo_location.DOMAIN, CONFIG)
await hass.async_block_till_done()
) as mock_feed, assert_setup_component(1, geo_location.DOMAIN):
assert await async_setup_component(hass, geo_location.DOMAIN, CONFIG)
await hass.async_block_till_done()
mock_feed.return_value.update.return_value = "OK", [mock_entry_1]
mock_feed.return_value.update.return_value = "OK", [mock_entry_1]
# Artificially trigger update.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
# Artificially trigger update.
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 1
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 1
all_states = hass.states.async_all()
assert len(all_states) == 1
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 1
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 1
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = "ERROR", None
async_fire_time_changed(hass, utcnow + SCAN_INTERVAL)
await hass.async_block_till_done()
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = "ERROR", None
async_fire_time_changed(hass, utcnow + SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 0
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 0
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 0
all_states = hass.states.async_all()
assert len(all_states) == 0
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 0
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 0
# Simulate an update - 1 entry
mock_feed.return_value.update.return_value = "OK", [mock_entry_1]
async_fire_time_changed(hass, utcnow + 2 * SCAN_INTERVAL)
await hass.async_block_till_done()
# Simulate an update - 1 entry
mock_feed.return_value.update.return_value = "OK", [mock_entry_1]
async_fire_time_changed(hass, utcnow + 2 * SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 1
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 1
all_states = hass.states.async_all()
assert len(all_states) == 1
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 1
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 1
# Simulate an update - 1 entry
mock_feed.return_value.update.return_value = "OK", [mock_entry_1]
async_fire_time_changed(hass, utcnow + 3 * SCAN_INTERVAL)
await hass.async_block_till_done()
# Simulate an update - 1 entry
mock_feed.return_value.update.return_value = "OK", [mock_entry_1]
async_fire_time_changed(hass, utcnow + 3 * SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 1
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 1
all_states = hass.states.async_all()
assert len(all_states) == 1
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 1
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 1
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = "ERROR", None
async_fire_time_changed(hass, utcnow + 4 * SCAN_INTERVAL)
await hass.async_block_till_done()
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = "ERROR", None
async_fire_time_changed(hass, utcnow + 4 * SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 0
# Ensure that delete and update signal targets are now empty.
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 0
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 0
all_states = hass.states.async_all()
assert len(all_states) == 0
# Ensure that delete and update signal targets are now empty.
assert len(hass.data[DATA_DISPATCHER][delete_signal]) == 0
assert len(hass.data[DATA_DISPATCHER][update_signal]) == 0

85
tests/components/geo_rss_events/test_sensor.py
View file

@ -68,54 +68,55 @@ async def test_setup(hass, mock_feed):
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, sensor.DOMAIN):
assert await async_setup_component(hass, sensor.DOMAIN, VALID_CONFIG)
# Artificially trigger update.
hass.bus.fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
with patch(
"homeassistant.util.dt.utcnow", return_value=utcnow
), assert_setup_component(1, sensor.DOMAIN):
assert await async_setup_component(hass, sensor.DOMAIN, VALID_CONFIG)
# Artificially trigger update.
hass.bus.fire(EVENT_HOMEASSISTANT_START)
# Collect events.
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
all_states = hass.states.async_all()
assert len(all_states) == 1
state = hass.states.get("sensor.event_service_any")
assert state is not None
assert state.name == "Event Service Any"
assert int(state.state) == 2
assert state.attributes == {
ATTR_FRIENDLY_NAME: "Event Service Any",
ATTR_UNIT_OF_MEASUREMENT: "Events",
ATTR_ICON: "mdi:alert",
"Title 1": "16km",
"Title 2": "20km",
}
state = hass.states.get("sensor.event_service_any")
assert state is not None
assert state.name == "Event Service Any"
assert int(state.state) == 2
assert state.attributes == {
ATTR_FRIENDLY_NAME: "Event Service Any",
ATTR_UNIT_OF_MEASUREMENT: "Events",
ATTR_ICON: "mdi:alert",
"Title 1": "16km",
"Title 2": "20km",
}
# Simulate an update - empty data, but successful update,
# so no changes to entities.
mock_feed.return_value.update.return_value = "OK_NO_DATA", None
async_fire_time_changed(hass, utcnow + geo_rss_events.SCAN_INTERVAL)
await hass.async_block_till_done()
# Simulate an update - empty data, but successful update,
# so no changes to entities.
mock_feed.return_value.update.return_value = "OK_NO_DATA", None
async_fire_time_changed(hass, utcnow + geo_rss_events.SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
state = hass.states.get("sensor.event_service_any")
assert int(state.state) == 2
all_states = hass.states.async_all()
assert len(all_states) == 1
state = hass.states.get("sensor.event_service_any")
assert int(state.state) == 2
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = "ERROR", None
async_fire_time_changed(hass, utcnow + 2 * geo_rss_events.SCAN_INTERVAL)
await hass.async_block_till_done()
# Simulate an update - empty data, removes all entities
mock_feed.return_value.update.return_value = "ERROR", None
async_fire_time_changed(hass, utcnow + 2 * geo_rss_events.SCAN_INTERVAL)
await hass.async_block_till_done()
all_states = hass.states.async_all()
assert len(all_states) == 1
state = hass.states.get("sensor.event_service_any")
assert int(state.state) == 0
assert state.attributes == {
ATTR_FRIENDLY_NAME: "Event Service Any",
ATTR_UNIT_OF_MEASUREMENT: "Events",
ATTR_ICON: "mdi:alert",
}
all_states = hass.states.async_all()
assert len(all_states) == 1
state = hass.states.get("sensor.event_service_any")
assert int(state.state) == 0
assert state.attributes == {
ATTR_FRIENDLY_NAME: "Event Service Any",
ATTR_UNIT_OF_MEASUREMENT: "Events",
ATTR_ICON: "mdi:alert",
}
async def test_setup_with_categories(hass, mock_feed):

17
tests/components/graphite/test_init.py
View file

@ -220,11 +220,12 @@ class TestGraphite(unittest.TestCase):
runs.append(1)
return event
with mock.patch.object(self.gf, "_queue") as mock_queue:
with mock.patch.object(self.gf, "_report_attributes") as mock_r:
mock_queue.get.side_effect = fake_get
self.gf.run()
# Twice for two events, once for the stop
assert mock_queue.task_done.call_count == 3
assert mock_r.call_count == 1
assert mock_r.call_args == mock.call("entity", event.data["new_state"])
with mock.patch.object(self.gf, "_queue") as mock_queue, mock.patch.object(
self.gf, "_report_attributes"
) as mock_r:
mock_queue.get.side_effect = fake_get
self.gf.run()
# Twice for two events, once for the stop
assert mock_queue.task_done.call_count == 3
assert mock_r.call_count == 1
assert mock_r.call_args == mock.call("entity", event.data["new_state"])

69
tests/components/hisense_aehw4a1/test_init.py
View file

@ -13,24 +13,21 @@ async def test_creating_entry_sets_up_climate_discovery(hass):
with patch(
"homeassistant.components.hisense_aehw4a1.config_flow.AehW4a1.discovery",
return_value=["1.2.3.4"],
):
with patch(
"homeassistant.components.hisense_aehw4a1.climate.async_setup_entry",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
hisense_aehw4a1.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
), patch(
"homeassistant.components.hisense_aehw4a1.climate.async_setup_entry",
return_value=True,
) as mock_setup:
result = await hass.config_entries.flow.async_init(
hisense_aehw4a1.DOMAIN, context={"source": config_entries.SOURCE_USER}
)
# Confirmation form
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
# Confirmation form
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
await hass.async_block_till_done()
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
@ -40,17 +37,16 @@ async def test_configuring_hisense_w4a1_create_entry(hass):
with patch(
"homeassistant.components.hisense_aehw4a1.config_flow.AehW4a1.check",
return_value=True,
):
with patch(
"homeassistant.components.hisense_aehw4a1.async_setup_entry",
return_value=True,
) as mock_setup:
await async_setup_component(
hass,
hisense_aehw4a1.DOMAIN,
{"hisense_aehw4a1": {"ip_address": ["1.2.3.4"]}},
)
await hass.async_block_till_done()
), patch(
"homeassistant.components.hisense_aehw4a1.async_setup_entry",
return_value=True,
) as mock_setup:
await async_setup_component(
hass,
hisense_aehw4a1.DOMAIN,
{"hisense_aehw4a1": {"ip_address": ["1.2.3.4"]}},
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
@ -60,17 +56,16 @@ async def test_configuring_hisense_w4a1_not_creates_entry_for_device_not_found(h
with patch(
"homeassistant.components.hisense_aehw4a1.config_flow.AehW4a1.check",
side_effect=exceptions.ConnectionError,
):
with patch(
"homeassistant.components.hisense_aehw4a1.async_setup_entry",
return_value=True,
) as mock_setup:
await async_setup_component(
hass,
hisense_aehw4a1.DOMAIN,
{"hisense_aehw4a1": {"ip_address": ["1.2.3.4"]}},
)
await hass.async_block_till_done()
), patch(
"homeassistant.components.hisense_aehw4a1.async_setup_entry",
return_value=True,
) as mock_setup:
await async_setup_component(
hass,
hisense_aehw4a1.DOMAIN,
{"hisense_aehw4a1": {"ip_address": ["1.2.3.4"]}},
)
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 0

22
tests/components/history_stats/test_sensor.py
View file

@ -162,12 +162,11 @@ class TestHistoryStatsSensor(unittest.TestCase):
with patch(
"homeassistant.components.history.state_changes_during_period",
return_value=fake_states,
):
with patch("homeassistant.components.history.get_state", return_value=None):
sensor1.update()
sensor2.update()
sensor3.update()
sensor4.update()
), patch("homeassistant.components.history.get_state", return_value=None):
sensor1.update()
sensor2.update()
sensor3.update()
sensor4.update()
assert sensor1.state == 0.5
assert sensor2.state is None
@ -246,12 +245,11 @@ class TestHistoryStatsSensor(unittest.TestCase):
with patch(
"homeassistant.components.history.state_changes_during_period",
return_value=fake_states,
):
with patch("homeassistant.components.history.get_state", return_value=None):
sensor1.update()
sensor2.update()
sensor3.update()
sensor4.update()
), patch("homeassistant.components.history.get_state", return_value=None):
sensor1.update()
sensor2.update()
sensor3.update()
sensor4.update()
assert sensor1.state == 0.5
assert sensor2.state is None

15
tests/components/influxdb/test_init.py
View file

@ -254,14 +254,15 @@ async def test_setup_config_ssl(
config = {"influxdb": config_base.copy()}
config["influxdb"].update(config_ext)
with patch("os.access", return_value=True):
with patch("os.path.isfile", return_value=True):
assert await async_setup_component(hass, influxdb.DOMAIN, config)
await hass.async_block_till_done()
with patch("os.access", return_value=True), patch(
"os.path.isfile", return_value=True
):
assert await async_setup_component(hass, influxdb.DOMAIN, config)
await hass.async_block_till_done()
assert hass.bus.listen.called
assert hass.bus.listen.call_args_list[0][0][0] == EVENT_STATE_CHANGED
assert expected_client_args.items() <= mock_client.call_args.kwargs.items()
assert hass.bus.listen.called
assert hass.bus.listen.call_args_list[0][0][0] == EVENT_STATE_CHANGED
assert expected_client_args.items() <= mock_client.call_args.kwargs.items()
@pytest.mark.parametrize(

9
tests/components/lyric/test_config_flow.py
View file

@ -90,11 +90,10 @@ async def test_full_flow(
},
)
with patch("homeassistant.components.lyric.api.ConfigEntryLyricClient"):
with patch(
"homeassistant.components.lyric.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
with patch("homeassistant.components.lyric.api.ConfigEntryLyricClient"), patch(
"homeassistant.components.lyric.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["data"]["auth_implementation"] == DOMAIN

75
tests/components/melissa/test_climate.py
View file

@ -288,19 +288,18 @@ async def test_update(hass):
"""Test update."""
with patch(
"homeassistant.components.melissa.climate._LOGGER.warning"
) as mocked_warning:
with patch("homeassistant.components.melissa"):
api = melissa_mock()
device = (await api.async_fetch_devices())[_SERIAL]
thermostat = MelissaClimate(api, _SERIAL, device)
await thermostat.async_update()
assert thermostat.fan_mode == SPEED_LOW
assert thermostat.state == HVAC_MODE_HEAT
api.async_status = AsyncMock(side_effect=KeyError("boom"))
await thermostat.async_update()
mocked_warning.assert_called_once_with(
"Unable to update entity %s", thermostat.entity_id
)
) as mocked_warning, patch("homeassistant.components.melissa"):
api = melissa_mock()
device = (await api.async_fetch_devices())[_SERIAL]
thermostat = MelissaClimate(api, _SERIAL, device)
await thermostat.async_update()
assert thermostat.fan_mode == SPEED_LOW
assert thermostat.state == HVAC_MODE_HEAT
api.async_status = AsyncMock(side_effect=KeyError("boom"))
await thermostat.async_update()
mocked_warning.assert_called_once_with(
"Unable to update entity %s", thermostat.entity_id
)
async def test_melissa_op_to_hass(hass):
@ -333,35 +332,33 @@ async def test_hass_mode_to_melissa(hass):
"""Test for hass operations to melssa."""
with patch(
"homeassistant.components.melissa.climate._LOGGER.warning"
) as mocked_warning:
with patch("homeassistant.components.melissa"):
api = melissa_mock()
device = (await api.async_fetch_devices())[_SERIAL]
thermostat = MelissaClimate(api, _SERIAL, device)
assert thermostat.hass_mode_to_melissa(HVAC_MODE_FAN_ONLY) == 1
assert thermostat.hass_mode_to_melissa(HVAC_MODE_HEAT) == 2
assert thermostat.hass_mode_to_melissa(HVAC_MODE_COOL) == 3
assert thermostat.hass_mode_to_melissa(HVAC_MODE_DRY) == 4
thermostat.hass_mode_to_melissa("test")
mocked_warning.assert_called_once_with(
"Melissa have no setting for %s mode", "test"
)
) as mocked_warning, patch("homeassistant.components.melissa"):
api = melissa_mock()
device = (await api.async_fetch_devices())[_SERIAL]
thermostat = MelissaClimate(api, _SERIAL, device)
assert thermostat.hass_mode_to_melissa(HVAC_MODE_FAN_ONLY) == 1
assert thermostat.hass_mode_to_melissa(HVAC_MODE_HEAT) == 2
assert thermostat.hass_mode_to_melissa(HVAC_MODE_COOL) == 3
assert thermostat.hass_mode_to_melissa(HVAC_MODE_DRY) == 4
thermostat.hass_mode_to_melissa("test")
mocked_warning.assert_called_once_with(
"Melissa have no setting for %s mode", "test"
)
async def test_hass_fan_to_melissa(hass):
"""Test for translate melissa states to hass."""
with patch(
"homeassistant.components.melissa.climate._LOGGER.warning"
) as mocked_warning:
with patch("homeassistant.components.melissa"):
api = melissa_mock()
device = (await api.async_fetch_devices())[_SERIAL]
thermostat = MelissaClimate(api, _SERIAL, device)
assert thermostat.hass_fan_to_melissa("auto") == 0
assert thermostat.hass_fan_to_melissa(SPEED_LOW) == 1
assert thermostat.hass_fan_to_melissa(SPEED_MEDIUM) == 2
assert thermostat.hass_fan_to_melissa(SPEED_HIGH) == 3
thermostat.hass_fan_to_melissa("test")
mocked_warning.assert_called_once_with(
"Melissa have no setting for %s fan mode", "test"
)
) as mocked_warning, patch("homeassistant.components.melissa"):
api = melissa_mock()
device = (await api.async_fetch_devices())[_SERIAL]
thermostat = MelissaClimate(api, _SERIAL, device)
assert thermostat.hass_fan_to_melissa("auto") == 0
assert thermostat.hass_fan_to_melissa(SPEED_LOW) == 1
assert thermostat.hass_fan_to_melissa(SPEED_MEDIUM) == 2
assert thermostat.hass_fan_to_melissa(SPEED_HIGH) == 3
thermostat.hass_fan_to_melissa("test")
mocked_warning.assert_called_once_with(
"Melissa have no setting for %s fan mode", "test"
)

161
tests/components/minecraft_server/test_config_flow.py
View file

@ -103,31 +103,27 @@ async def test_invalid_ip(hass: HomeAssistantType) -> None:
async def test_same_host(hass: HomeAssistantType) -> None:
"""Test abort in case of same host name."""
with patch(
"aiodns.DNSResolver.query",
side_effect=aiodns.error.DNSError,
with patch("aiodns.DNSResolver.query", side_effect=aiodns.error.DNSError,), patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
with patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
unique_id = "mc.dummyserver.com-25565"
config_data = {
CONF_NAME: DEFAULT_NAME,
CONF_HOST: "mc.dummyserver.com",
CONF_PORT: DEFAULT_PORT,
}
mock_config_entry = MockConfigEntry(
domain=DOMAIN, unique_id=unique_id, data=config_data
)
mock_config_entry.add_to_hass(hass)
unique_id = "mc.dummyserver.com-25565"
config_data = {
CONF_NAME: DEFAULT_NAME,
CONF_HOST: "mc.dummyserver.com",
CONF_PORT: DEFAULT_PORT,
}
mock_config_entry = MockConfigEntry(
domain=DOMAIN, unique_id=unique_id, data=config_data
)
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_port_too_small(hass: HomeAssistantType) -> None:
@ -163,93 +159,80 @@ async def test_connection_failed(hass: HomeAssistantType) -> None:
with patch(
"aiodns.DNSResolver.query",
side_effect=aiodns.error.DNSError,
):
with patch("mcstatus.server.MinecraftServer.status", side_effect=OSError):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT
)
), patch("mcstatus.server.MinecraftServer.status", side_effect=OSError):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_connection_succeeded_with_srv_record(hass: HomeAssistantType) -> None:
"""Test config entry in case of a successful connection with a SRV record."""
with patch(
"aiodns.DNSResolver.query",
return_value=SRV_RECORDS,
with patch("aiodns.DNSResolver.query", return_value=SRV_RECORDS,), patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
with patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT_SRV
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT_SRV
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT_SRV[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT_SRV[CONF_NAME]
assert result["data"][CONF_HOST] == USER_INPUT_SRV[CONF_HOST]
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT_SRV[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT_SRV[CONF_NAME]
assert result["data"][CONF_HOST] == USER_INPUT_SRV[CONF_HOST]
async def test_connection_succeeded_with_host(hass: HomeAssistantType) -> None:
"""Test config entry in case of a successful connection with a host name."""
with patch(
"aiodns.DNSResolver.query",
side_effect=aiodns.error.DNSError,
with patch("aiodns.DNSResolver.query", side_effect=aiodns.error.DNSError,), patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
with patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT[CONF_NAME]
assert result["data"][CONF_HOST] == "mc.dummyserver.com"
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT[CONF_NAME]
assert result["data"][CONF_HOST] == "mc.dummyserver.com"
async def test_connection_succeeded_with_ip4(hass: HomeAssistantType) -> None:
"""Test config entry in case of a successful connection with an IPv4 address."""
with patch("getmac.get_mac_address", return_value="01:23:45:67:89:ab"):
with patch(
"aiodns.DNSResolver.query",
side_effect=aiodns.error.DNSError,
):
with patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT_IPV4
)
with patch("getmac.get_mac_address", return_value="01:23:45:67:89:ab"), patch(
"aiodns.DNSResolver.query",
side_effect=aiodns.error.DNSError,
), patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT_IPV4
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT_IPV4[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT_IPV4[CONF_NAME]
assert result["data"][CONF_HOST] == "1.1.1.1"
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT_IPV4[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT_IPV4[CONF_NAME]
assert result["data"][CONF_HOST] == "1.1.1.1"
async def test_connection_succeeded_with_ip6(hass: HomeAssistantType) -> None:
"""Test config entry in case of a successful connection with an IPv6 address."""
with patch("getmac.get_mac_address", return_value="01:23:45:67:89:ab"):
with patch(
"aiodns.DNSResolver.query",
side_effect=aiodns.error.DNSError,
):
with patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT_IPV6
)
with patch("getmac.get_mac_address", return_value="01:23:45:67:89:ab"), patch(
"aiodns.DNSResolver.query",
side_effect=aiodns.error.DNSError,
), patch(
"mcstatus.server.MinecraftServer.status",
return_value=PingResponse(STATUS_RESPONSE_RAW),
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data=USER_INPUT_IPV6
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT_IPV6[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT_IPV6[CONF_NAME]
assert result["data"][CONF_HOST] == "::ffff:0101:0101"
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == USER_INPUT_IPV6[CONF_HOST]
assert result["data"][CONF_NAME] == USER_INPUT_IPV6[CONF_NAME]
assert result["data"][CONF_HOST] == "::ffff:0101:0101"

7
tests/components/mqtt/test_light.py
View file

@ -743,10 +743,9 @@ async def test_sending_mqtt_commands_and_optimistic(hass, mqtt_mock):
with patch(
"homeassistant.helpers.restore_state.RestoreEntity.async_get_last_state",
return_value=fake_state,
):
with assert_setup_component(1, light.DOMAIN):
assert await async_setup_component(hass, light.DOMAIN, config)
await hass.async_block_till_done()
), assert_setup_component(1, light.DOMAIN):
assert await async_setup_component(hass, light.DOMAIN, config)
await hass.async_block_till_done()
state = hass.states.get("light.test")
assert state.state == STATE_ON

65
tests/components/mqtt/test_light_template.py
View file

@ -298,39 +298,38 @@ async def test_sending_mqtt_commands_and_optimistic(hass, mqtt_mock):
with patch(
"homeassistant.helpers.restore_state.RestoreEntity.async_get_last_state",
return_value=fake_state,
):
with assert_setup_component(1, light.DOMAIN):
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"schema": "template",
"name": "test",
"command_topic": "test_light_rgb/set",
"command_on_template": "on,"
"{{ brightness|d }},"
"{{ color_temp|d }},"
"{{ white_value|d }},"
"{{ red|d }}-"
"{{ green|d }}-"
"{{ blue|d }}",
"command_off_template": "off",
"effect_list": ["colorloop", "random"],
"optimistic": True,
"state_template": '{{ value.split(",")[0] }}',
"color_temp_template": '{{ value.split(",")[2] }}',
"white_value_template": '{{ value.split(",")[3] }}',
"red_template": '{{ value.split(",")[4].' 'split("-")[0] }}',
"green_template": '{{ value.split(",")[4].' 'split("-")[1] }}',
"blue_template": '{{ value.split(",")[4].' 'split("-")[2] }}',
"effect_template": '{{ value.split(",")[5] }}',
"qos": 2,
}
},
)
await hass.async_block_till_done()
), assert_setup_component(1, light.DOMAIN):
assert await async_setup_component(
hass,
light.DOMAIN,
{
light.DOMAIN: {
"platform": "mqtt",
"schema": "template",
"name": "test",
"command_topic": "test_light_rgb/set",
"command_on_template": "on,"
"{{ brightness|d }},"
"{{ color_temp|d }},"
"{{ white_value|d }},"
"{{ red|d }}-"
"{{ green|d }}-"
"{{ blue|d }}",
"command_off_template": "off",
"effect_list": ["colorloop", "random"],
"optimistic": True,
"state_template": '{{ value.split(",")[0] }}',
"color_temp_template": '{{ value.split(",")[2] }}',
"white_value_template": '{{ value.split(",")[3] }}',
"red_template": '{{ value.split(",")[4].' 'split("-")[0] }}',
"green_template": '{{ value.split(",")[4].' 'split("-")[1] }}',
"blue_template": '{{ value.split(",")[4].' 'split("-")[2] }}',
"effect_template": '{{ value.split(",")[5] }}',
"qos": 2,
}
},
)
await hass.async_block_till_done()
state = hass.states.get("light.test")
assert state.state == STATE_ON

69
tests/components/pilight/test_init.py
View file

@ -64,29 +64,31 @@ class PilightDaemonSim:
@patch("homeassistant.components.pilight._LOGGER.error")
async def test_connection_failed_error(mock_error, hass):
"""Try to connect at 127.0.0.1:5001 with socket error."""
with assert_setup_component(4):
with patch("pilight.pilight.Client", side_effect=socket.error) as mock_client:
assert not await async_setup_component(
hass, pilight.DOMAIN, {pilight.DOMAIN: {}}
)
mock_client.assert_called_once_with(
host=pilight.DEFAULT_HOST, port=pilight.DEFAULT_PORT
)
assert mock_error.call_count == 1
with assert_setup_component(4), patch(
"pilight.pilight.Client", side_effect=socket.error
) as mock_client:
assert not await async_setup_component(
hass, pilight.DOMAIN, {pilight.DOMAIN: {}}
)
mock_client.assert_called_once_with(
host=pilight.DEFAULT_HOST, port=pilight.DEFAULT_PORT
)
assert mock_error.call_count == 1
@patch("homeassistant.components.pilight._LOGGER.error")
async def test_connection_timeout_error(mock_error, hass):
"""Try to connect at 127.0.0.1:5001 with socket timeout."""
with assert_setup_component(4):
with patch("pilight.pilight.Client", side_effect=socket.timeout) as mock_client:
assert not await async_setup_component(
hass, pilight.DOMAIN, {pilight.DOMAIN: {}}
)
mock_client.assert_called_once_with(
host=pilight.DEFAULT_HOST, port=pilight.DEFAULT_PORT
)
assert mock_error.call_count == 1
with assert_setup_component(4), patch(
"pilight.pilight.Client", side_effect=socket.timeout
) as mock_client:
assert not await async_setup_component(
hass, pilight.DOMAIN, {pilight.DOMAIN: {}}
)
mock_client.assert_called_once_with(
host=pilight.DEFAULT_HOST, port=pilight.DEFAULT_PORT
)
assert mock_error.call_count == 1
@patch("pilight.pilight.Client", PilightDaemonSim)
@ -134,23 +136,22 @@ async def test_send_code(mock_pilight_error, hass):
@patch("homeassistant.components.pilight._LOGGER.error")
async def test_send_code_fail(mock_pilight_error, hass):
"""Check IOError exception error message."""
with assert_setup_component(4):
with patch("pilight.pilight.Client.send_code", side_effect=IOError):
assert await async_setup_component(
hass, pilight.DOMAIN, {pilight.DOMAIN: {}}
)
with assert_setup_component(4), patch(
"pilight.pilight.Client.send_code", side_effect=IOError
):
assert await async_setup_component(hass, pilight.DOMAIN, {pilight.DOMAIN: {}})
# Call with protocol info, should not give error
service_data = {"protocol": "test", "value": 42}
await hass.services.async_call(
pilight.DOMAIN,
pilight.SERVICE_NAME,
service_data=service_data,
blocking=True,
)
await hass.async_block_till_done()
error_log_call = mock_pilight_error.call_args_list[-1]
assert "Pilight send failed" in str(error_log_call)
# Call with protocol info, should not give error
service_data = {"protocol": "test", "value": 42}
await hass.services.async_call(
pilight.DOMAIN,
pilight.SERVICE_NAME,
service_data=service_data,
blocking=True,
)
await hass.async_block_till_done()
error_log_call = mock_pilight_error.call_args_list[-1]
assert "Pilight send failed" in str(error_log_call)
@patch("homeassistant.components.pilight._LOGGER.error")

22
tests/components/signal_messenger/test_notify.py
View file

@ -87,12 +87,11 @@ class TestSignalMesssenger(unittest.TestCase):
)
with self.assertLogs(
"homeassistant.components.signal_messenger.notify", level="WARNING"
) as context:
with tempfile.NamedTemporaryFile(
suffix=".png", prefix=os.path.basename(__file__)
) as tf:
data = {"data": {"attachment": tf.name}}
self._signalmessenger.send_message(message, **data)
) as context, tempfile.NamedTemporaryFile(
suffix=".png", prefix=os.path.basename(__file__)
) as tf:
data = {"data": {"attachment": tf.name}}
self._signalmessenger.send_message(message, **data)
self.assertIn(
"The 'attachment' option is deprecated, please replace it with 'attachments'. This option will become invalid in version 0.108",
context.output[0],
@ -117,12 +116,11 @@ class TestSignalMesssenger(unittest.TestCase):
)
with self.assertLogs(
"homeassistant.components.signal_messenger.notify", level="DEBUG"
) as context:
with tempfile.NamedTemporaryFile(
suffix=".png", prefix=os.path.basename(__file__)
) as tf:
data = {"data": {"attachments": [tf.name]}}
self._signalmessenger.send_message(message, **data)
) as context, tempfile.NamedTemporaryFile(
suffix=".png", prefix=os.path.basename(__file__)
) as tf:
data = {"data": {"attachments": [tf.name]}}
self._signalmessenger.send_message(message, **data)
self.assertIn("Sending signal message", context.output[0])
self.assertTrue(mock.called)
self.assertEqual(mock.call_count, 2)

25
tests/components/system_log/test_init.py
View file

@ -291,20 +291,19 @@ async def async_log_error_from_test_path(hass, path, sq):
call_path = "internal_path.py"
with patch.object(
_LOGGER, "findCaller", MagicMock(return_value=(call_path, 0, None, None))
), patch(
"traceback.extract_stack",
MagicMock(
return_value=[
get_frame("main_path/main.py"),
get_frame(path),
get_frame(call_path),
get_frame("venv_path/logging/log.py"),
]
),
):
with patch(
"traceback.extract_stack",
MagicMock(
return_value=[
get_frame("main_path/main.py"),
get_frame(path),
get_frame(call_path),
get_frame("venv_path/logging/log.py"),
]
),
):
_LOGGER.error("error message")
await _async_block_until_queue_empty(hass, sq)
_LOGGER.error("error message")
await _async_block_until_queue_empty(hass, sq)
async def test_homeassistant_path(hass, simple_queue, hass_client):

9
tests/components/zha/test_channels.py
View file

@ -415,10 +415,11 @@ async def test_ep_channels_configure(channel):
claimed = {ch_1.id: ch_1, ch_2.id: ch_2, ch_3.id: ch_3}
client_chans = {ch_4.id: ch_4, ch_5.id: ch_5}
with mock.patch.dict(ep_channels.claimed_channels, claimed, clear=True):
with mock.patch.dict(ep_channels.client_channels, client_chans, clear=True):
await ep_channels.async_configure()
await ep_channels.async_initialize(mock.sentinel.from_cache)
with mock.patch.dict(
ep_channels.claimed_channels, claimed, clear=True
), mock.patch.dict(ep_channels.client_channels, client_chans, clear=True):
await ep_channels.async_configure()
await ep_channels.async_initialize(mock.sentinel.from_cache)
for ch in [*claimed.values(), *client_chans.values()]:
assert ch.async_initialize.call_count == 1

19
tests/components/zha/test_discover.py
View file

@ -212,17 +212,14 @@ def test_discover_by_device_type_override():
with mock.patch(
"homeassistant.components.zha.core.registries.ZHA_ENTITIES.get_entity",
get_entity_mock,
):
with mock.patch.dict(disc.PROBE._device_configs, overrides, clear=True):
disc.PROBE.discover_by_device_type(ep_channels)
assert get_entity_mock.call_count == 1
assert ep_channels.claim_channels.call_count == 1
assert ep_channels.claim_channels.call_args[0][0] is mock.sentinel.claimed
assert ep_channels.async_new_entity.call_count == 1
assert ep_channels.async_new_entity.call_args[0][0] == zha_const.SWITCH
assert (
ep_channels.async_new_entity.call_args[0][1] == mock.sentinel.entity_cls
)
), mock.patch.dict(disc.PROBE._device_configs, overrides, clear=True):
disc.PROBE.discover_by_device_type(ep_channels)
assert get_entity_mock.call_count == 1
assert ep_channels.claim_channels.call_count == 1
assert ep_channels.claim_channels.call_args[0][0] is mock.sentinel.claimed
assert ep_channels.async_new_entity.call_count == 1
assert ep_channels.async_new_entity.call_args[0][0] == zha_const.SWITCH
assert ep_channels.async_new_entity.call_args[0][1] == mock.sentinel.entity_cls
def test_discover_probe_single_cluster():

45
tests/components/zwave/test_init.py
View file

@ -202,20 +202,17 @@ async def test_zwave_ready_wait(hass, mock_openzwave, zwave_setup):
sleeps.append(duration)
await asyncio_sleep(0)
with patch("homeassistant.components.zwave.dt_util.utcnow", new=utcnow):
with patch("asyncio.sleep", new=sleep):
with patch.object(zwave, "_LOGGER") as mock_logger:
hass.data[DATA_NETWORK].state = MockNetwork.STATE_STARTED
with patch("homeassistant.components.zwave.dt_util.utcnow", new=utcnow), patch(
"asyncio.sleep", new=sleep
), patch.object(zwave, "_LOGGER") as mock_logger:
hass.data[DATA_NETWORK].state = MockNetwork.STATE_STARTED
await hass.async_start()
await hass.async_start()
assert len(sleeps) == const.NETWORK_READY_WAIT_SECS
assert mock_logger.warning.called
assert len(mock_logger.warning.mock_calls) == 1
assert (
mock_logger.warning.mock_calls[0][1][1]
== const.NETWORK_READY_WAIT_SECS
)
assert len(sleeps) == const.NETWORK_READY_WAIT_SECS
assert mock_logger.warning.called
assert len(mock_logger.warning.mock_calls) == 1
assert mock_logger.warning.mock_calls[0][1][1] == const.NETWORK_READY_WAIT_SECS
async def test_device_entity(hass, mock_openzwave):
@ -341,19 +338,19 @@ async def test_unparsed_node_discovery(hass, mock_openzwave):
sleeps.append(duration)
await asyncio_sleep(0)
with patch("homeassistant.components.zwave.dt_util.utcnow", new=utcnow):
with patch("asyncio.sleep", new=sleep):
with patch.object(zwave, "_LOGGER") as mock_logger:
await hass.async_add_executor_job(mock_receivers[0], node)
await hass.async_block_till_done()
with patch("homeassistant.components.zwave.dt_util.utcnow", new=utcnow), patch(
"asyncio.sleep", new=sleep
), patch.object(zwave, "_LOGGER") as mock_logger:
await hass.async_add_executor_job(mock_receivers[0], node)
await hass.async_block_till_done()
assert len(sleeps) == const.NODE_READY_WAIT_SECS
assert mock_logger.warning.called
assert len(mock_logger.warning.mock_calls) == 1
assert mock_logger.warning.mock_calls[0][1][1:] == (
14,
const.NODE_READY_WAIT_SECS,
)
assert len(sleeps) == const.NODE_READY_WAIT_SECS
assert mock_logger.warning.called
assert len(mock_logger.warning.mock_calls) == 1
assert mock_logger.warning.mock_calls[0][1][1:] == (
14,
const.NODE_READY_WAIT_SECS,
)
assert hass.states.get("zwave.unknown_node_14").state == "unknown"

7
tests/helpers/test_config_entry_oauth2_flow.py
View file

@ -99,9 +99,10 @@ def test_inherit_enforces_domain_set():
"""Return logger."""
return logging.getLogger(__name__)
with patch.dict(config_entries.HANDLERS, {TEST_DOMAIN: TestFlowHandler}):
with pytest.raises(TypeError):
TestFlowHandler()
with patch.dict(
config_entries.HANDLERS, {TEST_DOMAIN: TestFlowHandler}
), pytest.raises(TypeError):
TestFlowHandler()
async def test_abort_if_no_implementation(hass, flow_handler):

5
tests/helpers/test_network.py
View file

@ -377,9 +377,8 @@ async def test_get_cloud_url(hass: HomeAssistant):
hass.components.cloud,
"async_remote_ui_url",
side_effect=cloud.CloudNotAvailable,
):
with pytest.raises(NoURLAvailableError):
_get_cloud_url(hass)
), pytest.raises(NoURLAvailableError):
_get_cloud_url(hass)
async def test_get_external_url_cloud_fallback(hass: HomeAssistant):

31
tests/helpers/test_service.py
View file

@ -561,22 +561,21 @@ async def test_call_context_target_specific_no_auth(
hass, mock_handle_entity_call, mock_entities
):
"""Check targeting specific entities without auth."""
with pytest.raises(exceptions.Unauthorized) as err:
with patch(
"homeassistant.auth.AuthManager.async_get_user",
return_value=Mock(permissions=PolicyPermissions({}, None)),
):
await service.entity_service_call(
hass,
[Mock(entities=mock_entities)],
Mock(),
ha.ServiceCall(
"test_domain",
"test_service",
{"entity_id": "light.kitchen"},
context=ha.Context(user_id="mock-id"),
),
)
with pytest.raises(exceptions.Unauthorized) as err, patch(
"homeassistant.auth.AuthManager.async_get_user",
return_value=Mock(permissions=PolicyPermissions({}, None)),
):
await service.entity_service_call(
hass,
[Mock(entities=mock_entities)],
Mock(),
ha.ServiceCall(
"test_domain",
"test_service",
{"entity_id": "light.kitchen"},
context=ha.Context(user_id="mock-id"),
),
)
assert err.value.context.user_id == "mock-id"
assert err.value.entity_id == "light.kitchen"

11
tests/test_config_entries.py
View file

@ -1221,12 +1221,11 @@ async def test_init_custom_integration(hass):
None,
{"name": "Hue", "dependencies": [], "requirements": [], "domain": "hue"},
)
with pytest.raises(data_entry_flow.UnknownHandler):
with patch(
"homeassistant.loader.async_get_integration",
return_value=integration,
):
await hass.config_entries.flow.async_init("bla")
with pytest.raises(data_entry_flow.UnknownHandler), patch(
"homeassistant.loader.async_get_integration",
return_value=integration,
):
await hass.config_entries.flow.async_init("bla")
async def test_support_entry_unload(hass):

5
tests/test_requirements.py
View file

@ -84,9 +84,8 @@ async def test_install_missing_package(hass):
"""Test an install attempt on an existing package."""
with patch(
"homeassistant.util.package.install_package", return_value=False
) as mock_inst:
with pytest.raises(RequirementsNotFound):
await async_process_requirements(hass, "test_component", ["hello==1.0.0"])
) as mock_inst, pytest.raises(RequirementsNotFound):
await async_process_requirements(hass, "test_component", ["hello==1.0.0"])
assert len(mock_inst.mock_calls) == 1

7
tests/util/test_async.py
View file

@ -206,8 +206,9 @@ async def test_callback_is_always_scheduled(hass):
callback = MagicMock()
hasync.shutdown_run_callback_threadsafe(hass.loop)
with patch.object(hass.loop, "call_soon_threadsafe") as mock_call_soon_threadsafe:
with pytest.raises(RuntimeError):
hasync.run_callback_threadsafe(hass.loop, callback)
with patch.object(
hass.loop, "call_soon_threadsafe"
) as mock_call_soon_threadsafe, pytest.raises(RuntimeError):
hasync.run_callback_threadsafe(hass.loop, callback)
mock_call_soon_threadsafe.assert_called_once()

5
tests/util/yaml/test_init.py
View file

@ -65,9 +65,8 @@ def test_environment_variable_default():
def test_invalid_environment_variable():
"""Test config file with no environment variable sat."""
conf = "password: !env_var PASSWORD"
with pytest.raises(HomeAssistantError):
with io.StringIO(conf) as file:
yaml_loader.yaml.load(file, Loader=yaml_loader.SafeLineLoader)
with pytest.raises(HomeAssistantError), io.StringIO(conf) as file:
yaml_loader.yaml.load(file, Loader=yaml_loader.SafeLineLoader)
def test_include_yaml():