Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Enable Ruff SIM117 (#86783)

pull/86788/head
Franck Nijhof 2023-01-27 11:52:49 +01:00 committed by GitHub
parent 57cf11f067
commit a79885ceaf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 441 additions and 453 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
homeassistant/components/cert_expiry/helper.py
View File

@ -21,10 +21,11 @@ def get_cert(
"""Get the certificate for the host and port combination."""
ctx = ssl.create_default_context()
address = (host, port)
with socket.create_connection(address, timeout=TIMEOUT) as sock:
with ctx.wrap_socket(sock, server_hostname=address[0]) as ssock:
cert = ssock.getpeercert()
return cert
with socket.create_connection(address, timeout=TIMEOUT) as sock, ctx.wrap_socket(
sock, server_hostname=address[0]
) as ssock:
cert = ssock.getpeercert()
return cert
async def get_cert_expiry_timestamp(

40
homeassistant/components/recorder/migration.py
View File

@ -661,18 +661,19 @@ def _apply_update( # noqa: C901
),
table,
)
with contextlib.suppress(SQLAlchemyError):
with session_scope(session=session_maker()) as session:
connection = session.connection()
connection.execute(
# Using LOCK=EXCLUSIVE to prevent
# the database from corrupting
# https://github.com/home-assistant/core/issues/56104
text(
f"ALTER TABLE {table} CONVERT TO CHARACTER SET utf8mb4"
" COLLATE utf8mb4_unicode_ci, LOCK=EXCLUSIVE"
)
with contextlib.suppress(SQLAlchemyError), session_scope(
session=session_maker()
) as session:
connection = session.connection()
connection.execute(
# Using LOCK=EXCLUSIVE to prevent
# the database from corrupting
# https://github.com/home-assistant/core/issues/56104
text(
f"ALTER TABLE {table} CONVERT TO CHARACTER SET utf8mb4"
" COLLATE utf8mb4_unicode_ci, LOCK=EXCLUSIVE"
)
)
elif new_version == 22:
# Recreate the all statistics tables for Oracle DB with Identity columns
#
@ -804,14 +805,15 @@ def _apply_update( # noqa: C901
if engine.dialect.name == SupportedDialect.MYSQL:
# Ensure the row format is dynamic or the index
# unique will be too large
with contextlib.suppress(SQLAlchemyError):
with session_scope(session=session_maker()) as session:
connection = session.connection()
# This is safe to run multiple times and fast
# since the table is small.
connection.execute(
text("ALTER TABLE statistics_meta ROW_FORMAT=DYNAMIC")
)
with contextlib.suppress(SQLAlchemyError), session_scope(
session=session_maker()
) as session:
connection = session.connection()
# This is safe to run multiple times and fast
# since the table is small.
connection.execute(
text("ALTER TABLE statistics_meta ROW_FORMAT=DYNAMIC")
)
try:
_create_index(
session_maker, "statistics_meta", "ix_statistics_meta_statistic_id"

21
homeassistant/components/recorder/statistics.py
View File

@ -2422,17 +2422,18 @@ def correct_db_schema(
),
"statistics_meta",
)
with contextlib.suppress(SQLAlchemyError):
with session_scope(session=session_maker()) as session:
connection = session.connection()
connection.execute(
# Using LOCK=EXCLUSIVE to prevent the database from corrupting
# https://github.com/home-assistant/core/issues/56104
text(
"ALTER TABLE statistics_meta CONVERT TO CHARACTER SET utf8mb4"
" COLLATE utf8mb4_unicode_ci, LOCK=EXCLUSIVE"
)
with contextlib.suppress(SQLAlchemyError), session_scope(
session=session_maker()
) as session:
connection = session.connection()
connection.execute(
# Using LOCK=EXCLUSIVE to prevent the database from corrupting
# https://github.com/home-assistant/core/issues/56104
text(
"ALTER TABLE statistics_meta CONVERT TO CHARACTER SET utf8mb4"
" COLLATE utf8mb4_unicode_ci, LOCK=EXCLUSIVE"
)
)
tables: tuple[type[Statistics | StatisticsShortTerm], ...] = (
Statistics,

1
pyproject.toml
View File

@ -248,6 +248,7 @@ select = [
"F", # pyflakes/autoflake
"PGH004", # Use specific rule codes when using noqa
"SIM105", # Use contextlib.suppress({exception}) instead of try-except-pass
"SIM117", # Merge with-statements that use the same scope
"T20", # flake8-print
"UP", # pyupgrade
"W", # pycodestyle

64
tests/components/airthings_ble/test_config_flow.py
View File

@ -24,15 +24,14 @@ from tests.common import MockConfigEntry
async def test_bluetooth_discovery(hass: HomeAssistant):
"""Test discovery via bluetooth with a valid device."""
with patch_async_ble_device_from_address(WAVE_SERVICE_INFO):
with patch_airthings_ble(
AirthingsDevice(name="Airthings Wave+", identifier="123456")
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WAVE_SERVICE_INFO,
)
with patch_async_ble_device_from_address(WAVE_SERVICE_INFO), patch_airthings_ble(
AirthingsDevice(name="Airthings Wave+", identifier="123456")
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WAVE_SERVICE_INFO,
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "bluetooth_confirm"
@ -66,13 +65,14 @@ async def test_bluetooth_discovery_airthings_ble_update_failed(
"""Test discovery via bluetooth but there's an exception from airthings-ble."""
for loop in [(Exception(), "unknown"), (BleakError(), "cannot_connect")]:
exc, reason = loop
with patch_async_ble_device_from_address(WAVE_SERVICE_INFO):
with patch_airthings_ble(side_effect=exc):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WAVE_SERVICE_INFO,
)
with patch_async_ble_device_from_address(
WAVE_SERVICE_INFO
), patch_airthings_ble(side_effect=exc):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_BLUETOOTH},
data=WAVE_SERVICE_INFO,
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == reason
@ -99,14 +99,12 @@ async def test_user_setup(hass: HomeAssistant):
with patch(
"homeassistant.components.airthings_ble.config_flow.async_discovered_service_info",
return_value=[WAVE_SERVICE_INFO],
), patch_async_ble_device_from_address(WAVE_SERVICE_INFO), patch_airthings_ble(
AirthingsDevice(name="Airthings Wave+", identifier="123456")
):
with patch_async_ble_device_from_address(WAVE_SERVICE_INFO):
with patch_airthings_ble(
AirthingsDevice(name="Airthings Wave+", identifier="123456")
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] is None
@ -167,12 +165,12 @@ async def test_user_setup_unknown_error(hass: HomeAssistant):
with patch(
"homeassistant.components.airthings_ble.config_flow.async_discovered_service_info",
return_value=[WAVE_SERVICE_INFO],
), patch_async_ble_device_from_address(WAVE_SERVICE_INFO), patch_airthings_ble(
None, Exception()
):
with patch_async_ble_device_from_address(WAVE_SERVICE_INFO):
with patch_airthings_ble(None, Exception()):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unknown"
@ -183,12 +181,12 @@ async def test_user_setup_unable_to_connect(hass: HomeAssistant):
with patch(
"homeassistant.components.airthings_ble.config_flow.async_discovered_service_info",
return_value=[WAVE_SERVICE_INFO],
), patch_async_ble_device_from_address(WAVE_SERVICE_INFO), patch_airthings_ble(
side_effect=BleakError("An error")
):
with patch_async_ble_device_from_address(WAVE_SERVICE_INFO):
with patch_airthings_ble(side_effect=BleakError("An error")):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "cannot_connect"

5
tests/components/flo/test_device.py
View File

@ -85,6 +85,5 @@ async def test_device(hass, config_entry, aioclient_mock_fixture, aioclient_mock
with patch(
"homeassistant.components.flo.device.FloDeviceDataUpdateCoordinator.send_presence_ping",
side_effect=RequestError,
):
with pytest.raises(UpdateFailed):
await valve._async_update_data()
), pytest.raises(UpdateFailed):
await valve._async_update_data()

188
tests/components/history_stats/test_sensor.py
View File

@ -629,30 +629,28 @@ async def test_async_start_from_history_and_switch_to_watching_state_changes_sin
with patch(
"homeassistant.components.recorder.history.state_changes_during_period",
_fake_states,
):
), freeze_time(start_time):
await async_setup_component(
hass,
"sensor",
{
"sensor": [
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor1",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "time",
}
]
},
)
await hass.async_block_till_done()
with freeze_time(start_time):
await async_setup_component(
hass,
"sensor",
{
"sensor": [
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor1",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "time",
}
]
},
)
await hass.async_block_till_done()
await async_update_entity(hass, "sensor.sensor1")
await hass.async_block_till_done()
await async_update_entity(hass, "sensor.sensor1")
await hass.async_block_till_done()
assert hass.states.get("sensor.sensor1").state == "0.0"
@ -729,30 +727,28 @@ async def test_async_start_from_history_and_switch_to_watching_state_changes_sin
with patch(
"homeassistant.components.recorder.history.state_changes_during_period",
_fake_states,
):
), freeze_time(start_time):
await async_setup_component(
hass,
"sensor",
{
"sensor": [
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor1",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"end": "{{ utcnow() }}",
"type": "time",
}
]
},
)
await hass.async_block_till_done()
with freeze_time(start_time):
await async_setup_component(
hass,
"sensor",
{
"sensor": [
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor1",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"end": "{{ utcnow() }}",
"type": "time",
}
]
},
)
await hass.async_block_till_done()
await async_update_entity(hass, "sensor.sensor1")
await hass.async_block_till_done()
await async_update_entity(hass, "sensor.sensor1")
await hass.async_block_till_done()
assert hass.states.get("sensor.sensor1").state == "0.0"
@ -828,58 +824,56 @@ async def test_async_start_from_history_and_switch_to_watching_state_changes_mul
with patch(
"homeassistant.components.recorder.history.state_changes_during_period",
_fake_states,
):
), freeze_time(start_time):
await async_setup_component(
hass,
"sensor",
{
"sensor": [
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor1",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "time",
},
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor2",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "time",
},
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor3",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "count",
},
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor4",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "ratio",
},
]
},
)
await hass.async_block_till_done()
with freeze_time(start_time):
await async_setup_component(
hass,
"sensor",
{
"sensor": [
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor1",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "time",
},
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor2",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "time",
},
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor3",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "count",
},
{
"platform": "history_stats",
"entity_id": "binary_sensor.state",
"name": "sensor4",
"state": "on",
"start": "{{ utcnow().replace(hour=0, minute=0, second=0) }}",
"duration": {"hours": 2},
"type": "ratio",
},
]
},
)
await hass.async_block_till_done()
for i in range(1, 5):
await async_update_entity(hass, f"sensor.sensor{i}")
await hass.async_block_till_done()
for i in range(1, 5):
await async_update_entity(hass, f"sensor.sensor{i}")
await hass.async_block_till_done()
assert hass.states.get("sensor.sensor1").state == "0.0"
assert hass.states.get("sensor.sensor2").state == "0.0"

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

