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

Changes after late upnp review (#72241)

Browse source 
* Changes after review of #70008, part 1

* Changes after review from #70008

* Revert to UpnpDataUpdateCoordinator._async_update_data
This commit is contained in:
Steven Looman 2022-05-24 21:37:37 +02:00 • committed by GitHub
parent a5e100176b
commit 2e36a79357
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 132 additions and 319 deletions
Download patch file Download diff file Expand all files Collapse all files

27
homeassistant/components/upnp/__init__.py
View file

@ -39,7 +39,7 @@ from .const import (
DOMAIN, DOMAIN,
LOGGER, LOGGER,
) )
from .device import Device, async_get_mac_address_from_host from .device import Device, async_create_device, async_get_mac_address_from_host
NOTIFICATION_ID = "upnp_notification" NOTIFICATION_ID = "upnp_notification"
NOTIFICATION_TITLE = "UPnP/IGD Setup" NOTIFICATION_TITLE = "UPnP/IGD Setup"
@ -113,8 +113,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
try: try:
await asyncio.wait_for(device_discovered_event.wait(), timeout=10) await asyncio.wait_for(device_discovered_event.wait(), timeout=10)
except asyncio.TimeoutError as err: except asyncio.TimeoutError as err:
LOGGER.debug("Device not discovered: %s", usn) raise ConfigEntryNotReady(f"Device not discovered: {usn}") from err
raise ConfigEntryNotReady from err
finally: finally:
cancel_discovered_callback() cancel_discovered_callback()
@ -123,12 +122,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
assert discovery_info.ssdp_location is not None assert discovery_info.ssdp_location is not None
location = discovery_info.ssdp_location location = discovery_info.ssdp_location
try: try:
device = await Device.async_create_device(hass, location) device = await async_create_device(hass, location)
except UpnpConnectionError as err: except UpnpConnectionError as err:
LOGGER.debug( raise ConfigEntryNotReady(
"Error connecting to device at location: %s, err: %s", location, err f"Error connecting to device at location: {location}, err: {err}"
) ) from err
raise ConfigEntryNotReady from err
# Track the original UDN such that existing sensors do not change their unique_id. # Track the original UDN such that existing sensors do not change their unique_id.
if CONFIG_ENTRY_ORIGINAL_UDN not in entry.data: if CONFIG_ENTRY_ORIGINAL_UDN not in entry.data:
@ -255,21 +253,15 @@ class UpnpDataUpdateCoordinator(DataUpdateCoordinator):
LOGGER, LOGGER,
name=device.name, name=device.name,
update_interval=update_interval, update_interval=update_interval,
update_method=self._async_fetch_data,
) )
async def _async_fetch_data(self) -> Mapping[str, Any]: async def _async_update_data(self) -> Mapping[str, Any]:
"""Update data.""" """Update data."""
try: try:
update_values = await asyncio.gather( update_values = await asyncio.gather(
self.device.async_get_traffic_data(), self.device.async_get_traffic_data(),
self.device.async_get_status(), self.device.async_get_status(),
) )
return {
**update_values[0],
**update_values[1],
}
except UpnpCommunicationError as exception: except UpnpCommunicationError as exception:
LOGGER.debug( LOGGER.debug(
"Caught exception when updating device: %s, exception: %s", "Caught exception when updating device: %s, exception: %s",
@ -280,6 +272,11 @@ class UpnpDataUpdateCoordinator(DataUpdateCoordinator):
f"Unable to communicate with IGD at: {self.device.device_url}" f"Unable to communicate with IGD at: {self.device.device_url}"
) from exception ) from exception
return {
**update_values[0],
**update_values[1],
}
class UpnpEntity(CoordinatorEntity[UpnpDataUpdateCoordinator]): class UpnpEntity(CoordinatorEntity[UpnpDataUpdateCoordinator]):
"""Base class for UPnP/IGD entities.""" """Base class for UPnP/IGD entities."""

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

