Detect and discard stale iCloud locations

homeassistant/components/icloud/account.py

@@ -1,6 +1,6 @@
 """iCloud account."""
 
-from datetime import timedelta
+from datetime import UTC, datetime, timedelta
 import logging
 import operator
 from typing import TYPE_CHECKING, Any
@@ -42,8 +42,10 @@
     DEVICE_ID,
     DEVICE_LOCATION,
     DEVICE_LOCATION_HORIZONTAL_ACCURACY,
+    DEVICE_LOCATION_IS_OLD,
     DEVICE_LOCATION_LATITUDE,
     DEVICE_LOCATION_LONGITUDE,
+    DEVICE_LOCATION_TIMESTAMP,
     DEVICE_LOST_MODE_CAPABLE,
     DEVICE_LOW_POWER_MODE,
     DEVICE_NAME,
@@ -361,6 +363,11 @@ def fetch_interval(self) -> float:
         """Return the account fetch interval."""
         return self._fetch_interval
 
+    @property
+    def max_interval(self) -> int:
+        """Return the longest interval between two fetches, in minutes."""
+        return self._max_interval
+
     @property
     def devices(self) -> dict[str, Any]:
         """Return the account devices."""
@@ -429,12 +436,56 @@ def update(self, status) -> None:
 
             if (
                 self._status[DEVICE_LOCATION]
-                and self._status[DEVICE_LOCATION][DEVICE_LOCATION_LATITUDE]
+                and self._status[DEVICE_LOCATION][DEVICE_LOCATION_LATITUDE] is not None
+                and self._status[DEVICE_LOCATION][DEVICE_LOCATION_LONGITUDE] is not None
             ):
                 location = self._status[DEVICE_LOCATION]
-                if self._location is None:
-                    dispatcher_send(self._account.hass, self._account.signal_device_new)
-                self._location = location
+                if self._is_stale(location):
+                    # Without a location the tracker reports "unknown" rather
+                    # than the place the device has since left.
+                    self._location = None
+                else:
+                    if self._location is None:
+                        dispatcher_send(
+                            self._account.hass, self._account.signal_device_new
+                        )
+                    self._location = location
+
+    def _is_stale(self, location: dict[str, Any]) -> bool:
+        """Return whether a location fix is too old to be trusted.
+
+        Both signals are needed, because neither is sufficient alone:
+
+        `isOld` says iCloud served a cached fix rather than a fresh one, but
+        not why. A device that was briefly unreachable looks exactly like one
+        that has genuinely moved away, so acting on the flag by itself would
+        drop locations that are still perfectly good.
+
+        The age of the fix alone is no better: a device sitting still keeps
+        reporting the same old timestamp while iCloud still considers that fix
+        current, and discarding those would throw away the only location such
+        a device ever reports.
+
+        Requiring both means a fix is only discarded once iCloud has marked it
+        cached *and* it is older than the longest gap between two fetches - by
+        which point a fresher one should already have superseded it. The half
+        interval of headroom keeps a fix that merely lands late from being
+        treated as stale.
+
+        The configured maximum is used rather than the current fetch interval,
+        which is recomputed after this runs and drops as low as fifteen
+        seconds while devices are still pending - a threshold that short would
+        discard every cached fix there is.
+        """
+        if not location.get(DEVICE_LOCATION_IS_OLD):
+            return False
+
+        if (timestamp := location.get(DEVICE_LOCATION_TIMESTAMP)) is None:
+            return False
+
+        # iCloud reports the fix time in milliseconds.
+        age = utcnow() - datetime.fromtimestamp(timestamp / 1000, tz=UTC)
+        return age.total_seconds() > self._account.max_interval * 60 * 1.5
 
     def play_sound(self) -> None:
         """Play sound on the device."""

homeassistant/components/icloud/const.py

@@ -28,8 +28,10 @@
 DEVICE_ID = "id"
 DEVICE_LOCATION = "location"
 DEVICE_LOCATION_HORIZONTAL_ACCURACY = "horizontalAccuracy"
+DEVICE_LOCATION_IS_OLD = "isOld"
 DEVICE_LOCATION_LATITUDE = "latitude"
 DEVICE_LOCATION_LONGITUDE = "longitude"
+DEVICE_LOCATION_TIMESTAMP = "timeStamp"
 DEVICE_LOST_MODE_CAPABLE = "lostModeCapable"
 DEVICE_LOW_POWER_MODE = "lowPowerMode"
 DEVICE_NAME = "name"

homeassistant/components/icloud/device_tracker.py

@@ -1,6 +1,6 @@
 """Support for tracking for iCloud devices."""
 
-from typing import TYPE_CHECKING, Any, override
+from typing import Any, override
 
 from homeassistant.components.device_tracker import TrackerEntity
 from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