@ -140,11 +140,10 @@ async def test_reauthentication_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["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_successful"

17
tests/components/nextdns/test_switch.py
View File

@ -997,11 +997,12 @@ async def test_switch_failure(hass: HomeAssistant, exc: Exception) -> None:
"""Tests that the turn on/off service throws HomeAssistantError."""
await init_integration(hass)
with patch("homeassistant.components.nextdns.NextDns.set_setting", side_effect=exc):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.fake_profile_block_page"},
blocking=True,
)
with patch(
"homeassistant.components.nextdns.NextDns.set_setting", side_effect=exc
), pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.fake_profile_block_page"},
blocking=True,
)

15
tests/components/peco/test_init.py
View File

@ -29,15 +29,14 @@ async def test_unload_entry(hass: HomeAssistant) -> None:
outage_count=0,
customers_served=350394,
),
), patch(
"peco.PecoOutageApi.get_map_alerts",
return_value=AlertResults(
alert_content="Testing 1234", alert_title="Testing 4321"
),
):
with patch(
"peco.PecoOutageApi.get_map_alerts",
return_value=AlertResults(
alert_content="Testing 1234", alert_title="Testing 4321"
),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.data[DOMAIN]
entries = hass.config_entries.async_entries(DOMAIN)

30
tests/components/peco/test_sensor.py
View File

@ -41,15 +41,14 @@ async def test_sensor_available(
outage_count=456,
customers_served=789,
),
), patch(
"peco.PecoOutageApi.get_map_alerts",
return_value=AlertResults(
alert_content="Testing 1234", alert_title="Testing 4321"
),
):
with patch(
"peco.PecoOutageApi.get_map_alerts",
return_value=AlertResults(
alert_content="Testing 1234", alert_title="Testing 4321"
),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert hass.data[DOMAIN]
entries = hass.config_entries.async_entries(DOMAIN)
@ -74,15 +73,14 @@ async def test_sensor_available(
outage_count=456,
customers_served=789,
),
), patch(
"peco.PecoOutageApi.get_map_alerts",
return_value=AlertResults(
alert_content="Testing 1234", alert_title="Testing 4321"
),
):
with patch(
"peco.PecoOutageApi.get_map_alerts",
return_value=AlertResults(
alert_content="Testing 1234", alert_title="Testing 4321"
),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 2

18
tests/components/plex/test_config_flow.py
View File

@ -832,11 +832,10 @@ async def test_client_request_missing(hass):
with patch("plexauth.PlexAuth.initiate_auth"), patch(
"plexauth.PlexAuth.token", return_value=None
):
with pytest.raises(RuntimeError):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
), pytest.raises(RuntimeError):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
async def test_client_header_issues(hass, current_request_with_host):
@ -855,8 +854,9 @@ async def test_client_header_issues(hass, current_request_with_host):
"plexauth.PlexAuth.token", return_value=None
), patch(
"homeassistant.components.http.current_request.get", return_value=MockRequest()
), pytest.raises(
RuntimeError
):
with pytest.raises(RuntimeError):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)

