add enphase_envoy interface mac to device registry (#143758)

* add enphase_envoy interface mac to device registry

* Test for capitalized error log entry.

* increase mac collection delay from 17 to 34 sec
This commit is contained in:
Arie Catsman 2025-04-28 11:20:11 +02:00 committed by GitHub
parent 84f07ee992
commit d1236a53b8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 686 additions and 3 deletions

View File

@ -9,12 +9,14 @@ import logging
from typing import Any
from pyenphase import Envoy, EnvoyError, EnvoyTokenAuth
from pyenphase.models.home import EnvoyInterfaceInformation
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_TOKEN, CONF_USERNAME
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.event import async_call_later, async_track_time_interval
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
@ -26,7 +28,7 @@ TOKEN_REFRESH_CHECK_INTERVAL = timedelta(days=1)
STALE_TOKEN_THRESHOLD = timedelta(days=30).total_seconds()
NOTIFICATION_ID = "enphase_envoy_notification"
FIRMWARE_REFRESH_INTERVAL = timedelta(hours=4)
MAC_VERIFICATION_DELAY = timedelta(seconds=34)
_LOGGER = logging.getLogger(__name__)
@ -39,6 +41,7 @@ class EnphaseUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
envoy_serial_number: str
envoy_firmware: str
config_entry: EnphaseConfigEntry
interface: EnvoyInterfaceInformation | None
def __init__(
self, hass: HomeAssistant, envoy: Envoy, entry: EnphaseConfigEntry
@ -50,8 +53,10 @@ class EnphaseUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self.password = entry_data[CONF_PASSWORD]
self._setup_complete = False
self.envoy_firmware = ""
self.interface = None
self._cancel_token_refresh: CALLBACK_TYPE | None = None
self._cancel_firmware_refresh: CALLBACK_TYPE | None = None
self._cancel_mac_verification: CALLBACK_TYPE | None = None
super().__init__(
hass,
_LOGGER,
@ -121,6 +126,66 @@ class EnphaseUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
self.hass.config_entries.async_reload(self.config_entry.entry_id)
)
def _schedule_mac_verification(
self, delay: timedelta = MAC_VERIFICATION_DELAY
) -> None:
"""Schedule one time job to verify envoy mac address."""
self.async_cancel_mac_verification()
self._cancel_mac_verification = async_call_later(
self.hass,
delay,
self._async_verify_mac,
)
@callback
def _async_verify_mac(self, now: datetime.datetime) -> None:
"""Verify Envoy active interface mac address in background."""
self.hass.async_create_background_task(
self._async_fetch_and_compare_mac(), "{name} verify envoy mac address"
)
async def _async_fetch_and_compare_mac(self) -> None:
"""Get Envoy interface information and update mac in device connections."""
interface: (
EnvoyInterfaceInformation | None
) = await self.envoy.interface_settings()
if interface is None:
_LOGGER.debug("%s: interface information returned None", self.name)
return
# remember interface information so diagnostics can include in report
self.interface = interface
# Add to or update device registry connections as needed
device_registry = dr.async_get(self.hass)
envoy_device = device_registry.async_get_device(
identifiers={
(
DOMAIN,
self.envoy_serial_number,
)
}
)
if envoy_device is None:
_LOGGER.error(
"No envoy device found in device registry: %s %s",
DOMAIN,
self.envoy_serial_number,
)
return
connection = (dr.CONNECTION_NETWORK_MAC, interface.mac)
if connection in envoy_device.connections:
_LOGGER.debug(
"connection verified as existing: %s in %s", connection, self.name
)
return
device_registry.async_update_device(
device_id=envoy_device.id,
new_connections={connection},
)
_LOGGER.debug("added connection: %s to %s", connection, self.name)
@callback
def _async_mark_setup_complete(self) -> None:
"""Mark setup as complete and setup firmware checks and token refresh if needed."""
@ -132,6 +197,7 @@ class EnphaseUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
FIRMWARE_REFRESH_INTERVAL,
cancel_on_shutdown=True,
)
self._schedule_mac_verification()
self.async_cancel_token_refresh()
if not isinstance(self.envoy.auth, EnvoyTokenAuth):
return
@ -252,3 +318,10 @@ class EnphaseUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
if self._cancel_firmware_refresh:
self._cancel_firmware_refresh()
self._cancel_firmware_refresh = None
@callback
def async_cancel_mac_verification(self) -> None:
"""Cancel mac verification."""
if self._cancel_mac_verification:
self._cancel_mac_verification()
self._cancel_mac_verification = None

View File

@ -3,6 +3,7 @@
from __future__ import annotations
import copy
from datetime import datetime
from typing import TYPE_CHECKING, Any
from attr import asdict
@ -63,6 +64,7 @@ async def _get_fixture_collection(envoy: Envoy, serial: str) -> dict[str, Any]:
"/ivp/ensemble/generator",
"/ivp/meters",
"/ivp/meters/readings",
"/home,",
]
for end_point in end_points:
@ -146,11 +148,25 @@ async def async_get_config_entry_diagnostics(
"inverters": envoy_data.inverters,
"tariff": envoy_data.tariff,
}
# Add Envoy active interface information to report
active_interface: dict[str, Any] = {}
if coordinator.interface:
active_interface = {
"name": (interface := coordinator.interface).primary_interface,
"interface type": interface.interface_type,
"mac": interface.mac,
"uses dhcp": interface.dhcp,
"firmware build date": datetime.fromtimestamp(
interface.software_build_epoch
).strftime("%Y-%m-%d %H:%M:%S"),
"envoy timezone": interface.timezone,
}
envoy_properties: dict[str, Any] = {
"envoy_firmware": envoy.firmware,
"part_number": envoy.part_number,
"envoy_model": envoy.envoy_model,
"active interface": active_interface,
"supported_features": [feature.name for feature in envoy.supported_features],
"phase_mode": envoy.phase_mode,
"phase_count": envoy.phase_count,

View File

@ -20,6 +20,7 @@ from pyenphase import (
)
from pyenphase.const import SupportedFeatures
from pyenphase.models.dry_contacts import EnvoyDryContactSettings, EnvoyDryContactStatus
from pyenphase.models.home import EnvoyInterfaceInformation
from pyenphase.models.meters import EnvoyMeterData
from pyenphase.models.tariff import EnvoyStorageSettings, EnvoyTariff
import pytest
@ -145,6 +146,11 @@ def load_envoy_fixture(mock_envoy: AsyncMock, fixture_name: str) -> None:
_load_json_2_encharge_enpower_data(mock_envoy.data, json_fixture)
_load_json_2_raw_data(mock_envoy.data, json_fixture)
if item := json_fixture.get("interface_information"):
mock_envoy.interface_settings.return_value = EnvoyInterfaceInformation(**item)
else:
mock_envoy.interface_settings.return_value = None
def _load_json_2_production_data(
mocked_data: EnvoyData, json_fixture: dict[str, Any]

View File

@ -47,5 +47,13 @@
"raw": {
"varies_by": "firmware_version"
}
},
"interface_information": {
"primary_interface": "eth0",
"interface_type": "ethernet",
"mac": "00:11:22:33:44:55",
"dhcp": true,
"software_build_epoch": 1719503966,
"timezone": "Europe/Amsterdam"
}
}