@@ -70,24 +70,24 @@ def __init__(self, account: IcloudAccount, device: IcloudDevice) -> None:
     @override
     def location_accuracy(self) -> float:
         """Return the location accuracy of the device."""
-        if TYPE_CHECKING:
-            assert self._device.location is not None
+        if self._device.location is None:
+            return 0
         return self._device.location[DEVICE_LOCATION_HORIZONTAL_ACCURACY]
 
     @property
     @override
-    def latitude(self) -> float:
+    def latitude(self) -> float | None:
         """Return latitude value of the device."""
-        if TYPE_CHECKING:
-            assert self._device.location is not None
+        if self._device.location is None:
+            return None
         return self._device.location[DEVICE_LOCATION_LATITUDE]
 
     @property
     @override
-    def longitude(self) -> float:
+    def longitude(self) -> float | None:
         """Return longitude value of the device."""
-        if TYPE_CHECKING:
-            assert self._device.location is not None
+        if self._device.location is None:
+            return None
         return self._device.location[DEVICE_LOCATION_LONGITUDE]
 
     @property

tests/components/icloud/__init__.py

@@ -1 +1,41 @@
 """Tests for the iCloud component."""
+
+
+class MockAppleDevice:
+    """Mock Apple device implementing .status() used by the account."""
+
+    def __init__(self, status_dict) -> None:
+        """Set status."""
+        self._status = status_dict
+
+    def status(self, key):
+        """Return current status."""
+        return self._status
+
+    def __getitem__(self, key):
+        """Allow indexing to proxy into the raw status dict."""
+        return self._status.get(key)
+
+
+class MockDevicesContainer:
+    """Mock devices container, iterable and indexable."""
+
+    def __init__(self, userinfo, devices) -> None:
+        """Initialize with userinfo and list of device objects."""
+        self.user_info = userinfo
+        self._devices = devices
+
+    def __iter__(self):
+        """Iterate returns device objects (each must have .status(...))."""
+        return iter(self._devices)
+
+    def __len__(self):
+        """Return number of devices."""
+        return len(self._devices)
+
+    def __getitem__(self, idx):
+        """Indexing returns device object (which must have .status(...))."""
+        dev = self._devices[idx]
+        if hasattr(dev, "status"):
+            return dev.status(None)
+        return dev

tests/components/icloud/conftest.py

@@ -7,6 +7,10 @@
 import pytest
 
 from homeassistant.components.icloud.const import DOMAIN
+from homeassistant.util import dt as dt_util
+
+from . import MockAppleDevice, MockDevicesContainer
+from .const import DEVICE, USER_INFO
 
 from tests.common import AsyncMock, MockConfigEntry
 from tests.typing import MagicMock
@@ -139,3 +143,25 @@ def mock_config_entry() -> MockConfigEntry:
             "gps_accuracy_threshold": 0,
         },
     )
+
+
+@pytest.fixture(name="locating_service")
+def mock_locating_service() -> Generator[tuple[MagicMock, dict]]:
+    """Mock an account with one device that reports a location."""
+    status = {
+        **DEVICE,
+        "location": {
+            "latitude": 1.0,
+            "longitude": 2.0,
+            "horizontalAccuracy": 10,
+            "isOld": False,
+            "timeStamp": dt_util.utcnow().timestamp() * 1000,
+        },
+    }
+    with patch(
+        "homeassistant.components.icloud.account.PyiCloudService"
+    ) as service_mock:
+        service = service_mock.return_value
+        service.requires_2fa = False
+        service.devices = MockDevicesContainer(USER_INFO, [MockAppleDevice(status)])
+        yield service, status

tests/components/icloud/test_account.py

@@ -16,6 +16,7 @@
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers.storage import Store
 
+from . import MockAppleDevice, MockDevicesContainer
 from .const import DEVICE, MOCK_CONFIG, USER_INFO, USERNAME
 
 from tests.common import MockConfigEntry
