Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00
Code Activity

Disable owning integrations for the entire firmware interaction process (#157082)

Browse Source
This commit is contained in:
puddly
2025-12-01 12:18:32 -05:00
committed by GitHub
parent 758a30eebc
commit e8acced335
7 changed files with 214 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
homeassistant/components/homeassistant_hardware/firmware_config_flow.py
View File

@@ -33,13 +33,14 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.hassio import is_hassio
from .const import OTBR_DOMAIN, Z2M_EMBER_DOCS_URL, ZHA_DOMAIN
from .const import DOMAIN, OTBR_DOMAIN, Z2M_EMBER_DOCS_URL, ZHA_DOMAIN
from .util import (
ApplicationType,
FirmwareInfo,
OwningAddon,
OwningIntegration,
ResetTarget,
async_firmware_flashing_context,
async_flash_silabs_firmware,
get_otbr_addon_manager,
guess_firmware_info,
@@ -228,83 +229,95 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
# Keep track of the firmware we're working with, for error messages
self.installing_firmware_name = firmware_name
# Installing new firmware is only truly required if the wrong type is
# installed: upgrading to the latest release of the current firmware type
# isn't strictly necessary for functionality.
self._probed_firmware_info = await probe_silabs_firmware_info(
self._device,
bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
application_probe_methods=self.APPLICATION_PROBE_METHODS,
)
firmware_install_required = self._probed_firmware_info is None or (
self._probed_firmware_info.firmware_type != expected_installed_firmware_type
)
session = async_get_clientsession(self.hass)
client = FirmwareUpdateClient(fw_update_url, session)
try:
manifest = await client.async_update_data()
fw_manifest = next(
fw for fw in manifest.firmwares if fw.filename.startswith(fw_type)
# For the duration of firmware flashing, hint to other integrations (i.e. ZHA)
# that the hardware is in use and should not be accessed. This is separate from
# locking the serial port itself, since a momentary release of the port may
# still allow for ZHA to reclaim the device.
async with async_firmware_flashing_context(self.hass, self._device, DOMAIN):
# Installing new firmware is only truly required if the wrong type is
# installed: upgrading to the latest release of the current firmware type
# isn't strictly necessary for functionality.
self._probed_firmware_info = await probe_silabs_firmware_info(
self._device,
bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
application_probe_methods=self.APPLICATION_PROBE_METHODS,
)
except (StopIteration, TimeoutError, ClientError, ManifestMissing) as err:
_LOGGER.warning("Failed to fetch firmware update manifest", exc_info=True)
# Not having internet access should not prevent setup
if not firmware_install_required:
_LOGGER.debug("Skipping firmware upgrade due to index download failure")
return
firmware_install_required = self._probed_firmware_info is None or (
self._probed_firmware_info.firmware_type
!= expected_installed_firmware_type
)
raise AbortFlow(
reason="fw_download_failed",
description_placeholders=self._get_translation_placeholders(),
) from err
session = async_get_clientsession(self.hass)
client = FirmwareUpdateClient(fw_update_url, session)
if not firmware_install_required:
assert self._probed_firmware_info is not None
# Make sure we do not downgrade the firmware
fw_metadata = NabuCasaMetadata.from_json(fw_manifest.metadata)
fw_version = fw_metadata.get_public_version()
probed_fw_version = Version(self._probed_firmware_info.firmware_version)
if probed_fw_version >= fw_version:
_LOGGER.debug(
"Not downgrading firmware, installed %s is newer than available %s",
probed_fw_version,
fw_version,
try:
manifest = await client.async_update_data()
fw_manifest = next(
fw for fw in manifest.firmwares if fw.filename.startswith(fw_type)
)
except (StopIteration, TimeoutError, ClientError, ManifestMissing) as err:
_LOGGER.warning(
"Failed to fetch firmware update manifest", exc_info=True
)
return
try:
fw_data = await client.async_fetch_firmware(fw_manifest)
except (TimeoutError, ClientError, ValueError) as err:
_LOGGER.warning("Failed to fetch firmware update", exc_info=True)
# Not having internet access should not prevent setup
if not firmware_install_required:
_LOGGER.debug(
"Skipping firmware upgrade due to index download failure"
)
return
raise AbortFlow(
reason="fw_download_failed",
description_placeholders=self._get_translation_placeholders(),
) from err
# If we cannot download new firmware, we shouldn't block setup
if not firmware_install_required:
_LOGGER.debug("Skipping firmware upgrade due to image download failure")
return
assert self._probed_firmware_info is not None
# Otherwise, fail
raise AbortFlow(
reason="fw_download_failed",
description_placeholders=self._get_translation_placeholders(),
) from err
# Make sure we do not downgrade the firmware
fw_metadata = NabuCasaMetadata.from_json(fw_manifest.metadata)
fw_version = fw_metadata.get_public_version()
probed_fw_version = Version(self._probed_firmware_info.firmware_version)
self._probed_firmware_info = await async_flash_silabs_firmware(
hass=self.hass,
device=self._device,
fw_data=fw_data,
expected_installed_firmware_type=expected_installed_firmware_type,
bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
application_probe_methods=self.APPLICATION_PROBE_METHODS,
progress_callback=lambda offset, total: self.async_update_progress(
offset / total
),
)
if probed_fw_version >= fw_version:
_LOGGER.debug(
"Not downgrading firmware, installed %s is newer than available %s",
probed_fw_version,
fw_version,
)
return
try:
fw_data = await client.async_fetch_firmware(fw_manifest)
except (TimeoutError, ClientError, ValueError) as err:
_LOGGER.warning("Failed to fetch firmware update", exc_info=True)
# If we cannot download new firmware, we shouldn't block setup
if not firmware_install_required:
_LOGGER.debug(
"Skipping firmware upgrade due to image download failure"
)
return
# Otherwise, fail
raise AbortFlow(
reason="fw_download_failed",
description_placeholders=self._get_translation_placeholders(),
) from err
self._probed_firmware_info = await async_flash_silabs_firmware(
hass=self.hass,
device=self._device,
fw_data=fw_data,
expected_installed_firmware_type=expected_installed_firmware_type,
bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
application_probe_methods=self.APPLICATION_PROBE_METHODS,
progress_callback=lambda offset, total: self.async_update_progress(
offset / total
),
)
async def _configure_and_start_otbr_addon(self) -> None:
"""Configure and start the OTBR addon."""

23
homeassistant/components/homeassistant_hardware/update.py
View File

@@ -26,6 +26,7 @@ from .util import (
ApplicationType,
FirmwareInfo,
ResetTarget,
async_firmware_flashing_context,
async_flash_silabs_firmware,
)
@@ -274,16 +275,18 @@ class BaseFirmwareUpdateEntity(
)
try:
firmware_info = await async_flash_silabs_firmware(
hass=self.hass,
device=self._current_device,
fw_data=fw_data,
expected_installed_firmware_type=self.entity_description.expected_firmware_type,
bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
application_probe_methods=self.APPLICATION_PROBE_METHODS,
progress_callback=self._update_progress,
domain=self._config_entry.domain,
)
async with async_firmware_flashing_context(
self.hass, self._current_device, self._config_entry.domain
):
firmware_info = await async_flash_silabs_firmware(
hass=self.hass,
device=self._current_device,
fw_data=fw_data,
expected_installed_firmware_type=self.entity_description.expected_firmware_type,
bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
application_probe_methods=self.APPLICATION_PROBE_METHODS,
progress_callback=self._update_progress,
)
finally:
self._attr_in_progress = False
self.async_write_ha_state()

104
homeassistant/components/homeassistant_hardware/util.py
View File

@@ -26,7 +26,6 @@ from homeassistant.helpers.singleton import singleton
from . import DATA_COMPONENT
from .const import (
DOMAIN,
OTBR_ADDON_MANAGER_DATA,
OTBR_ADDON_NAME,
OTBR_ADDON_SLUG,
@@ -366,6 +365,22 @@ async def probe_silabs_firmware_type(
return fw_info.firmware_type
@asynccontextmanager
async def async_firmware_flashing_context(
hass: HomeAssistant, device: str, source_domain: str
) -> AsyncIterator[None]:
"""Register a device as having its firmware being actively interacted with."""
async with async_firmware_update_context(hass, device, source_domain):
firmware_info = await guess_firmware_info(hass, device)
_LOGGER.debug("Guessed firmware info before update: %s", firmware_info)
async with AsyncExitStack() as stack:
for owner in firmware_info.owners:
await stack.enter_async_context(owner.temporarily_stop(hass))
yield
async def async_flash_silabs_firmware(
hass: HomeAssistant,
device: str,
@@ -374,10 +389,11 @@ async def async_flash_silabs_firmware(
bootloader_reset_methods: Sequence[ResetTarget],
application_probe_methods: Sequence[tuple[ApplicationType, int]],
progress_callback: Callable[[int, int], None] | None = None,
*,
domain: str = DOMAIN,
) -> FirmwareInfo:
"""Flash firmware to the SiLabs device."""
"""Flash firmware to the SiLabs device.
This function is meant to be used within a firmware update context.
"""
if not any(
method == expected_installed_firmware_type
for method, _ in application_probe_methods
@@ -387,54 +403,44 @@ async def async_flash_silabs_firmware(
f" not in application probe methods {application_probe_methods!r}"
)
async with async_firmware_update_context(hass, device, domain):
firmware_info = await guess_firmware_info(hass, device)
_LOGGER.debug("Identified firmware info: %s", firmware_info)
fw_image = await hass.async_add_executor_job(parse_firmware_image, fw_data)
fw_image = await hass.async_add_executor_job(parse_firmware_image, fw_data)
flasher = Flasher(
device=device,
probe_methods=tuple(
(m.as_flasher_application_type(), baudrate)
for m, baudrate in application_probe_methods
),
bootloader_reset=tuple(
m.as_flasher_reset_target() for m in bootloader_reset_methods
),
)
flasher = Flasher(
device=device,
probe_methods=tuple(
(m.as_flasher_application_type(), baudrate)
for m, baudrate in application_probe_methods
),
bootloader_reset=tuple(
m.as_flasher_reset_target() for m in bootloader_reset_methods
),
)
try:
# Enter the bootloader with indeterminate progress
await flasher.enter_bootloader()
async with AsyncExitStack() as stack:
for owner in firmware_info.owners:
await stack.enter_async_context(owner.temporarily_stop(hass))
# Flash the firmware, with progress
await flasher.flash_firmware(fw_image, progress_callback=progress_callback)
except PermissionError as err:
raise HomeAssistantError(
"Failed to flash firmware: Device is used by another application"
) from err
except Exception as err:
raise HomeAssistantError("Failed to flash firmware") from err
try:
# Enter the bootloader with indeterminate progress
await flasher.enter_bootloader()
probed_firmware_info = await probe_silabs_firmware_info(
device,
bootloader_reset_methods=bootloader_reset_methods,
# Only probe for the expected installed firmware type
application_probe_methods=[
(method, baudrate)
for method, baudrate in application_probe_methods
if method == expected_installed_firmware_type
],
)
# Flash the firmware, with progress
await flasher.flash_firmware(
fw_image, progress_callback=progress_callback
)
except PermissionError as err:
raise HomeAssistantError(
"Failed to flash firmware: Device is used by another application"
) from err
except Exception as err:
raise HomeAssistantError("Failed to flash firmware") from err
if probed_firmware_info is None:
raise HomeAssistantError("Failed to probe the firmware after flashing")
probed_firmware_info = await probe_silabs_firmware_info(
device,
bootloader_reset_methods=bootloader_reset_methods,
# Only probe for the expected installed firmware type
application_probe_methods=[
(method, baudrate)
for method, baudrate in application_probe_methods
if method == expected_installed_firmware_type
],
)
if probed_firmware_info is None:
raise HomeAssistantError("Failed to probe the firmware after flashing")
return probed_firmware_info
return probed_firmware_info

36
tests/components/homeassistant_hardware/test_config_flow.py
View File

@@ -23,9 +23,6 @@ from homeassistant.components.homeassistant_hardware.firmware_config_flow import
BaseFirmwareConfigFlow,
BaseFirmwareOptionsFlow,
)
from homeassistant.components.homeassistant_hardware.helpers import (
async_firmware_update_context,
)
from homeassistant.components.homeassistant_hardware.util import (
ApplicationType,
FirmwareInfo,
@@ -204,11 +201,6 @@ async def mock_test_firmware_platform(
yield
@pytest.fixture(autouse=True)
async def fixture_mock_supervisor_client(supervisor_client: AsyncMock):
"""Mock supervisor client in tests."""
def delayed_side_effect() -> Callable[..., Awaitable[None]]:
"""Slows down eager tasks by delaying for an event loop tick."""
@@ -307,21 +299,18 @@ def mock_firmware_info(
bootloader_reset_methods: Sequence[ResetTarget] = (),
application_probe_methods: Sequence[tuple[ApplicationType, int]] = (),
progress_callback: Callable[[int, int], None] | None = None,
*,
domain: str = "homeassistant_hardware",
) -> FirmwareInfo:
async with async_firmware_update_context(hass, device, domain):
await asyncio.sleep(0)
progress_callback(0, 100)
await asyncio.sleep(0)
progress_callback(50, 100)
await asyncio.sleep(0)
progress_callback(100, 100)
await asyncio.sleep(0)
progress_callback(0, 100)
await asyncio.sleep(0)
progress_callback(50, 100)
await asyncio.sleep(0)
progress_callback(100, 100)
if flashed_firmware_info is None:
raise HomeAssistantError("Failed to probe the firmware after flashing")
if flashed_firmware_info is None:
raise HomeAssistantError("Failed to probe the firmware after flashing")
return flashed_firmware_info
return flashed_firmware_info
with (
patch(
@@ -373,6 +362,7 @@ async def consume_progress_flow(
return result
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_zigbee_recommended(hass: HomeAssistant) -> None:
"""Test flow with recommended Zigbee installation type."""
init_result = await hass.config_entries.flow.async_init(
@@ -447,6 +437,7 @@ async def test_config_flow_zigbee_recommended(hass: HomeAssistant) -> None:
}
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_zigbee_custom_zha(hass: HomeAssistant) -> None:
"""Test flow with custom Zigbee installation type and ZHA selected."""
init_result = await hass.config_entries.flow.async_init(
@@ -539,6 +530,7 @@ async def test_config_flow_zigbee_custom_zha(hass: HomeAssistant) -> None:
}
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_zigbee_custom_other(hass: HomeAssistant) -> None:
"""Test flow with custom Zigbee installation type and Other selected."""
init_result = await hass.config_entries.flow.async_init(
@@ -611,6 +603,7 @@ async def test_config_flow_zigbee_custom_other(hass: HomeAssistant) -> None:
assert flows == []
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_firmware_index_download_fails_but_not_required(
hass: HomeAssistant,
) -> None:
@@ -647,6 +640,7 @@ async def test_config_flow_firmware_index_download_fails_but_not_required(
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_firmware_download_fails_but_not_required(
hass: HomeAssistant,
) -> None:
@@ -682,6 +676,7 @@ async def test_config_flow_firmware_download_fails_but_not_required(
assert result["type"] is FlowResultType.CREATE_ENTRY
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_doesnt_downgrade(
hass: HomeAssistant,
) -> None:
@@ -720,6 +715,7 @@ async def test_config_flow_doesnt_downgrade(
assert len(mock_async_flash_silabs_firmware.mock_calls) == 0
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_zigbee_skip_step_if_installed(hass: HomeAssistant) -> None:
"""Test skip installing the firmware if not needed."""
result = await hass.config_entries.flow.async_init(

7
tests/components/homeassistant_hardware/test_config_flow_failures.py
View File

@@ -31,11 +31,6 @@ from .test_config_flow import (
from tests.common import MockConfigEntry
@pytest.fixture(autouse=True)
async def fixture_mock_supervisor_client(supervisor_client: AsyncMock):
"""Mock supervisor client in tests."""
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
@@ -312,6 +307,7 @@ async def test_config_flow_thread_confirmation_fails(hass: HomeAssistant) -> Non
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains", ["test_firmware_domain"]
)
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_firmware_index_download_fails_and_required(
hass: HomeAssistant,
) -> None:
@@ -354,6 +350,7 @@ async def test_config_flow_firmware_index_download_fails_and_required(
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains", ["test_firmware_domain"]
)
@pytest.mark.usefixtures("addon_store_info", "addon_info")
async def test_config_flow_firmware_download_fails_and_required(
hass: HomeAssistant,
) -> None:

2
tests/components/homeassistant_hardware/test_update.py
View File

@@ -327,8 +327,6 @@ async def test_update_entity_installation(
bootloader_reset_methods: Sequence[ResetTarget] = (),
application_probe_methods: Sequence[tuple[ApplicationType, int]] = (),
progress_callback: Callable[[int, int], None] | None = None,
*,
domain: str = "homeassistant_hardware",
) -> FirmwareInfo:
await asyncio.sleep(0)
progress_callback(0, 100)

84
tests/components/homeassistant_hardware/test_util.py
View File

@@ -24,6 +24,7 @@ from homeassistant.components.homeassistant_hardware.util import (
OwningAddon,
OwningIntegration,
ResetTarget,
async_firmware_flashing_context,
async_flash_silabs_firmware,
get_otbr_addon_firmware_info,
guess_firmware_info,
@@ -606,18 +607,21 @@ async def test_async_flash_silabs_firmware(hass: HomeAssistant) -> None:
return_value=expected_firmware_info,
),
):
after_flash_info = await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_methods=[ResetTarget.RTS_DTR],
application_probe_methods=[
(ApplicationType.EZSP, 460800),
(ApplicationType.SPINEL, 460800),
],
progress_callback=progress_callback,
)
async with async_firmware_flashing_context(
hass, "/dev/ttyUSB0", "homeassistant_hardware"
):
after_flash_info = await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_methods=[ResetTarget.RTS_DTR],
application_probe_methods=[
(ApplicationType.EZSP, 460800),
(ApplicationType.SPINEL, 460800),
],
progress_callback=progress_callback,
)
assert progress_callback.mock_calls == [call(0, 100), call(50, 100), call(100, 100)]
assert after_flash_info == expected_firmware_info
@@ -712,17 +716,20 @@ async def test_async_flash_silabs_firmware_flash_failure(
),
pytest.raises(HomeAssistantError, match=expected_error_msg) as exc,
):
await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_methods=[ResetTarget.RTS_DTR],
application_probe_methods=[
(ApplicationType.EZSP, 460800),
(ApplicationType.SPINEL, 460800),
],
)
async with async_firmware_flashing_context(
hass, "/dev/ttyUSB0", "homeassistant_hardware"
):
await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_methods=[ResetTarget.RTS_DTR],
application_probe_methods=[
(ApplicationType.EZSP, 460800),
(ApplicationType.SPINEL, 460800),
],
)
# Both owning integrations/addons are stopped and restarted
assert owner1.temporarily_stop.mock_calls == [
@@ -774,30 +781,33 @@ async def test_async_flash_silabs_firmware_probe_failure(hass: HomeAssistant) ->
),
pytest.raises(
HomeAssistantError, match="Failed to probe the firmware after flashing"
),
) as exc,
):
await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_methods=[ResetTarget.RTS_DTR],
application_probe_methods=[
(ApplicationType.EZSP, 460800),
(ApplicationType.SPINEL, 460800),
],
)
async with async_firmware_flashing_context(
hass, "/dev/ttyUSB0", "homeassistant_hardware"
):
await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_methods=[ResetTarget.RTS_DTR],
application_probe_methods=[
(ApplicationType.EZSP, 460800),
(ApplicationType.SPINEL, 460800),
],
)
# Both owning integrations/addons are stopped and restarted
assert owner1.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
call().__aexit__(ANY, None, None, None),
call().__aexit__(ANY, HomeAssistantError, exc.value, ANY),
]
assert owner2.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
call().__aexit__(ANY, None, None, None),
call().__aexit__(ANY, HomeAssistantError, exc.value, ANY),
]