Compare commits

...

5 Commits

Author SHA1 Message Date
Erik 0674179b90 Add entity option to associate scanner tracker with any zone 2026-05-25 16:33:11 +02:00
Erik 44a5f0fa76 Improve test 2026-05-25 08:56:55 +02:00
Erik 7fbfaf06ed Adjust type annotations 2026-05-25 08:46:56 +02:00
Erik c8bd5ab97d Update test snapshots 2026-05-25 08:44:17 +02:00
Erik 3e0307fd88 Add state attribute in_zones to BaseScannerEntity 2026-05-22 11:01:06 +02:00
12 changed files with 883 additions and 9 deletions
@@ -31,6 +31,7 @@ from .const import ( # noqa: F401
ATTR_LOCATION_NAME,
ATTR_MAC,
ATTR_SOURCE_TYPE,
CONF_ASSOCIATED_ZONE,
CONF_CONSIDER_HOME,
CONF_NEW_DEVICE_DEFAULTS,
CONF_SCAN_INTERVAL,
@@ -16,8 +16,19 @@ from homeassistant.const import (
STATE_NOT_HOME,
EntityCategory,
)
from homeassistant.core import Event, HomeAssistant, State, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.core import (
CALLBACK_TYPE,
Event,
EventStateChangedData,
HomeAssistant,
State,
callback,
)
from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.helpers.device_registry import (
DeviceInfo,
EventDeviceRegistryUpdatedData,
@@ -26,6 +37,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.entity import Entity, EntityDescription
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.entity_platform import EntityPlatform
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers.typing import StateType
from homeassistant.util.hass_dict import HassKey
@@ -35,6 +47,7 @@ from .const import (
ATTR_IP,
ATTR_MAC,
ATTR_SOURCE_TYPE,
CONF_ASSOCIATED_ZONE,
CONNECTED_DEVICE_REGISTERED,
DOMAIN,
LOGGER,
@@ -315,20 +328,148 @@ class BaseScannerEntity(BaseTrackerEntity):
addresses being used to identify the device.
"""
_scanner_option_associated_zone: str = zone.ENTITY_ID_HOME
_scanner_option_associated_zone_unsub: CALLBACK_TYPE | None = None
async def async_internal_added_to_hass(self) -> None:
"""Call when the scanner entity is added to hass."""
await super().async_internal_added_to_hass()
if not self.registry_entry:
return
self._async_read_entity_options()
async def async_internal_will_remove_from_hass(self) -> None:
"""Call when the scanner entity is about to be removed from hass."""
if self._scanner_option_associated_zone_unsub is not None:
self._scanner_option_associated_zone_unsub()
self._scanner_option_associated_zone_unsub = None
self._async_clear_associated_zone_issue()
await super().async_internal_will_remove_from_hass()
@callback
def async_registry_entry_updated(self) -> None:
"""Run when the entity registry entry has been updated."""
self._async_read_entity_options()
@callback
def _async_read_entity_options(self) -> None:
"""Read entity options from the entity registry.
Called when the entity registry entry has been updated and before the
scanner entity is added to the state machine.
"""
assert self.registry_entry
if (scanner_options := self.registry_entry.options.get(DOMAIN)) and (
associated_zone := scanner_options.get(CONF_ASSOCIATED_ZONE)
):
new_zone = associated_zone
else:
new_zone = zone.ENTITY_ID_HOME
if new_zone == self._scanner_option_associated_zone:
return
# Tear down tracking for the previous zone.
if self._scanner_option_associated_zone_unsub is not None:
self._scanner_option_associated_zone_unsub()
self._scanner_option_associated_zone_unsub = None
self._async_clear_associated_zone_issue()
self._scanner_option_associated_zone = new_zone
# zone.home is always present so no tracking or issue handling needed.
if new_zone == zone.ENTITY_ID_HOME:
return
self._scanner_option_associated_zone_unsub = async_track_state_change_event(
self.hass, new_zone, self._async_associated_zone_state_changed
)
if self.hass.states.get(new_zone) is None:
self._async_create_associated_zone_issue()
@callback
def _async_associated_zone_state_changed(
self, event: Event[EventStateChangedData]
) -> None:
"""Open or clear the repair issue when the associated zone appears or disappears."""
if event.data["new_state"] is None:
self._async_create_associated_zone_issue()
else:
self._async_clear_associated_zone_issue()
self.async_write_ha_state()
@callback
def _async_create_associated_zone_issue(self) -> None:
"""Create a repair issue prompting the user to reconfigure the scanner."""
ir.async_create_issue(
self.hass,
DOMAIN,
self._associated_zone_issue_id,
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="associated_zone_missing",
translation_placeholders={
"entity_id": self.entity_id,
"zone": self._scanner_option_associated_zone,
},
)
@callback
def _async_clear_associated_zone_issue(self) -> None:
"""Clear the associated-zone-missing repair issue if it exists."""
ir.async_delete_issue(self.hass, DOMAIN, self._associated_zone_issue_id)
@property
def _associated_zone_issue_id(self) -> str:
"""Return the issue id for the associated-zone-missing repair."""
return f"associated_zone_missing_{self.entity_id}"
@property
def state(self) -> str | None:
"""Return the state of the device."""
if self.is_connected is None:
return None
if self.is_connected:
if not self.is_connected:
return STATE_NOT_HOME
associated_zone = self._scanner_option_associated_zone
if associated_zone == zone.ENTITY_ID_HOME:
return STATE_HOME
return STATE_NOT_HOME
if zone_state := self.hass.states.get(associated_zone):
return zone_state.name
# Configured zone has been removed; state is unknown.
return None
@property
def is_connected(self) -> bool | None:
"""Return true if the device is connected."""
raise NotImplementedError
@final
@property
def state_attributes(self) -> dict[str, Any]:
"""Return the device state attributes."""
attr: dict[str, Any] = {ATTR_IN_ZONES: []}
attr.update(super().state_attributes)
if not self.is_connected:
return attr
associated_zone = self._scanner_option_associated_zone
# If the configured zone has been removed, in_zones stays empty so the
# attribute does not claim membership in a zone that no longer exists.
if (
associated_zone != zone.ENTITY_ID_HOME
and self.hass.states.get(associated_zone) is None
):
return attr
attr[ATTR_IN_ZONES] = [
associated_zone,
*zone.async_get_enclosing_zones(self.hass, associated_zone),
]
return attr
class ScannerEntityDescription(EntityDescription, frozen_or_thawed=True):
"""A class that describes tracker entities."""
@@ -456,9 +597,12 @@ class ScannerEntity(
# Do this last or else the entity registry update listener has been installed
await super().async_internal_added_to_hass()
@final
# BaseScannerEntity.state_attributes is @final to keep external subclasses
# from tampering with it; ScannerEntity is an in-tree subclass that
# intentionally extends it with ip/mac/hostname.
@final # type: ignore[misc]
@property
def state_attributes(self) -> dict[str, StateType]:
def state_attributes(self) -> dict[str, Any]:
"""Return the device state attributes."""
attr = super().state_attributes
@@ -36,6 +36,8 @@ DEFAULT_CONSIDER_HOME: Final = timedelta(seconds=180)
CONF_NEW_DEVICE_DEFAULTS: Final = "new_device_defaults"
CONF_ASSOCIATED_ZONE: Final = "associated_zone"
ATTR_ATTRIBUTES: Final = "attributes"
ATTR_BATTERY: Final = "battery"
ATTR_DEV_ID: Final = "dev_id"
@@ -44,6 +44,12 @@
}
}
},
"issues": {
"associated_zone_missing": {
"description": "The scanner entity `{entity_id}` is associated with the zone `{zone}`, but that zone has been removed.\n\nTo fix this, reconfigure the scanner to use a different zone or recreate the missing zone.",
"title": "Scanner is associated with a removed zone"
}
},
"services": {
"see": {
"description": "Manually update the records of a seen legacy device tracker in the known_devices.yaml file.",
+57
View File
@@ -180,6 +180,63 @@ def async_in_zones(
return (closest, [itm[0] for itm in zones])
def async_get_enclosing_zones(hass: HomeAssistant, zone_entity_id: str) -> list[str]:
"""Find zones which fully contain the given zone.
Returns a list of zone entity_ids whose interior contains the given zone
(``zone_dist + input_radius <= other_zone_radius``); a zone whose edge
touches another zone's edge from the inside counts as contained. Passive
zones are included. The queried zone itself is excluded from the result.
The list is sorted by radius then distance, so the smallest enclosing zone
is first.
Returns an empty list if the zone does not exist or is unavailable.
This method must be run in the event loop.
"""
if (
not (input_zone := hass.states.get(zone_entity_id))
or input_zone.state == STATE_UNAVAILABLE
):
return []
input_attrs = input_zone.attributes
input_latitude: float = input_attrs[ATTR_LATITUDE]
input_longitude: float = input_attrs[ATTR_LONGITUDE]
input_radius: float = input_attrs[ATTR_RADIUS]
zones: list[tuple[str, float, float]] = []
# This can be called before async_setup by device tracker
zone_entity_ids = hass.data.get(DATA_ZONE_ENTITY_IDS, ())
for entity_id in zone_entity_ids:
if entity_id == zone_entity_id:
continue
if (
not (zone := hass.states.get(entity_id))
# Skip unavailable zones
or zone.state == STATE_UNAVAILABLE
):
continue
zone_attrs = zone.attributes
if (
zone_dist := distance(
input_latitude,
input_longitude,
zone_attrs[ATTR_LATITUDE],
zone_attrs[ATTR_LONGITUDE],
)
) is None:
continue
zone_radius = zone_attrs[ATTR_RADIUS]
if not zone_dist + input_radius <= zone_radius:
continue
zones.append((zone.entity_id, zone_dist, zone_radius))
zones.sort(key=lambda x: (x[2], x[1]))
return [itm[0] for itm in zones]
def async_active_zone(
hass: HomeAssistant, latitude: float, longitude: float, radius: float = 0
) -> State | None:
@@ -11,6 +11,7 @@ from homeassistant.components.device_tracker import (
ATTR_IP,
ATTR_MAC,
ATTR_SOURCE_TYPE,
CONF_ASSOCIATED_ZONE,
DOMAIN,
SourceType,
)
@@ -35,7 +36,11 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers import (
device_registry as dr,
entity_registry as er,
issue_registry as ir,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
@@ -599,6 +604,7 @@ async def test_base_scanner_entity_state(
assert entity_state
assert entity_state.attributes == {
ATTR_SOURCE_TYPE: SourceType.BLUETOOTH_LE,
ATTR_IN_ZONES: [],
}
assert entity_state.state == STATE_NOT_HOME
@@ -608,6 +614,12 @@ async def test_base_scanner_entity_state(
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_HOME
# No zone.home in the test state machine, so only the canonical home
# entity_id is reported.
assert entity_state.attributes == {
ATTR_SOURCE_TYPE: SourceType.BLUETOOTH_LE,
ATTR_IN_ZONES: ["zone.home"],
}
base_scanner_entity.set_connected(None)
await hass.async_block_till_done()
@@ -615,6 +627,420 @@ async def test_base_scanner_entity_state(
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_UNKNOWN
# is_connected is None -> empty in_zones (always reported).
assert entity_state.attributes == {
ATTR_SOURCE_TYPE: SourceType.BLUETOOTH_LE,
ATTR_IN_ZONES: [],
}
@pytest.mark.parametrize(
("zones", "expected_in_zones"),
[
pytest.param(
[("zone.home", 50.0, 60.0, 100)],
["zone.home"],
id="home_only",
),
pytest.param(
[
("zone.home", 50.0, 60.0, 100),
("zone.neighborhood", 50.0, 60.0, 500),
],
["zone.home", "zone.neighborhood"],
id="strictly_containing_zone",
),
pytest.param(
[
("zone.home", 50.0, 60.0, 100),
("zone.huge", 50.0, 60.0, 10000),
("zone.medium", 50.0, 60.0, 500),
],
["zone.home", "zone.medium", "zone.huge"],
id="multiple_containing_zones_sorted_by_radius",
),
pytest.param(
[
("zone.home", 50.0, 60.0, 100),
("zone.tiny", 50.0, 60.0, 50),
],
["zone.home"],
id="zone_smaller_than_home_excluded",
),
pytest.param(
[
("zone.home", 50.0, 60.0, 100),
("zone.equal", 50.0, 60.0, 100),
],
# Same center and radius as home: included under the <= predicate.
# zone.home stays first because the strict-result zone.home entry
# is filtered out, and zone.equal is the next entry.
["zone.home", "zone.equal"],
id="zone_equal_to_home_included",
),
pytest.param(
[
("zone.home", 50.0, 60.0, 100),
# Small offset, the home zone is fully inside
# the other zone (~330m + 100 < 500).
("zone.nearby", 50.0030, 60.0, 500),
# Offset by enough that the home zone is not fully inside
# the other zone (~440m + 100 > 500).
("zone.further_away", 50.0040, 60.0, 500),
# Offset by a very large amount, no overlap
# the other zone (~130km + 100 > 500).
("zone.faraway", 51.0, 61.0, 500),
],
["zone.home", "zone.nearby"],
id="offset_zone_excluded",
),
],
)
async def test_base_scanner_entity_in_zones_when_connected(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
zones: list[tuple[str, float, float, int]],
expected_in_zones: list[str],
) -> None:
"""Test in_zones content for a connected BaseScannerEntity across zone setups."""
base_scanner_entity._connected = True
for entity, latitude, longitude, radius in zones:
hass.states.async_set(
entity,
"0",
{ATTR_LATITUDE: latitude, ATTR_LONGITUDE: longitude, ATTR_RADIUS: radius},
)
await hass.async_block_till_done()
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_HOME
assert entity_state.attributes == {
ATTR_SOURCE_TYPE: SourceType.BLUETOOTH_LE,
ATTR_IN_ZONES: expected_in_zones,
}
@pytest.mark.parametrize("unique_id", ["unique_scanner"])
async def test_base_scanner_entity_associated_zone_option(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test the associated_zone entity option overrides which zone in_zones reports.
The scanner reports being connected to a non-default zone; state and in_zones
must follow the configured zone, and a zone enclosing the configured one is
included in in_zones too.
"""
hass.states.async_set(
"zone.home",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 1000},
)
hass.states.async_set(
"zone.kitchen",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 50},
)
await hass.async_block_till_done()
base_scanner_entity._connected = True
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
# Default: no option set -> associated with zone.home.
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_HOME
assert entity_state.attributes[ATTR_IN_ZONES] == ["zone.home"]
# Set the option -> associated_zone replaces zone.home; zone.home now shows
# up via the enclosing-zones lookup.
entity_registry.async_update_entity_options(
entity_id,
DOMAIN,
{CONF_ASSOCIATED_ZONE: "zone.kitchen"},
)
await hass.async_block_till_done()
assert base_scanner_entity._scanner_option_associated_zone == "zone.kitchen"
entity_state = hass.states.get(entity_id)
assert entity_state
# zone.kitchen is the configured zone -> state is the zone's name.
assert entity_state.state == "kitchen"
assert entity_state.attributes[ATTR_IN_ZONES] == ["zone.kitchen", "zone.home"]
# Clearing the option falls back to zone.home.
entity_registry.async_update_entity_options(entity_id, DOMAIN, None)
await hass.async_block_till_done()
assert base_scanner_entity._scanner_option_associated_zone == "zone.home"
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_HOME
assert entity_state.attributes[ATTR_IN_ZONES] == ["zone.home"]
@pytest.mark.parametrize("unique_id", ["unique_scanner"])
async def test_base_scanner_entity_associated_zone_removed_after_set(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
issue_registry: ir.IssueRegistry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test scanner state and repair issue when associated zone is removed.
When the user picks a zone via the associated_zone option and then deletes
that zone, the scanner falls back to ``state == "unknown"`` and a repair
issue is opened prompting the user to reconfigure.
"""
hass.states.async_set(
"zone.home",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 1000},
)
hass.states.async_set(
"zone.kitchen",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 50},
)
await hass.async_block_till_done()
base_scanner_entity._connected = True
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
entity_registry.async_update_entity_options(
entity_id,
DOMAIN,
{CONF_ASSOCIATED_ZONE: "zone.kitchen"},
)
await hass.async_block_till_done()
# Sanity check before removal.
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == "kitchen"
assert entity_state.attributes[ATTR_IN_ZONES] == ["zone.kitchen", "zone.home"]
issue_id = f"associated_zone_missing_{entity_id}"
assert issue_registry.async_get_issue(DOMAIN, issue_id) is None
# Remove the associated zone.
hass.states.async_remove("zone.kitchen")
await hass.async_block_till_done()
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_UNKNOWN
assert entity_state.attributes[ATTR_IN_ZONES] == []
issue = issue_registry.async_get_issue(DOMAIN, issue_id)
assert issue is not None
assert issue.severity is ir.IssueSeverity.WARNING
assert issue.translation_key == "associated_zone_missing"
assert issue.translation_placeholders == {
"entity_id": entity_id,
"zone": "zone.kitchen",
}
# Restore the zone -> issue is cleared, state recovers.
hass.states.async_set(
"zone.kitchen",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 50},
)
await hass.async_block_till_done()
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == "kitchen"
assert entity_state.attributes[ATTR_IN_ZONES] == ["zone.kitchen", "zone.home"]
assert issue_registry.async_get_issue(DOMAIN, issue_id) is None
@pytest.mark.parametrize("unique_id", ["unique_scanner"])
async def test_base_scanner_entity_associated_zone_missing_at_setup(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
issue_registry: ir.IssueRegistry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test repair issue is created when the configured zone is missing at setup."""
hass.states.async_set(
"zone.home",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 1000},
)
await hass.async_block_till_done()
# Pre-register the entity option pointing at a zone that does not exist.
entity_registry.async_get_or_create(
DOMAIN,
TEST_DOMAIN,
base_scanner_entity.unique_id,
suggested_object_id="entity1",
)
entity_registry.async_update_entity_options(
entity_id,
DOMAIN,
{CONF_ASSOCIATED_ZONE: "zone.never_existed"},
)
base_scanner_entity._connected = True
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_UNKNOWN
assert entity_state.attributes[ATTR_IN_ZONES] == []
issue_id = f"associated_zone_missing_{entity_id}"
issue = issue_registry.async_get_issue(DOMAIN, issue_id)
assert issue is not None
assert issue.translation_placeholders == {
"entity_id": entity_id,
"zone": "zone.never_existed",
}
@pytest.mark.parametrize("unique_id", ["unique_scanner"])
async def test_base_scanner_entity_associated_zone_issue_cleared_on_option_change(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
issue_registry: ir.IssueRegistry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test the repair issue is cleared when the user clears the option."""
hass.states.async_set(
"zone.home",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 1000},
)
await hass.async_block_till_done()
base_scanner_entity._connected = True
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
entity_registry.async_update_entity_options(
entity_id,
DOMAIN,
{CONF_ASSOCIATED_ZONE: "zone.never_existed"},
)
await hass.async_block_till_done()
issue_id = f"associated_zone_missing_{entity_id}"
assert issue_registry.async_get_issue(DOMAIN, issue_id) is not None
# Clearing the option restores the default and clears the repair issue.
entity_registry.async_update_entity_options(entity_id, DOMAIN, None)
await hass.async_block_till_done()
assert issue_registry.async_get_issue(DOMAIN, issue_id) is None
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == STATE_HOME
@pytest.mark.parametrize("unique_id", ["unique_scanner"])
async def test_base_scanner_entity_associated_zone_issue_cleared_on_unload(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
issue_registry: ir.IssueRegistry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test the repair issue is cleared when the entity is removed from hass."""
hass.states.async_set(
"zone.home",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 1000},
)
await hass.async_block_till_done()
base_scanner_entity._connected = True
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
entity_registry.async_update_entity_options(
entity_id,
DOMAIN,
{CONF_ASSOCIATED_ZONE: "zone.never_existed"},
)
await hass.async_block_till_done()
issue_id = f"associated_zone_missing_{entity_id}"
assert issue_registry.async_get_issue(DOMAIN, issue_id) is not None
assert await hass.config_entries.async_unload(config_entry.entry_id)
await hass.async_block_till_done()
assert issue_registry.async_get_issue(DOMAIN, issue_id) is None
@pytest.mark.parametrize("unique_id", ["unique_scanner"])
async def test_base_scanner_entity_associated_zone_option_set_before_add(
hass: HomeAssistant,
config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
entity_id: str,
base_scanner_entity: MockBaseScannerEntity,
) -> None:
"""Test associated_zone option set before the entity is added is honored."""
hass.states.async_set(
"zone.home",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 1000},
)
hass.states.async_set(
"zone.kitchen",
"0",
{ATTR_LATITUDE: 50.0, ATTR_LONGITUDE: 60.0, ATTR_RADIUS: 50},
)
await hass.async_block_till_done()
# Pre-register the entity with the option set before the platform is set up.
entity_registry.async_get_or_create(
DOMAIN,
TEST_DOMAIN,
base_scanner_entity.unique_id,
suggested_object_id="entity1",
)
entity_registry.async_update_entity_options(
entity_id,
DOMAIN,
{CONF_ASSOCIATED_ZONE: "zone.kitchen"},
)
base_scanner_entity._connected = True
config_entry = await create_mock_platform(hass, config_entry, [base_scanner_entity])
assert config_entry.state is ConfigEntryState.LOADED
assert base_scanner_entity._scanner_option_associated_zone == "zone.kitchen"
entity_state = hass.states.get(entity_id)
assert entity_state
assert entity_state.state == "kitchen"
assert entity_state.attributes[ATTR_IN_ZONES] == ["zone.kitchen", "zone.home"]
@pytest.mark.parametrize(
@@ -648,6 +1074,7 @@ async def test_scanner_entity_state(
assert entity_state
assert entity_state.attributes == {
ATTR_SOURCE_TYPE: SourceType.ROUTER,
ATTR_IN_ZONES: [],
ATTR_IP: ip_address,
ATTR_MAC: mac_address,
ATTR_HOST_NAME: hostname,
@@ -4,6 +4,9 @@
'attributes': ReadOnlyDict({
'band': '5 GHz',
'friendly_name': '00:00:5E:00:53:01',
'in_zones': list([
'zone.home',
]),
'mac': '00:00:5E:00:53:01',
'source_type': <SourceType.ROUTER: 'router'>,
'wifi': 'Main',
@@ -41,6 +41,9 @@
'attributes': ReadOnlyDict({
'friendly_name': 'FreeBSD router',
'icon': 'mdi:router-network',
'in_zones': list([
'zone.home',
]),
'ip': '192.168.50.1',
'mac': '00:00:00:00:00:01',
'source_type': <SourceType.ROUTER: 'router'>,
@@ -95,6 +98,9 @@
'attributes': ReadOnlyDict({
'friendly_name': 'PC_HOME',
'icon': 'mdi:desktop-tower',
'in_zones': list([
'zone.home',
]),
'ip': '192.168.50.3',
'mac': '00:00:00:00:00:03',
'source_type': <SourceType.ROUTER: 'router'>,
@@ -149,6 +155,9 @@
'attributes': ReadOnlyDict({
'friendly_name': 'Samsung The Frame 55',
'icon': 'mdi:television',
'in_zones': list([
'zone.home',
]),
'ip': '192.168.50.2',
'mac': '00:00:00:00:00:02',
'source_type': <SourceType.ROUTER: 'router'>,
@@ -4,6 +4,9 @@
'attributes': ReadOnlyDict({
'friendly_name': 'Banana',
'host_name': 'testhost',
'in_zones': list([
'zone.home',
]),
'ip': '192.168.1.102',
'mac': '2C-71-FF-ED-34-83',
'source_type': <SourceType.ROUTER: 'router'>,
@@ -20,6 +23,8 @@
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Banana',
'in_zones': list([
]),
'mac': '2C-71-FF-ED-34-83',
'source_type': <SourceType.ROUTER: 'router'>,
}),
@@ -40,6 +40,9 @@
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device 6 Switch 1',
'in_zones': list([
'zone.home',
]),
'ip': '10.0.1.1',
'mac': '00:00:00:00:01:01',
'source_type': <SourceType.ROUTER: 'router'>,
@@ -94,6 +97,8 @@
'attributes': ReadOnlyDict({
'friendly_name': 'Device 1 wd_client_1',
'host_name': 'wd_client_1',
'in_zones': list([
]),
'mac': '00:00:00:00:00:02',
'source_type': <SourceType.ROUTER: 'router'>,
}),
@@ -147,6 +152,8 @@
'attributes': ReadOnlyDict({
'friendly_name': 'Device 0 ws_client_1',
'host_name': 'ws_client_1',
'in_zones': list([
]),
'ip': '10.0.0.1',
'mac': '00:00:00:00:00:01',
'source_type': <SourceType.ROUTER: 'router'>,
@@ -41,6 +41,8 @@
'attributes': ReadOnlyDict({
'friendly_name': 'LanDevice1',
'host_name': 'LanDevice1',
'in_zones': list([
]),
'ip': '192.168.1.11',
'mac': 'yy:yy:yy:yy:yy:yy',
'source_type': <SourceType.ROUTER: 'router'>,
@@ -95,6 +97,9 @@
'attributes': ReadOnlyDict({
'friendly_name': 'WifiDevice0',
'host_name': 'WifiDevice0',
'in_zones': list([
'zone.home',
]),
'ip': '192.168.1.10',
'mac': 'xx:xx:xx:xx:xx:xx',
'source_type': <SourceType.ROUTER: 'router'>,
+210 -2
View File
@@ -7,11 +7,13 @@ import pytest
from homeassistant import setup
from homeassistant.components import zone
from homeassistant.components.zone import DOMAIN
from homeassistant.components.zone import ATTR_RADIUS, DOMAIN
from homeassistant.const import (
ATTR_EDITABLE,
ATTR_FRIENDLY_NAME,
ATTR_ICON,
ATTR_LATITUDE,
ATTR_LONGITUDE,
ATTR_NAME,
ATTR_PERSONS,
SERVICE_RELOAD,
@@ -580,7 +582,16 @@ async def test_zone_empty_setup(hass: HomeAssistant) -> None:
async def test_unavailable_zone(hass: HomeAssistant) -> None:
"""Test active zone with unavailable zones."""
"""Test active zone with unavailable zones.
Simulates the startup window where a zone has been pre-filled by the entity
registry as ``unavailable`` (``restored: True``) before the zone integration
has had a chance to write the zone's real state. Storage-created zones, and
YAML zones with an explicit ``id:``, get a unique_id and so are registered
in the entity registry; on ``EVENT_HOMEASSISTANT_START`` the registry writes
``unavailable`` for any such registered entity that does not yet have a
state. The zone helpers must skip these placeholder states.
"""
assert await setup.async_setup_component(hass, DOMAIN, {"zone": {}})
hass.states.async_set("zone.bla", "unavailable", {"restored": True})
@@ -592,6 +603,203 @@ async def test_unavailable_zone(hass: HomeAssistant) -> None:
assert zone.in_zone(hass.states.get("zone.bla"), 0, 0) is False
async def test_async_get_enclosing_zones(hass: HomeAssistant) -> None:
"""Test async_get_enclosing_zones returns zones that fully contain the given zone."""
latitude = 32.880600
longitude = -117.237561
assert await setup.async_setup_component(
hass,
zone.DOMAIN,
{
"zone": [
{
"name": "Small Zone",
"latitude": latitude,
"longitude": longitude,
"radius": 100,
},
{
"name": "Medium Zone",
"latitude": latitude,
"longitude": longitude,
"radius": 500,
},
{
"name": "Big Zone",
"latitude": latitude,
"longitude": longitude,
"radius": 2000,
},
{
"name": "Passive Big Zone",
"latitude": latitude,
"longitude": longitude,
"radius": 3000,
"passive": True,
},
]
},
)
# Small zone is enclosed by every larger concentric zone, sorted by radius.
# The queried zone itself is excluded.
assert zone.async_get_enclosing_zones(hass, "zone.small_zone") == [
"zone.medium_zone",
"zone.big_zone",
"zone.passive_big_zone",
]
# Medium zone is enclosed only by the two bigger zones.
assert zone.async_get_enclosing_zones(hass, "zone.medium_zone") == [
"zone.big_zone",
"zone.passive_big_zone",
]
# The biggest active zone is enclosed by the passive zone only.
assert zone.async_get_enclosing_zones(hass, "zone.big_zone") == [
"zone.passive_big_zone",
]
# The largest zone of all is enclosed by nothing.
assert zone.async_get_enclosing_zones(hass, "zone.passive_big_zone") == []
async def test_async_get_enclosing_zones_equal_radius(hass: HomeAssistant) -> None:
"""Test that same-center same-radius zones enclose each other (<= predicate)."""
latitude = 32.880600
longitude = -117.237561
assert await setup.async_setup_component(
hass,
zone.DOMAIN,
{
"zone": [
{
"name": "Zone A",
"latitude": latitude,
"longitude": longitude,
"radius": 1000,
},
{
"name": "Zone B",
"latitude": latitude,
"longitude": longitude,
"radius": 1000,
},
]
},
)
assert zone.async_get_enclosing_zones(hass, "zone.zone_a") == ["zone.zone_b"]
assert zone.async_get_enclosing_zones(hass, "zone.zone_b") == ["zone.zone_a"]
async def test_async_get_enclosing_zones_with_offset(hass: HomeAssistant) -> None:
"""Test full containment accounts for distance from zone center."""
latitude = 32.880600
longitude = -117.237561
assert await setup.async_setup_component(
hass,
zone.DOMAIN,
{
"zone": [
{
"name": "Inner",
"latitude": latitude,
"longitude": longitude,
"radius": 100,
},
{
"name": "Centered",
"latitude": latitude,
"longitude": longitude,
"radius": 1000,
},
{
"name": "Offset",
"latitude": 32.880600,
"longitude": -117.245561, # ~750m to the west
"radius": 1000,
},
]
},
)
# 100m radius zone at the center: both 1000m zones enclose it
# (centered trivially; offset has ~750 + 100 = 850 <= 1000).
assert zone.async_get_enclosing_zones(hass, "zone.inner") == [
"zone.centered",
"zone.offset",
]
async def test_async_get_enclosing_zones_missing_zone(hass: HomeAssistant) -> None:
"""Test async_get_enclosing_zones returns [] for an unknown zone."""
assert await setup.async_setup_component(hass, DOMAIN, {"zone": {}})
assert zone.async_get_enclosing_zones(hass, "zone.does_not_exist") == []
async def test_async_get_enclosing_zones_unavailable_input(hass: HomeAssistant) -> None:
"""Test async_get_enclosing_zones returns [] when the input zone is unavailable.
Simulates the startup window where a zone has been pre-filled by the entity
registry as ``unavailable`` (``restored: True``) before the zone integration
has had a chance to write the zone's real state. See ``test_unavailable_zone``
for the full explanation.
"""
assert await setup.async_setup_component(hass, DOMAIN, {"zone": {}})
hass.states.async_set(
"zone.bla",
"unavailable",
{"restored": True},
)
assert zone.async_get_enclosing_zones(hass, "zone.bla") == []
async def test_async_get_enclosing_zones_skips_unavailable_other(
hass: HomeAssistant,
) -> None:
"""Test other zones that are unavailable are skipped when searching containers.
See ``test_unavailable_zone`` for why an unavailable zone can appear in the
state machine.
"""
latitude = 32.880600
longitude = -117.237561
assert await setup.async_setup_component(
hass,
zone.DOMAIN,
{
"zone": [
{
"name": "Inner",
"latitude": latitude,
"longitude": longitude,
"radius": 100,
},
]
},
)
# A zone that would otherwise enclose Inner, but is unavailable.
hass.states.async_set(
"zone.bigger",
"unavailable",
{
ATTR_LATITUDE: latitude,
ATTR_LONGITUDE: longitude,
ATTR_RADIUS: 1000,
"restored": True,
},
)
assert zone.async_get_enclosing_zones(hass, "zone.inner") == []
async def test_state(hass: HomeAssistant) -> None:
"""Test the state of a zone."""
info = {