Update pylint to 2.8.0 (#49637)

homeassistant/__main__.py

@@ -145,6 +145,7 @@ def daemonize() -> None:
         sys.exit(0)
 
     # redirect standard file descriptors to devnull
+    # pylint: disable=consider-using-with
     infd = open(os.devnull)
     outfd = open(os.devnull, "a+")
     sys.stdout.flush()

homeassistant/components/alexa/handlers.py

@@ -1374,10 +1374,7 @@ async def async_api_seek(hass, config, directive, context):
         msg = f"{entity} did not return the current media position."
         raise AlexaVideoActionNotPermittedForContentError(msg)
 
-    seek_position = int(current_position) + int(position_delta / 1000)
+    seek_position = max(int(current_position) + int(position_delta / 1000), 0)
-
-    if seek_position < 0:
-        seek_position = 0
 
     media_duration = entity.attributes.get(media_player.ATTR_MEDIA_DURATION)
     if media_duration and 0 < int(media_duration) < seek_position:

homeassistant/components/command_line/notify.py

@@ -7,6 +7,7 @@ import voluptuous as vol
 from homeassistant.components.notify import PLATFORM_SCHEMA, BaseNotificationService
 from homeassistant.const import CONF_COMMAND, CONF_NAME
 import homeassistant.helpers.config_validation as cv
+from homeassistant.util.process import kill_subprocess
 
 from .const import CONF_COMMAND_TIMEOUT, DEFAULT_TIMEOUT
 
@@ -39,17 +40,18 @@ class CommandLineNotificationService(BaseNotificationService):
 
     def send_message(self, message="", **kwargs):
         """Send a message to a command line."""
-        try:
+        with subprocess.Popen(
-            proc = subprocess.Popen(
+            self.command,
-                self.command,
+            universal_newlines=True,
-                universal_newlines=True,
+            stdin=subprocess.PIPE,
-                stdin=subprocess.PIPE,
+            shell=True,  # nosec # shell by design
-                shell=True,  # nosec # shell by design
+        ) as proc:
-            )
+            try:
-            proc.communicate(input=message, timeout=self._timeout)
+                proc.communicate(input=message, timeout=self._timeout)
-            if proc.returncode != 0:
+                if proc.returncode != 0:
-                _LOGGER.error("Command failed: %s", self.command)
+                    _LOGGER.error("Command failed: %s", self.command)
-        except subprocess.TimeoutExpired:
+            except subprocess.TimeoutExpired:
-            _LOGGER.error("Timeout for command: %s", self.command)
+                _LOGGER.error("Timeout for command: %s", self.command)
-        except subprocess.SubprocessError:
+                kill_subprocess(proc)
-            _LOGGER.error("Error trying to exec command: %s", self.command)
+            except subprocess.SubprocessError:
+                _LOGGER.error("Error trying to exec command: %s", self.command)

homeassistant/components/denonavr/media_player.py

@@ -153,7 +153,7 @@ class DenonDevice(MediaPlayerEntity):
         )
         self._available = True
 
