Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix broken upnp derived sensors reporting b/s instead of kb/s (#57681)

pull/57805/head
Steven Looman 2021-10-15 22:03:03 +02:00 committed by GitHub
parent 12d1dfdaf9
commit b75f1b8951
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 191 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/upnp/__init__.py
View File

@ -264,5 +264,5 @@ class UpnpEntity(CoordinatorEntity):
def available(self) -> bool:
"""Return if entity is available."""
return super().available and (
self.coordinator.data.get(self.entity_description.key) or False
self.coordinator.data.get(self.entity_description.key) is not None
)

5
homeassistant/components/upnp/sensor.py
View File

@ -190,7 +190,10 @@ class DerivedUpnpSensor(UpnpSensor):
# Calculate derivative.
delta_value = current_value - self._last_value
if self.entity_description.native_unit_of_measurement == DATA_BYTES:
if (
self.entity_description.native_unit_of_measurement
== DATA_RATE_KIBIBYTES_PER_SECOND
):
delta_value /= KIBIBYTE
delta_time = current_timestamp - self._last_timestamp
if delta_time.total_seconds() == 0:

30
tests/components/upnp/conftest.py
View File

@ -9,6 +9,9 @@ from homeassistant.components import ssdp
from homeassistant.components.upnp.const import (
BYTES_RECEIVED,
BYTES_SENT,
CONFIG_ENTRY_ST,
CONFIG_ENTRY_UDN,
DOMAIN,
PACKETS_RECEIVED,
PACKETS_SENT,
ROUTER_IP,
@ -19,6 +22,8 @@ from homeassistant.components.upnp.const import (
from homeassistant.core import HomeAssistant
from homeassistant.util import dt
from tests.common import MockConfigEntry
TEST_UDN = "uuid:device"
TEST_ST = "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
TEST_USN = f"{TEST_UDN}::{TEST_ST}"
@ -115,8 +120,8 @@ class MockDevice:
self.status_times_polled += 1
return {
WAN_STATUS: "Connected",
ROUTER_UPTIME: 0,
ROUTER_IP: "192.168.0.1",
ROUTER_UPTIME: 10,
ROUTER_IP: "8.9.10.11",
}
@ -185,3 +190,24 @@ async def ssdp_no_discovery():
return_value=[],
) as mock_get_info:
yield (mock_register, mock_get_info)
@pytest.fixture
async def setup_integration(
hass: HomeAssistant, mock_get_source_ip, ssdp_instant_discovery, mock_upnp_device
):
"""Create an initialized integration."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONFIG_ENTRY_UDN: TEST_UDN,
CONFIG_ENTRY_ST: TEST_ST,
},
)
# Load config_entry.
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
yield entry

42
tests/components/upnp/test_binary_sensor.py Normal file
View File

@ -0,0 +1,42 @@
"""Tests for UPnP/IGD binary_sensor."""
from datetime import timedelta
from unittest.mock import AsyncMock
from homeassistant.components.upnp.const import (
DOMAIN,
ROUTER_IP,
ROUTER_UPTIME,
WAN_STATUS,
)
from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util
from .conftest import MockDevice
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_upnp_binary_sensors(
hass: HomeAssistant, setup_integration: MockConfigEntry
):
"""Test normal sensors."""
mock_device: MockDevice = hass.data[DOMAIN][setup_integration.entry_id].device
# First poll.
wan_status_state = hass.states.get("binary_sensor.mock_name_wan_status")
assert wan_status_state.state == "on"
# Second poll.
mock_device.async_get_status = AsyncMock(
return_value={
WAN_STATUS: "Disconnected",
ROUTER_UPTIME: 100,
ROUTER_IP: "",
}
)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=31))
await hass.async_block_till_done()
wan_status_state = hass.states.get("binary_sensor.mock_name_wan_status")
assert wan_status_state.state == "off"

3
tests/components/upnp/test_config_flow.py
View File

@ -25,6 +25,7 @@ from .conftest import (
TEST_ST,
TEST_UDN,
TEST_USN,
MockDevice,
)
from tests.common import MockConfigEntry, async_fire_time_changed
@ -196,7 +197,7 @@ async def test_options_flow(hass: HomeAssistant):
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id) is True
await hass.async_block_till_done()
mock_device = hass.data[DOMAIN][config_entry.entry_id].device
mock_device: MockDevice = hass.data[DOMAIN][config_entry.entry_id].device
# Reset.
mock_device.traffic_times_polled = 0

5
tests/components/upnp/test_init.py
View File

@ -9,7 +9,6 @@ from homeassistant.components.upnp.const import (
DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import TEST_ST, TEST_UDN
@ -28,10 +27,6 @@ async def test_async_setup_entry_default(hass: HomeAssistant):
},
)
# Initialisation of component, no device discovered.
await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
# Load config_entry.
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id) is True

114
tests/components/upnp/test_sensor.py Normal file
View File

@ -0,0 +1,114 @@
"""Tests for UPnP/IGD sensor."""
from datetime import timedelta
from unittest.mock import AsyncMock
from homeassistant.components.upnp.const import (
BYTES_RECEIVED,
BYTES_SENT,
DOMAIN,
PACKETS_RECEIVED,
PACKETS_SENT,
ROUTER_IP,
ROUTER_UPTIME,
TIMESTAMP,
UPDATE_INTERVAL,
WAN_STATUS,
)
from homeassistant.core import HomeAssistant
import homeassistant.util.dt as dt_util
from .conftest import MockDevice
from tests.common import MockConfigEntry, async_fire_time_changed
async def test_upnp_sensors(hass: HomeAssistant, setup_integration: MockConfigEntry):
"""Test normal sensors."""
mock_device: MockDevice = hass.data[DOMAIN][setup_integration.entry_id].device
# First poll.
b_received_state = hass.states.get("sensor.mock_name_b_received")
b_sent_state = hass.states.get("sensor.mock_name_b_sent")
packets_received_state = hass.states.get("sensor.mock_name_packets_received")
packets_sent_state = hass.states.get("sensor.mock_name_packets_sent")
external_ip_state = hass.states.get("sensor.mock_name_external_ip")
wan_status_state = hass.states.get("sensor.mock_name_wan_status")
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.
mock_device.async_get_traffic_data = AsyncMock(
return_value={
TIMESTAMP: dt_util.utcnow() + UPDATE_INTERVAL,
BYTES_RECEIVED: 10240,
BYTES_SENT: 20480,
PACKETS_RECEIVED: 30,
PACKETS_SENT: 40,
}
)
mock_device.async_get_status = AsyncMock(
return_value={
WAN_STATUS: "Disconnected",
ROUTER_UPTIME: 100,
ROUTER_IP: "",
}
)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=31))
await hass.async_block_till_done()
b_received_state = hass.states.get("sensor.mock_name_b_received")
b_sent_state = hass.states.get("sensor.mock_name_b_sent")
packets_received_state = hass.states.get("sensor.mock_name_packets_received")
packets_sent_state = hass.states.get("sensor.mock_name_packets_sent")
external_ip_state = hass.states.get("sensor.mock_name_external_ip")
wan_status_state = hass.states.get("sensor.mock_name_wan_status")
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, setup_integration: MockConfigEntry
):
"""Test derived sensors."""
mock_device: MockDevice = hass.data[DOMAIN][setup_integration.entry_id].device
# First poll.
kib_s_received_state = hass.states.get("sensor.mock_name_kib_s_received")
kib_s_sent_state = hass.states.get("sensor.mock_name_kib_s_sent")
packets_s_received_state = hass.states.get("sensor.mock_name_packets_s_received")
packets_s_sent_state = hass.states.get("sensor.mock_name_packets_s_sent")
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.
mock_device.async_get_traffic_data = AsyncMock(
return_value={
TIMESTAMP: dt_util.utcnow() + UPDATE_INTERVAL,
BYTES_RECEIVED: int(10240 * UPDATE_INTERVAL.total_seconds()),
BYTES_SENT: int(20480 * UPDATE_INTERVAL.total_seconds()),
PACKETS_RECEIVED: int(30 * UPDATE_INTERVAL.total_seconds()),
PACKETS_SENT: int(40 * UPDATE_INTERVAL.total_seconds()),
}
)
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=31))
await hass.async_block_till_done()
kib_s_received_state = hass.states.get("sensor.mock_name_kib_s_received")
kib_s_sent_state = hass.states.get("sensor.mock_name_kib_s_sent")
packets_s_received_state = hass.states.get("sensor.mock_name_packets_s_received")
packets_s_sent_state = hass.states.get("sensor.mock_name_packets_s_sent")
assert kib_s_received_state.state == "10.0"
assert kib_s_sent_state.state == "20.0"
assert packets_s_received_state.state == "30.0"
assert packets_s_sent_state.state == "40.0"