diff --git a/homeassistant/components/otbr/config_flow.py b/homeassistant/components/otbr/config_flow.py index 0a554806c32..c8ab8246c8b 100644 --- a/homeassistant/components/otbr/config_flow.py +++ b/homeassistant/components/otbr/config_flow.py @@ -3,10 +3,12 @@ from __future__ import annotations import asyncio import logging +from typing import cast import aiohttp import python_otbr_api from python_otbr_api import tlv_parser +from python_otbr_api.tlv_parser import MeshcopTLVType import voluptuous as vol import yarl @@ -38,8 +40,8 @@ class OTBRConfigFlow(ConfigFlow, domain=DOMAIN): thread_dataset_tlv = await async_get_preferred_dataset(self.hass) if thread_dataset_tlv: dataset = tlv_parser.parse_tlv(thread_dataset_tlv) - if channel_str := dataset.get(tlv_parser.MeshcopTLVType.CHANNEL): - thread_dataset_channel = int(channel_str, base=16) + if channel := dataset.get(MeshcopTLVType.CHANNEL): + thread_dataset_channel = cast(tlv_parser.Channel, channel).channel if thread_dataset_tlv is not None and ( not allowed_channel or allowed_channel == thread_dataset_channel diff --git a/homeassistant/components/otbr/manifest.json b/homeassistant/components/otbr/manifest.json index f092bff48b9..f04e15a549c 100644 --- a/homeassistant/components/otbr/manifest.json +++ b/homeassistant/components/otbr/manifest.json @@ -8,5 +8,5 @@ "documentation": "https://www.home-assistant.io/integrations/otbr", "integration_type": "service", "iot_class": "local_polling", - "requirements": ["python-otbr-api==2.0.0"] + "requirements": ["python-otbr-api==2.1.0"] } diff --git a/homeassistant/components/otbr/util.py b/homeassistant/components/otbr/util.py index a625ba09018..5541ecb6874 100644 --- a/homeassistant/components/otbr/util.py +++ b/homeassistant/components/otbr/util.py @@ -5,11 +5,12 @@ from collections.abc import Callable, Coroutine import contextlib import dataclasses from functools import wraps -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate, ParamSpec, TypeVar, cast import python_otbr_api from python_otbr_api import tlv_parser from python_otbr_api.pskc import compute_pskc +from python_otbr_api.tlv_parser import MeshcopTLVType from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( is_multiprotocol_url, @@ -146,14 +147,10 @@ async def _warn_on_channel_collision( dataset = tlv_parser.parse_tlv(dataset_tlvs.hex()) - if (channel_s := dataset.get(tlv_parser.MeshcopTLVType.CHANNEL)) is None: - delete_issue() - return - try: - channel = int(channel_s, 16) - except ValueError: + if (channel_s := dataset.get(MeshcopTLVType.CHANNEL)) is None: delete_issue() return + channel = cast(tlv_parser.Channel, channel_s).channel if channel == allowed_channel: delete_issue() @@ -186,20 +183,20 @@ def _warn_on_default_network_settings( insecure = False if ( - network_key := dataset.get(tlv_parser.MeshcopTLVType.NETWORKKEY) - ) is not None and bytes.fromhex(network_key) in INSECURE_NETWORK_KEYS: + network_key := dataset.get(MeshcopTLVType.NETWORKKEY) + ) is not None and network_key.data in INSECURE_NETWORK_KEYS: insecure = True if ( not insecure - and tlv_parser.MeshcopTLVType.EXTPANID in dataset - and tlv_parser.MeshcopTLVType.NETWORKNAME in dataset - and tlv_parser.MeshcopTLVType.PSKC in dataset + and MeshcopTLVType.EXTPANID in dataset + and MeshcopTLVType.NETWORKNAME in dataset + and MeshcopTLVType.PSKC in dataset ): - ext_pan_id = dataset[tlv_parser.MeshcopTLVType.EXTPANID] - network_name = dataset[tlv_parser.MeshcopTLVType.NETWORKNAME] - pskc = bytes.fromhex(dataset[tlv_parser.MeshcopTLVType.PSKC]) + ext_pan_id = dataset[MeshcopTLVType.EXTPANID] + network_name = cast(tlv_parser.NetworkName, dataset[MeshcopTLVType.NETWORKNAME]) + pskc = dataset[MeshcopTLVType.PSKC].data for passphrase in INSECURE_PASSPHRASES: - if pskc == compute_pskc(ext_pan_id, network_name, passphrase): + if pskc == compute_pskc(ext_pan_id.data, network_name.name, passphrase): insecure = True break diff --git a/homeassistant/components/otbr/websocket_api.py b/homeassistant/components/otbr/websocket_api.py index 7fd0f8187b2..0dcce288348 100644 --- a/homeassistant/components/otbr/websocket_api.py +++ b/homeassistant/components/otbr/websocket_api.py @@ -1,7 +1,10 @@ """Websocket API for OTBR.""" +from typing import cast + import python_otbr_api from python_otbr_api import tlv_parser +from python_otbr_api.tlv_parser import MeshcopTLVType import voluptuous as vol from homeassistant.components import websocket_api @@ -133,8 +136,8 @@ async def websocket_set_network( connection.send_error(msg["id"], "unknown_dataset", "Unknown dataset") return dataset = tlv_parser.parse_tlv(dataset_tlv) - if channel_str := dataset.get(tlv_parser.MeshcopTLVType.CHANNEL): - thread_dataset_channel = int(channel_str, base=16) + if channel := dataset.get(MeshcopTLVType.CHANNEL): + thread_dataset_channel = cast(tlv_parser.Channel, channel).channel data: OTBRData = hass.data[DOMAIN] allowed_channel = await get_allowed_channel(hass, data.url) diff --git a/homeassistant/components/thread/dataset_store.py b/homeassistant/components/thread/dataset_store.py index bcadf3ca5fb..a277ceb16a2 100644 --- a/homeassistant/components/thread/dataset_store.py +++ b/homeassistant/components/thread/dataset_store.py @@ -1,13 +1,13 @@ """Persistently store thread datasets.""" from __future__ import annotations -from contextlib import suppress import dataclasses from datetime import datetime from functools import cached_property from typing import Any, cast from python_otbr_api import tlv_parser +from python_otbr_api.tlv_parser import MeshcopTLVType from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import HomeAssistantError @@ -39,31 +39,33 @@ class DatasetEntry: @property def channel(self) -> int | None: """Return channel as an integer.""" - if (channel := self.dataset.get(tlv_parser.MeshcopTLVType.CHANNEL)) is None: + if (channel := self.dataset.get(MeshcopTLVType.CHANNEL)) is None: return None - with suppress(ValueError): - return int(channel, 16) - return None + return cast(tlv_parser.Channel, channel).channel @cached_property - def dataset(self) -> dict[tlv_parser.MeshcopTLVType, str]: + def dataset(self) -> dict[MeshcopTLVType, tlv_parser.MeshcopTLVItem]: """Return the dataset in dict format.""" return tlv_parser.parse_tlv(self.tlv) @property def extended_pan_id(self) -> str | None: """Return extended PAN ID as a hex string.""" - return self.dataset.get(tlv_parser.MeshcopTLVType.EXTPANID) + if (ext_pan_id := self.dataset.get(MeshcopTLVType.EXTPANID)) is None: + return None + return str(ext_pan_id) @property def network_name(self) -> str | None: """Return network name as a string.""" - return self.dataset.get(tlv_parser.MeshcopTLVType.NETWORKNAME) + if (name := self.dataset.get(MeshcopTLVType.NETWORKNAME)) is None: + return None + return cast(tlv_parser.NetworkName, name).name @property def pan_id(self) -> str | None: """Return PAN ID as a hex string.""" - return self.dataset.get(tlv_parser.MeshcopTLVType.PANID) + return str(self.dataset.get(MeshcopTLVType.PANID)) def to_json(self) -> dict[str, Any]: """Return a JSON serializable representation for storage.""" diff --git a/homeassistant/components/thread/diagnostics.py b/homeassistant/components/thread/diagnostics.py index 8dc5dd43041..e6149214af4 100644 --- a/homeassistant/components/thread/diagnostics.py +++ b/homeassistant/components/thread/diagnostics.py @@ -148,7 +148,8 @@ async def async_get_config_entry_diagnostics( "unexpected_routers": set(), }, ) - if mlp := record.dataset.get(MeshcopTLVType.MESHLOCALPREFIX): + if mlp_item := record.dataset.get(MeshcopTLVType.MESHLOCALPREFIX): + mlp = str(mlp_item) network["prefixes"].add(f"{mlp[0:4]}:{mlp[4:8]}:{mlp[8:12]}:{mlp[12:16]}") # Find all routes currently act that might be thread related, so we can match them to diff --git a/homeassistant/components/thread/manifest.json b/homeassistant/components/thread/manifest.json index da142be49aa..9a6a64481cd 100644 --- a/homeassistant/components/thread/manifest.json +++ b/homeassistant/components/thread/manifest.json @@ -7,6 +7,6 @@ "documentation": "https://www.home-assistant.io/integrations/thread", "integration_type": "service", "iot_class": "local_polling", - "requirements": ["python-otbr-api==2.0.0", "pyroute2==0.7.5"], + "requirements": ["python-otbr-api==2.1.0", "pyroute2==0.7.5"], "zeroconf": ["_meshcop._udp.local."] } diff --git a/requirements_all.txt b/requirements_all.txt index a0493a79d35..e224ec3294b 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -2109,7 +2109,7 @@ python-opensky==0.0.7 # homeassistant.components.otbr # homeassistant.components.thread -python-otbr-api==2.0.0 +python-otbr-api==2.1.0 # homeassistant.components.picnic python-picnic-api==1.1.0 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index f64d37ff45f..c526b1b8859 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1532,7 +1532,7 @@ python-nest==4.2.0 # homeassistant.components.otbr # homeassistant.components.thread -python-otbr-api==2.0.0 +python-otbr-api==2.1.0 # homeassistant.components.picnic python-picnic-api==1.1.0 diff --git a/tests/components/otbr/test_init.py b/tests/components/otbr/test_init.py index bf481488a0a..3d646287ce1 100644 --- a/tests/components/otbr/test_init.py +++ b/tests/components/otbr/test_init.py @@ -24,12 +24,6 @@ from . import ( from tests.common import MockConfigEntry from tests.test_util.aiohttp import AiohttpClientMocker -DATASET_BAD_CHANNEL = bytes.fromhex( - "0E080000000000010000000035060004001FFFE00208F642646DA209B1C00708FDF57B5A" - "0FE2AAF60510DE98B5BA1A528FEE049D4B4B01835375030D4F70656E5468726561642048410102" - "25A40410F5DD18371BFD29E1A601EF6FFAD94C030C0402A0F7F8" -) - DATASET_NO_CHANNEL = bytes.fromhex( "0E08000000000001000035060004001FFFE00208F642646DA209B1C00708FDF57B5A" "0FE2AAF60510DE98B5BA1A528FEE049D4B4B01835375030D4F70656E5468726561642048410102" @@ -103,9 +97,7 @@ async def test_import_share_radio_channel_collision(hass: HomeAssistant) -> None ) -@pytest.mark.parametrize( - "dataset", [DATASET_BAD_CHANNEL, DATASET_CH15, DATASET_NO_CHANNEL] -) +@pytest.mark.parametrize("dataset", [DATASET_CH15, DATASET_NO_CHANNEL]) async def test_import_share_radio_no_channel_collision( hass: HomeAssistant, dataset: bytes ) -> None: diff --git a/tests/components/thread/test_dataset_store.py b/tests/components/thread/test_dataset_store.py index 212db0de06f..45efd232ba7 100644 --- a/tests/components/thread/test_dataset_store.py +++ b/tests/components/thread/test_dataset_store.py @@ -121,7 +121,6 @@ async def test_dataset_properties(hass: HomeAssistant) -> None: {"source": "Google", "tlv": DATASET_1}, {"source": "Multipan", "tlv": DATASET_2}, {"source": "🎅", "tlv": DATASET_3}, - {"source": "test1", "tlv": DATASET_1_BAD_CHANNEL}, {"source": "test2", "tlv": DATASET_1_NO_CHANNEL}, ] @@ -136,10 +135,8 @@ async def test_dataset_properties(hass: HomeAssistant) -> None: dataset_2 = dataset if dataset.source == "🎅": dataset_3 = dataset - if dataset.source == "test1": - dataset_4 = dataset if dataset.source == "test2": - dataset_5 = dataset + dataset_4 = dataset dataset = store.async_get(dataset_1.id) assert dataset == dataset_1 @@ -166,10 +163,6 @@ async def test_dataset_properties(hass: HomeAssistant) -> None: assert dataset == dataset_4 assert dataset.channel is None - dataset = store.async_get(dataset_5.id) - assert dataset == dataset_5 - assert dataset.channel is None - async def test_load_datasets(hass: HomeAssistant) -> None: """Make sure that we can load/save data correctly."""