25
tests/components/plex/test_media_search.py
View File

@ -33,18 +33,19 @@ async def test_media_lookups(hass, mock_plex_server, requests_mock, playqueue_cr
},
True,
)
with pytest.raises(MediaNotFound) as excinfo:
with patch("plexapi.server.PlexServer.fetchItem", side_effect=NotFound):
assert await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: DOMAIN,
ATTR_MEDIA_CONTENT_ID: 123,
},
True,
)
with pytest.raises(MediaNotFound) as excinfo, patch(
"plexapi.server.PlexServer.fetchItem", side_effect=NotFound
):
assert await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player_id,
ATTR_MEDIA_CONTENT_TYPE: DOMAIN,
ATTR_MEDIA_CONTENT_ID: 123,
},
True,
)
assert "Media for key 123 not found" in str(excinfo.value)
# TV show searches

27
tests/components/plex/test_playback.py
View File

@ -62,19 +62,20 @@ async def test_media_player_playback(
# Test media lookup failure
payload = '{"library_name": "Movies", "title": "Movie 1" }'
with patch("plexapi.library.LibrarySection.search", return_value=None):
with pytest.raises(HomeAssistantError) as excinfo:
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MediaType.MOVIE,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert not playmedia_mock.called
with patch(
"plexapi.library.LibrarySection.search", return_value=None
), pytest.raises(HomeAssistantError) as excinfo:
assert await hass.services.async_call(
MP_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: media_player,
ATTR_MEDIA_CONTENT_TYPE: MediaType.MOVIE,
ATTR_MEDIA_CONTENT_ID: payload,
},
True,
)
assert not playmedia_mock.called
assert f"No {MediaType.MOVIE} results in 'Movies' for" in str(excinfo.value)
movie1 = MockPlexMedia("Movie", "movie")

