jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve code quality command_line (#65333)

Browse Source
This commit is contained in:
G Johansson 2022-02-12 15:19:37 +01:00 committed by GitHub
parent db6969739f
commit 3771c154fa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 190 additions and 205 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
homeassistant/components/command_line/__init__.py
View File

@ -1,4 +1,5 @@
"""The command_line component."""
from __future__ import annotations
import logging
import subprocess
@ -6,7 +7,9 @@ import subprocess
_LOGGER = logging.getLogger(__name__)
def call_shell_with_timeout(command, timeout, *, log_return_code=True):
def call_shell_with_timeout(
command: str, timeout: int, *, log_return_code: bool = True
) -> int:
"""Run a shell command with a timeout.
If log_return_code is set to False, it will not print an error if a non-zero
@ -30,7 +33,7 @@ def call_shell_with_timeout(command, timeout, *, log_return_code=True):
return -1
def check_output_or_log(command, timeout):
def check_output_or_log(command: str, timeout: int) -> str | None:
"""Run a shell command with a timeout and return the output."""
try:
return_value = subprocess.check_output(

63
homeassistant/components/command_line/binary_sensor.py
View File

@ -23,6 +23,7 @@ from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT, DOMAIN, PLATFORMS
@ -59,14 +60,14 @@ def setup_platform(
setup_reload_service(hass, DOMAIN, PLATFORMS)
name = config.get(CONF_NAME)
command = config.get(CONF_COMMAND)
payload_off = config.get(CONF_PAYLOAD_OFF)
payload_on = config.get(CONF_PAYLOAD_ON)
device_class = config.get(CONF_DEVICE_CLASS)
value_template = config.get(CONF_VALUE_TEMPLATE)
command_timeout = config.get(CONF_COMMAND_TIMEOUT)
unique_id = config.get(CONF_UNIQUE_ID)
name: str = config[CONF_NAME]
command: str = config[CONF_COMMAND]
payload_off: str = config[CONF_PAYLOAD_OFF]
payload_on: str = config[CONF_PAYLOAD_ON]
device_class: str | None = config.get(CONF_DEVICE_CLASS)
value_template: Template | None = config.get(CONF_VALUE_TEMPLATE)
command_timeout: int = config[CONF_COMMAND_TIMEOUT]
unique_id: str | None = config.get(CONF_UNIQUE_ID)
if value_template is not None:
value_template.hass = hass
data = CommandSensorData(hass, command, command_timeout)
@ -74,7 +75,6 @@ def setup_platform(
add_entities(
[
CommandBinarySensor(
hass,
data,
name,
device_class,
@ -93,42 +93,25 @@ class CommandBinarySensor(BinarySensorEntity):
def __init__(
self,
hass,
data,
name,
device_class,
payload_on,
payload_off,
value_template,
unique_id,
):
data: CommandSensorData,
name: str,
device_class: str | None,
payload_on: str,
payload_off: str,
value_template: Template | None,
unique_id: str | None,
) -> None:
"""Initialize the Command line binary sensor."""
self._hass = hass
self.data = data
self._name = name
self._device_class = device_class
self._state = False
self._attr_name = name
self._attr_device_class = device_class
self._attr_is_on = None
self._payload_on = payload_on
self._payload_off = payload_off
self._value_template = value_template
self._attr_unique_id = unique_id
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def is_on(self):
"""Return true if the binary sensor is on."""
return self._state
@property
def device_class(self):
"""Return the class of the binary sensor."""
return self._device_class
def update(self):
def update(self) -> None:
"""Get the latest data and updates the state."""
self.data.update()
value = self.data.value
@ -136,6 +119,6 @@ class CommandBinarySensor(BinarySensorEntity):
if self._value_template is not None:
value = self._value_template.render_with_possible_json_value(value, False)
if value == self._payload_on:
self._state = True
self._attr_is_on = True
elif value == self._payload_off:
self._state = False
self._attr_is_on = False

66
homeassistant/components/command_line/cover.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import voluptuous as vol
@ -20,6 +21,7 @@ from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import call_shell_with_timeout, check_output_or_log
@ -55,17 +57,16 @@ def setup_platform(
setup_reload_service(hass, DOMAIN, PLATFORMS)
devices = config.get(CONF_COVERS, {})
devices: dict[str, Any] = config.get(CONF_COVERS, {})
covers = []
for device_name, device_config in devices.items():
value_template = device_config.get(CONF_VALUE_TEMPLATE)
value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE)
if value_template is not None:
value_template.hass = hass
covers.append(
CommandCover(
hass,
device_config.get(CONF_FRIENDLY_NAME, device_name),
device_config[CONF_COMMAND_OPEN],
device_config[CONF_COMMAND_CLOSE],
@ -89,20 +90,18 @@ class CommandCover(CoverEntity):
def __init__(
self,
hass,
name,
command_open,
command_close,
command_stop,
command_state,
value_template,
timeout,
unique_id,
):
name: str,
command_open: str,
command_close: str,
command_stop: str,
command_state: str | None,
value_template: Template | None,
timeout: int,
unique_id: str | None,
) -> None:
"""Initialize the cover."""
self._hass = hass
self._name = name
self._state = None
self._attr_name = name
self._state: int | None = None
self._command_open = command_open
self._command_close = command_close
self._command_stop = command_stop
@ -110,8 +109,9 @@ class CommandCover(CoverEntity):
self._value_template = value_template
self._timeout = timeout
self._attr_unique_id = unique_id
self._attr_should_poll = bool(command_state)
def _move_cover(self, command):
def _move_cover(self, command: str) -> bool:
"""Execute the actual commands."""
_LOGGER.info("Running command: %s", command)
@ -123,35 +123,29 @@ class CommandCover(CoverEntity):
return success
@property
def should_poll(self):
"""Only poll if we have state command."""
return self._command_state is not None
@property
def name(self):
"""Return the name of the cover."""
return self._name
@property
def is_closed(self):
def is_closed(self) -> bool | None:
"""Return if the cover is closed."""
if self.current_cover_position is not None:
return self.current_cover_position == 0
return None
@property
def current_cover_position(self):
def current_cover_position(self) -> int | None:
"""Return current position of cover.
None is unknown, 0 is closed, 100 is fully open.
"""
return self._state
def _query_state(self):
def _query_state(self) -> str | None:
"""Query for the state."""
_LOGGER.info("Running state value command: %s", self._command_state)
return check_output_or_log(self._command_state, self._timeout)
if self._command_state:
_LOGGER.info("Running state value command: %s", self._command_state)
return check_output_or_log(self._command_state, self._timeout)
if TYPE_CHECKING:
return None
def update(self):
def update(self) -> None:
"""Update device state."""
if self._command_state:
payload = str(self._query_state())
@ -159,14 +153,14 @@ class CommandCover(CoverEntity):
payload = self._value_template.render_with_possible_json_value(payload)
self._state = int(payload)
def open_cover(self, **kwargs):
def open_cover(self, **kwargs) -> None:
"""Open the cover."""
self._move_cover(self._command_open)
def close_cover(self, **kwargs):
def close_cover(self, **kwargs) -> None:
"""Close the cover."""
self._move_cover(self._command_close)
def stop_cover(self, **kwargs):
def stop_cover(self, **kwargs) -> None:
"""Stop the cover."""
self._move_cover(self._command_stop)

18
homeassistant/components/command_line/notify.py
View File

@ -1,4 +1,6 @@
"""Support for command line notification services."""
from __future__ import annotations
import logging
import subprocess
@ -6,7 +8,9 @@ import voluptuous as vol
from homeassistant.components.notify import PLATFORM_SCHEMA, BaseNotificationService
from homeassistant.const import CONF_COMMAND, CONF_NAME
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.process import kill_subprocess
from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT
@ -22,10 +26,14 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def get_service(hass, config, discovery_info=None):
def get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> CommandLineNotificationService:
"""Get the Command Line notification service."""
command = config[CONF_COMMAND]
timeout = config[CONF_COMMAND_TIMEOUT]
command: str = config[CONF_COMMAND]
timeout: int = config[CONF_COMMAND_TIMEOUT]
return CommandLineNotificationService(command, timeout)
@ -33,12 +41,12 @@ def get_service(hass, config, discovery_info=None):
class CommandLineNotificationService(BaseNotificationService):
"""Implement the notification service for the Command Line service."""
def __init__(self, command, timeout):
def __init__(self, command: str, timeout: int) -> None:
"""Initialize the service."""
self.command = command
self._timeout = timeout
def send_message(self, message="", **kwargs):
def send_message(self, message="", **kwargs) -> None:
"""Send a message to a command line."""
with subprocess.Popen(
self.command,

88
homeassistant/components/command_line/sensor.py
View File

@ -19,10 +19,10 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import TemplateError
from homeassistant.helpers import template
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import check_output_or_log
@ -59,23 +59,19 @@ def setup_platform(
setup_reload_service(hass, DOMAIN, PLATFORMS)
name = config.get(CONF_NAME)
command = config.get(CONF_COMMAND)
unit = config.get(CONF_UNIT_OF_MEASUREMENT)
value_template = config.get(CONF_VALUE_TEMPLATE)
command_timeout = config.get(CONF_COMMAND_TIMEOUT)
unique_id = config.get(CONF_UNIQUE_ID)
name: str = config[CONF_NAME]
command: str = config[CONF_COMMAND]
unit: str | None = config.get(CONF_UNIT_OF_MEASUREMENT)
value_template: Template | None = config.get(CONF_VALUE_TEMPLATE)
command_timeout: int = config[CONF_COMMAND_TIMEOUT]
unique_id: str | None = config.get(CONF_UNIQUE_ID)
if value_template is not None:
value_template.hass = hass
json_attributes = config.get(CONF_JSON_ATTRIBUTES)
json_attributes: list[str] | None = config.get(CONF_JSON_ATTRIBUTES)
data = CommandSensorData(hass, command, command_timeout)
add_entities(
[
CommandSensor(
hass, data, name, unit, value_template, json_attributes, unique_id
)
],
[CommandSensor(data, name, unit, value_template, json_attributes, unique_id)],
True,
)
@ -85,57 +81,35 @@ class CommandSensor(SensorEntity):
def __init__(
self,
hass,
data,
name,
unit_of_measurement,
value_template,
json_attributes,
unique_id,
):
data: CommandSensorData,
name: str,
unit_of_measurement: str | None,
value_template: Template | None,
json_attributes: list[str] | None,
unique_id: str | None,
) -> None:
"""Initialize the sensor."""
self._hass = hass
self.data = data
self._attributes = None
self._attr_extra_state_attributes = {}
self._json_attributes = json_attributes
self._name = name
self._state = None
self._unit_of_measurement = unit_of_measurement
self._attr_name = name
self._attr_native_value = None
self._attr_native_unit_of_measurement = unit_of_measurement
self._value_template = value_template
self._attr_unique_id = unique_id
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def native_unit_of_measurement(self):
"""Return the unit the value is expressed in."""
return self._unit_of_measurement
@property
def native_value(self):
"""Return the state of the device."""
return self._state
@property
def extra_state_attributes(self):
"""Return the state attributes."""
return self._attributes
def update(self):
def update(self) -> None:
"""Get the latest data and updates the state."""
self.data.update()
value = self.data.value
if self._json_attributes:
self._attributes = {}
self._attr_extra_state_attributes = {}
if value:
try:
json_dict = json.loads(value)
if isinstance(json_dict, Mapping):
self._attributes = {
self._attr_extra_state_attributes = {
k: json_dict[k]
for k in self._json_attributes
if k in json_dict
@ -150,24 +124,26 @@ class CommandSensor(SensorEntity):
if value is None:
value = STATE_UNKNOWN
elif self._value_template is not None:
self._state = self._value_template.render_with_possible_json_value(
value, STATE_UNKNOWN
self._attr_native_value = (
self._value_template.render_with_possible_json_value(
value, STATE_UNKNOWN
)
)
else:
self._state = value
self._attr_native_value = value
class CommandSensorData:
"""The class for handling the data retrieval."""
def __init__(self, hass, command, command_timeout):
def __init__(self, hass: HomeAssistant, command: str, command_timeout: int) -> None:
"""Initialize the data object."""
self.value = None
self.value: str | None = None
self.hass = hass
self.command = command
self.timeout = command_timeout
def update(self):
def update(self) -> None:
"""Get the latest data with a shell command."""
command = self.command
@ -177,7 +153,7 @@ class CommandSensorData:
args_compiled = None
else:
prog, args = command.split(" ", 1)
args_compiled = template.Template(args, self.hass)
args_compiled = Template(args, self.hass)
if args_compiled:
try:

82
homeassistant/components/command_line/switch.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import voluptuous as vol
@ -24,6 +25,7 @@ from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import call_shell_with_timeout, check_output_or_log
@ -59,22 +61,21 @@ def setup_platform(
setup_reload_service(hass, DOMAIN, PLATFORMS)
devices = config.get(CONF_SWITCHES, {})
devices: dict[str, Any] = config.get(CONF_SWITCHES, {})
switches = []
for object_id, device_config in devices.items():
value_template = device_config.get(CONF_VALUE_TEMPLATE)
value_template: Template | None = device_config.get(CONF_VALUE_TEMPLATE)
if value_template is not None:
value_template.hass = hass
icon_template = device_config.get(CONF_ICON_TEMPLATE)
icon_template: Template | None = device_config.get(CONF_ICON_TEMPLATE)
if icon_template is not None:
icon_template.hass = hass
switches.append(
CommandSwitch(
hass,
object_id,
device_config.get(CONF_FRIENDLY_NAME, object_id),
device_config[CONF_COMMAND_ON],
@ -99,22 +100,20 @@ class CommandSwitch(SwitchEntity):
def __init__(
self,
hass,
object_id,
friendly_name,
command_on,
command_off,
command_state,
icon_template,
value_template,
timeout,
unique_id,
):
object_id: str,
friendly_name: str,
command_on: str,
command_off: str,
command_state: str | None,
icon_template: Template | None,
value_template: Template | None,
timeout: int,
unique_id: str | None,
) -> None:
"""Initialize the switch."""
self._hass = hass
self.entity_id = ENTITY_ID_FORMAT.format(object_id)
self._name = friendly_name
self._state = False
self._attr_name = friendly_name
self._attr_is_on = False
self._command_on = command_on
self._command_off = command_off
self._command_state = command_state
@ -122,8 +121,9 @@ class CommandSwitch(SwitchEntity):
self._value_template = value_template
self._timeout = timeout
self._attr_unique_id = unique_id
self._attr_should_poll = bool(command_state)
def _switch(self, command):
def _switch(self, command: str) -> bool:
"""Execute the actual commands."""
_LOGGER.info("Running command: %s", command)
@ -134,12 +134,12 @@ class CommandSwitch(SwitchEntity):
return success
def _query_state_value(self, command):
def _query_state_value(self, command: str) -> str | None:
"""Execute state command for return value."""
_LOGGER.info("Running state value command: %s", command)
return check_output_or_log(command, self._timeout)
def _query_state_code(self, command):
def _query_state_code(self, command: str) -> bool:
"""Execute state command for return code."""
_LOGGER.info("Running state code command: %s", command)
return (
@ -147,32 +147,20 @@ class CommandSwitch(SwitchEntity):
)
@property
def should_poll(self):
"""Only poll if we have state command."""
return self._command_state is not None
@property
def name(self):
"""Return the name of the switch."""
return self._name
@property
def is_on(self):
"""Return true if device is on."""
return self._state
@property
def assumed_state(self):
def assumed_state(self) -> bool:
"""Return true if we do optimistic updates."""
return self._command_state is None
def _query_state(self):
def _query_state(self) -> str | int | None:
"""Query for state."""
if self._value_template:
return self._query_state_value(self._command_state)
return self._query_state_code(self._command_state)
if self._command_state:
if self._value_template:
return self._query_state_value(self._command_state)
return self._query_state_code(self._command_state)
if TYPE_CHECKING:
return None
def update(self):
def update(self) -> None:
"""Update device state."""
if self._command_state:
payload = str(self._query_state())
@ -182,16 +170,16 @@ class CommandSwitch(SwitchEntity):
)
if self._value_template:
payload = self._value_template.render_with_possible_json_value(payload)
self._state = payload.lower() == "true"
self._attr_is_on = payload.lower() == "true"
def turn_on(self, **kwargs):
def turn_on(self, **kwargs) -> None:
"""Turn the device on."""
if self._switch(self._command_on) and not self._command_state:
self._state = True
self._attr_is_on = True
self.schedule_update_ha_state()
def turn_off(self, **kwargs):
def turn_off(self, **kwargs) -> None:
"""Turn the device off."""
if self._switch(self._command_off) and not self._command_state:
self._state = False
self._attr_is_on = False
self.schedule_update_ha_state()

4
tests/components/command_line/test_binary_sensor.py
View File

@ -51,6 +51,7 @@ async def test_template(hass: HomeAssistant) -> None:
)
entity_state = hass.states.get("binary_sensor.test")
assert entity_state
assert entity_state.state == STATE_ON
@ -65,10 +66,11 @@ async def test_sensor_off(hass: HomeAssistant) -> None:
},
)
entity_state = hass.states.get("binary_sensor.test")
assert entity_state
assert entity_state.state == STATE_OFF
async def test_unique_id(hass):
async def test_unique_id(hass: HomeAssistant) -> None:
"""Test unique_id option and if it only creates one binary sensor per id."""
assert await setup.async_setup_component(
hass,

10
tests/components/command_line/test_cover.py
View File

@ -6,6 +6,8 @@ import tempfile
from typing import Any
from unittest.mock import patch
from pytest import LogCaptureFixture
from homeassistant import config as hass_config, setup
from homeassistant.components.cover import DOMAIN, SCAN_INTERVAL
from homeassistant.const import (
@ -36,7 +38,7 @@ async def setup_test_entity(hass: HomeAssistant, config_dict: dict[str, Any]) ->
await hass.async_block_till_done()
async def test_no_covers(caplog: Any, hass: HomeAssistant) -> None:
async def test_no_covers(caplog: LogCaptureFixture, hass: HomeAssistant) -> None:
"""Test that the cover does not polls when there's no state command."""
with patch(
@ -150,7 +152,9 @@ async def test_reload(hass: HomeAssistant) -> None:
assert hass.states.get("cover.from_yaml")
async def test_move_cover_failure(caplog: Any, hass: HomeAssistant) -> None:
async def test_move_cover_failure(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test with state value."""
await setup_test_entity(
@ -163,7 +167,7 @@ async def test_move_cover_failure(caplog: Any, hass: HomeAssistant) -> None:
assert "Command failed" in caplog.text
async def test_unique_id(hass):
async def test_unique_id(hass: HomeAssistant) -> None:
"""Test unique_id option and if it only creates one cover per id."""
await setup_test_entity(
hass,

12
tests/components/command_line/test_notify.py
View File

@ -7,6 +7,8 @@ import tempfile
from typing import Any
from unittest.mock import patch
from pytest import LogCaptureFixture
from homeassistant import setup
from homeassistant.components.notify import DOMAIN
from homeassistant.core import HomeAssistant
@ -60,7 +62,9 @@ async def test_command_line_output(hass: HomeAssistant) -> None:
assert message == handle.read()
async def test_error_for_none_zero_exit_code(caplog: Any, hass: HomeAssistant) -> None:
async def test_error_for_none_zero_exit_code(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test if an error is logged for non zero exit codes."""
await setup_test_service(
hass,
@ -75,7 +79,7 @@ async def test_error_for_none_zero_exit_code(caplog: Any, hass: HomeAssistant) -
assert "Command failed" in caplog.text
async def test_timeout(caplog: Any, hass: HomeAssistant) -> None:
async def test_timeout(caplog: LogCaptureFixture, hass: HomeAssistant) -> None:
"""Test blocking is not forever."""
await setup_test_service(
hass,
@ -90,7 +94,9 @@ async def test_timeout(caplog: Any, hass: HomeAssistant) -> None:
assert "Timeout" in caplog.text
async def test_subprocess_exceptions(caplog: Any, hass: HomeAssistant) -> None:
async def test_subprocess_exceptions(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test that notify subprocess exceptions are handled correctly."""
with patch(

28
tests/components/command_line/test_sensor.py
View File

@ -4,6 +4,8 @@ from __future__ import annotations
from typing import Any
from unittest.mock import patch
from pytest import LogCaptureFixture
from homeassistant import setup
from homeassistant.components.sensor import DOMAIN
from homeassistant.core import HomeAssistant
@ -98,7 +100,9 @@ async def test_template_render_with_quote(hass: HomeAssistant) -> None:
)
async def test_bad_template_render(caplog: Any, hass: HomeAssistant) -> None:
async def test_bad_template_render(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test rendering a broken template."""
await setup_test_entities(
@ -141,7 +145,9 @@ async def test_update_with_json_attrs(hass: HomeAssistant) -> None:
assert entity_state.attributes["key_three"] == "value_three"
async def test_update_with_json_attrs_no_data(caplog, hass: HomeAssistant) -> None: # type: ignore[no-untyped-def]
async def test_update_with_json_attrs_no_data(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test attributes when no JSON result fetched."""
await setup_test_entities(
@ -157,7 +163,9 @@ async def test_update_with_json_attrs_no_data(caplog, hass: HomeAssistant) -> No
assert "Empty reply found when expecting JSON data" in caplog.text
async def test_update_with_json_attrs_not_dict(caplog, hass: HomeAssistant) -> None: # type: ignore[no-untyped-def]
async def test_update_with_json_attrs_not_dict(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test attributes when the return value not a dict."""
await setup_test_entities(
@ -173,7 +181,9 @@ async def test_update_with_json_attrs_not_dict(caplog, hass: HomeAssistant) -> N
assert "JSON result was not a dictionary" in caplog.text
async def test_update_with_json_attrs_bad_json(caplog, hass: HomeAssistant) -> None: # type: ignore[no-untyped-def]
async def test_update_with_json_attrs_bad_json(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test attributes when the return value is invalid JSON."""
await setup_test_entities(
@ -189,7 +199,9 @@ async def test_update_with_json_attrs_bad_json(caplog, hass: HomeAssistant) -> N
assert "Unable to parse output as JSON" in caplog.text
async def test_update_with_missing_json_attrs(caplog, hass: HomeAssistant) -> None: # type: ignore[no-untyped-def]
async def test_update_with_missing_json_attrs(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test attributes when an expected key is missing."""
await setup_test_entities(
@ -208,7 +220,9 @@ async def test_update_with_missing_json_attrs(caplog, hass: HomeAssistant) -> No
assert "missing_key" not in entity_state.attributes
async def test_update_with_unnecessary_json_attrs(caplog, hass: HomeAssistant) -> None: # type: ignore[no-untyped-def]
async def test_update_with_unnecessary_json_attrs(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test attributes when an expected key is missing."""
await setup_test_entities(
@ -226,7 +240,7 @@ async def test_update_with_unnecessary_json_attrs(caplog, hass: HomeAssistant) -
assert "key_three" not in entity_state.attributes
async def test_unique_id(hass):
async def test_unique_id(hass: HomeAssistant) -> None:
"""Test unique_id option and if it only creates one sensor per id."""
assert await setup.async_setup_component(
hass,

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

@ -8,6 +8,8 @@ import tempfile
from typing import Any
from unittest.mock import patch
from pytest import LogCaptureFixture
from homeassistant import setup
from homeassistant.components.switch import DOMAIN, SCAN_INTERVAL
from homeassistant.const import (
@ -269,10 +271,13 @@ async def test_name_is_set_correctly(hass: HomeAssistant) -> None:
)
entity_state = hass.states.get("switch.test")
assert entity_state
assert entity_state.name == "Test friendly name!"
async def test_switch_command_state_fail(caplog: Any, hass: HomeAssistant) -> None:
async def test_switch_command_state_fail(
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test that switch failures are handled correctly."""
await setup_test_entity(
hass,
@ -289,6 +294,7 @@ async def test_switch_command_state_fail(caplog: Any, hass: HomeAssistant) -> No
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test")
assert entity_state
assert entity_state.state == "on"
await hass.services.async_call(
@ -300,13 +306,14 @@ async def test_switch_command_state_fail(caplog: Any, hass: HomeAssistant) -> No
await hass.async_block_till_done()
entity_state = hass.states.get("switch.test")
assert entity_state
assert entity_state.state == "on"
assert "Command failed" in caplog.text
async def test_switch_command_state_code_exceptions(
caplog: Any, hass: HomeAssistant
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test that switch state code exceptions are handled correctly."""
@ -339,7 +346,7 @@ async def test_switch_command_state_code_exceptions(
async def test_switch_command_state_value_exceptions(
caplog: Any, hass: HomeAssistant
caplog: LogCaptureFixture, hass: HomeAssistant
) -> None:
"""Test that switch state value exceptions are handled correctly."""
@ -372,14 +379,14 @@ async def test_switch_command_state_value_exceptions(
assert "Error trying to exec command" in caplog.text
async def test_no_switches(caplog: Any, hass: HomeAssistant) -> None:
async def test_no_switches(caplog: LogCaptureFixture, hass: HomeAssistant) -> None:
"""Test with no switches."""
await setup_test_entity(hass, {})
assert "No switches" in caplog.text
async def test_unique_id(hass):
async def test_unique_id(hass: HomeAssistant) -> None:
"""Test unique_id option and if it only creates one switch per id."""
await setup_test_entity(
hass,