-    def async_log_errors(  # pylint: disable=no-self-argument
+    def async_log_errors(
         func: Coroutine,
     ) -> Coroutine:
         """
@@ -168,7 +168,7 @@ class DenonDevice(MediaPlayerEntity):
             # pylint: disable=protected-access
             available = True
             try:
-                return await func(self, *args, **kwargs)  # pylint: disable=not-callable
+                return await func(self, *args, **kwargs)
             except AvrTimoutError:
                 available = False
                 if self._available is True:
@@ -203,7 +203,7 @@ class DenonDevice(MediaPlayerEntity):
                 _LOGGER.error(
                     "Error %s occurred in method %s for Denon AVR receiver",
                     err,
-                    func.__name__,  # pylint: disable=no-member
+                    func.__name__,
                     exc_info=True,
                 )
             finally:

homeassistant/components/devolo_home_control/config_flow.py

@@ -7,12 +7,7 @@ from homeassistant.core import callback
 from homeassistant.helpers.typing import DiscoveryInfoType
 
 from . import configure_mydevolo
-from .const import (  # pylint:disable=unused-import
+from .const import CONF_MYDEVOLO, DEFAULT_MYDEVOLO, DOMAIN, SUPPORTED_MODEL_TYPES
-    CONF_MYDEVOLO,
-    DEFAULT_MYDEVOLO,
-    DOMAIN,
-    SUPPORTED_MODEL_TYPES,
-)
 from .exceptions import CredentialsInvalid
 
 

homeassistant/components/hangouts/hangouts_bot.py

@@ -289,6 +289,7 @@ class HangoutsBot:
                 uri = data.get("image_file")
                 if self.hass.config.is_allowed_path(uri):
                     try:
+                        # pylint: disable=consider-using-with
                         image_file = open(uri, "rb")
                     except OSError as error:
                         _LOGGER.error(

homeassistant/components/met_eireann/config_flow.py

@@ -6,7 +6,6 @@ from homeassistant import config_entries
 from homeassistant.const import CONF_ELEVATION, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
 import homeassistant.helpers.config_validation as cv
 
-# pylint:disable=unused-import
 from .const import DOMAIN, HOME_LOCATION_NAME
 
 

homeassistant/components/motioneye/config_flow.py

@@ -21,7 +21,7 @@ from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.typing import ConfigType
 
 from . import create_motioneye_client
-from .const import (  # pylint:disable=unused-import
+from .const import (
     CONF_ADMIN_PASSWORD,
     CONF_ADMIN_USERNAME,
     CONF_CONFIG_ENTRY,

homeassistant/components/nfandroidtv/notify.py

@@ -266,7 +266,7 @@ class NFAndroidTVNotificationService(BaseNotificationService):
             if local_path is not None:
                 # Check whether path is whitelisted in configuration.yaml
                 if self.is_allowed_path(local_path):
-                    return open(local_path, "rb")
+                    return open(local_path, "rb")  # pylint: disable=consider-using-with
                 _LOGGER.warning("'%s' is not secure to load data from!", local_path)
             else:
                 _LOGGER.warning("Neither URL nor local path found in params!")

homeassistant/components/openalpr_local/image_processing.py

@@ -183,7 +183,6 @@ class OpenAlprLocalEntity(ImageProcessingAlprEntity):
 
         alpr = await asyncio.create_subprocess_exec(
             *self._cmd,
-            loop=self.hass.loop,
             stdin=asyncio.subprocess.PIPE,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.DEVNULL,

homeassistant/components/picnic/config_flow.py

@@ -1,6 +1,7 @@
 """Config flow for Picnic integration."""
+from __future__ import annotations
+
 import logging
-from typing import Tuple
 
 from python_picnic_api import PicnicAPI
 from python_picnic_api.session import PicnicAuthError
@@ -10,11 +11,7 @@ import voluptuous as vol
 from homeassistant import config_entries, core, exceptions
 from homeassistant.const import CONF_ACCESS_TOKEN, CONF_PASSWORD, CONF_USERNAME
 
-from .const import (  # pylint: disable=unused-import
+from .const import CONF_COUNTRY_CODE, COUNTRY_CODES, DOMAIN
-    CONF_COUNTRY_CODE,
-    COUNTRY_CODES,
-    DOMAIN,
-)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -33,7 +30,7 @@ class PicnicHub:
     """Hub class to test user authentication."""
 
     @staticmethod
-    def authenticate(username, password, country_code) -> Tuple[str, dict]:
+    def authenticate(username, password, country_code) -> tuple[str, dict]:
         """Test if we can authenticate with the Picnic API."""
         picnic = PicnicAPI(username, password, country_code)
         return picnic.session.auth_token, picnic.get_user()

homeassistant/components/picnic/sensor.py

@@ -1,6 +1,7 @@
 """Definition of Picnic sensors."""
+from __future__ import annotations
 
-from typing import Any, Optional
+from typing import Any
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import ATTR_ATTRIBUTION
@@ -48,17 +49,17 @@ class PicnicSensor(CoordinatorEntity):
         self._service_unique_id = config_entry.unique_id
 
     @property
-    def unit_of_measurement(self) -> Optional[str]:
+    def unit_of_measurement(self) -> str | None:
         """Return the unit this state is expressed in."""
         return self.properties.get("unit")
 
     @property
-    def unique_id(self) -> Optional[str]:
+    def unique_id(self) -> str | None:
         """Return a unique ID."""
         return f"{self._service_unique_id}.{self.sensor_type}"
 
     @property
-    def name(self) -> Optional[str]:
+    def name(self) -> str | None:
         """Return the name of the entity."""
         return self._to_capitalized_name(self.sensor_type)
 
@@ -69,12 +70,12 @@ class PicnicSensor(CoordinatorEntity):
         return self.properties["state"](data_set)
 
     @property
-    def device_class(self) -> Optional[str]:
+    def device_class(self) -> str | None:
         """Return the class of this device, from component DEVICE_CLASSES."""
         return self.properties.get("class")
 
     @property
-    def icon(self) -> Optional[str]:
+    def icon(self) -> str | None:
         """Return the icon to use in the frontend, if any."""
         return self.properties["icon"]
 

homeassistant/components/ping/device_tracker.py

@@ -54,18 +54,18 @@ class HostSubProcess:
 
     def ping(self):
         """Send an ICMP echo request and return True if success."""
-        pinger = subprocess.Popen(
+        with subprocess.Popen(
             self._ping_cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL
-        )
+        ) as pinger:
-        try:
+            try:
-            pinger.communicate(timeout=1 + PING_TIMEOUT)
+                pinger.communicate(timeout=1 + PING_TIMEOUT)
-            return pinger.returncode == 0
+                return pinger.returncode == 0
-        except subprocess.TimeoutExpired:
+            except subprocess.TimeoutExpired:
-            kill_subprocess(pinger)
+                kill_subprocess(pinger)
-            return False
+                return False
 
-        except subprocess.CalledProcessError:
+            except subprocess.CalledProcessError:
-            return False
+                return False
 
     def update(self) -> bool:
         """Update device state by sending one or more ping messages."""

homeassistant/components/pushover/notify.py

@@ -75,6 +75,7 @@ class PushoverNotificationService(BaseNotificationService):
             if self._hass.config.is_allowed_path(data[ATTR_ATTACHMENT]):
                 # try to open it as a normal file.
                 try:
+                    # pylint: disable=consider-using-with
                     file_handle = open(data[ATTR_ATTACHMENT], "rb")
                     # Replace the attachment identifier with file object.
                     image = file_handle

homeassistant/components/rpi_camera/camera.py

@@ -56,9 +56,8 @@ def setup_platform(hass, config, add_entities, discovery_info=None):
 
     # If no file path is defined, use a temporary file
     if file_path is None:
-        temp_file = NamedTemporaryFile(suffix=".jpg", delete=False)
+        with NamedTemporaryFile(suffix=".jpg", delete=False) as temp_file:
-        temp_file.close()
+            file_path = temp_file.name
-        file_path = temp_file.name
         setup_config[CONF_FILE_PATH] = file_path
         hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, delete_temp_file)
 

homeassistant/components/seven_segments/image_processing.py

@@ -124,14 +124,14 @@ class ImageProcessingSsocr(ImageProcessingEntity):
         img = Image.open(stream)
         img.save(self.filepath, "png")
 
-        ocr = subprocess.Popen(
+        with subprocess.Popen(
             self._command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
-        )
+        ) as ocr:
-        out = ocr.communicate()
+            out = ocr.communicate()
-        if out[0] != b"":
+            if out[0] != b"":
-            self._state = out[0].strip().decode("utf-8")
+                self._state = out[0].strip().decode("utf-8")
-        else:
+            else:
-            self._state = None
+                self._state = None
-            _LOGGER.warning(
+                _LOGGER.warning(
-                "Unable to detect value: %s", out[1].strip().decode("utf-8")
+                    "Unable to detect value: %s", out[1].strip().decode("utf-8")
-            )
+                )

homeassistant/components/telegram_bot/__init__.py

@@ -282,7 +282,7 @@ def load_data(
             _LOGGER.warning("Can't load data in %s after %s retries", url, retry_num)
         elif filepath is not None:
             if hass.config.is_allowed_path(filepath):
-                return open(filepath, "rb")
+                return open(filepath, "rb")  # pylint: disable=consider-using-with
 
             _LOGGER.warning("'%s' are not secure to load data from!", filepath)
         else:

homeassistant/components/tradfri/light.py

@@ -185,8 +185,7 @@ class TradfriLight(TradfriBaseDevice, LightEntity):
         dimmer_command = None
         if ATTR_BRIGHTNESS in kwargs:
             brightness = kwargs[ATTR_BRIGHTNESS]
-            if brightness > 254:
+            brightness = min(brightness, 254)
-                brightness = 254
             dimmer_data = {
                 ATTR_DIMMER: brightness,
                 ATTR_TRANSITION_TIME: transition_time,

homeassistant/components/zeroconf/__init__.py

@@ -1,6 +1,7 @@
 """Support for exposing Home Assistant via Zeroconf."""
 from __future__ import annotations
 
+from collections.abc import Iterable
 from contextlib import suppress
 import fnmatch
 from functools import partial
@@ -8,7 +9,7 @@ import ipaddress
 from ipaddress import ip_address
 import logging
 import socket
-from typing import Any, Iterable, TypedDict, cast
+from typing import Any, TypedDict, cast
 
 from pyroute2 import IPRoute
 import voluptuous as vol

homeassistant/components/zone/__init__.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Dict, cast
+from typing import Any, cast
 
 import voluptuous as vol
 
@@ -163,7 +163,7 @@ class ZoneStorageCollection(collection.StorageCollection):
 
     async def _process_create_data(self, data: dict) -> dict:
         """Validate the config is valid."""
-        return cast(Dict, self.CREATE_SCHEMA(data))
+        return cast(dict, self.CREATE_SCHEMA(data))
 
     @callback
     def _get_suggested_id(self, info: dict) -> str:
@@ -291,7 +291,7 @@ class Zone(entity.Entity):
         """Return entity instance initialized from yaml storage."""
         zone = cls(config)
         zone.editable = False
-        zone._generate_attrs()  # pylint:disable=protected-access
+        zone._generate_attrs()
         return zone
 
     @property

homeassistant/core.py

@@ -7,7 +7,7 @@ of entities and react to changes.
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Awaitable, Collection, Iterable, Mapping
+from collections.abc import Awaitable, Collection, Coroutine, Iterable, Mapping
 import datetime
 import enum
 import functools
@@ -18,7 +18,7 @@ import re
 import threading
 from time import monotonic
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, Callable, Coroutine, Optional, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Callable, Optional, TypeVar, cast
 
 import attr
 import voluptuous as vol

homeassistant/helpers/config_validation.py

@@ -483,7 +483,7 @@ def schema_with_slug_keys(
         for key in value.keys():
             slug_validator(key)
 
-        return cast(Dict, schema(value))
+        return cast(dict, schema(value))
 
     return verify
 

homeassistant/helpers/selector.py

@@ -1,7 +1,7 @@
 """Selectors for Home Assistant."""
 from __future__ import annotations
 
-from typing import Any, Callable, Dict, cast
+from typing import Any, Callable, cast
 
 import voluptuous as vol
 
@@ -31,7 +31,7 @@ def validate_selector(config: Any) -> dict:
         return {selector_type: {}}
 
     return {
-        selector_type: cast(Dict, selector_class.CONFIG_SCHEMA(config[selector_type]))
+        selector_type: cast(dict, selector_class.CONFIG_SCHEMA(config[selector_type]))
     }
 
 

homeassistant/helpers/template.py

@@ -216,7 +216,6 @@ class RenderInfo:
         self.exception: TemplateError | None = None
         self.all_states = False
         self.all_states_lifecycle = False
-        # pylint: disable=unsubscriptable-object  # for abc.Set, https://github.com/PyCQA/pylint/pull/4275
         self.domains: collections.abc.Set[str] = set()
         self.domains_lifecycle: collections.abc.Set[str] = set()
         self.entities: collections.abc.Set[str] = set()

homeassistant/requirements.py

@@ -81,7 +81,7 @@ async def async_get_integration_with_requirements(
 
     try:
         await _async_process_integration(hass, integration, done)
-    except Exception:  # pylint: disable=broad-except
+    except Exception:
         del cache[domain]
         event.set()
         raise

homeassistant/util/package.py

@@ -90,15 +90,15 @@ def install_package(
             # Workaround for incompatible prefix setting
             # See http://stackoverflow.com/a/4495175
             args += ["--prefix="]
-    process = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, env=env)
+    with Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, env=env) as process:
-    _, stderr = process.communicate()
+        _, stderr = process.communicate()
-    if process.returncode != 0:
+        if process.returncode != 0:
-        _LOGGER.error(
+            _LOGGER.error(
-            "Unable to install package %s: %s",
+                "Unable to install package %s: %s",
-            package,
+                package,
-            stderr.decode("utf-8").lstrip().strip(),
+                stderr.decode("utf-8").lstrip().strip(),
-        )
+            )
-        return False
+            return False
 
     return True
 

homeassistant/util/ruamel_yaml.py

@@ -6,7 +6,7 @@ from contextlib import suppress
 import logging
 import os
 from os import O_CREAT, O_TRUNC, O_WRONLY, stat_result
-from typing import Dict, List, Union
+from typing import Union
 
 import ruamel.yaml
 from ruamel.yaml import YAML  # type: ignore
@@ -19,7 +19,7 @@ from homeassistant.util.yaml import secret_yaml
 
 _LOGGER = logging.getLogger(__name__)
 
-JSON_TYPE = Union[List, Dict, str]  # pylint: disable=invalid-name
+JSON_TYPE = Union[list, dict, str]  # pylint: disable=invalid-name
 
 
 class ExtSafeConstructor(SafeConstructor):

homeassistant/util/yaml/loader.py

@@ -7,7 +7,7 @@ import fnmatch
 import logging
 import os
 from pathlib import Path
-from typing import Any, Dict, List, TextIO, TypeVar, Union, overload
+from typing import Any, TextIO, TypeVar, Union, overload
 
 import yaml
 
@@ -18,8 +18,8 @@ from .objects import Input, NodeListClass, NodeStrClass
 
 # mypy: allow-untyped-calls, no-warn-return-any
 
-JSON_TYPE = Union[List, Dict, str]  # pylint: disable=invalid-name
+JSON_TYPE = Union[list, dict, str]  # pylint: disable=invalid-name
-DICT_T = TypeVar("DICT_T", bound=Dict)  # pylint: disable=invalid-name
+DICT_T = TypeVar("DICT_T", bound=dict)  # pylint: disable=invalid-name
 
 _LOGGER = logging.getLogger(__name__)
 

pyproject.toml

@@ -25,6 +25,7 @@ ignore = [
 jobs = 2
 init-hook='from pylint.config.find_default_config_files import find_default_config_files; from pathlib import Path; import sys; sys.path.append(str(Path(Path(list(find_default_config_files())[0]).parent, "pylint/plugins")))'
 load-plugins = [
+    "pylint.extensions.typing",
     "pylint_strict_informational",
     "hass_logger"
 ]
@@ -109,6 +110,10 @@ overgeneral-exceptions = [
     "HomeAssistantError",
 ]
 
+[tool.pylint.TYPING]
+py-version = "3.8"
+runtime-typing = false
+
 [tool.pytest.ini_options]
 testpaths = [
     "tests",

requirements_test.txt

@@ -10,8 +10,8 @@ jsonpickle==1.4.1
 mock-open==1.4.0
 mypy==0.812
 pre-commit==2.12.1
-pylint==2.7.4
+pylint==2.8.0
-astroid==2.5.2
+astroid==2.5.5
 pipdeptree==1.0.0
 pylint-strict-informational==0.1
 pytest-aiohttp==0.3.0

tests/components/command_line/test_notify.py

@@ -94,21 +94,24 @@ async def test_subprocess_exceptions(caplog: Any, hass: HomeAssistant) -> None:
     """Test that notify subprocess exceptions are handled correctly."""
 
     with patch(
-        "homeassistant.components.command_line.notify.subprocess.Popen",
+        "homeassistant.components.command_line.notify.subprocess.Popen"
-        side_effect=[
-            subprocess.TimeoutExpired("cmd", 10),
-            subprocess.SubprocessError(),
-        ],
     ) as check_output:
+        check_output.return_value.__enter__ = check_output
+        check_output.return_value.communicate.side_effect = [
+            subprocess.TimeoutExpired("cmd", 10),
+            None,
+            subprocess.SubprocessError(),
+        ]
+
         await setup_test_service(hass, {"command": "exit 0"})
         assert await hass.services.async_call(
             DOMAIN, "test", {"message": "error"}, blocking=True
         )
-        assert check_output.call_count == 1
+        assert check_output.call_count == 2
         assert "Timeout for command" in caplog.text
 
         assert await hass.services.async_call(
             DOMAIN, "test", {"message": "error"}, blocking=True
         )
-        assert check_output.call_count == 2
+        assert check_output.call_count == 4
         assert "Error trying to exec command" in caplog.text

tests/util/test_package.py

@@ -46,6 +46,7 @@ def lib_dir(deps_dir):
 def mock_popen(lib_dir):
     """Return a Popen mock."""
     with patch("homeassistant.util.package.Popen") as popen_mock:
+        popen_mock.return_value.__enter__ = popen_mock
         popen_mock.return_value.communicate.return_value = (
             bytes(lib_dir, "utf-8"),
             b"error",
@@ -87,8 +88,8 @@ def test_install(mock_sys, mock_popen, mock_env_copy, mock_venv):
     """Test an install attempt on a package that doesn't exist."""
     env = mock_env_copy()
     assert package.install_package(TEST_NEW_REQ, False)
-    assert mock_popen.call_count == 1
+    assert mock_popen.call_count == 2
-    assert mock_popen.call_args == call(
+    assert mock_popen.mock_calls[0] == call(
         [mock_sys.executable, "-m", "pip", "install", "--quiet", TEST_NEW_REQ],
         stdin=PIPE,
         stdout=PIPE,
@@ -102,8 +103,8 @@ def test_install_upgrade(mock_sys, mock_popen, mock_env_copy, mock_venv):
     """Test an upgrade attempt on a package."""
     env = mock_env_copy()
     assert package.install_package(TEST_NEW_REQ)
-    assert mock_popen.call_count == 1
+    assert mock_popen.call_count == 2
-    assert mock_popen.call_args == call(
+    assert mock_popen.mock_calls[0] == call(
         [
             mock_sys.executable,
             "-m",
@@ -140,8 +141,8 @@ def test_install_target(mock_sys, mock_popen, mock_env_copy, mock_venv):
     ]
 
     assert package.install_package(TEST_NEW_REQ, False, target=target)
-    assert mock_popen.call_count == 1
+    assert mock_popen.call_count == 2
-    assert mock_popen.call_args == call(
+    assert mock_popen.mock_calls[0] == call(
         args, stdin=PIPE, stdout=PIPE, stderr=PIPE, env=env
     )
     assert mock_popen.return_value.communicate.call_count == 1
@@ -169,8 +170,8 @@ def test_install_constraint(mock_sys, mock_popen, mock_env_copy, mock_venv):
     env = mock_env_copy()
     constraints = "constraints_file.txt"
     assert package.install_package(TEST_NEW_REQ, False, constraints=constraints)
-    assert mock_popen.call_count == 1
+    assert mock_popen.call_count == 2
-    assert mock_popen.call_args == call(
+    assert mock_popen.mock_calls[0] == call(
         [
             mock_sys.executable,
             "-m",
@@ -194,8 +195,8 @@ def test_install_find_links(mock_sys, mock_popen, mock_env_copy, mock_venv):
     env = mock_env_copy()
     link = "https://wheels-repository"
     assert package.install_package(TEST_NEW_REQ, False, find_links=link)
-    assert mock_popen.call_count == 1
+    assert mock_popen.call_count == 2
-    assert mock_popen.call_args == call(
+    assert mock_popen.mock_calls[0] == call(
         [
             mock_sys.executable,
             "-m",