mirror of
https://github.com/python-kasa/python-kasa.git
synced 2026-02-26 04:39:56 +00:00
Implement IOT Time Module Failover (#1583)
Some checks failed
CI / Perform Lint Checks (3.13) (push) Has been cancelled
CodeQL Checks / Analyze (python) (push) Has been cancelled
CI / Python 3.11 on macos-latest (push) Has been cancelled
CI / Python 3.12 on macos-latest (push) Has been cancelled
CI / Python 3.13 on macos-latest (push) Has been cancelled
CI / Python 3.11 on ubuntu-latest (push) Has been cancelled
CI / Python 3.12 on ubuntu-latest (push) Has been cancelled
CI / Python 3.13 on ubuntu-latest (push) Has been cancelled
CI / Python 3.11 on windows-latest (push) Has been cancelled
CI / Python 3.12 on windows-latest (push) Has been cancelled
CI / Python 3.13 on windows-latest (push) Has been cancelled
Stale / stale (push) Has been cancelled
Some checks failed
CI / Perform Lint Checks (3.13) (push) Has been cancelled
CodeQL Checks / Analyze (python) (push) Has been cancelled
CI / Python 3.11 on macos-latest (push) Has been cancelled
CI / Python 3.12 on macos-latest (push) Has been cancelled
CI / Python 3.13 on macos-latest (push) Has been cancelled
CI / Python 3.11 on ubuntu-latest (push) Has been cancelled
CI / Python 3.12 on ubuntu-latest (push) Has been cancelled
CI / Python 3.13 on ubuntu-latest (push) Has been cancelled
CI / Python 3.11 on windows-latest (push) Has been cancelled
CI / Python 3.12 on windows-latest (push) Has been cancelled
CI / Python 3.13 on windows-latest (push) Has been cancelled
Stale / stale (push) Has been cancelled
Adds a failover in the IOT Time Module to handle issues where the system default time zone files don't have the correct time zone info.
This commit is contained in:
@@ -3,9 +3,9 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timedelta, tzinfo
|
||||
from datetime import UTC, datetime, timedelta, timezone, tzinfo
|
||||
from typing import cast
|
||||
from zoneinfo import ZoneInfo
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
|
||||
from ..cachedzoneinfo import CachedZoneInfo
|
||||
|
||||
@@ -14,7 +14,7 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
async def get_timezone(index: int) -> tzinfo:
|
||||
"""Get the timezone from the index."""
|
||||
if index > 109:
|
||||
if index < 0 or index > 109:
|
||||
_LOGGER.error(
|
||||
"Unexpected index %s not configured as a timezone, defaulting to UTC", index
|
||||
)
|
||||
@@ -25,7 +25,12 @@ async def get_timezone(index: int) -> tzinfo:
|
||||
|
||||
|
||||
async def get_timezone_index(tzone: tzinfo) -> int:
|
||||
"""Return the iot firmware index for a valid IANA timezone key."""
|
||||
"""Return the iot firmware index for a valid IANA timezone key.
|
||||
|
||||
If tzinfo is a ZoneInfo and its key is in TIMEZONE_INDEX, return that index.
|
||||
Otherwise, compare annual offset behavior to find the best match.
|
||||
Indices that cannot be loaded on this host are skipped.
|
||||
"""
|
||||
if isinstance(tzone, ZoneInfo):
|
||||
name = tzone.key
|
||||
rev = {val: key for key, val in TIMEZONE_INDEX.items()}
|
||||
@@ -33,14 +38,23 @@ async def get_timezone_index(tzone: tzinfo) -> int:
|
||||
return rev[name]
|
||||
|
||||
for i in range(110):
|
||||
if _is_same_timezone(tzone, await get_timezone(i)):
|
||||
try:
|
||||
cand = await get_timezone(i)
|
||||
except ZoneInfoNotFoundError:
|
||||
continue
|
||||
if _is_same_timezone(tzone, cand):
|
||||
return i
|
||||
raise ValueError("Device does not support timezone %s", name)
|
||||
raise ValueError(
|
||||
f"Device does not support timezone {getattr(tzone, 'key', tzone)!r}"
|
||||
)
|
||||
|
||||
|
||||
async def get_matching_timezones(tzone: tzinfo) -> list[str]:
|
||||
"""Return the iot firmware index for a valid IANA timezone key."""
|
||||
matches = []
|
||||
"""Return available IANA keys from TIMEZONE_INDEX that match the given tzinfo.
|
||||
|
||||
Skips zones that cannot be resolved on the host.
|
||||
"""
|
||||
matches: list[str] = []
|
||||
if isinstance(tzone, ZoneInfo):
|
||||
name = tzone.key
|
||||
vals = {val for val in TIMEZONE_INDEX.values()}
|
||||
@@ -48,7 +62,10 @@ async def get_matching_timezones(tzone: tzinfo) -> list[str]:
|
||||
matches.append(name)
|
||||
|
||||
for i in range(110):
|
||||
fw_tz = await get_timezone(i)
|
||||
try:
|
||||
fw_tz = await get_timezone(i)
|
||||
except ZoneInfoNotFoundError:
|
||||
continue
|
||||
if _is_same_timezone(tzone, fw_tz):
|
||||
match_key = cast(ZoneInfo, fw_tz).key
|
||||
if match_key not in matches:
|
||||
@@ -57,11 +74,7 @@ async def get_matching_timezones(tzone: tzinfo) -> list[str]:
|
||||
|
||||
|
||||
def _is_same_timezone(tzone1: tzinfo, tzone2: tzinfo) -> bool:
|
||||
"""Return true if the timezones have the same utcffset and dst offset.
|
||||
|
||||
Iot devices only support a limited static list of IANA timezones; this is used to
|
||||
check if a static timezone matches the same utc offset and dst settings.
|
||||
"""
|
||||
"""Return true if the timezones have the same UTC offset each day of the year."""
|
||||
now = datetime.now()
|
||||
start_day = datetime(now.year, 1, 1, 12)
|
||||
for i in range(365):
|
||||
@@ -71,6 +84,83 @@ def _is_same_timezone(tzone1: tzinfo, tzone2: tzinfo) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def _dst_expected_from_key(key: str) -> bool | None:
|
||||
"""Infer if a zone key implies DST behavior (heuristic, no manual map).
|
||||
|
||||
- Posix-style keys with two abbreviations like 'CST6CDT', 'MST7MDT' -> True
|
||||
- Fixed abbreviation keys like 'EST', 'MST', 'HST' -> False
|
||||
- 'Etc/*' zones are fixed-offset -> False
|
||||
- Otherwise unknown -> None
|
||||
"""
|
||||
k = key.upper()
|
||||
if k.startswith("ETC/"):
|
||||
return False
|
||||
# Two abbreviations with a number in between (e.g., CST6CDT)
|
||||
if any(ch.isdigit() for ch in k) and any(
|
||||
x in k for x in ("CDT", "PDT", "MDT", "EDT")
|
||||
):
|
||||
return True
|
||||
if k in {"UTC", "UCT", "GMT", "EST", "MST", "HST", "PST"}:
|
||||
return False
|
||||
return None
|
||||
|
||||
|
||||
def _expected_dst_behavior_for_index(index: int) -> bool | None:
|
||||
"""Return whether the given index implies a DST-observing zone."""
|
||||
key = TIMEZONE_INDEX[index]
|
||||
return _dst_expected_from_key(key)
|
||||
|
||||
|
||||
async def _guess_timezone_by_offset(
|
||||
offset: timedelta, when_utc: datetime, dst_expected: bool | None = None
|
||||
) -> tzinfo:
|
||||
"""Pick a ZoneInfo from TIMEZONE_INDEX that exists on this host and matches.
|
||||
|
||||
- offset: device's UTC offset at 'when_utc'
|
||||
- when_utc: reference instant; naive is treated as UTC
|
||||
- dst_expected: if True/False, prefer candidates that do/do not observe DST annually
|
||||
|
||||
Returns the lowest-index matching ZoneInfo for determinism.
|
||||
If none match, returns a fixed-offset timezone as a last resort.
|
||||
"""
|
||||
if when_utc.tzinfo is None:
|
||||
when_utc = when_utc.replace(tzinfo=UTC)
|
||||
else:
|
||||
when_utc = when_utc.astimezone(UTC)
|
||||
|
||||
year = when_utc.year
|
||||
# Reference mid-winter and mid-summer dates to detect DST-observing candidates
|
||||
jan_ref = datetime(year, 1, 15, 12, tzinfo=UTC)
|
||||
jul_ref = datetime(year, 7, 15, 12, tzinfo=UTC)
|
||||
|
||||
candidates: list[tuple[int, tzinfo, bool]] = []
|
||||
for idx, name in TIMEZONE_INDEX.items():
|
||||
try:
|
||||
tz = await CachedZoneInfo.get_cached_zone_info(name)
|
||||
except ZoneInfoNotFoundError:
|
||||
continue
|
||||
|
||||
cand_offset_now = when_utc.astimezone(tz).utcoffset()
|
||||
if cand_offset_now != offset:
|
||||
continue
|
||||
|
||||
# Determine if this candidate observes DST (offset differs between Jan and Jul)
|
||||
jan_off = jan_ref.astimezone(tz).utcoffset()
|
||||
jul_off = jul_ref.astimezone(tz).utcoffset()
|
||||
cand_observes_dst = jan_off != jul_off
|
||||
|
||||
if dst_expected is None or cand_observes_dst == dst_expected:
|
||||
candidates.append((idx, tz, cand_observes_dst))
|
||||
|
||||
if candidates:
|
||||
candidates.sort(key=lambda it: it[0])
|
||||
chosen = candidates[0][1]
|
||||
return chosen
|
||||
|
||||
# No ZoneInfo matched; return fixed offset as a last resort
|
||||
return timezone(offset)
|
||||
|
||||
|
||||
TIMEZONE_INDEX = {
|
||||
0: "Etc/GMT+12",
|
||||
1: "Pacific/Samoa",
|
||||
|
||||
@@ -2,12 +2,19 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime, tzinfo
|
||||
import contextlib
|
||||
from datetime import UTC, datetime, timedelta, tzinfo
|
||||
from zoneinfo import ZoneInfoNotFoundError
|
||||
|
||||
from ...exceptions import KasaException
|
||||
from ...interfaces import Time as TimeInterface
|
||||
from ..iotmodule import IotModule, merge
|
||||
from ..iottimezone import get_timezone, get_timezone_index
|
||||
from ..iottimezone import (
|
||||
_expected_dst_behavior_for_index,
|
||||
_guess_timezone_by_offset,
|
||||
get_timezone,
|
||||
get_timezone_index,
|
||||
)
|
||||
|
||||
|
||||
class Time(IotModule, TimeInterface):
|
||||
@@ -23,9 +30,46 @@ class Time(IotModule, TimeInterface):
|
||||
return q
|
||||
|
||||
async def _post_update_hook(self) -> None:
|
||||
"""Perform actions after a device update."""
|
||||
"""Perform actions after a device update.
|
||||
|
||||
If the configured zone is not available on this host, compute the device's
|
||||
current UTC offset and choose a best-match available zone, preferring DST-
|
||||
observing candidates when the original index implies DST. As a last resort,
|
||||
use a fixed-offset timezone.
|
||||
"""
|
||||
if res := self.data.get("get_timezone"):
|
||||
self._timezone = await get_timezone(res.get("index"))
|
||||
idx = res.get("index")
|
||||
try:
|
||||
self._timezone = await get_timezone(idx)
|
||||
return
|
||||
except ZoneInfoNotFoundError:
|
||||
pass # fall through to offset-based match
|
||||
|
||||
gt = self.data.get("get_time")
|
||||
if gt:
|
||||
device_local = datetime(
|
||||
gt["year"],
|
||||
gt["month"],
|
||||
gt["mday"],
|
||||
gt["hour"],
|
||||
gt["min"],
|
||||
gt["sec"],
|
||||
)
|
||||
now_utc = datetime.now(UTC)
|
||||
delta = device_local - now_utc.replace(tzinfo=None)
|
||||
rounded = timedelta(seconds=60 * round(delta.total_seconds() / 60))
|
||||
|
||||
dst_expected = None
|
||||
if res := self.data.get("get_timezone"):
|
||||
idx = res.get("index")
|
||||
with contextlib.suppress(KeyError):
|
||||
dst_expected = _expected_dst_behavior_for_index(idx)
|
||||
|
||||
self._timezone = await _guess_timezone_by_offset(
|
||||
rounded, when_utc=now_utc, dst_expected=dst_expected
|
||||
)
|
||||
else:
|
||||
self._timezone = UTC
|
||||
|
||||
@property
|
||||
def time(self) -> datetime:
|
||||
|
||||
194
tests/iot/test_iottimezone.py
Normal file
194
tests/iot/test_iottimezone.py
Normal file
@@ -0,0 +1,194 @@
|
||||
from datetime import UTC, datetime, timedelta, timezone
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
|
||||
def test_expected_dst_behavior_for_index_cases():
|
||||
"""Exercise _expected_dst_behavior_for_index for several representative indices."""
|
||||
from kasa.iot.iottimezone import _expected_dst_behavior_for_index
|
||||
|
||||
# Posix-style DST zones
|
||||
assert _expected_dst_behavior_for_index(10) is True # MST7MDT
|
||||
assert _expected_dst_behavior_for_index(13) is True # CST6CDT
|
||||
# Fixed-offset or fixed-abbreviation zones
|
||||
assert _expected_dst_behavior_for_index(34) is False # Etc/GMT+2
|
||||
assert _expected_dst_behavior_for_index(18) is False # EST
|
||||
# Invalid index should raise KeyError
|
||||
with pytest.raises(KeyError):
|
||||
_expected_dst_behavior_for_index(999)
|
||||
|
||||
|
||||
async def test_guess_timezone_by_offset_fixed_fallback_unit():
|
||||
"""When no ZoneInfo matches, return a fixed-offset tzinfo."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
year = datetime.now(UTC).year
|
||||
when = datetime(year, 1, 15, 12, tzinfo=UTC)
|
||||
offset = timedelta(minutes=2) # unlikely to match any real zone
|
||||
tz = await tzmod._guess_timezone_by_offset(offset, when_utc=when)
|
||||
assert tz.utcoffset(when) == offset
|
||||
|
||||
|
||||
async def test_guess_timezone_by_offset_candidates_unit():
|
||||
"""Cover naive when_utc branch and candidate selection path (non-empty candidates)."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
# naive datetime hits the 'naive -> UTC' branch
|
||||
when = datetime(2025, 1, 15, 12)
|
||||
offset = timedelta(0)
|
||||
tz = await tzmod._guess_timezone_by_offset(offset, when_utc=when)
|
||||
|
||||
# Should choose a ZoneInfo candidate (not the fixed-offset fallback), with matching offset
|
||||
assert isinstance(tz, ZoneInfo)
|
||||
assert tz.utcoffset(when.replace(tzinfo=UTC)) == offset
|
||||
|
||||
|
||||
async def test_guess_timezone_by_offset_dst_expected_true_filters(
|
||||
mocker: MockerFixture,
|
||||
):
|
||||
"""dst_expected=True should prefer a DST-observing zone when possible."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
when = datetime(datetime.now(UTC).year, 1, 15, 12, tzinfo=UTC)
|
||||
tz = await tzmod._guess_timezone_by_offset(
|
||||
timedelta(0), when_utc=when, dst_expected=True
|
||||
)
|
||||
assert tz.utcoffset(when) == timedelta(0)
|
||||
if isinstance(tz, ZoneInfo):
|
||||
jan = datetime(when.year, 1, 15, 12, tzinfo=UTC).astimezone(tz).utcoffset()
|
||||
jul = datetime(when.year, 7, 15, 12, tzinfo=UTC).astimezone(tz).utcoffset()
|
||||
assert jan != jul # observes DST
|
||||
|
||||
|
||||
async def test_guess_timezone_by_offset_dst_expected_false_prefers_non_dst():
|
||||
"""dst_expected=False should prefer a non-DST zone and skip DST candidates (covers False branch)."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
when = datetime(datetime.now(UTC).year, 1, 15, 12, tzinfo=UTC)
|
||||
tz = await tzmod._guess_timezone_by_offset(
|
||||
timedelta(0), when_utc=when, dst_expected=False
|
||||
)
|
||||
assert tz.utcoffset(when) == timedelta(0)
|
||||
if isinstance(tz, ZoneInfo):
|
||||
jan = datetime(when.year, 1, 15, 12, tzinfo=UTC).astimezone(tz).utcoffset()
|
||||
jul = datetime(when.year, 7, 15, 12, tzinfo=UTC).astimezone(tz).utcoffset()
|
||||
assert jan == jul # non-DST zone chosen
|
||||
|
||||
|
||||
async def test_guess_timezone_by_offset_handles_missing_zoneinfo_unit(
|
||||
mocker: MockerFixture,
|
||||
):
|
||||
"""Cover the ZoneInfoNotFoundError continue path within guess_timezone_by_offset."""
|
||||
from zoneinfo import ZoneInfoNotFoundError as ZNF
|
||||
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
original = tzmod.CachedZoneInfo.get_cached_zone_info
|
||||
|
||||
async def flaky_get(name: str):
|
||||
# Force the first entry to raise to exercise the except path (143-144)
|
||||
first_name = next(iter(tzmod.TIMEZONE_INDEX.values()))
|
||||
if name == first_name:
|
||||
raise ZNF("unavailable on host")
|
||||
return await original(name)
|
||||
|
||||
mocker.patch.object(tzmod.CachedZoneInfo, "get_cached_zone_info", new=flaky_get)
|
||||
|
||||
when = datetime(datetime.now(UTC).year, 1, 15, 12, tzinfo=UTC)
|
||||
tz = await tzmod._guess_timezone_by_offset(timedelta(0), when_utc=when)
|
||||
assert tz.utcoffset(when) == timedelta(0)
|
||||
|
||||
|
||||
async def test_get_timezone_index_direct_match():
|
||||
"""If ZoneInfo key is in TIMEZONE_INDEX, return index directly."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
idx = await tzmod.get_timezone_index(ZoneInfo("GB"))
|
||||
assert idx == 39 # "GB" is mapped to index 39
|
||||
|
||||
|
||||
async def test_get_timezone_index_non_zoneinfo_unit():
|
||||
"""Exercise get_timezone_index path when input tzinfo is not a ZoneInfo instance."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
# Fixed offset +0 should match a valid index (e.g., UCT/Africa/Monrovia)
|
||||
idx = await tzmod.get_timezone_index(timezone(timedelta(0)))
|
||||
assert isinstance(idx, int)
|
||||
assert 0 <= idx <= 109
|
||||
|
||||
|
||||
async def test_get_timezone_index_skips_missing_unit(mocker: MockerFixture):
|
||||
"""Cover ZoneInfoNotFoundError path in get_timezone_index loop and successful match."""
|
||||
from zoneinfo import ZoneInfoNotFoundError as ZNF
|
||||
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
original_get_tz = tzmod.get_timezone
|
||||
|
||||
async def side_effect(i: int):
|
||||
if i < 5:
|
||||
raise ZNF("unavailable on host")
|
||||
return await original_get_tz(i)
|
||||
|
||||
mocker.patch("kasa.iot.iottimezone.get_timezone", new=side_effect)
|
||||
|
||||
# Use a ZoneInfo not directly present in TIMEZONE_INDEX values to avoid early return
|
||||
idx = await tzmod.get_timezone_index(ZoneInfo("Europe/London"))
|
||||
assert isinstance(idx, int)
|
||||
assert 0 <= idx <= 109
|
||||
assert idx >= 5
|
||||
|
||||
|
||||
async def test_get_timezone_index_raises_for_unmatched_unit():
|
||||
"""Ensure get_timezone_index completes loop and raises when no match exists (covers raise branch)."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
# Uncommon 2-minute offset won't match any real zone in TIMEZONE_INDEX
|
||||
with pytest.raises(ValueError, match="Device does not support timezone"):
|
||||
await tzmod.get_timezone_index(timezone(timedelta(minutes=2)))
|
||||
|
||||
|
||||
async def test_get_matching_timezones_branches_unit(mocker: MockerFixture):
|
||||
"""Cover initial append, except path, and duplicate suppression in get_matching_timezones."""
|
||||
from zoneinfo import ZoneInfoNotFoundError as ZNF
|
||||
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
original_get_tz = tzmod.get_timezone
|
||||
|
||||
async def side_effect(i: int):
|
||||
# Force one miss to hit the except path
|
||||
if i == 0:
|
||||
raise ZNF("unavailable on host")
|
||||
return await original_get_tz(i)
|
||||
|
||||
mocker.patch("kasa.iot.iottimezone.get_timezone", new=side_effect)
|
||||
|
||||
# 'GB' is in TIMEZONE_INDEX; passing ZoneInfo('GB') will trigger initial append
|
||||
matches = await tzmod.get_matching_timezones(ZoneInfo("GB"))
|
||||
assert "GB" in matches # initial append done
|
||||
# Loop should find GB again but not duplicate it
|
||||
|
||||
|
||||
async def test_get_matching_timezones_non_zoneinfo_unit():
|
||||
"""Exercise get_matching_timezones when input tzinfo is not a ZoneInfo (skips initial append)."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
matches = await tzmod.get_matching_timezones(timezone(timedelta(0)))
|
||||
assert isinstance(matches, list)
|
||||
assert len(matches) > 0
|
||||
|
||||
|
||||
async def test_get_timezone_out_of_range_defaults_to_utc():
|
||||
"""Out-of-range index should log and default to UTC."""
|
||||
import kasa.iot.iottimezone as tzmod
|
||||
|
||||
tz = await tzmod.get_timezone(-1)
|
||||
assert isinstance(tz, ZoneInfo)
|
||||
assert tz.key in ("Etc/UTC", "UTC") # platform alias acceptable
|
||||
|
||||
tz2 = await tzmod.get_timezone(999)
|
||||
assert isinstance(tz2, ZoneInfo)
|
||||
assert tz2.key in ("Etc/UTC", "UTC")
|
||||
@@ -2,14 +2,15 @@ import importlib
|
||||
import inspect
|
||||
import pkgutil
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from datetime import UTC, datetime, timedelta, timezone
|
||||
from unittest.mock import AsyncMock
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
import kasa.interfaces
|
||||
from kasa import Device, LightState, Module, ThermostatState
|
||||
from kasa import Device, KasaException, LightState, Module, ThermostatState
|
||||
from kasa.module import _get_feature_attribute
|
||||
|
||||
from .device_fixtures import (
|
||||
@@ -456,3 +457,214 @@ async def test_set_time(dev: Device):
|
||||
await time_mod.set_time(original_time)
|
||||
await dev.update()
|
||||
assert time_mod.time == original_time
|
||||
|
||||
|
||||
async def test_time_post_update_no_time_uses_utc_unit(monkeypatch: pytest.MonkeyPatch):
|
||||
"""If neither get_timezone nor get_time are present, timezone falls back to UTC."""
|
||||
from kasa.iot.modules.time import Time as TimeModule
|
||||
|
||||
inst = object.__new__(TimeModule)
|
||||
monkeypatch.setattr(TimeModule, "data", property(lambda self: {}))
|
||||
|
||||
await TimeModule._post_update_hook(inst)
|
||||
assert inst.timezone is UTC
|
||||
|
||||
|
||||
async def test_time_post_update_uses_offset_when_index_missing_unit(
|
||||
monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture
|
||||
):
|
||||
"""When index present but zone not on host, fall back to offset-based guess."""
|
||||
from zoneinfo import ZoneInfoNotFoundError
|
||||
|
||||
from kasa.iot.modules.time import Time as TimeModule
|
||||
|
||||
inst = object.__new__(TimeModule)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
data = {
|
||||
"get_timezone": {"index": 39}, # any index; we'll force failure to load it
|
||||
"get_time": {
|
||||
"year": now.year,
|
||||
"month": now.month,
|
||||
"mday": now.day,
|
||||
"hour": now.hour,
|
||||
"min": now.minute,
|
||||
"sec": now.second,
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr(TimeModule, "data", property(lambda self: data))
|
||||
|
||||
mocker.patch(
|
||||
"kasa.iot.modules.time.get_timezone",
|
||||
new=AsyncMock(side_effect=ZoneInfoNotFoundError("missing on host")),
|
||||
)
|
||||
mock_guess = mocker.patch(
|
||||
"kasa.iot.modules.time._guess_timezone_by_offset",
|
||||
new=AsyncMock(return_value=timezone(timedelta(0))),
|
||||
)
|
||||
|
||||
await TimeModule._post_update_hook(inst)
|
||||
mock_guess.assert_awaited_once()
|
||||
# timezone should be set to a valid tzinfo after fallback
|
||||
assert inst.timezone.utcoffset(now) == timedelta(0)
|
||||
|
||||
|
||||
async def test_time_get_time_exception_returns_none_unit(mocker: MockerFixture):
|
||||
"""Cover Time.get_time exception path (unit test of iot Time)."""
|
||||
from kasa.iot.modules.time import Time as TimeModule
|
||||
|
||||
inst = object.__new__(TimeModule)
|
||||
mocker.patch.object(inst, "call", new=AsyncMock(side_effect=KasaException("boom")))
|
||||
|
||||
assert await TimeModule.get_time(inst) is None
|
||||
|
||||
|
||||
async def test_time_get_time_success_unit(mocker: MockerFixture):
|
||||
"""Cover the success path of Time.get_time."""
|
||||
from kasa.iot.modules.time import Time as TimeModule
|
||||
|
||||
inst = object.__new__(TimeModule)
|
||||
# Ensure timezone is available on the instance
|
||||
inst._timezone = UTC
|
||||
ret = {
|
||||
"year": 2024,
|
||||
"month": 1,
|
||||
"mday": 2,
|
||||
"hour": 3,
|
||||
"min": 4,
|
||||
"sec": 5,
|
||||
}
|
||||
mocker.patch.object(inst, "call", new=AsyncMock(return_value=ret))
|
||||
|
||||
dt = await TimeModule.get_time(inst)
|
||||
assert dt is not None
|
||||
assert (dt.year, dt.month, dt.day, dt.hour, dt.minute, dt.second) == (
|
||||
2024,
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
4,
|
||||
5,
|
||||
)
|
||||
assert dt.tzinfo == inst.timezone
|
||||
|
||||
|
||||
async def test_time_post_update_with_time_no_tz_uses_guess_unit(
|
||||
monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture
|
||||
):
|
||||
"""When get_time is present but get_timezone is missing, use offset-based guess (dst_expected None)."""
|
||||
from kasa.iot.modules.time import Time as TimeModule
|
||||
|
||||
inst = object.__new__(TimeModule)
|
||||
now = datetime.now(UTC)
|
||||
data = {
|
||||
"get_time": {
|
||||
"year": now.year,
|
||||
"month": now.month,
|
||||
"mday": now.day,
|
||||
"hour": now.hour,
|
||||
"min": now.minute,
|
||||
"sec": now.second,
|
||||
}
|
||||
# Note: no "get_timezone" key
|
||||
}
|
||||
monkeypatch.setattr(TimeModule, "data", property(lambda self: data))
|
||||
|
||||
mock_guess = mocker.patch(
|
||||
"kasa.iot.modules.time._guess_timezone_by_offset",
|
||||
new=AsyncMock(return_value=timezone(timedelta(hours=2))),
|
||||
)
|
||||
|
||||
await TimeModule._post_update_hook(inst)
|
||||
mock_guess.assert_awaited_once()
|
||||
assert inst.timezone.utcoffset(now) == timedelta(hours=2)
|
||||
|
||||
|
||||
async def test_time_set_time_wraps_exception_unit(
|
||||
monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture
|
||||
):
|
||||
"""Cover exception wrapping in Time.set_time (unit test of iot Time)."""
|
||||
from kasa.iot.modules.time import Time as TimeModule
|
||||
|
||||
inst = object.__new__(TimeModule)
|
||||
# Keep data empty so set_time path is chosen (no timezone change)
|
||||
monkeypatch.setattr(TimeModule, "data", property(lambda self: {}))
|
||||
mocker.patch.object(inst, "call", new=AsyncMock(side_effect=RuntimeError("err")))
|
||||
|
||||
with pytest.raises(KasaException):
|
||||
await TimeModule.set_time(inst, datetime.now())
|
||||
|
||||
|
||||
# New tests to cover remaining smart and smartcam time.py branches
|
||||
|
||||
|
||||
async def test_smart_time_set_time_no_region_added_when_tzname_none_unit(
|
||||
mocker: MockerFixture,
|
||||
):
|
||||
"""In smart Time.set_time, ensure we cover the branch where tzname() returns None, so 'region' is omitted."""
|
||||
from datetime import tzinfo as _tzinfo
|
||||
|
||||
from kasa.smart.modules.time import Time as SmartTimeModule
|
||||
|
||||
class NullNameTZ(_tzinfo):
|
||||
def utcoffset(self, dt):
|
||||
return timedelta(hours=1)
|
||||
|
||||
def dst(self, dt):
|
||||
return timedelta(0)
|
||||
|
||||
def tzname(self, dt):
|
||||
return None
|
||||
|
||||
inst = object.__new__(SmartTimeModule)
|
||||
call_mock = mocker.patch.object(inst, "call", new=AsyncMock(return_value={}))
|
||||
|
||||
aware_dt = datetime(2024, 1, 1, 12, 0, 0, tzinfo=NullNameTZ())
|
||||
await SmartTimeModule.set_time(inst, aware_dt)
|
||||
|
||||
call_mock.assert_awaited_once()
|
||||
args, _ = call_mock.call_args
|
||||
assert args[0] == "set_device_time"
|
||||
params = args[1]
|
||||
# 'region' must not be present when tzname() is None
|
||||
assert "region" not in params
|
||||
# sanity: timestamp and time_diff still provided
|
||||
assert isinstance(params["timestamp"], int)
|
||||
assert isinstance(params["time_diff"], int)
|
||||
|
||||
|
||||
async def test_smartcam_time_post_update_fallback_parses_timezone_str_unit(
|
||||
monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture
|
||||
):
|
||||
"""Exercise smartcam Time._post_update_hook fallback when ZoneInfo not found, parsing 'timezone' string."""
|
||||
from zoneinfo import ZoneInfoNotFoundError
|
||||
|
||||
from kasa.smartcam.modules.time import Time as CamTimeModule
|
||||
|
||||
inst = object.__new__(CamTimeModule)
|
||||
# Provide data with an unknown zone_id but with a 'timezone' string like 'UTC+02:00'
|
||||
ts = 1_700_000_000
|
||||
data = {
|
||||
"getClockStatus": {"system": {"clock_status": {"seconds_from_1970": ts}}},
|
||||
"getTimezone": {
|
||||
"system": {"basic": {"zone_id": "Nowhere/Unknown", "timezone": "UTC+02:00"}}
|
||||
},
|
||||
}
|
||||
monkeypatch.setattr(CamTimeModule, "data", property(lambda self: data))
|
||||
|
||||
# Patch directly via the module path instead of sys.modules lookup
|
||||
mocker.patch(
|
||||
"kasa.smartcam.modules.time.CachedZoneInfo.get_cached_zone_info",
|
||||
new=AsyncMock(side_effect=ZoneInfoNotFoundError("missing on host")),
|
||||
)
|
||||
|
||||
await CamTimeModule._post_update_hook(inst)
|
||||
|
||||
# Check timezone fallback parsed to +02:00
|
||||
now_local = datetime.now(inst.timezone)
|
||||
assert inst.timezone.utcoffset(now_local) == timedelta(hours=2)
|
||||
|
||||
# Check time set from seconds_from_1970 and is tz-aware with the chosen tz
|
||||
assert isinstance(inst.time, datetime)
|
||||
assert inst.time.tzinfo == inst.timezone
|
||||
assert int(inst.time.timestamp()) == ts
|
||||
|
||||
Reference in New Issue
Block a user