View File

@ -423,6 +423,8 @@
'tariff': None,
}),
'envoy_properties': dict({
'active interface': dict({
}),
'active_phasecount': 0,
'ct_consumption_meter': None,
'ct_count': 0,
@ -870,6 +872,8 @@
'tariff': None,
}),
'envoy_properties': dict({
'active interface': dict({
}),
'active_phasecount': 0,
'ct_consumption_meter': None,
'ct_count': 0,
@ -892,6 +896,8 @@
'/api/v1/production/inverters': 'Testing request replies.',
'/api/v1/production/inverters_log': '{"headers":{"Hello":"World"},"code":200}',
'/api/v1/production_log': '{"headers":{"Hello":"World"},"code":200}',
'/home,': 'Testing request replies.',
'/home,_log': '{"headers":{"Hello":"World"},"code":200}',
'/info': 'Testing request replies.',
'/info_log': '{"headers":{"Hello":"World"},"code":200}',
'/ivp/ensemble/dry_contacts': 'Testing request replies.',
@ -1357,6 +1363,8 @@
'tariff': None,
}),
'envoy_properties': dict({
'active interface': dict({
}),
'active_phasecount': 0,
'ct_consumption_meter': None,
'ct_count': 0,
@ -1382,6 +1390,9 @@
'/api/v1/production_log': dict({
'Error': "EnvoyError('Test')",
}),
'/home,_log': dict({
'Error': "EnvoyError('Test')",
}),
'/info_log': dict({
'Error': "EnvoyError('Test')",
}),
@ -1439,3 +1450,461 @@
}),
})
# ---
# name: test_entry_diagnostics_with_interface_information
dict({
'config_entry': dict({
'data': dict({
'host': '1.1.1.1',
'name': '**REDACTED**',
'password': '**REDACTED**',
'token': '**REDACTED**',
'username': '**REDACTED**',
}),
'disabled_by': None,
'discovery_keys': dict({
}),
'domain': 'enphase_envoy',
'entry_id': '45a36e55aaddb2007c5f6602e0c38e72',
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'subentries': list([
]),
'title': '**REDACTED**',
'unique_id': '**REDACTED**',
'version': 1,
}),
'envoy_entities_by_device': list([
dict({
'device': dict({
'area_id': None,
'config_entries': list([
'45a36e55aaddb2007c5f6602e0c38e72',
]),
'config_entries_subentries': dict({
'45a36e55aaddb2007c5f6602e0c38e72': list([
None,
]),
}),
'configuration_url': None,
'connections': list([
]),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'identifiers': list([
list([
'enphase_envoy',
'1',
]),
]),
'is_new': False,
'labels': list([
]),
'manufacturer': 'Enphase',
'model': 'Inverter',
'model_id': None,
'name': 'Inverter 1',
'name_by_user': None,
'primary_config_entry': '45a36e55aaddb2007c5f6602e0c38e72',
'serial_number': None,
'suggested_area': None,
'sw_version': None,
}),
'entities': list([
dict({
'entity': dict({
'aliases': list([
]),
'area_id': None,
'capabilities': dict({
'state_class': 'measurement',
}),
'categories': dict({
}),
'config_entry_id': '45a36e55aaddb2007c5f6602e0c38e72',
'config_subentry_id': None,
'device_class': None,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.inverter_1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'labels': list([
]),
'name': None,
'options': dict({
}),
'original_device_class': 'power',
'original_icon': None,
'original_name': None,
'platform': 'enphase_envoy',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '1',
'unit_of_measurement': 'W',
}),
'state': dict({
'attributes': dict({
'device_class': 'power',
'friendly_name': 'Inverter 1',
'state_class': 'measurement',
'unit_of_measurement': 'W',
}),
'entity_id': 'sensor.inverter_1',
'state': '1',
}),
}),
dict({
'entity': dict({
'aliases': list([
]),
'area_id': None,
'capabilities': None,
'categories': dict({
}),
'config_entry_id': '45a36e55aaddb2007c5f6602e0c38e72',
'config_subentry_id': None,
'device_class': None,
'disabled_by': 'integration',
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.inverter_1_last_reported',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'labels': list([
]),
'name': None,
'options': dict({
}),
'original_device_class': 'timestamp',
'original_icon': None,
'original_name': 'Last reported',
'platform': 'enphase_envoy',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'last_reported',
'unique_id': '1_last_reported',
'unit_of_measurement': None,
}),
'state': None,
}),
]),
}),
dict({
'device': dict({
'area_id': None,
'config_entries': list([
'45a36e55aaddb2007c5f6602e0c38e72',
]),
'config_entries_subentries': dict({
'45a36e55aaddb2007c5f6602e0c38e72': list([
None,
]),
}),
'configuration_url': None,
'connections': list([
list([
'mac',
'00:11:22:33:44:55',
]),
]),
'disabled_by': None,
'entry_type': None,
'hw_version': '<<envoyserial>>56789',
'identifiers': list([
list([
'enphase_envoy',
'<<envoyserial>>',
]),
]),
'is_new': False,
'labels': list([
]),
'manufacturer': 'Enphase',
'model': 'Envoy',
'model_id': None,
'name': 'Envoy <<envoyserial>>',
'name_by_user': None,
'primary_config_entry': '45a36e55aaddb2007c5f6602e0c38e72',
'serial_number': '<<envoyserial>>',
'suggested_area': None,
'sw_version': '7.6.175',
}),
'entities': list([
dict({
'entity': dict({
'aliases': list([
]),
'area_id': None,
'capabilities': dict({
'state_class': 'measurement',
}),
'categories': dict({
}),
'config_entry_id': '45a36e55aaddb2007c5f6602e0c38e72',
'config_subentry_id': None,
'device_class': None,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.envoy_<<envoyserial>>_current_power_production',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'labels': list([
]),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 3,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': 'kW',
}),
}),
'original_device_class': 'power',
'original_icon': None,
'original_name': 'Current power production',
'platform': 'enphase_envoy',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'current_power_production',
'unique_id': '<<envoyserial>>_production',
'unit_of_measurement': 'kW',
}),
'state': dict({
'attributes': dict({
'device_class': 'power',
'friendly_name': 'Envoy <<envoyserial>> Current power production',
'state_class': 'measurement',
'unit_of_measurement': 'kW',
}),
'entity_id': 'sensor.envoy_<<envoyserial>>_current_power_production',
'state': '1.234',
}),
}),
dict({
'entity': dict({
'aliases': list([
]),
'area_id': None,
'capabilities': dict({
'state_class': 'total_increasing',
}),
'categories': dict({
}),
'config_entry_id': '45a36e55aaddb2007c5f6602e0c38e72',
'config_subentry_id': None,
'device_class': None,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.envoy_<<envoyserial>>_energy_production_today',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'labels': list([
]),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 2,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': 'kWh',
}),
}),
'original_device_class': 'energy',
'original_icon': None,
'original_name': 'Energy production today',
'platform': 'enphase_envoy',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'daily_production',
'unique_id': '<<envoyserial>>_daily_production',
'unit_of_measurement': 'kWh',
}),
'state': dict({
'attributes': dict({
'device_class': 'energy',
'friendly_name': 'Envoy <<envoyserial>> Energy production today',
'state_class': 'total_increasing',
'unit_of_measurement': 'kWh',
}),
'entity_id': 'sensor.envoy_<<envoyserial>>_energy_production_today',
'state': '1.234',
}),
}),
dict({
'entity': dict({
'aliases': list([
]),
'area_id': None,
'capabilities': None,
'categories': dict({
}),
'config_entry_id': '45a36e55aaddb2007c5f6602e0c38e72',
'config_subentry_id': None,
'device_class': None,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.envoy_<<envoyserial>>_energy_production_last_seven_days',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'labels': list([
]),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': 'kWh',
}),
}),
'original_device_class': 'energy',
'original_icon': None,
'original_name': 'Energy production last seven days',
'platform': 'enphase_envoy',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'seven_days_production',
'unique_id': '<<envoyserial>>_seven_days_production',
'unit_of_measurement': 'kWh',
}),
'state': dict({
'attributes': dict({
'device_class': 'energy',
'friendly_name': 'Envoy <<envoyserial>> Energy production last seven days',
'unit_of_measurement': 'kWh',
}),
'entity_id': 'sensor.envoy_<<envoyserial>>_energy_production_last_seven_days',
'state': '1.234',
}),
}),
dict({
'entity': dict({
'aliases': list([
]),
'area_id': None,
'capabilities': dict({
'state_class': 'total_increasing',
}),
'categories': dict({
}),
'config_entry_id': '45a36e55aaddb2007c5f6602e0c38e72',
'config_subentry_id': None,
'device_class': None,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.envoy_<<envoyserial>>_lifetime_energy_production',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'labels': list([
]),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 3,
}),
'sensor.private': dict({
'suggested_unit_of_measurement': 'MWh',
}),
}),
'original_device_class': 'energy',
'original_icon': None,
'original_name': 'Lifetime energy production',
'platform': 'enphase_envoy',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'lifetime_production',
'unique_id': '<<envoyserial>>_lifetime_production',
'unit_of_measurement': 'MWh',
}),
'state': dict({
'attributes': dict({
'device_class': 'energy',
'friendly_name': 'Envoy <<envoyserial>> Lifetime energy production',
'state_class': 'total_increasing',
'unit_of_measurement': 'MWh',
}),
'entity_id': 'sensor.envoy_<<envoyserial>>_lifetime_energy_production',
'state': '0.00<<envoyserial>>',
}),
}),
]),
}),
]),
'envoy_model_data': dict({
'ctmeter_consumption': None,
'ctmeter_consumption_phases': None,
'ctmeter_production': None,
'ctmeter_production_phases': None,
'ctmeter_storage': None,
'ctmeter_storage_phases': None,
'dry_contact_settings': dict({
}),
'dry_contact_status': dict({
}),
'encharge_aggregate': None,
'encharge_inventory': None,
'encharge_power': None,
'enpower': None,
'inverters': dict({
'1': dict({
'__type': "<class 'pyenphase.models.inverter.EnvoyInverter'>",
'repr': "EnvoyInverter(serial_number='1', last_report_date=1, last_report_watts=1, max_report_watts=1)",
}),
}),
'system_consumption': None,
'system_consumption_phases': None,
'system_production': dict({
'__type': "<class 'pyenphase.models.system_production.EnvoySystemProduction'>",
'repr': 'EnvoySystemProduction(watt_hours_lifetime=1234, watt_hours_last_7_days=1234, watt_hours_today=1234, watts_now=1234)',
}),
'system_production_phases': None,
'tariff': None,
}),
'envoy_properties': dict({
'active interface': dict({
'envoy timezone': 'Europe/Amsterdam',
'firmware build date': '2024-06-27 15:59:26',
'interface type': 'ethernet',
'mac': '00:11:22:33:44:55',
'name': 'eth0',
'uses dhcp': True,
}),
'active_phasecount': 0,
'ct_consumption_meter': None,
'ct_count': 0,
'ct_production_meter': None,
'ct_storage_meter': None,
'envoy_firmware': '7.6.175',
'envoy_model': 'Envoy',
'part_number': '123456789',
'phase_count': 1,
'phase_mode': None,
'supported_features': list([
'INVERTERS',
'PRODUCTION',
]),
}),
'fixtures': dict({
}),
'raw_data': dict({
'varies_by': 'firmware_version',
}),
})
# ---