22
tests/components/recorder/test_util.py
View File

@ -37,9 +37,8 @@ def test_session_scope_not_setup(hass_recorder):
hass = hass_recorder()
with patch.object(
util.get_instance(hass), "get_session", return_value=None
), pytest.raises(RuntimeError):
with util.session_scope(hass=hass):
pass
), pytest.raises(RuntimeError), util.session_scope(hass=hass):
pass
def test_recorder_bad_commit(hass_recorder, recorder_db_url):
@ -689,16 +688,15 @@ async def test_write_lock_db(
with instance.engine.connect() as connection:
connection.execute(text("DROP TABLE events;"))
with util.write_lock_db_sqlite(instance):
with util.write_lock_db_sqlite(instance), pytest.raises(OperationalError):
# Database should be locked now, try writing SQL command
with pytest.raises(OperationalError):
# This needs to be called in another thread since
# the lock method is BEGIN IMMEDIATE and since we have
# a connection per thread with sqlite now, we cannot do it
# in the same thread as the one holding the lock since it
# would be allowed to proceed as the goal is to prevent
# all the other threads from accessing the database
await hass.async_add_executor_job(_drop_table)
# This needs to be called in another thread since
# the lock method is BEGIN IMMEDIATE and since we have
# a connection per thread with sqlite now, we cannot do it
# in the same thread as the one holding the lock since it
# would be allowed to proceed as the goal is to prevent
# all the other threads from accessing the database
await hass.async_add_executor_job(_drop_table)
def test_is_second_sunday():

9
tests/components/renault/test_services.py
View File

@ -92,11 +92,10 @@ async def test_service_set_ac_cancel(
with patch(
"renault_api.renault_vehicle.RenaultVehicle.set_ac_stop",
side_effect=RenaultException("Didn't work"),
) as mock_action:
with pytest.raises(HomeAssistantError, match="Didn't work"):
await hass.services.async_call(
DOMAIN, SERVICE_AC_CANCEL, service_data=data, blocking=True
)
) as mock_action, pytest.raises(HomeAssistantError, match="Didn't work"):
await hass.services.async_call(
DOMAIN, SERVICE_AC_CANCEL, service_data=data, blocking=True
)
assert len(mock_action.mock_calls) == 1
assert mock_action.mock_calls[0][1] == ()

19
tests/components/sensibo/test_button.py
View File

@ -97,13 +97,14 @@ async def test_button_failure(
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_reset_filter",
return_value={"status": "failure"},
), pytest.raises(
HomeAssistantError
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{
ATTR_ENTITY_ID: state_button.entity_id,
},
blocking=True,
)
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{
ATTR_ENTITY_ID: state_button.entity_id,
},
blocking=True,
)

