Fix failing UniFi tests related to utcnow (#119131)

* test

* Fix missed test
This commit is contained in:
Robert Svensson 2024-06-08 17:58:47 +02:00 committed by GitHub
parent c49ca5ed56
commit 915658daa1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 92 additions and 115 deletions

View File

@ -704,32 +704,11 @@ async def test_option_track_devices(
assert hass.states.get("device_tracker.device") assert hass.states.get("device_tracker.device")
@pytest.mark.parametrize(
"client_payload",
[
[
{
"essid": "ssid",
"hostname": "client",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
},
{
"essid": "ssid2",
"hostname": "client_on_ssid2",
"is_wired": False,
"last_seen": 1562600145,
"mac": "00:00:00:00:00:02",
},
]
],
)
@pytest.mark.usefixtures("mock_device_registry") @pytest.mark.usefixtures("mock_device_registry")
async def test_option_ssid_filter( async def test_option_ssid_filter(
hass: HomeAssistant, hass: HomeAssistant,
mock_unifi_websocket, mock_unifi_websocket,
config_entry_setup: ConfigEntry, config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]], client_payload: list[dict[str, Any]],
) -> None: ) -> None:
"""Test the SSID filter works. """Test the SSID filter works.
@ -737,13 +716,31 @@ async def test_option_ssid_filter(
Client will travel from a supported SSID to an unsupported ssid. Client will travel from a supported SSID to an unsupported ssid.
Client on SSID2 will be removed on change of options. Client on SSID2 will be removed on change of options.
""" """
client_payload += [
{
"essid": "ssid",
"hostname": "client",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
},
{
"essid": "ssid2",
"hostname": "client_on_ssid2",
"is_wired": False,
"last_seen": 1562600145,
"mac": "00:00:00:00:00:02",
},
]
config_entry = await config_entry_factory()
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2 assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 2
assert hass.states.get("device_tracker.client").state == STATE_HOME assert hass.states.get("device_tracker.client").state == STATE_HOME
assert hass.states.get("device_tracker.client_on_ssid2").state == STATE_NOT_HOME assert hass.states.get("device_tracker.client_on_ssid2").state == STATE_NOT_HOME
# Setting SSID filter will remove clients outside of filter # Setting SSID filter will remove clients outside of filter
hass.config_entries.async_update_entry( hass.config_entries.async_update_entry(
config_entry_setup, options={CONF_SSID_FILTER: ["ssid"]} config_entry, options={CONF_SSID_FILTER: ["ssid"]}
) )
await hass.async_block_till_done() await hass.async_block_till_done()
@ -766,8 +763,7 @@ async def test_option_ssid_filter(
new_time = dt_util.utcnow() + timedelta( new_time = dt_util.utcnow() + timedelta(
seconds=( seconds=(
config_entry_setup.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME) config_entry.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME) + 1
+ 1
) )
) )
with freeze_time(new_time): with freeze_time(new_time):
@ -781,9 +777,7 @@ async def test_option_ssid_filter(
assert not hass.states.get("device_tracker.client_on_ssid2") assert not hass.states.get("device_tracker.client_on_ssid2")
# Remove SSID filter # Remove SSID filter
hass.config_entries.async_update_entry( hass.config_entries.async_update_entry(config_entry, options={CONF_SSID_FILTER: []})
config_entry_setup, options={CONF_SSID_FILTER: []}
)
await hass.async_block_till_done() await hass.async_block_till_done()
client["last_seen"] += 1 client["last_seen"] += 1
@ -797,8 +791,7 @@ async def test_option_ssid_filter(
# Time pass to mark client as away # Time pass to mark client as away
new_time += timedelta( new_time += timedelta(
seconds=( seconds=(
config_entry_setup.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME) config_entry.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME) + 1
+ 1
) )
) )
with freeze_time(new_time): with freeze_time(new_time):
@ -820,9 +813,7 @@ async def test_option_ssid_filter(
await hass.async_block_till_done() await hass.async_block_till_done()
new_time += timedelta( new_time += timedelta(
seconds=( seconds=(config_entry.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME))
config_entry_setup.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME)
)
) )
with freeze_time(new_time): with freeze_time(new_time):
async_fire_time_changed(hass, new_time) async_fire_time_changed(hass, new_time)
@ -831,32 +822,29 @@ async def test_option_ssid_filter(
assert hass.states.get("device_tracker.client_on_ssid2").state == STATE_NOT_HOME assert hass.states.get("device_tracker.client_on_ssid2").state == STATE_NOT_HOME
@pytest.mark.parametrize(
"client_payload",
[
[
{
"essid": "ssid",
"hostname": "client",
"ip": "10.0.0.1",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
}
]
],
)
@pytest.mark.usefixtures("mock_device_registry") @pytest.mark.usefixtures("mock_device_registry")
async def test_wireless_client_go_wired_issue( async def test_wireless_client_go_wired_issue(
hass: HomeAssistant, hass: HomeAssistant,
mock_unifi_websocket, mock_unifi_websocket,
config_entry_setup: ConfigEntry, config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]], client_payload: list[dict[str, Any]],
) -> None: ) -> None:
"""Test the solution to catch wireless device go wired UniFi issue. """Test the solution to catch wireless device go wired UniFi issue.
UniFi Network has a known issue that when a wireless device goes away it sometimes gets marked as wired. UniFi Network has a known issue that when a wireless device goes away it sometimes gets marked as wired.
""" """
client_payload.append(
{
"essid": "ssid",
"hostname": "client",
"ip": "10.0.0.1",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
}
)
config_entry = await config_entry_factory()
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 1
# Client is wireless # Client is wireless
@ -876,9 +864,7 @@ async def test_wireless_client_go_wired_issue(
# Pass time # Pass time
new_time = dt_util.utcnow() + timedelta( new_time = dt_util.utcnow() + timedelta(
seconds=( seconds=(config_entry.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME))
config_entry_setup.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME)
)
) )
with freeze_time(new_time): with freeze_time(new_time):
async_fire_time_changed(hass, new_time) async_fire_time_changed(hass, new_time)
@ -909,30 +895,27 @@ async def test_wireless_client_go_wired_issue(
@pytest.mark.parametrize("config_entry_options", [{CONF_IGNORE_WIRED_BUG: True}]) @pytest.mark.parametrize("config_entry_options", [{CONF_IGNORE_WIRED_BUG: True}])
@pytest.mark.parametrize(
"client_payload",
[
[
{
"ap_mac": "00:00:00:00:02:01",
"essid": "ssid",
"hostname": "client",
"ip": "10.0.0.1",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
}
]
],
)
@pytest.mark.usefixtures("mock_device_registry") @pytest.mark.usefixtures("mock_device_registry")
async def test_option_ignore_wired_bug( async def test_option_ignore_wired_bug(
hass: HomeAssistant, hass: HomeAssistant,
mock_unifi_websocket, mock_unifi_websocket,
config_entry_setup: ConfigEntry, config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]], client_payload: list[dict[str, Any]],
) -> None: ) -> None:
"""Test option to ignore wired bug.""" """Test option to ignore wired bug."""
client_payload.append(
{
"ap_mac": "00:00:00:00:02:01",
"essid": "ssid",
"hostname": "client",
"ip": "10.0.0.1",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
}
)
config_entry = await config_entry_factory()
assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(TRACKER_DOMAIN)) == 1
# Client is wireless # Client is wireless
@ -951,9 +934,7 @@ async def test_option_ignore_wired_bug(
# pass time # pass time
new_time = dt_util.utcnow() + timedelta( new_time = dt_util.utcnow() + timedelta(
seconds=config_entry_setup.options.get( seconds=config_entry.options.get(CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME)
CONF_DETECTION_TIME, DEFAULT_DETECTION_TIME
)
) )
with freeze_time(new_time): with freeze_time(new_time):
async_fire_time_changed(hass, new_time) async_fire_time_changed(hass, new_time)

View File

@ -4,6 +4,7 @@ from collections.abc import Callable
from copy import deepcopy from copy import deepcopy
from datetime import timedelta from datetime import timedelta
from http import HTTPStatus from http import HTTPStatus
from typing import Any
from unittest.mock import patch from unittest.mock import patch
import aiounifi import aiounifi
@ -313,27 +314,25 @@ async def test_reset_fails(
assert config_entry.state is ConfigEntryState.LOADED assert config_entry.state is ConfigEntryState.LOADED
@pytest.mark.parametrize(
"client_payload",
[
[
{
"hostname": "client",
"ip": "10.0.0.1",
"is_wired": True,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
},
]
],
)
async def test_connection_state_signalling( async def test_connection_state_signalling(
hass: HomeAssistant, hass: HomeAssistant,
mock_device_registry, mock_device_registry,
config_entry_setup: ConfigEntry,
websocket_mock, websocket_mock,
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]],
) -> None: ) -> None:
"""Verify connection statesignalling and connection state are working.""" """Verify connection statesignalling and connection state are working."""
client_payload.append(
{
"hostname": "client",
"ip": "10.0.0.1",
"is_wired": True,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
}
)
await config_entry_factory()
# Controller is connected # Controller is connected
assert hass.states.get("device_tracker.client").state == "home" assert hass.states.get("device_tracker.client").state == "home"