@@ -76,46 +77,6 @@ async def test_setup_fails_when_userinfo_missing(
         account.setup()
 
 
-class MockAppleDevice:
-    """Mock Apple device implementing .status() used by the account."""
-
-    def __init__(self, status_dict) -> None:
-        """Set status."""
-        self._status = status_dict
-
-    def status(self, key):
-        """Return current status."""
-        return self._status
-
-    def __getitem__(self, key):
-        """Allow indexing to proxy into the raw status dict."""
-        return self._status.get(key)
-
-
-class MockDevicesContainer:
-    """Mock devices container, iterable and indexable."""
-
-    def __init__(self, userinfo, devices) -> None:
-        """Initialize with userinfo and list of device objects."""
-        self.user_info = userinfo
-        self._devices = devices
-
-    def __iter__(self):
-        """Iterate returns device objects (each must have .status(...))."""
-        return iter(self._devices)
-
-    def __len__(self):
-        """Return number of devices."""
-        return len(self._devices)
-
-    def __getitem__(self, idx):
-        """Indexing returns device object (which must have .status(...))."""
-        dev = self._devices[idx]
-        if hasattr(dev, "status"):
-            return dev.status(None)
-        return dev
-
-
 @pytest.fixture(name="mock_icloud_service")
 def mock_icloud_service_fixture():
     """Mock PyiCloudService with iterable and indexable devices."""

tests/components/icloud/test_device_tracker.py

@@ -0,0 +1,121 @@
+"""Tests for the iCloud device tracker."""
+
+from datetime import timedelta
+
+from freezegun.api import FrozenDateTimeFactory
+import pytest
+
+from homeassistant.components.icloud.const import DEFAULT_MAX_INTERVAL, DOMAIN
+from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE, STATE_UNKNOWN
+from homeassistant.core import HomeAssistant
+from homeassistant.util import dt as dt_util
+
+from .const import MOCK_CONFIG, USERNAME
+
+from tests.common import MockConfigEntry, async_fire_time_changed
+
+ENTITY_ID = "device_tracker.iphone"
+
+
+def location(*, is_old: bool, age: timedelta) -> dict:
+    """Return a location fix of a given age, as iCloud reports it."""
+    fixed_at = dt_util.utcnow() - age
+    return {
+        "latitude": 1.0,
+        "longitude": 2.0,
+        "horizontalAccuracy": 10,
+        "isOld": is_old,
+        "timeStamp": fixed_at.timestamp() * 1000,
+    }
+
+
+async def setup_account(hass: HomeAssistant) -> MockConfigEntry:
+    """Set up the iCloud integration."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN, data=MOCK_CONFIG, entry_id="test", unique_id=USERNAME
+    )
+    config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+    return config_entry
+
+
+async def next_fetch(hass: HomeAssistant, freezer: FrozenDateTimeFactory) -> None:
+    """Let the account poll iCloud again."""
+    freezer.tick(timedelta(minutes=DEFAULT_MAX_INTERVAL + 1))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+
+@pytest.mark.usefixtures("locating_service")
+async def test_location_reported(hass: HomeAssistant) -> None:
+    """Test that a fresh fix is reported."""
+    await setup_account(hass)
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.attributes[ATTR_LATITUDE] == 1.0
+    assert state.attributes[ATTR_LONGITUDE] == 2.0
+
+
+async def test_stale_location_cleared(
+    hass: HomeAssistant,
+    locating_service: tuple,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that a cached fix older than a polling cycle is discarded."""
+    _, status = locating_service
+    await setup_account(hass)
+    assert ATTR_LATITUDE in hass.states.get(ENTITY_ID).attributes
+
+    # iCloud keeps serving the same fix, now a cached one and older than a
+    # full polling cycle: the device has moved on without telling us.
+    status["location"] = location(
+        is_old=True, age=timedelta(minutes=DEFAULT_MAX_INTERVAL * 3)
+    )
+    await next_fetch(hass, freezer)
+
+    state = hass.states.get(ENTITY_ID)
+    assert ATTR_LATITUDE not in state.attributes
+    # No coordinates and no zone, so the tracker reports that it does not
+    # know where the device is rather than claiming it is away.
+    assert state.state == STATE_UNKNOWN
+
+
+async def test_recent_cached_location_kept(
+    hass: HomeAssistant,
+    locating_service: tuple,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that a cached but recent fix is kept.
+
+    A device that was only briefly unreachable is reported the same way as one
+    that has moved away, so the flag alone must not discard a location.
+    """
+    _, status = locating_service
+    await setup_account(hass)
+
+    status["location"] = location(is_old=True, age=timedelta(seconds=30))
+    await next_fetch(hass, freezer)
+
+    assert hass.states.get(ENTITY_ID).attributes[ATTR_LATITUDE] == 1.0
+
+
+async def test_old_uncached_location_kept(
+    hass: HomeAssistant,
+    locating_service: tuple,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test that an old fix iCloud still considers current is kept.
+
+    A device that is sitting still reports an old timestamp indefinitely, so
+    age alone must not be enough to discard a location either.
+    """
+    _, status = locating_service
+    await setup_account(hass)
+
+    status["location"] = location(
+        is_old=False, age=timedelta(minutes=DEFAULT_MAX_INTERVAL * 3)
+    )
+    await next_fetch(hass, freezer)
+
+    assert hass.states.get(ENTITY_ID).attributes[ATTR_LATITUDE] == 1.0