209
tests/components/sensibo/test_climate.py
View File

@ -168,14 +168,13 @@ async def test_climate_fan(
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_FAN_MODE,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_FAN_MODE: "low"},
blocking=True,
)
), pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_FAN_MODE,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_FAN_MODE: "low"},
blocking=True,
)
await hass.async_block_till_done()
state3 = hass.states.get("climate.hallway")
@ -235,14 +234,13 @@ async def test_climate_swing(
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_SWING_MODE,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_SWING_MODE: "fixedtop"},
blocking=True,
)
), pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_SWING_MODE,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_SWING_MODE: "fixedtop"},
blocking=True,
)
await hass.async_block_till_done()
state3 = hass.states.get("climate.hallway")
@ -353,14 +351,13 @@ async def test_climate_temperatures(
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
return_value={"result": {"status": "Success"}},
):
with pytest.raises(MultipleInvalid):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: state1.entity_id},
blocking=True,
)
), pytest.raises(MultipleInvalid):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: state1.entity_id},
blocking=True,
)
await hass.async_block_till_done()
state2 = hass.states.get("climate.hallway")
@ -391,14 +388,13 @@ async def test_climate_temperatures(
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
return_value={"result": {"status": "Success"}},
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_TEMPERATURE: 20},
blocking=True,
)
), pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_TEMPERATURE: 20},
blocking=True,
)
await hass.async_block_till_done()
state2 = hass.states.get("climate.hallway")
@ -447,18 +443,17 @@ async def test_climate_temperature_is_none(
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
):
with pytest.raises(ValueError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: state1.entity_id,
ATTR_TARGET_TEMP_HIGH: 30,
ATTR_TARGET_TEMP_LOW: 20,
},
blocking=True,
)
), pytest.raises(ValueError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: state1.entity_id,
ATTR_TARGET_TEMP_HIGH: 30,
ATTR_TARGET_TEMP_LOW: 20,
},
blocking=True,
)
await hass.async_block_till_done()
state2 = hass.states.get("climate.hallway")
@ -634,14 +629,13 @@ async def test_climate_service_failed(
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
return_value={"result": {"status": "Error", "failureReason": "Did not work"}},
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: state1.entity_id},
blocking=True,
)
), pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: state1.entity_id},
blocking=True,
)
await hass.async_block_till_done()
state2 = hass.states.get("climate.hallway")
@ -752,16 +746,17 @@ async def test_climate_set_timer(
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_timer",
return_value={"status": "failure"},
), pytest.raises(
MultipleInvalid
):
with pytest.raises(MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_TIMER,
{
ATTR_ENTITY_ID: state_climate.entity_id,
},
blocking=True,
)
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_TIMER,
{
ATTR_ENTITY_ID: state_climate.entity_id,
},
blocking=True,
)
await hass.async_block_till_done()
with patch(
@ -770,17 +765,18 @@ async def test_climate_set_timer(
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_timer",
return_value={"status": "failure"},
), pytest.raises(
HomeAssistantError
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_TIMER,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_MINUTES: 30,
},
blocking=True,
)
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_TIMER,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_MINUTES: 30,
},
blocking=True,
)
await hass.async_block_till_done()
with patch(
@ -853,19 +849,20 @@ async def test_climate_pure_boost(
"homeassistant.components.sensibo.util.SensiboClient.async_get_devices_data",
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_pureboost",
), pytest.raises(
MultipleInvalid
):
with pytest.raises(MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_PURE_BOOST,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_INDOOR_INTEGRATION: True,
ATTR_OUTDOOR_INTEGRATION: True,
ATTR_SENSITIVITY: "Sensitive",
},
blocking=True,
)
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_PURE_BOOST,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_INDOOR_INTEGRATION: True,
ATTR_OUTDOOR_INTEGRATION: True,
ATTR_SENSITIVITY: "Sensitive",
},
blocking=True,
)
await hass.async_block_till_done()
with patch(
@ -954,19 +951,20 @@ async def test_climate_climate_react(
"homeassistant.components.sensibo.util.SensiboClient.async_get_devices_data",
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_climate_react",
), pytest.raises(
MultipleInvalid
):
with pytest.raises(MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_PURE_BOOST,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_LOW_TEMPERATURE_THRESHOLD: 0.2,
ATTR_HIGH_TEMPERATURE_THRESHOLD: 30.3,
ATTR_SMART_TYPE: "temperature",
},
blocking=True,
)
await hass.services.async_call(
DOMAIN,
SERVICE_ENABLE_PURE_BOOST,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_LOW_TEMPERATURE_THRESHOLD: 0.2,
ATTR_HIGH_TEMPERATURE_THRESHOLD: 30.3,
ATTR_SMART_TYPE: "temperature",
},
blocking=True,
)
await hass.async_block_till_done()
with patch(
@ -1260,17 +1258,18 @@ async def test_climate_full_ac_state(
"homeassistant.components.sensibo.util.SensiboClient.async_get_devices_data",
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_states",
), pytest.raises(
MultipleInvalid
):
with pytest.raises(MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_FULL_STATE,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_TARGET_TEMPERATURE: 22,
},
blocking=True,
)
await hass.services.async_call(
DOMAIN,
SERVICE_FULL_STATE,
{
ATTR_ENTITY_ID: state_climate.entity_id,
ATTR_TARGET_TEMPERATURE: 22,
},
blocking=True,
)
await hass.async_block_till_done()
with patch(

15
tests/components/sensibo/test_entity.py
View File

@ -75,14 +75,13 @@ async def test_entity_failed_service_calls(
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
side_effect=p_error,
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_FAN_MODE,
{ATTR_ENTITY_ID: state.entity_id, ATTR_FAN_MODE: "low"},
blocking=True,
)
), pytest.raises(HomeAssistantError):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_FAN_MODE,
{ATTR_ENTITY_ID: state.entity_id, ATTR_FAN_MODE: "low"},
blocking=True,
)
state = hass.states.get("climate.hallway")
assert state.attributes["fan_mode"] == "low"