View File

@ -2,6 +2,7 @@
from unittest.mock import AsyncMock
from freezegun.api import FrozenDateTimeFactory
from pyenphase.exceptions import EnvoyError
import pytest
from syrupy.assertion import SnapshotAssertion
@ -10,11 +11,12 @@ from homeassistant.components.enphase_envoy.const import (
DOMAIN,
OPTION_DIAGNOSTICS_INCLUDE_FIXTURES,
)
from homeassistant.components.enphase_envoy.coordinator import MAC_VERIFICATION_DELAY
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.diagnostics import get_diagnostics_for_config_entry
from tests.typing import ClientSessionGenerator
@ -90,3 +92,24 @@ async def test_entry_diagnostics_with_fixtures_with_error(
assert await get_diagnostics_for_config_entry(
hass, hass_client, config_entry_options
) == snapshot(exclude=limit_diagnostic_attrs)
async def test_entry_diagnostics_with_interface_information(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
mock_envoy: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test config entry diagnostics including interface data."""
await setup_integration(hass, config_entry)
# move time forward so interface information is collected
freezer.tick(MAC_VERIFICATION_DELAY)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert await get_diagnostics_for_config_entry(
hass, hass_client, config_entry
) == snapshot(exclude=limit_diagnostic_attrs)

View File

@ -19,6 +19,7 @@ from homeassistant.components.enphase_envoy.const import (
)
from homeassistant.components.enphase_envoy.coordinator import (
FIRMWARE_REFRESH_INTERVAL,
MAC_VERIFICATION_DELAY,
SCAN_INTERVAL,
)
from homeassistant.config_entries import ConfigEntryState
@ -443,3 +444,90 @@ async def test_coordinator_firmware_refresh_with_envoy_error(
await hass.async_block_till_done(wait_background_tasks=True)
assert "Error reading firmware:" in caplog.text
@respx.mock
async def test_coordinator_interface_information(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_envoy: AsyncMock,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test coordinator interface mac verification."""
await setup_integration(hass, config_entry)
caplog.set_level(logging.DEBUG)
logging.getLogger("homeassistant.components.enphase_envoy.coordinator").setLevel(
logging.DEBUG
)
# move time forward so interface information is fetched
freezer.tick(MAC_VERIFICATION_DELAY)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
# verify first time add of mac to connections is in log
assert "added connection" in caplog.text
# trigger integration reload by changing options
hass.config_entries.async_update_entry(
config_entry,
options={
OPTION_DIAGNOSTICS_INCLUDE_FIXTURES: False,
OPTION_DISABLE_KEEP_ALIVE: True,
},
)
await hass.async_block_till_done(wait_background_tasks=True)
assert config_entry.state is ConfigEntryState.LOADED
caplog.clear()
# envoy reloaded and device registry still has connection info
# force mac verification again to test existing connection is verified
freezer.tick(MAC_VERIFICATION_DELAY)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
# verify existing connection is verified in log
assert "connection verified as existing" in caplog.text
@respx.mock
async def test_coordinator_interface_information_no_device(
hass: HomeAssistant,
config_entry: MockConfigEntry,
mock_envoy: AsyncMock,
freezer: FrozenDateTimeFactory,
caplog: pytest.LogCaptureFixture,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test coordinator interface mac verification full code cov."""
await setup_integration(hass, config_entry)
caplog.set_level(logging.DEBUG)
logging.getLogger("homeassistant.components.enphase_envoy.coordinator").setLevel(
logging.DEBUG
)
# update device to force no device found in mac verification
device_registry = dr.async_get(hass)
envoy_device = device_registry.async_get_device(
identifiers={
(
DOMAIN,
mock_envoy.serial_number,
)
}
)
device_registry.async_update_device(
device_id=envoy_device.id,
new_identifiers={(DOMAIN, "9999")},
)
# move time forward so interface information is fetched
freezer.tick(MAC_VERIFICATION_DELAY)
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
# verify no device found message in log
assert "No envoy device found in device registry" in caplog.text