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

Isolate systemmonitor from psutil shared state (#111110)

Browse source 
* Isolate systemmonitor from psutil shared state

* Mods

* typing

* Fix tests

* Fix read temp issue

---------

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
G Johansson 2024-02-23 20:53:48 +01:00 • committed by GitHub
parent 2ff0102bce
commit 9e46c2e2b3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 148 additions and 181 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
homeassistant/components/systemmonitor/__init__.py
View file

@ -2,12 +2,16 @@
import logging
import psutil_home_assistant as ha_psutil
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
@ -15,7 +19,8 @@ PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up System Monitor from a config entry."""
psutil_wrapper = await hass.async_add_executor_job(ha_psutil.PsutilWrapper)
hass.data[DOMAIN] = psutil_wrapper
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
return True

19
homeassistant/components/systemmonitor/binary_sensor.py
View file

@ -9,7 +9,8 @@ import logging
import sys
from typing import Generic, Literal
import psutil
from psutil import NoSuchProcess, Process
import psutil_home_assistant as ha_psutil
from homeassistant.components.binary_sensor import (
DOMAIN as BINARY_SENSOR_DOMAIN,
@ -50,7 +51,7 @@ def get_cpu_icon() -> Literal["mdi:cpu-64-bit", "mdi:cpu-32-bit"]:
return "mdi:cpu-32-bit"
def get_process(entity: SystemMonitorSensor[list[psutil.Process]]) -> bool:
def get_process(entity: SystemMonitorSensor[list[Process]]) -> bool:
"""Return process."""
state = False
for proc in entity.coordinator.data:
@ -59,7 +60,7 @@ def get_process(entity: SystemMonitorSensor[list[psutil.Process]]) -> bool:
if entity.argument == proc.name():
state = True
break
except psutil.NoSuchProcess as err:
except NoSuchProcess as err:
_LOGGER.warning(
"Failed to load process with ID: %s, old name: %s",
err.pid,
@ -77,10 +78,8 @@ class SysMonitorBinarySensorEntityDescription(
value_fn: Callable[[SystemMonitorSensor[dataT]], bool]
SENSOR_TYPES: tuple[
SysMonitorBinarySensorEntityDescription[list[psutil.Process]], ...
] = (
SysMonitorBinarySensorEntityDescription[list[psutil.Process]](
SENSOR_TYPES: tuple[SysMonitorBinarySensorEntityDescription[list[Process]], ...] = (
SysMonitorBinarySensorEntityDescription[list[Process]](
key="binary_process",
translation_key="process",
icon=get_cpu_icon(),
@ -94,8 +93,12 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up System Montor binary sensors based on a config entry."""
psutil_wrapper: ha_psutil.PsutilWrapper = hass.data[DOMAIN]
entities: list[SystemMonitorSensor] = []
process_coordinator = SystemMonitorProcessCoordinator(hass, "Process coordinator")
process_coordinator = SystemMonitorProcessCoordinator(
hass, psutil_wrapper, "Process coordinator"
)
await process_coordinator.async_request_refresh()
for sensor_description in SENSOR_TYPES:

2
homeassistant/components/systemmonitor/config_flow.py
View file

@ -86,7 +86,7 @@ async def validate_import_sensor_setup(
async def get_sensor_setup_schema(handler: SchemaCommonFlowHandler) -> vol.Schema:
"""Return process sensor setup schema."""
hass = handler.parent_handler.hass
processes = list(await hass.async_add_executor_job(get_all_running_processes))
processes = list(await hass.async_add_executor_job(get_all_running_processes, hass))
return vol.Schema(
{
vol.Required(CONF_PROCESS): SelectSelector(

67
homeassistant/components/systemmonitor/coordinator.py
View file

@ -8,8 +8,9 @@ import logging
import os
from typing import NamedTuple, TypeVar
import psutil
from psutil import Process
from psutil._common import sdiskusage, shwtemp, snetio, snicaddr, sswap
import psutil_home_assistant as ha_psutil
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_component import DEFAULT_SCAN_INTERVAL
@ -40,7 +41,7 @@ dataT = TypeVar(
| dict[str, list[snicaddr]]
| dict[str, snetio]
| float
| list[psutil.Process]
| list[Process]
| sswap
| VirtualMemory
| tuple[float, float, float]
@ -52,7 +53,9 @@ dataT = TypeVar(
class MonitorCoordinator(DataUpdateCoordinator[dataT]):
"""A System monitor Base Data Update Coordinator."""
def __init__(self, hass: HomeAssistant, name: str) -> None:
def __init__(
self, hass: HomeAssistant, psutil_wrapper: ha_psutil.PsutilWrapper, name: str
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
@ -61,6 +64,7 @@ class MonitorCoordinator(DataUpdateCoordinator[dataT]):
update_interval=DEFAULT_SCAN_INTERVAL,
always_update=False,
)
self._psutil = psutil_wrapper.psutil
async def _async_update_data(self) -> dataT:
"""Fetch data."""
@ -74,15 +78,22 @@ class MonitorCoordinator(DataUpdateCoordinator[dataT]):
class SystemMonitorDiskCoordinator(MonitorCoordinator[sdiskusage]):
"""A System monitor Disk Data Update Coordinator."""
def __init__(self, hass: HomeAssistant, name: str, argument: str) -> None:
def __init__(
self,
hass: HomeAssistant,
psutil_wrapper: ha_psutil.PsutilWrapper,
name: str,
argument: str,
) -> None:
"""Initialize the disk coordinator."""
super().__init__(hass, name)
super().__init__(hass, psutil_wrapper, name)
self._argument = argument
def update_data(self) -> sdiskusage:
"""Fetch data."""
try:
return psutil.disk_usage(self._argument)
usage: sdiskusage = self._psutil.disk_usage(self._argument)
return usage
except PermissionError as err:
raise UpdateFailed(f"No permission to access {self._argument}") from err
except OSError as err:
@ -94,7 +105,8 @@ class SystemMonitorSwapCoordinator(MonitorCoordinator[sswap]):
def update_data(self) -> sswap:
"""Fetch data."""
return psutil.swap_memory()
swap: sswap = self._psutil.swap_memory()
return swap
class SystemMonitorMemoryCoordinator(MonitorCoordinator[VirtualMemory]):
@ -102,7 +114,7 @@ class SystemMonitorMemoryCoordinator(MonitorCoordinator[VirtualMemory]):
def update_data(self) -> VirtualMemory:
"""Fetch data."""
memory = psutil.virtual_memory()
memory = self._psutil.virtual_memory()
return VirtualMemory(
memory.total, memory.available, memory.percent, memory.used, memory.free
)
@ -113,7 +125,8 @@ class SystemMonitorNetIOCoordinator(MonitorCoordinator[dict[str, snetio]]):
def update_data(self) -> dict[str, snetio]:
"""Fetch data."""
return psutil.net_io_counters(pernic=True)
io_counters: dict[str, snetio] = self._psutil.net_io_counters(pernic=True)
return io_counters
class SystemMonitorNetAddrCoordinator(MonitorCoordinator[dict[str, list[snicaddr]]]):
@ -121,13 +134,19 @@ class SystemMonitorNetAddrCoordinator(MonitorCoordinator[dict[str, list[snicaddr
def update_data(self) -> dict[str, list[snicaddr]]:
"""Fetch data."""
return psutil.net_if_addrs()
addresses: dict[str, list[snicaddr]] = self._psutil.net_if_addrs()
return addresses
class SystemMonitorLoadCoordinator(MonitorCoordinator[tuple[float, float, float]]):
class SystemMonitorLoadCoordinator(
MonitorCoordinator[tuple[float, float, float] | None]
):
"""A System monitor Load Data Update Coordinator."""
def update_data(self) -> tuple[float, float, float]:
def update_data(self) -> tuple[float, float, float] | None:
"""Coordinator is not async."""
async def _async_update_data(self) -> tuple[float, float, float] | None:
"""Fetch data."""
return os.getloadavg()
@ -136,8 +155,17 @@ class SystemMonitorProcessorCoordinator(MonitorCoordinator[float | None]):
"""A System monitor Processor Data Update Coordinator."""
def update_data(self) -> float | None:
"""Fetch data."""
cpu_percent = psutil.cpu_percent(interval=None)
"""Coordinator is not async."""
async def _async_update_data(self) -> float | None:
"""Get cpu usage.
Unlikely the rest of the coordinators, this one is async
since it does not block and we need to make sure it runs
in the same thread every time as psutil checks the thread
tid and compares it against the previous one.
"""
cpu_percent: float = self._psutil.cpu_percent(interval=None)
if cpu_percent > 0.0:
return cpu_percent
return None
@ -148,15 +176,15 @@ class SystemMonitorBootTimeCoordinator(MonitorCoordinator[datetime]):
def update_data(self) -> datetime:
"""Fetch data."""
return dt_util.utc_from_timestamp(psutil.boot_time())
return dt_util.utc_from_timestamp(self._psutil.boot_time())
class SystemMonitorProcessCoordinator(MonitorCoordinator[list[psutil.Process]]):
class SystemMonitorProcessCoordinator(MonitorCoordinator[list[Process]]):
"""A System monitor Process Data Update Coordinator."""
def update_data(self) -> list[psutil.Process]:
def update_data(self) -> list[Process]:
"""Fetch data."""
processes = psutil.process_iter()
processes = self._psutil.process_iter()
return list(processes)
@ -166,6 +194,7 @@ class SystemMonitorCPUtempCoordinator(MonitorCoordinator[dict[str, list[shwtemp]
def update_data(self) -> dict[str, list[shwtemp]]:
"""Fetch data."""
try:
return psutil.sensors_temperatures()
temps: dict[str, list[shwtemp]] = self._psutil.sensors_temperatures()
return temps
except AttributeError as err:
raise UpdateFailed("OS does not provide temperature sensors") from err

2
homeassistant/components/systemmonitor/manifest.json
View file

@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/systemmonitor",
"iot_class": "local_push",
"loggers": ["psutil"],
"requirements": ["psutil==5.9.8"]
"requirements": ["psutil-home-assistant==0.0.1", "psutil==5.9.8"]
}

53
homeassistant/components/systemmonitor/sensor.py
View file

@ -12,8 +12,9 @@ import sys
import time
from typing import Any, Generic, Literal
import psutil
from psutil import NoSuchProcess, Process
from psutil._common import sdiskusage, shwtemp, snetio, snicaddr, sswap
import psutil_home_assistant as ha_psutil
import voluptuous as vol
from homeassistant.components.sensor import (
@ -89,10 +90,10 @@ def get_processor_temperature(
entity: SystemMonitorSensor[dict[str, list[shwtemp]]],
) -> float | None:
"""Return processor temperature."""
return read_cpu_temperature(entity.coordinator.data)
return read_cpu_temperature(entity.hass, entity.coordinator.data)
def get_process(entity: SystemMonitorSensor[list[psutil.Process]]) -> str:
def get_process(entity: SystemMonitorSensor[list[Process]]) -> str:
"""Return process."""
state = STATE_OFF
for proc in entity.coordinator.data:
@ -101,7 +102,7 @@ def get_process(entity: SystemMonitorSensor[list[psutil.Process]]) -> str:
if entity.argument == proc.name():
state = STATE_ON
break
except psutil.NoSuchProcess as err:
except NoSuchProcess as err:
_LOGGER.warning(
"Failed to load process with ID: %s, old name: %s",
err.pid,
@ -332,7 +333,7 @@ SENSOR_TYPES: dict[str, SysMonitorSensorEntityDescription[Any]] = {
mandatory_arg=True,
value_fn=get_throughput,
),
"process": SysMonitorSensorEntityDescription[list[psutil.Process]](
"process": SysMonitorSensorEntityDescription[list[Process]](
key="process",
translation_key="process",
placeholder="process",
@ -487,12 +488,16 @@ async def async_setup_entry( # noqa: C901
entities: list[SystemMonitorSensor] = []
legacy_resources: set[str] = set(entry.options.get("resources", []))
loaded_resources: set[str] = set()
psutil_wrapper: ha_psutil.PsutilWrapper = hass.data[DOMAIN]
def get_arguments() -> dict[str, Any]:
"""Return startup information."""
disk_arguments = get_all_disk_mounts()
network_arguments = get_all_network_interfaces()
cpu_temperature = read_cpu_temperature()
disk_arguments = get_all_disk_mounts(hass)
network_arguments = get_all_network_interfaces(hass)
try:
cpu_temperature = read_cpu_temperature(hass)
except AttributeError:
cpu_temperature = 0.0
return {
"disk_arguments": disk_arguments,
"network_arguments": network_arguments,
@ -504,31 +509,39 @@ async def async_setup_entry( # noqa: C901
disk_coordinators: dict[str, SystemMonitorDiskCoordinator] = {}
for argument in startup_arguments["disk_arguments"]:
disk_coordinators[argument] = SystemMonitorDiskCoordinator(
hass, f"Disk {argument} coordinator", argument
hass, psutil_wrapper, f"Disk {argument} coordinator", argument
)
swap_coordinator = SystemMonitorSwapCoordinator(hass, "Swap coordinator")
memory_coordinator = SystemMonitorMemoryCoordinator(hass, "Memory coordinator")
net_io_coordinator = SystemMonitorNetIOCoordinator(hass, "Net IO coordnator")
swap_coordinator = SystemMonitorSwapCoordinator(
hass, psutil_wrapper, "Swap coordinator"
)
memory_coordinator = SystemMonitorMemoryCoordinator(
hass, psutil_wrapper, "Memory coordinator"
)
net_io_coordinator = SystemMonitorNetIOCoordinator(
hass, psutil_wrapper, "Net IO coordnator"
)
net_addr_coordinator = SystemMonitorNetAddrCoordinator(
hass, "Net address coordinator"
hass, psutil_wrapper, "Net address coordinator"
)
system_load_coordinator = SystemMonitorLoadCoordinator(
hass, "System load coordinator"
hass, psutil_wrapper, "System load coordinator"
)
processor_coordinator = SystemMonitorProcessorCoordinator(
hass, "Processor coordinator"
hass, psutil_wrapper, "Processor coordinator"
)
boot_time_coordinator = SystemMonitorBootTimeCoordinator(
hass, "Boot time coordinator"
hass, psutil_wrapper, "Boot time coordinator"
)
process_coordinator = SystemMonitorProcessCoordinator(
hass, psutil_wrapper, "Process coordinator"
)
process_coordinator = SystemMonitorProcessCoordinator(hass, "Process coordinator")
cpu_temp_coordinator = SystemMonitorCPUtempCoordinator(
hass, "CPU temperature coordinator"
hass, psutil_wrapper, "CPU temperature coordinator"
)
for argument in startup_arguments["disk_arguments"]:
disk_coordinators[argument] = SystemMonitorDiskCoordinator(
hass, f"Disk {argument} coordinator", argument
hass, psutil_wrapper, f"Disk {argument} coordinator", argument
)
_LOGGER.debug("Setup from options %s", entry.options)
@ -722,7 +735,7 @@ async def async_setup_entry( # noqa: C901
_LOGGER.debug("Loading legacy %s with argument %s", _type, argument)
if not disk_coordinators.get(argument):
disk_coordinators[argument] = SystemMonitorDiskCoordinator(
hass, f"Disk {argument} coordinator", argument
hass, psutil_wrapper, f"Disk {argument} coordinator", argument
)
entities.append(
SystemMonitorSensor(

32
homeassistant/components/systemmonitor/util.py
View file

@ -3,20 +3,23 @@
import logging
import os
import psutil
from psutil._common import shwtemp
import psutil_home_assistant as ha_psutil
from .const import CPU_SENSOR_PREFIXES
from homeassistant.core import HomeAssistant
from .const import CPU_SENSOR_PREFIXES, DOMAIN
_LOGGER = logging.getLogger(__name__)
SKIP_DISK_TYPES = {"proc", "tmpfs", "devtmpfs"}
def get_all_disk_mounts() -> set[str]:
def get_all_disk_mounts(hass: HomeAssistant) -> set[str]:
"""Return all disk mount points on system."""
psutil_wrapper: ha_psutil = hass.data[DOMAIN]
disks: set[str] = set()
for part in psutil.disk_partitions(all=True):
for part in psutil_wrapper.psutil.disk_partitions(all=True):
if os.name == "nt":
if "cdrom" in part.opts or part.fstype == "":
# skip cd-rom drives with no disk in it; they may raise
@ -27,7 +30,7 @@ def get_all_disk_mounts() -> set[str]:
# Ignore disks which are memory
continue
try:
usage = psutil.disk_usage(part.mountpoint)
usage = psutil_wrapper.psutil.disk_usage(part.mountpoint)
except PermissionError:
_LOGGER.debug(
"No permission for running user to access %s", part.mountpoint
@ -44,10 +47,11 @@ def get_all_disk_mounts() -> set[str]:
return disks
def get_all_network_interfaces() -> set[str]:
def get_all_network_interfaces(hass: HomeAssistant) -> set[str]:
"""Return all network interfaces on system."""
psutil_wrapper: ha_psutil = hass.data[DOMAIN]
interfaces: set[str] = set()
for interface, _ in psutil.net_if_addrs().items():
for interface, _ in psutil_wrapper.psutil.net_if_addrs().items():
if interface.startswith("veth"):
# Don't load docker virtual network interfaces
continue
@ -56,20 +60,24 @@ def get_all_network_interfaces() -> set[str]:
return interfaces
def get_all_running_processes() -> set[str]:
def get_all_running_processes(hass: HomeAssistant) -> set[str]:
"""Return all running processes on system."""
psutil_wrapper: ha_psutil = hass.data.get(DOMAIN, ha_psutil.PsutilWrapper())
processes: set[str] = set()
for proc in psutil.process_iter(["name"]):
for proc in psutil_wrapper.psutil.process_iter(["name"]):
if proc.name() not in processes:
processes.add(proc.name())
_LOGGER.debug("Running processes: %s", ", ".join(processes))
return processes
def read_cpu_temperature(temps: dict[str, list[shwtemp]] | None = None) -> float | None:
def read_cpu_temperature(
hass: HomeAssistant, temps: dict[str, list[shwtemp]] | None = None
) -> float | None:
"""Attempt to read CPU / processor temperature."""
if not temps:
temps = psutil.sensors_temperatures()
if temps is None:
psutil_wrapper: ha_psutil = hass.data[DOMAIN]
temps = psutil_wrapper.psutil.sensors_temperatures()
entry: shwtemp
_LOGGER.debug("CPU Temperatures: %s", temps)

1
requirements_all.txt
View file

@ -1573,6 +1573,7 @@ proxmoxer==2.0.1
# homeassistant.components.hardware
# homeassistant.components.recorder
# homeassistant.components.systemmonitor
psutil-home-assistant==0.0.1
# homeassistant.components.systemmonitor

1
requirements_test_all.txt
View file

@ -1229,6 +1229,7 @@ prometheus-client==0.17.1
# homeassistant.components.hardware
# homeassistant.components.recorder
# homeassistant.components.systemmonitor
psutil-home-assistant==0.0.1
# homeassistant.components.systemmonitor

96
tests/components/systemmonitor/conftest.py
View file

@ -1,38 +1,20 @@
"""Fixtures for the System Monitor integration."""
from __future__ import annotations
from collections import namedtuple
from collections.abc import Generator
import socket
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock, Mock, NonCallableMock, patch
from psutil import NoSuchProcess, Process
from psutil._common import sdiskpart, sdiskusage, shwtemp, snetio, snicaddr, sswap
import pytest
from homeassistant.components.systemmonitor.const import DOMAIN
from homeassistant.components.systemmonitor.coordinator import VirtualMemory
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
# Different depending on platform so making according to Linux
svmem = namedtuple(
"svmem",
[
"total",
"available",
"percent",
"used",
"free",
"active",
"inactive",
"buffers",
"cached",
"shared",
"slab",
],
)
@pytest.fixture(autouse=True)
def mock_sys_platform() -> Generator[None, None, None]:
@ -92,7 +74,6 @@ async def mock_added_config_entry(
hass: HomeAssistant,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
mock_config_entry: MockConfigEntry,
) -> MockConfigEntry:
"""Mock ConfigEntry that's been added to HA."""
@ -112,30 +93,26 @@ def mock_process() -> list[MockProcess]:
@pytest.fixture
def mock_psutil(mock_process: list[MockProcess]) -> Mock:
def mock_psutil(mock_process: list[MockProcess]) -> Generator:
"""Mock psutil."""
with patch(
"homeassistant.components.systemmonitor.coordinator.psutil",
autospec=True,
) as mock_psutil:
"homeassistant.components.systemmonitor.ha_psutil.PsutilWrapper",
) as psutil_wrapper:
_wrapper = psutil_wrapper.return_value
_wrapper.psutil = NonCallableMock()
mock_psutil = _wrapper.psutil
mock_psutil.disk_usage.return_value = sdiskusage(
500 * 1024**3, 300 * 1024**3, 200 * 1024**3, 60.0
)
mock_psutil.swap_memory.return_value = sswap(
100 * 1024**2, 60 * 1024**2, 40 * 1024**2, 60.0, 1, 1
)
mock_psutil.virtual_memory.return_value = svmem(
mock_psutil.virtual_memory.return_value = VirtualMemory(
100 * 1024**2,
40 * 1024**2,
40.0,
60 * 1024**2,
30 * 1024**2,
1,
1,
1,
1,
1,
1,
)
mock_psutil.net_io_counters.return_value = {
"eth0": snetio(100 * 1024**2, 100 * 1024**2, 50, 50, 0, 0, 0, 0),
@ -180,65 +157,18 @@ def mock_psutil(mock_process: list[MockProcess]) -> Mock:
mock_psutil.sensors_temperatures.return_value = {
"cpu0-thermal": [shwtemp("cpu0-thermal", 50.0, 60.0, 70.0)]
}
mock_psutil.NoSuchProcess = NoSuchProcess
yield mock_psutil
@pytest.fixture
def mock_util(mock_process) -> Mock:
"""Mock psutil."""
with patch(
"homeassistant.components.systemmonitor.util.psutil", autospec=True
) as mock_util:
mock_util.net_if_addrs.return_value = {
"eth0": [
snicaddr(
socket.AF_INET,
"192.168.1.1",
"255.255.255.0",
"255.255.255.255",
None,
)
],
"eth1": [
snicaddr(
socket.AF_INET,
"192.168.10.1",
"255.255.255.0",
"255.255.255.255",
None,
)
],
"vethxyzxyz": [
snicaddr(
socket.AF_INET,
"172.16.10.1",
"255.255.255.0",
"255.255.255.255",
None,
)
],
}
mock_process = [MockProcess("python3")]
mock_util.process_iter.return_value = mock_process
# sensors_temperatures not available on MacOS so we
# need to override the spec
mock_util.sensors_temperatures = Mock()
mock_util.sensors_temperatures.return_value = {
"cpu0-thermal": [shwtemp("cpu0-thermal", 50.0, 60.0, 70.0)]
}
mock_util.disk_partitions.return_value = [
mock_psutil.disk_partitions.return_value = [
sdiskpart("test", "/", "ext4", "", 1, 1),
sdiskpart("test2", "/media/share", "ext4", "", 1, 1),
sdiskpart("test3", "/incorrect", "", "", 1, 1),
sdiskpart("proc", "/proc/run", "proc", "", 1, 1),
]
mock_util.disk_usage.return_value = sdiskusage(10, 10, 0, 0)
yield mock_util
mock_psutil.NoSuchProcess = NoSuchProcess
yield mock_psutil
@pytest.fixture
def mock_os() -> Mock:
def mock_os() -> Generator:
"""Mock os."""
with patch(
"homeassistant.components.systemmonitor.coordinator.os"

2
tests/components/systemmonitor/test_binary_sensor.py
View file

@ -24,7 +24,6 @@ async def test_binary_sensor(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
@ -65,7 +64,6 @@ async def test_binary_sensor(
async def test_binary_sensor_icon(
hass: HomeAssistant,
entity_registry_enabled_by_default: None,
mock_util: Mock,
mock_psutil: Mock,
mock_os: Mock,
mock_config_entry: MockConfigEntry,

1
tests/components/systemmonitor/test_init.py
View file

@ -71,7 +71,6 @@ async def test_migrate_process_sensor_to_binary_sensors(
hass: HomeAssistant,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
) -> None:

1
tests/components/systemmonitor/test_repairs.py
View file

@ -28,7 +28,6 @@ async def test_migrate_process_sensor(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,

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

@ -11,6 +11,7 @@ from syrupy.assertion import SnapshotAssertion
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.systemmonitor.const import DOMAIN
from homeassistant.components.systemmonitor.coordinator import VirtualMemory
from homeassistant.components.systemmonitor.sensor import get_cpu_icon
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
@ -18,7 +19,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component
from .conftest import MockProcess, svmem
from .conftest import MockProcess
from tests.common import MockConfigEntry, async_fire_time_changed
@ -28,7 +29,6 @@ async def test_sensor(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
@ -82,7 +82,6 @@ async def test_process_sensor_not_loaded(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
@ -128,7 +127,6 @@ async def test_sensor_not_loading_veth_networks(
async def test_sensor_icon(
hass: HomeAssistant,
entity_registry_enabled_by_default: None,
mock_util: Mock,
mock_psutil: Mock,
mock_os: Mock,
mock_config_entry: MockConfigEntry,
@ -150,7 +148,6 @@ async def test_sensor_yaml(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
) -> None:
"""Test the sensor imported from YAML."""
config = {
@ -182,7 +179,6 @@ async def test_sensor_yaml_fails_missing_argument(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
) -> None:
"""Test the sensor imported from YAML fails on missing mandatory argument."""
config = {
@ -203,7 +199,6 @@ async def test_sensor_updating(
hass: HomeAssistant,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test the sensor."""
@ -245,18 +240,12 @@ async def test_sensor_updating(
assert memory_sensor.state == STATE_UNAVAILABLE
mock_psutil.virtual_memory.side_effect = None
mock_psutil.virtual_memory.return_value = svmem(
mock_psutil.virtual_memory.return_value = VirtualMemory(
100 * 1024**2,
25 * 1024**2,
25.0,
60 * 1024**2,
30 * 1024**2,
1,
1,
1,
1,
1,
1,
)
freezer.tick(timedelta(minutes=1))
async_fire_time_changed(hass)
@ -271,7 +260,6 @@ async def test_sensor_process_fails(
hass: HomeAssistant,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
) -> None:
@ -394,7 +382,6 @@ async def test_sensor_network_sensors(
async def test_missing_cpu_temperature(
hass: HomeAssistant,
entity_registry_enabled_by_default: None,
mock_util: Mock,
mock_psutil: Mock,
mock_os: Mock,
mock_config_entry: MockConfigEntry,
@ -404,7 +391,7 @@ async def test_missing_cpu_temperature(
mock_psutil.sensors_temperatures.return_value = {
"not_exist": [shwtemp("not_exist", 50.0, 60.0, 70.0)]
}
mock_util.sensors_temperatures.return_value = {
mock_psutil.sensors_temperatures.return_value = {
"not_exist": [shwtemp("not_exist", 50.0, 60.0, 70.0)]
}
mock_config_entry.add_to_hass(hass)
@ -419,7 +406,6 @@ async def test_missing_cpu_temperature(
async def test_processor_temperature(
hass: HomeAssistant,
entity_registry_enabled_by_default: None,
mock_util: Mock,
mock_psutil: Mock,
mock_os: Mock,
mock_config_entry: MockConfigEntry,

23
tests/components/systemmonitor/test_util.py
View file

@ -1,6 +1,6 @@
"""Test System Monitor utils."""
from unittest.mock import Mock, patch
from unittest.mock import Mock
from psutil._common import sdiskpart
import pytest
@ -22,7 +22,6 @@ async def test_disk_setup_failure(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
mock_config_entry: MockConfigEntry,
side_effect: Exception,
error_text: str,
@ -30,18 +29,15 @@ async def test_disk_setup_failure(
) -> None:
"""Test the disk failures."""
with patch(
"homeassistant.components.systemmonitor.util.psutil.disk_usage",
side_effect=side_effect,
):
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_psutil.disk_usage.side_effect = side_effect
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
disk_sensor = hass.states.get("sensor.system_monitor_disk_free_media_share")
assert disk_sensor is None
disk_sensor = hass.states.get("sensor.system_monitor_disk_free_media_share")
assert disk_sensor is None
assert error_text in caplog.text
assert error_text in caplog.text
async def test_disk_util(
@ -49,12 +45,11 @@ async def test_disk_util(
entity_registry_enabled_by_default: None,
mock_psutil: Mock,
mock_os: Mock,
mock_util: Mock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the disk failures."""
mock_util.disk_partitions.return_value = [
mock_psutil.psutil.disk_partitions.return_value = [
sdiskpart("test", "/", "ext4", "", 1, 1), # Should be ok
sdiskpart("test2", "/media/share", "ext4", "", 1, 1), # Should be ok
sdiskpart("test3", "/incorrect", "", "", 1, 1), # Should be skipped as no type