30
tests/components/sensibo/test_select.py
View File

@ -89,14 +89,15 @@ async def test_select_set_option(
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
return_value={"result": {"status": "failed"}},
), pytest.raises(
HomeAssistantError
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_OPTION: "fixedleft"},
blocking=True,
)
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_OPTION: "fixedleft"},
blocking=True,
)
await hass.async_block_till_done()
state2 = hass.states.get("select.hallway_horizontal_swing")
@ -130,14 +131,15 @@ async def test_select_set_option(
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_ac_state_property",
return_value={"result": {"status": "Failed", "failureReason": "No connection"}},
), pytest.raises(
HomeAssistantError
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_OPTION: "fixedleft"},
blocking=True,
)
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{ATTR_ENTITY_ID: state1.entity_id, ATTR_OPTION: "fixedleft"},
blocking=True,
)
await hass.async_block_till_done()
state2 = hass.states.get("select.hallway_horizontal_swing")

38
tests/components/sensibo/test_switch.py
View File

@ -195,16 +195,17 @@ async def test_switch_command_failure(
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_set_timer",
return_value={"status": "failure"},
), pytest.raises(
HomeAssistantError
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: state1.entity_id,
},
blocking=True,
)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{
ATTR_ENTITY_ID: state1.entity_id,
},
blocking=True,
)
with patch(
"homeassistant.components.sensibo.util.SensiboClient.async_get_devices_data",
@ -212,16 +213,17 @@ async def test_switch_command_failure(
), patch(
"homeassistant.components.sensibo.util.SensiboClient.async_del_timer",
return_value={"status": "failure"},
), pytest.raises(
HomeAssistantError
):
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: state1.entity_id,
},
blocking=True,
)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{
ATTR_ENTITY_ID: state1.entity_id,
},
blocking=True,
)
async def test_switch_climate_react(

25
tests/components/signal_messenger/test_notify.py
View File

@ -72,9 +72,8 @@ def test_send_message_to_api_with_bad_data_throws_error(
signal_requests_mock = signal_requests_mock_factory(False)
with caplog.at_level(
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
):
with pytest.raises(SignalCliRestApiError) as exc:
signal_notification_service.send_message(MESSAGE)
), pytest.raises(SignalCliRestApiError) as exc:
signal_notification_service.send_message(MESSAGE)
assert "Sending signal message" in caplog.text
assert signal_requests_mock.called
@ -90,10 +89,9 @@ def test_send_message_with_bad_data_throws_vol_error(
"""Test sending a message with bad data throws an error."""
with caplog.at_level(
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
):
with pytest.raises(vol.Invalid) as exc:
data = {"test": "test"}
signal_notification_service.send_message(MESSAGE, **{"data": data})
), pytest.raises(vol.Invalid) as exc:
data = {"test": "test"}
signal_notification_service.send_message(MESSAGE, **{"data": data})
assert "Sending signal message" in caplog.text
assert "extra keys not allowed" in str(exc.value)
@ -108,13 +106,12 @@ def test_send_message_with_attachment(
signal_requests_mock = signal_requests_mock_factory()
with caplog.at_level(
logging.DEBUG, logger="homeassistant.components.signal_messenger.notify"
):
with tempfile.NamedTemporaryFile(
mode="w", suffix=".png", prefix=os.path.basename(__file__)
) as temp_file:
temp_file.write("attachment_data")
data = {"attachments": [temp_file.name]}
signal_notification_service.send_message(MESSAGE, **{"data": data})
), tempfile.NamedTemporaryFile(
mode="w", suffix=".png", prefix=os.path.basename(__file__)
) as temp_file:
temp_file.write("attachment_data")
data = {"attachments": [temp_file.name]}
signal_notification_service.send_message(MESSAGE, **{"data": data})
assert "Sending signal message" in caplog.text
assert signal_requests_mock.called

39
tests/components/wemo/test_init.py
View File

@ -125,27 +125,26 @@ async def test_discovery(hass, pywemo_registry):
# Setup the component and start discovery.
with patch(
"pywemo.discover_devices", return_value=pywemo_devices
) as mock_discovery:
with patch(
"homeassistant.components.wemo.WemoDiscovery.discover_statics"
) as mock_discover_statics:
assert await async_setup_component(
hass, DOMAIN, {DOMAIN: {CONF_DISCOVERY: True}}
)
await pywemo_registry.semaphore.acquire() # Returns after platform setup.
mock_discovery.assert_called()
mock_discover_statics.assert_called()
pywemo_devices.append(create_device(2))
) as mock_discovery, patch(
"homeassistant.components.wemo.WemoDiscovery.discover_statics"
) as mock_discover_statics:
assert await async_setup_component(
hass, DOMAIN, {DOMAIN: {CONF_DISCOVERY: True}}
)
await pywemo_registry.semaphore.acquire() # Returns after platform setup.
mock_discovery.assert_called()
mock_discover_statics.assert_called()
pywemo_devices.append(create_device(2))
# Test that discovery runs periodically and the async_dispatcher_send code works.
async_fire_time_changed(
hass,
dt.utcnow()
+ timedelta(seconds=WemoDiscovery.ADDITIONAL_SECONDS_BETWEEN_SCANS + 1),
)
await hass.async_block_till_done()
# Test that discover_statics runs during discovery
assert mock_discover_statics.call_count == 3
# Test that discovery runs periodically and the async_dispatcher_send code works.
async_fire_time_changed(
hass,
dt.utcnow()
+ timedelta(seconds=WemoDiscovery.ADDITIONAL_SECONDS_BETWEEN_SCANS + 1),
)
await hass.async_block_till_done()
# Test that discover_statics runs during discovery
assert mock_discover_statics.call_count == 3
# Verify that the expected number of devices were setup.
entity_reg = er.async_get(hass)

