Ensure async_get_system_info does not fail if supervisor is unavailable (#96492)

* Ensure async_get_system_info does not fail if supervisor is unavailable

fixes #96470

* fix i/o in the event loop

* fix tests

* handle some more failure cases

* more I/O here

* coverage

* coverage

* Update homeassistant/helpers/system_info.py

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>

* remove supervisor detection fallback

* Update tests/helpers/test_system_info.py

---------

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
J. Nick Koston 2023-07-16 05:10:07 -10:00 committed by GitHub
parent cd0e9839a0
commit 7ec506907c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 100 additions and 14 deletions

View File

@ -1,7 +1,9 @@
"""Helper to gather system info.""" """Helper to gather system info."""
from __future__ import annotations from __future__ import annotations
from functools import cache
from getpass import getuser from getpass import getuser
import logging
import os import os
import platform import platform
from typing import Any from typing import Any
@ -9,17 +11,32 @@ from typing import Any
from homeassistant.const import __version__ as current_version from homeassistant.const import __version__ as current_version
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.loader import bind_hass from homeassistant.loader import bind_hass
from homeassistant.util.package import is_virtual_env from homeassistant.util.package import is_docker_env, is_virtual_env
_LOGGER = logging.getLogger(__name__)
@cache
def is_official_image() -> bool:
"""Return True if Home Assistant is running in an official container."""
return os.path.isfile("/OFFICIAL_IMAGE")
# Cache the result of getuser() because it can call getpwuid() which
# can do blocking I/O to look up the username in /etc/passwd.
cached_get_user = cache(getuser)
@bind_hass @bind_hass
async def async_get_system_info(hass: HomeAssistant) -> dict[str, Any]: async def async_get_system_info(hass: HomeAssistant) -> dict[str, Any]:
"""Return info about the system.""" """Return info about the system."""
is_hassio = hass.components.hassio.is_hassio()
info_object = { info_object = {
"installation_type": "Unknown", "installation_type": "Unknown",
"version": current_version, "version": current_version,
"dev": "dev" in current_version, "dev": "dev" in current_version,
"hassio": hass.components.hassio.is_hassio(), "hassio": is_hassio,
"virtualenv": is_virtual_env(), "virtualenv": is_virtual_env(),
"python_version": platform.python_version(), "python_version": platform.python_version(),
"docker": False, "docker": False,
@ -30,18 +47,18 @@ async def async_get_system_info(hass: HomeAssistant) -> dict[str, Any]:
} }
try: try:
info_object["user"] = getuser() info_object["user"] = cached_get_user()
except KeyError: except KeyError:
info_object["user"] = None info_object["user"] = None
if platform.system() == "Darwin": if platform.system() == "Darwin":
info_object["os_version"] = platform.mac_ver()[0] info_object["os_version"] = platform.mac_ver()[0]
elif platform.system() == "Linux": elif platform.system() == "Linux":
info_object["docker"] = os.path.isfile("/.dockerenv") info_object["docker"] = is_docker_env()
# Determine installation type on current data # Determine installation type on current data
if info_object["docker"]: if info_object["docker"]:
if info_object["user"] == "root" and os.path.isfile("/OFFICIAL_IMAGE"): if info_object["user"] == "root" and is_official_image():
info_object["installation_type"] = "Home Assistant Container" info_object["installation_type"] = "Home Assistant Container"
else: else:
info_object["installation_type"] = "Unsupported Third Party Container" info_object["installation_type"] = "Unsupported Third Party Container"
@ -50,10 +67,12 @@ async def async_get_system_info(hass: HomeAssistant) -> dict[str, Any]:
info_object["installation_type"] = "Home Assistant Core" info_object["installation_type"] = "Home Assistant Core"
# Enrich with Supervisor information # Enrich with Supervisor information
if hass.components.hassio.is_hassio(): if is_hassio:
info = hass.components.hassio.get_info() if not (info := hass.components.hassio.get_info()):
host = hass.components.hassio.get_host_info() _LOGGER.warning("No Home Assistant Supervisor info available")
info = {}
host = hass.components.hassio.get_host_info() or {}
info_object["supervisor"] = info.get("supervisor") info_object["supervisor"] = info.get("supervisor")
info_object["host_os"] = host.get("operating_system") info_object["host_os"] = host.get("operating_system")
info_object["docker_version"] = info.get("docker") info_object["docker_version"] = info.get("docker")

View File

@ -1,10 +1,23 @@
"""Tests for the system info helper.""" """Tests for the system info helper."""
import json import json
import os
from unittest.mock import patch from unittest.mock import patch
import pytest
from homeassistant.const import __version__ as current_version from homeassistant.const import __version__ as current_version
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.system_info import async_get_system_info from homeassistant.helpers.system_info import async_get_system_info, is_official_image
async def test_is_official_image() -> None:
"""Test is_official_image."""
is_official_image.cache_clear()
with patch("homeassistant.helpers.system_info.os.path.isfile", return_value=True):
assert is_official_image() is True
is_official_image.cache_clear()
with patch("homeassistant.helpers.system_info.os.path.isfile", return_value=False):
assert is_official_image() is False
async def test_get_system_info(hass: HomeAssistant) -> None: async def test_get_system_info(hass: HomeAssistant) -> None:
@ -16,23 +29,77 @@ async def test_get_system_info(hass: HomeAssistant) -> None:
assert json.dumps(info) is not None assert json.dumps(info) is not None
async def test_get_system_info_supervisor_not_available(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
"""Test the get system info when supervisor is not available."""
hass.config.components.add("hassio")
with patch("platform.system", return_value="Linux"), patch(
"homeassistant.helpers.system_info.is_docker_env", return_value=True
), patch(
"homeassistant.helpers.system_info.is_official_image", return_value=True
), patch(
"homeassistant.components.hassio.is_hassio", return_value=True
), patch(
"homeassistant.components.hassio.get_info", return_value=None
), patch(
"homeassistant.helpers.system_info.cached_get_user", return_value="root"
):
info = await async_get_system_info(hass)
assert isinstance(info, dict)
assert info["version"] == current_version
assert info["user"] is not None
assert json.dumps(info) is not None
assert info["installation_type"] == "Home Assistant Supervised"
assert "No Home Assistant Supervisor info available" in caplog.text
async def test_get_system_info_supervisor_not_loaded(hass: HomeAssistant) -> None:
"""Test the get system info when supervisor is not loaded."""
with patch("platform.system", return_value="Linux"), patch(
"homeassistant.helpers.system_info.is_docker_env", return_value=True
), patch(
"homeassistant.helpers.system_info.is_official_image", return_value=True
), patch(
"homeassistant.components.hassio.get_info", return_value=None
), patch.dict(
os.environ, {"SUPERVISOR": "127.0.0.1"}
):
info = await async_get_system_info(hass)
assert isinstance(info, dict)
assert info["version"] == current_version
assert info["user"] is not None
assert json.dumps(info) is not None
assert info["installation_type"] == "Unsupported Third Party Container"
async def test_container_installationtype(hass: HomeAssistant) -> None: async def test_container_installationtype(hass: HomeAssistant) -> None:
"""Test container installation type.""" """Test container installation type."""
with patch("platform.system", return_value="Linux"), patch( with patch("platform.system", return_value="Linux"), patch(
"os.path.isfile", return_value=True "homeassistant.helpers.system_info.is_docker_env", return_value=True
), patch("homeassistant.helpers.system_info.getuser", return_value="root"): ), patch(
"homeassistant.helpers.system_info.is_official_image", return_value=True
), patch(
"homeassistant.helpers.system_info.cached_get_user", return_value="root"
):
info = await async_get_system_info(hass) info = await async_get_system_info(hass)
assert info["installation_type"] == "Home Assistant Container" assert info["installation_type"] == "Home Assistant Container"
with patch("platform.system", return_value="Linux"), patch( with patch("platform.system", return_value="Linux"), patch(
"os.path.isfile", side_effect=lambda file: file == "/.dockerenv" "homeassistant.helpers.system_info.is_docker_env", return_value=True
), patch("homeassistant.helpers.system_info.getuser", return_value="user"): ), patch(
"homeassistant.helpers.system_info.is_official_image", return_value=False
), patch(
"homeassistant.helpers.system_info.cached_get_user", return_value="user"
):
info = await async_get_system_info(hass) info = await async_get_system_info(hass)
assert info["installation_type"] == "Unsupported Third Party Container" assert info["installation_type"] == "Unsupported Third Party Container"
async def test_getuser_keyerror(hass: HomeAssistant) -> None: async def test_getuser_keyerror(hass: HomeAssistant) -> None:
"""Test getuser keyerror.""" """Test getuser keyerror."""
with patch("homeassistant.helpers.system_info.getuser", side_effect=KeyError): with patch(
"homeassistant.helpers.system_info.cached_get_user", side_effect=KeyError
):
info = await async_get_system_info(hass) info = await async_get_system_info(hass)
assert info["user"] is None assert info["user"] is None