View File

@ -682,43 +682,40 @@ async def test_poe_port_switches(
assert hass.states.get("sensor.mock_name_port_1_poe_power") assert hass.states.get("sensor.mock_name_port_1_poe_power")
@pytest.mark.parametrize(
"client_payload",
[
[
{
"essid": "SSID 1",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
"name": "Wireless client",
"oui": "Producer",
"rx_bytes-r": 2345000000,
"tx_bytes-r": 6789000000,
},
{
"essid": "SSID 2",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:02",
"name": "Wireless client2",
"oui": "Producer2",
"rx_bytes-r": 2345000000,
"tx_bytes-r": 6789000000,
},
]
],
)
@pytest.mark.parametrize("wlan_payload", [[WLAN]]) @pytest.mark.parametrize("wlan_payload", [[WLAN]])
@pytest.mark.usefixtures("config_entry_setup")
async def test_wlan_client_sensors( async def test_wlan_client_sensors(
hass: HomeAssistant, hass: HomeAssistant,
entity_registry: er.EntityRegistry, entity_registry: er.EntityRegistry,
mock_unifi_websocket, mock_unifi_websocket,
websocket_mock, websocket_mock,
config_entry_factory: Callable[[], ConfigEntry],
client_payload: list[dict[str, Any]], client_payload: list[dict[str, Any]],
) -> None: ) -> None:
"""Verify that WLAN client sensors are working as expected.""" """Verify that WLAN client sensors are working as expected."""
client_payload += [
{
"essid": "SSID 1",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:01",
"name": "Wireless client",
"oui": "Producer",
"rx_bytes-r": 2345000000,
"tx_bytes-r": 6789000000,
},
{
"essid": "SSID 2",
"is_wired": False,
"last_seen": dt_util.as_timestamp(dt_util.utcnow()),
"mac": "00:00:00:00:00:02",
"name": "Wireless client2",
"oui": "Producer2",
"rx_bytes-r": 2345000000,
"tx_bytes-r": 6789000000,
},
]
await config_entry_factory()
assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 1 assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 1
ent_reg_entry = entity_registry.async_get("sensor.ssid_1") ent_reg_entry = entity_registry.async_get("sensor.ssid_1")