9
tests/components/yolink/test_config_flow.py
View File

@ -192,11 +192,10 @@ async def test_reauthentication(
},
)
with patch("homeassistant.components.yolink.api.ConfigEntryAuth"):
with patch(
"homeassistant.components.yolink.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
with patch("homeassistant.components.yolink.api.ConfigEntryAuth"), patch(
"homeassistant.components.yolink.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
token_data = old_entry.data["token"]
assert token_data["access_token"] == "mock-access-token"
assert token_data["refresh_token"] == "mock-refresh-token"

10
tests/components/zha/test_gateway.py
View File

@ -256,9 +256,8 @@ async def test_gateway_initialize_failure(hass, device_light_1, coordinator):
with patch(
"bellows.zigbee.application.ControllerApplication.new",
side_effect=[asyncio.TimeoutError(), FileNotFoundError(), RuntimeError()],
) as mock_new:
with pytest.raises(RuntimeError):
await zha_gateway.async_initialize()
) as mock_new, pytest.raises(RuntimeError):
await zha_gateway.async_initialize()
assert mock_new.call_count == 3
@ -272,9 +271,8 @@ async def test_gateway_initialize_failure_transient(hass, device_light_1, coordi
with patch(
"bellows.zigbee.application.ControllerApplication.new",
side_effect=[RuntimeError(), zigpy.exceptions.TransientConnectionError()],
) as mock_new:
with pytest.raises(ConfigEntryNotReady):
await zha_gateway.async_initialize()
) as mock_new, pytest.raises(ConfigEntryNotReady):
await zha_gateway.async_initialize()
# Initialization immediately stops and is retried after TransientConnectionError
assert mock_new.call_count == 2