@ -54,7 +54,7 @@ async def _async_wait_for_discoveries(hass: HomeAssistant) -> bool:
async def device_discovered(info: SsdpServiceInfo, change: SsdpChange) -> None: async def device_discovered(info: SsdpServiceInfo, change: SsdpChange) -> None:
if change != SsdpChange.BYEBYE: if change != SsdpChange.BYEBYE:
LOGGER.info( LOGGER.debug(
"Device discovered: %s, at: %s", "Device discovered: %s, at: %s",
info.ssdp_usn, info.ssdp_usn,
info.ssdp_location, info.ssdp_location,

51
homeassistant/components/upnp/device.py
View file

@ -9,7 +9,6 @@ from typing import Any
from urllib.parse import urlparse from urllib.parse import urlparse
from async_upnp_client.aiohttp import AiohttpSessionRequester from async_upnp_client.aiohttp import AiohttpSessionRequester
from async_upnp_client.client import UpnpDevice
from async_upnp_client.client_factory import UpnpFactory from async_upnp_client.client_factory import UpnpFactory
from async_upnp_client.exceptions import UpnpError from async_upnp_client.exceptions import UpnpError
from async_upnp_client.profiles.igd import IgdDevice, StatusInfo from async_upnp_client.profiles.igd import IgdDevice, StatusInfo
@ -47,15 +46,19 @@ async def async_get_mac_address_from_host(hass: HomeAssistant, host: str) -> str
return mac_address return mac_address
async def async_create_upnp_device( async def async_create_device(hass: HomeAssistant, ssdp_location: str) -> Device:
hass: HomeAssistant, ssdp_location: str """Create UPnP/IGD device."""
) -> UpnpDevice:
"""Create UPnP device."""
session = async_get_clientsession(hass) session = async_get_clientsession(hass)
requester = AiohttpSessionRequester(session, with_sleep=True, timeout=20) requester = AiohttpSessionRequester(session, with_sleep=True, timeout=20)
factory = UpnpFactory(requester, disable_state_variable_validation=True) factory = UpnpFactory(requester, disable_state_variable_validation=True)
return await factory.async_create_device(ssdp_location) upnp_device = await factory.async_create_device(ssdp_location)
# Create profile wrapper.
igd_device = IgdDevice(upnp_device, None)
device = Device(hass, igd_device)
return device
class Device: class Device:
@ -66,40 +69,8 @@ class Device:
self.hass = hass self.hass = hass
self._igd_device = igd_device self._igd_device = igd_device
self.coordinator: DataUpdateCoordinator | None = None self.coordinator: DataUpdateCoordinator | None = None
self._mac_address: str | None = None self.mac_address: str | None = None
self.original_udn: str | None = None
@classmethod
async def async_create_device(
cls, hass: HomeAssistant, ssdp_location: str
) -> Device:
"""Create UPnP/IGD device."""
upnp_device = await async_create_upnp_device(hass, ssdp_location)
# Create profile wrapper.
igd_device = IgdDevice(upnp_device, None)
device = cls(hass, igd_device)
return device
@property
def mac_address(self) -> str | None:
"""Get the mac address."""
return self._mac_address
@mac_address.setter
def mac_address(self, mac_address: str) -> None:
"""Set the mac address."""
self._mac_address = mac_address
@property
def original_udn(self) -> str | None:
"""Get the mac address."""
return self._original_udn
@original_udn.setter
def original_udn(self, original_udn: str) -> None:
"""Set the original UDN."""
self._original_udn = original_udn
@property @property
def udn(self) -> str: def udn(self) -> str:

196
tests/components/upnp/conftest.py
View file

@ -1,33 +1,23 @@
"""Configuration for SSDP tests.""" """Configuration for SSDP tests."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence from unittest.mock import AsyncMock, MagicMock, create_autospec, patch
from unittest.mock import AsyncMock, MagicMock, patch
from urllib.parse import urlparse from urllib.parse import urlparse
from async_upnp_client.client import UpnpDevice from async_upnp_client.client import UpnpDevice
from async_upnp_client.event_handler import UpnpEventHandler from async_upnp_client.profiles.igd import IgdDevice, StatusInfo
from async_upnp_client.profiles.igd import StatusInfo
import pytest import pytest
from homeassistant.components import ssdp from homeassistant.components import ssdp
from homeassistant.components.upnp.const import ( from homeassistant.components.upnp.const import (
BYTES_RECEIVED,
BYTES_SENT,
CONFIG_ENTRY_LOCATION, CONFIG_ENTRY_LOCATION,
CONFIG_ENTRY_MAC_ADDRESS, CONFIG_ENTRY_MAC_ADDRESS,
CONFIG_ENTRY_ORIGINAL_UDN, CONFIG_ENTRY_ORIGINAL_UDN,
CONFIG_ENTRY_ST, CONFIG_ENTRY_ST,
CONFIG_ENTRY_UDN, CONFIG_ENTRY_UDN,
DOMAIN, DOMAIN,
PACKETS_RECEIVED,
PACKETS_SENT,
ROUTER_IP,
ROUTER_UPTIME,
WAN_STATUS,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.util import dt
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@ -59,160 +49,37 @@ TEST_DISCOVERY = ssdp.SsdpServiceInfo(
) )
class MockUpnpDevice:
"""Mock async_upnp_client UpnpDevice."""
def __init__(self, location: str) -> None:
"""Initialize."""
self.device_url = location
@property
def manufacturer(self) -> str:
"""Get manufacturer."""
return TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_MANUFACTURER]
@property
def name(self) -> str:
"""Get name."""
return TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME]
@property
def model_name(self) -> str:
"""Get the model name."""
return TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_MODEL_NAME]
@property
def device_type(self) -> str:
"""Get the device type."""
return TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_DEVICE_TYPE]
@property
def udn(self) -> str:
"""Get the UDN."""
return TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_UDN]
@property
def usn(self) -> str:
"""Get the USN."""
return f"{self.udn}::{self.device_type}"
@property
def unique_id(self) -> str:
"""Get the unique id."""
return self.usn
def reinit(self, new_upnp_device: UpnpDevice) -> None:
"""Reinitialize."""
self.device_url = new_upnp_device.device_url
class MockIgdDevice:
"""Mock async_upnp_client IgdDevice."""
def __init__(self, device: MockUpnpDevice, event_handler: UpnpEventHandler) -> None:
"""Initialize mock device."""
self.device = device
self.profile_device = device
self._timestamp = dt.utcnow()
self.traffic_times_polled = 0
self.status_times_polled = 0
self.traffic_data = {
BYTES_RECEIVED: 0,
BYTES_SENT: 0,
PACKETS_RECEIVED: 0,
PACKETS_SENT: 0,
}
self.status_data = {
WAN_STATUS: "Connected",
ROUTER_UPTIME: 10,
ROUTER_IP: "8.9.10.11",
}
@property
def name(self) -> str:
"""Get the name of the device."""
return self.profile_device.name
@property
def manufacturer(self) -> str:
"""Get the manufacturer of this device."""
return self.profile_device.manufacturer
@property
def model_name(self) -> str:
"""Get the model name of this device."""
return self.profile_device.model_name
@property
def udn(self) -> str:
"""Get the UDN of the device."""
return self.profile_device.udn
@property
def device_type(self) -> str:
"""Get the device type of this device."""
return self.profile_device.device_type
async def async_get_total_bytes_received(self) -> int | None:
"""Get total bytes received."""
self.traffic_times_polled += 1
return self.traffic_data[BYTES_RECEIVED]
async def async_get_total_bytes_sent(self) -> int | None:
"""Get total bytes sent."""
return self.traffic_data[BYTES_SENT]
async def async_get_total_packets_received(self) -> int | None:
"""Get total packets received."""
return self.traffic_data[PACKETS_RECEIVED]
async def async_get_total_packets_sent(self) -> int | None:
"""Get total packets sent."""
return self.traffic_data[PACKETS_SENT]
async def async_get_external_ip_address(
self, services: Sequence[str] | None = None
) -> str | None:
"""
Get the external IP address.
:param services List of service names to try to get action from, defaults to [WANIPC,WANPPP]
"""
return self.status_data[ROUTER_IP]
async def async_get_status_info(
self, services: Sequence[str] | None = None
) -> StatusInfo | None:
"""
Get status info.
:param services List of service names to try to get action from, defaults to [WANIPC,WANPPP]
"""
self.status_times_polled += 1
return StatusInfo(
self.status_data[WAN_STATUS], "", self.status_data[ROUTER_UPTIME]
)
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def mock_upnp_device(): def mock_igd_device() -> IgdDevice:
"""Mock homeassistant.components.upnp.Device.""" """Mock async_upnp_client device."""
mock_upnp_device = create_autospec(UpnpDevice, instance=True)
mock_upnp_device.device_url = TEST_DISCOVERY.ssdp_location
async def mock_async_create_upnp_device( mock_igd_device = create_autospec(IgdDevice)
hass: HomeAssistant, location: str mock_igd_device.name = TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_FRIENDLY_NAME]
) -> UpnpDevice: mock_igd_device.manufacturer = TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_MANUFACTURER]
"""Create UPnP device.""" mock_igd_device.model_name = TEST_DISCOVERY.upnp[ssdp.ATTR_UPNP_MODEL_NAME]
return MockUpnpDevice(location) mock_igd_device.udn = TEST_DISCOVERY.ssdp_udn
mock_igd_device.device = mock_upnp_device
mock_igd_device.async_get_total_bytes_received.return_value = 0
mock_igd_device.async_get_total_bytes_sent.return_value = 0
mock_igd_device.async_get_total_packets_received.return_value = 0
mock_igd_device.async_get_total_packets_sent.return_value = 0
mock_igd_device.async_get_status_info.return_value = StatusInfo(
"Connected",
"",
10,
)
mock_igd_device.async_get_external_ip_address.return_value = "8.9.10.11"
with patch( with patch(
"homeassistant.components.upnp.device.async_create_upnp_device", "homeassistant.components.upnp.device.UpnpFactory.async_create_device"
side_effect=mock_async_create_upnp_device, ), patch(
) as mock_async_create_upnp_device, patch( "homeassistant.components.upnp.device.IgdDevice.__new__",
"homeassistant.components.upnp.device.IgdDevice", new=MockIgdDevice return_value=mock_igd_device,
) as mock_igd_device: ):
yield mock_async_create_upnp_device, mock_igd_device yield mock_igd_device
@pytest.fixture @pytest.fixture
@ -297,11 +164,11 @@ async def ssdp_no_discovery():
@pytest.fixture @pytest.fixture
async def config_entry( async def mock_config_entry(
hass: HomeAssistant, hass: HomeAssistant,
mock_get_source_ip, mock_get_source_ip,
ssdp_instant_discovery, ssdp_instant_discovery,
mock_upnp_device, mock_igd_device: IgdDevice,
mock_mac_address_from_host, mock_mac_address_from_host,
): ):
"""Create an initialized integration.""" """Create an initialized integration."""
@ -316,6 +183,7 @@ async def config_entry(
CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS, CONFIG_ENTRY_MAC_ADDRESS: TEST_MAC_ADDRESS,
}, },
) )
entry.igd_device = mock_igd_device
# Load config_entry. # Load config_entry.
entry.add_to_hass(hass) entry.add_to_hass(hass)

34
tests/components/upnp/test_binary_sensor.py
View file

@ -2,36 +2,34 @@
from datetime import timedelta from datetime import timedelta
from homeassistant.components.upnp.const import ( from async_upnp_client.profiles.igd import StatusInfo
DOMAIN,
ROUTER_IP, from homeassistant.components.upnp.const import DEFAULT_SCAN_INTERVAL
ROUTER_UPTIME,
WAN_STATUS,
)
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from .conftest import MockIgdDevice
from tests.common import MockConfigEntry, async_fire_time_changed from tests.common import MockConfigEntry, async_fire_time_changed
async def test_upnp_binary_sensors(hass: HomeAssistant, config_entry: MockConfigEntry): async def test_upnp_binary_sensors(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
):
"""Test normal sensors.""" """Test normal sensors."""
# First poll. # First poll.
wan_status_state = hass.states.get("binary_sensor.mock_name_wan_status") wan_status_state = hass.states.get("binary_sensor.mock_name_wan_status")
assert wan_status_state.state == "on" assert wan_status_state.state == "on"
# Second poll. # Second poll.
mock_device: MockIgdDevice = hass.data[DOMAIN][ mock_igd_device = mock_config_entry.igd_device
config_entry.entry_id mock_igd_device.async_get_status_info.return_value = StatusInfo(
].device._igd_device "Disconnected",
mock_device.status_data = { "",
WAN_STATUS: "Disconnected", 40,
ROUTER_UPTIME: 100, )
ROUTER_IP: "",
} async_fire_time_changed(
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=31)) hass, dt_util.utcnow() + timedelta(seconds=DEFAULT_SCAN_INTERVAL)
)
await hass.async_block_till_done() await hass.async_block_till_done()
wan_status_state = hass.states.get("binary_sensor.mock_name_wan_status") wan_status_state = hass.states.get("binary_sensor.mock_name_wan_status")

141
tests/components/upnp/test_sensor.py
View file

@ -3,114 +3,93 @@
from datetime import timedelta from datetime import timedelta
from unittest.mock import patch from unittest.mock import patch
from async_upnp_client.profiles.igd import StatusInfo
import pytest import pytest
from homeassistant.components.upnp import UpnpDataUpdateCoordinator from homeassistant.components.upnp.const import DEFAULT_SCAN_INTERVAL
from homeassistant.components.upnp.const import (
BYTES_RECEIVED,
BYTES_SENT,
DEFAULT_SCAN_INTERVAL,
DOMAIN,
PACKETS_RECEIVED,
PACKETS_SENT,
ROUTER_IP,
ROUTER_UPTIME,
WAN_STATUS,
)
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from .conftest import MockIgdDevice
from tests.common import MockConfigEntry, async_fire_time_changed from tests.common import MockConfigEntry, async_fire_time_changed
async def test_upnp_sensors(hass: HomeAssistant, config_entry: MockConfigEntry): async def test_upnp_sensors(hass: HomeAssistant, mock_config_entry: MockConfigEntry):
"""Test normal sensors.""" """Test normal sensors."""
# First poll. # First poll.
b_received_state = hass.states.get("sensor.mock_name_b_received") assert hass.states.get("sensor.mock_name_b_received").state == "0"
b_sent_state = hass.states.get("sensor.mock_name_b_sent") assert hass.states.get("sensor.mock_name_b_sent").state == "0"
packets_received_state = hass.states.get("sensor.mock_name_packets_received") assert hass.states.get("sensor.mock_name_packets_received").state == "0"
packets_sent_state = hass.states.get("sensor.mock_name_packets_sent") assert hass.states.get("sensor.mock_name_packets_sent").state == "0"
external_ip_state = hass.states.get("sensor.mock_name_external_ip") assert hass.states.get("sensor.mock_name_external_ip").state == "8.9.10.11"
wan_status_state = hass.states.get("sensor.mock_name_wan_status") assert hass.states.get("sensor.mock_name_wan_status").state == "Connected"
assert b_received_state.state == "0"
assert b_sent_state.state == "0"
assert packets_received_state.state == "0"
assert packets_sent_state.state == "0"
assert external_ip_state.state == "8.9.10.11"
assert wan_status_state.state == "Connected"
# Second poll. # Second poll.
mock_device: MockIgdDevice = hass.data[DOMAIN][ mock_igd_device = mock_config_entry.igd_device
config_entry.entry_id mock_igd_device.async_get_total_bytes_received.return_value = 10240
].device._igd_device mock_igd_device.async_get_total_bytes_sent.return_value = 20480
mock_device.traffic_data = { mock_igd_device.async_get_total_packets_received.return_value = 30
BYTES_RECEIVED: 10240, mock_igd_device.async_get_total_packets_sent.return_value = 40
BYTES_SENT: 20480, mock_igd_device.async_get_status_info.return_value = StatusInfo(
PACKETS_RECEIVED: 30, "Disconnected",
PACKETS_SENT: 40, "",
} 40,
mock_device.status_data = { )
WAN_STATUS: "Disconnected", mock_igd_device.async_get_external_ip_address.return_value = ""
ROUTER_UPTIME: 100,
ROUTER_IP: "",
}
now = dt_util.utcnow() now = dt_util.utcnow()
async_fire_time_changed(hass, now + timedelta(seconds=DEFAULT_SCAN_INTERVAL)) async_fire_time_changed(hass, now + timedelta(seconds=DEFAULT_SCAN_INTERVAL))
await hass.async_block_till_done() await hass.async_block_till_done()
b_received_state = hass.states.get("sensor.mock_name_b_received") assert hass.states.get("sensor.mock_name_b_received").state == "10240"
b_sent_state = hass.states.get("sensor.mock_name_b_sent") assert hass.states.get("sensor.mock_name_b_sent").state == "20480"
packets_received_state = hass.states.get("sensor.mock_name_packets_received") assert hass.states.get("sensor.mock_name_packets_received").state == "30"
packets_sent_state = hass.states.get("sensor.mock_name_packets_sent") assert hass.states.get("sensor.mock_name_packets_sent").state == "40"
external_ip_state = hass.states.get("sensor.mock_name_external_ip") assert hass.states.get("sensor.mock_name_external_ip").state == ""
wan_status_state = hass.states.get("sensor.mock_name_wan_status") assert hass.states.get("sensor.mock_name_wan_status").state == "Disconnected"
assert b_received_state.state == "10240"
assert b_sent_state.state == "20480"
assert packets_received_state.state == "30"
assert packets_sent_state.state == "40"
assert external_ip_state.state == ""
assert wan_status_state.state == "Disconnected"
async def test_derived_upnp_sensors(hass: HomeAssistant, config_entry: MockConfigEntry): async def test_derived_upnp_sensors(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
):
"""Test derived sensors.""" """Test derived sensors."""
coordinator: UpnpDataUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
# First poll. # First poll.
kib_s_received_state = hass.states.get("sensor.mock_name_kib_s_received") assert hass.states.get("sensor.mock_name_kib_s_received").state == "unknown"
kib_s_sent_state = hass.states.get("sensor.mock_name_kib_s_sent") assert hass.states.get("sensor.mock_name_kib_s_sent").state == "unknown"
packets_s_received_state = hass.states.get("sensor.mock_name_packets_s_received") assert hass.states.get("sensor.mock_name_packets_s_received").state == "unknown"
packets_s_sent_state = hass.states.get("sensor.mock_name_packets_s_sent") assert hass.states.get("sensor.mock_name_packets_s_sent").state == "unknown"
assert kib_s_received_state.state == "unknown"
assert kib_s_sent_state.state == "unknown"
assert packets_s_received_state.state == "unknown"
assert packets_s_sent_state.state == "unknown"
# Second poll. # Second poll.
mock_igd_device = mock_config_entry.igd_device
mock_igd_device.async_get_total_bytes_received.return_value = int(
10240 * DEFAULT_SCAN_INTERVAL
)
mock_igd_device.async_get_total_bytes_sent.return_value = int(
20480 * DEFAULT_SCAN_INTERVAL
)
mock_igd_device.async_get_total_packets_received.return_value = int(
30 * DEFAULT_SCAN_INTERVAL
)
mock_igd_device.async_get_total_packets_sent.return_value = int(
40 * DEFAULT_SCAN_INTERVAL
)
now = dt_util.utcnow() now = dt_util.utcnow()
with patch( with patch(
"homeassistant.components.upnp.device.utcnow", "homeassistant.components.upnp.device.utcnow",
return_value=now + timedelta(seconds=DEFAULT_SCAN_INTERVAL), return_value=now + timedelta(seconds=DEFAULT_SCAN_INTERVAL),
): ):
mock_device: MockIgdDevice = coordinator.device._igd_device
mock_device.traffic_data = {
BYTES_RECEIVED: int(10240 * DEFAULT_SCAN_INTERVAL),
BYTES_SENT: int(20480 * DEFAULT_SCAN_INTERVAL),
PACKETS_RECEIVED: int(30 * DEFAULT_SCAN_INTERVAL),
PACKETS_SENT: int(40 * DEFAULT_SCAN_INTERVAL),
}
async_fire_time_changed(hass, now + timedelta(seconds=DEFAULT_SCAN_INTERVAL)) async_fire_time_changed(hass, now + timedelta(seconds=DEFAULT_SCAN_INTERVAL))
await hass.async_block_till_done() await hass.async_block_till_done()
kib_s_received_state = hass.states.get("sensor.mock_name_kib_s_received") assert float(
kib_s_sent_state = hass.states.get("sensor.mock_name_kib_s_sent") hass.states.get("sensor.mock_name_kib_s_received").state
packets_s_received_state = hass.states.get( ) == pytest.approx(10.0, rel=0.1)
"sensor.mock_name_packets_s_received" assert float(
) hass.states.get("sensor.mock_name_kib_s_sent").state
packets_s_sent_state = hass.states.get("sensor.mock_name_packets_s_sent") ) == pytest.approx(20.0, rel=0.1)
assert float(kib_s_received_state.state) == pytest.approx(10.0, rel=0.1) assert float(
assert float(kib_s_sent_state.state) == pytest.approx(20.0, rel=0.1) hass.states.get("sensor.mock_name_packets_s_received").state
assert float(packets_s_received_state.state) == pytest.approx(30.0, rel=0.1) ) == pytest.approx(30.0, rel=0.1)
assert float(packets_s_sent_state.state) == pytest.approx(40.0, rel=0.1) assert float(
hass.states.get("sensor.mock_name_packets_s_sent").state
) == pytest.approx(40.0, rel=0.1)