starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 14:29:21 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add WAN3 latency sensors to UniFi Network (#183620)

Browse Source 
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Sven Delmas
2026-09-30 10:28:33 +02:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 936a5ec119
commit b21bba68e7
3 changed files with 308 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/unifi/sensor.py
+8 -5
View File
@@ -67,6 +67,9 @@ from .hub import UnifiHub
PARALLEL_UPDATES = 0
# UniFi names the first WAN "WAN", not "WAN1".
type WanName = Literal["WAN", "WAN2", "WAN3"]
@callback
def async_bandwidth_sensor_allowed_fn(hub: UnifiHub, obj_id: str) -> bool:
@@ -256,7 +259,7 @@ def async_device_state_value_fn(hub: UnifiHub, device: Device) -> str | None:
@callback
def async_device_wan_latency_supported_fn(
wan: Literal["WAN", "WAN2"],
wan: WanName,
monitor_target: str,
hub: UnifiHub,
obj_id: str,
@@ -269,7 +272,7 @@ def async_device_wan_latency_supported_fn(
@callback
def async_device_wan_latency_value_fn(
wan: Literal["WAN", "WAN2"],
wan: WanName,
monitor_target: str,
hub: UnifiHub,
device: Device,
@@ -286,7 +289,7 @@ def async_device_wan_latency_value_fn(
@callback
def _device_wan_latency_monitor(
wan: Literal["WAN", "WAN2"], monitor_target: str, device: Device
wan: WanName, monitor_target: str, device: Device
) -> TypedDeviceUptimeStatsWanMonitor | None:
"""Return the target of the WAN latency monitor."""
if device.uptime_stats and (uptime_stats_wan := device.uptime_stats.get(wan)):
@@ -300,7 +303,7 @@ def make_wan_latency_sensors() -> tuple[UnifiSensorEntityDescription, ...]:
"""Create WAN latency sensors from WAN monitor data."""
def make_wan_latency_entity_description(
wan: Literal["WAN", "WAN2"], name: str, monitor_target: str
wan: WanName, name: str, monitor_target: str
) -> UnifiSensorEntityDescription:
name_wan = f"{name} {wan}"
return UnifiSensorEntityDescription[Devices, Device](
@@ -323,7 +326,7 @@ def make_wan_latency_sensors() -> tuple[UnifiSensorEntityDescription, ...]:
value_fn=partial(async_device_wan_latency_value_fn, wan, monitor_target),
)
wans: tuple[Literal["WAN"], Literal["WAN2"]] = ("WAN", "WAN2")
wans: tuple[WanName, ...] = ("WAN", "WAN2", "WAN3")
return tuple(
make_wan_latency_entity_description(wan, name, target)
for wan in wans
tests/components/unifi/snapshots/test_sensor.ambr
+174
View File
@@ -818,6 +818,64 @@
'state': 'unknown',
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_cloudflare_wan3_latency-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.mock_name_cloudflare_wan3_latency',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Cloudflare WAN3 latency',
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Cloudflare WAN3 latency',
'platform': 'unifi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'wan_latency',
'unique_id': 'cloudflare_wan3_latency-10:00:00:00:01:01',
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_cloudflare_wan3_latency-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'duration',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'mock-name Cloudflare WAN3 latency',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTime.MILLISECONDS: 'ms'>,
}),
'context': <ANY>,
'entity_id': 'sensor.mock_name_cloudflare_wan3_latency',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '16',
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_cloudflare_wan_latency-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -934,6 +992,64 @@
'state': 'unknown',
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_google_wan3_latency-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.mock_name_google_wan3_latency',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Google WAN3 latency',
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Google WAN3 latency',
'platform': 'unifi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'wan_latency',
'unique_id': 'google_wan3_latency-10:00:00:00:01:01',
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_google_wan3_latency-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'duration',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'mock-name Google WAN3 latency',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTime.MILLISECONDS: 'ms'>,
}),
'context': <ANY>,
'entity_id': 'sensor.mock_name_google_wan3_latency',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '32',
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_google_wan_latency-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
@@ -1050,6 +1166,64 @@
'state': 'unknown',
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_microsoft_wan3_latency-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.mock_name_microsoft_wan3_latency',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Microsoft WAN3 latency',
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.DURATION: 'duration'>,
'original_icon': None,
'original_name': 'Microsoft WAN3 latency',
'platform': 'unifi',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'wan_latency',
'unique_id': 'microsoft_wan3_latency-10:00:00:00:01:01',
'unit_of_measurement': <UnitOfTime.MILLISECONDS: 'ms'>,
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_microsoft_wan3_latency-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'duration',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'mock-name Microsoft WAN3 latency',
<SensorEntityCapabilityAttribute.STATE_CLASS: 'state_class'>: <SensorStateClass.MEASUREMENT: 'measurement'>,
<EntityStateAttribute.UNIT_OF_MEASUREMENT: 'unit_of_measurement'>: <UnitOfTime.MILLISECONDS: 'ms'>,
}),
'context': <ANY>,
'entity_id': 'sensor.mock_name_microsoft_wan3_latency',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '41',
})
# ---
# name: test_entity_and_device_data[wlan_payload0-device_payload0-client_payload0-config_entry_options0][sensor.mock_name_microsoft_wan_latency-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
tests/components/unifi/test_sensor.py
+126 -10
View File
@@ -528,6 +528,28 @@ UPS_DEVICE_2["vbms_table"]["battpool"]["device_bypass_voltage"] = 121.7
{"availability": 0.0, "target": "1.1.1.1", "type": "icmp"},
],
},
"WAN3": {
"monitors": [
{
"availability": 100.0,
"latency_average": 41,
"target": "www.microsoft.com",
"type": "icmp",
},
{
"availability": 100.0,
"latency_average": 32,
"target": "google.com",
"type": "icmp",
},
{
"availability": 100.0,
"latency_average": 16,
"target": "1.1.1.1",
"type": "icmp",
},
],
},
},
"state": 1,
"type": "usw",
@@ -1819,6 +1841,28 @@ async def test_device_uptime(
{"availability": 0.0, "target": "1.1.1.1", "type": "icmp"},
],
},
"WAN3": {
"monitors": [
{
"availability": 100.0,
"latency_average": 41,
"target": "www.microsoft.com",
"type": "icmp",
},
{
"availability": 100.0,
"latency_average": 32,
"target": "google.com",
"type": "icmp",
},
{
"availability": 100.0,
"latency_average": 16,
"target": "1.1.1.1",
"type": "icmp",
},
],
},
},
"state": 1,
"type": "usw",
@@ -1828,21 +1872,92 @@ async def test_device_uptime(
],
)
@pytest.mark.parametrize(
("monitor_id", "state", "index_to_update", "monitor_update", "updated_state"),
(
"wan",
"monitor_id",
"state",
"index_to_update",
"monitor_update",
"updated_state",
),
[
pytest.param(
"microsoft_wan", "56", 0, {"latency_average": 20}, "20", id="microsoft"
),
pytest.param("google_wan", "53", 1, {"latency_average": 90}, "90", id="google"),
pytest.param(
"cloudflare_wan", "30", 2, {"latency_average": 80}, "80", id="cloudflare"
"WAN",
"microsoft_wan",
"56",
0,
{"latency_average": 20},
"20",
id="microsoft",
),
pytest.param(
"microsoft_wan", "56", 0, {}, STATE_UNKNOWN, id="microsoft_no_response"
"WAN", "google_wan", "53", 1, {"latency_average": 90}, "90", id="google"
),
pytest.param("google_wan", "53", 1, {}, STATE_UNKNOWN, id="google_no_response"),
pytest.param(
"cloudflare_wan", "30", 2, {}, STATE_UNKNOWN, id="cloudflare_no_response"
"WAN",
"cloudflare_wan",
"30",
2,
{"latency_average": 80},
"80",
id="cloudflare",
),
pytest.param(
"WAN",
"microsoft_wan",
"56",
0,
{},
STATE_UNKNOWN,
id="microsoft_no_response",
),
pytest.param(
"WAN", "google_wan", "53", 1, {}, STATE_UNKNOWN, id="google_no_response"
),
pytest.param(
"WAN",
"cloudflare_wan",
"30",
2,
{},
STATE_UNKNOWN,
id="cloudflare_no_response",
),
pytest.param(
"WAN3",
"microsoft_wan3",
"41",
0,
{"latency_average": 25},
"25",
id="microsoft_wan3",
),
pytest.param(
"WAN3",
"google_wan3",
"32",
1,
{"latency_average": 60},
"60",
id="google_wan3",
),
pytest.param(
"WAN3",
"cloudflare_wan3",
"16",
2,
{"latency_average": 70},
"70",
id="cloudflare_wan3",
),
pytest.param(
"WAN3",
"cloudflare_wan3",
"16",
2,
{},
STATE_UNKNOWN,
id="cloudflare_wan3_no_response",
),
],
)
@@ -1852,6 +1967,7 @@ async def test_wan_monitor_latency(
entity_registry: er.EntityRegistry,
mock_websocket_message: WebsocketMessageMock,
device_payload: list[dict[str, Any]],
wan: str,
monitor_id: str,
state: str,
index_to_update: int,
@@ -1886,7 +2002,7 @@ async def test_wan_monitor_latency(
# Update state
device = deepcopy(device_payload[0])
monitor = device["uptime_stats"]["WAN"]["monitors"][index_to_update]
monitor = device["uptime_stats"][wan]["monitors"][index_to_update]
monitor.pop("latency_average")
monitor.update(monitor_update)
mock_websocket_message(message=MessageKey.DEVICE, data=device)