diff --git a/src/evohomeasync/main.py b/src/evohomeasync/main.py index 7bfbf30..a705509 100644 --- a/src/evohomeasync/main.py +++ b/src/evohomeasync/main.py @@ -69,9 +69,8 @@ async def update( /, *, _reset_config: bool = False, - # _dont_update_status: bool = False, ) -> list[EvoLocConfigDictT] | None: - """Retrieve the latest state of the installation and it's locations. + """Retrieve the latest state of the user's locations. If required, or when `_reset_config` is true, first retrieves the user information. diff --git a/src/evohomeasync/schemas.py b/src/evohomeasync/schemas.py index d6dede4..59245cb 100644 --- a/src/evohomeasync/schemas.py +++ b/src/evohomeasync/schemas.py @@ -3,7 +3,7 @@ from __future__ import annotations from enum import EnumCheck, StrEnum, verify -from typing import TYPE_CHECKING, Any, Final, NewType, TypedDict, TypeVar +from typing import TYPE_CHECKING, Any, Final, NewType, NotRequired, TypedDict, TypeVar import voluptuous as vol @@ -164,7 +164,7 @@ def _factory_location_response( vol.Required(fnc("locationOwnerName")): str, vol.Required(fnc("locationOwnerUserName")): vol.All(str, vol.Length(min=1)), vol.Required(fnc("canSearchForContractors")): bool, - vol.Required(fnc("contractor")): {str: dict}, # ContractorResponse + vol.Optional(fnc("contractor")): {str: dict}, # ContractorResponse }, extra=vol.ALLOW_EXTRA, ) @@ -311,7 +311,7 @@ class TccLocationResponseT(TypedDict): locationOwnerName: str locationOwnerUserName: str canSearchforcontractors: bool - contractor: dict[str, Any] # ContractorResponse + contractor: NotRequired[dict[str, Any]] # ContractorResponse class TccDeviceResponseT(TypedDict): @@ -456,7 +456,7 @@ class EvoLocConfigDictT(TypedDict): location_owner_name: str location_owner_user_name: str can_searchforcontractors: bool - contractor: dict[str, Any] # ContractorResponse + contractor: NotRequired[dict[str, Any]] # ContractorResponse class EvoGwyConfigDictT(TypedDict): diff --git a/src/evohomeasync2/gateway.py b/src/evohomeasync2/gateway.py index d5c6c3c..f803f6d 100644 --- a/src/evohomeasync2/gateway.py +++ b/src/evohomeasync2/gateway.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Final +from typing import TYPE_CHECKING, Final, NoReturn from evohome.helpers import camel_to_snake @@ -66,7 +66,18 @@ def gatewayId(self) -> str: # noqa: N802 def mac_address(self) -> str: return self._config[SZ_MAC] + async def _get_status(self) -> NoReturn: + """Get the latest state of the gateway and update its status attr. + + It is more efficient to call Location.update() as all descendants are updated + with a single GET. Returns the raw JSON of the latest state. + """ + + raise NotImplementedError + def _update_status(self, status: EvoGwyStatusResponseT) -> None: + """Update the GWY's status and cascade to its descendants.""" + self._update_faults(status["active_faults"]) self._status = status diff --git a/src/evohomeasync2/location.py b/src/evohomeasync2/location.py index acdc31a..ee58183 100644 --- a/src/evohomeasync2/location.py +++ b/src/evohomeasync2/location.py @@ -103,6 +103,28 @@ def country(self) -> str: def name(self) -> str: return self._config[SZ_NAME] + async def _get_config(self) -> EvoLocConfigResponseT: + """Get the latest state of the gateway and update its status attr. + + Usually called when DST starts/stops or the location's DST config changes (i.e. + there is no _update_config() method). Returns the raw JSON of the latest config. + """ + + # it is assumed that only the location's TZ/DST info can change + # so no ?includeTemperatureControlSystems=True + + config: EvoLocConfigResponseT = await self._auth.get( + f"location/{self._id}/installationInfo" # TODO: add schema + ) # type: ignore[assignment] + + self._config = config[SZ_LOCATION_INFO] + + # new TzInfo object, or update the existing one? + self._tzinfo = _create_tzinfo(self.time_zone_info, dst_enabled=self.dst_enabled) + # lf._tzinfo._update(time_zone_info=time_zone_info, use_dst_switching=use_dst_switching) + + return config + @property def time_zone_info(self) -> EvoTimeZoneInfoT: """Return the time zone information for the location. @@ -142,42 +164,37 @@ def now(self) -> dt: # always returns a TZ-aware dtm async def update( self, *, _update_time_zone_info: bool = False ) -> EvoLocStatusResponseT: - """Get the latest state of the location and update its status. + """Get the latest state of the location and update its status attrs. Will also update the status of its gateways, their TCSs, and their DHW/zones. - Returns the raw JSON of the latest state. """ if _update_time_zone_info: - await self._update_config() + await self._get_config() - status: EvoLocStatusResponseT = await self._auth.get( - f"{self._TYPE}/{self.id}/status?includeTemperatureControlSystems=True", - schema=self.SCH_STATUS, - ) # type: ignore[assignment] + status = await self._get_status() self._update_status(status) return status - async def _update_config(self) -> None: - """Usually called when DST starts/stops or the location's DST config changes.""" + async def _get_status(self) -> EvoLocStatusResponseT: + """Get the latest state of the location and update its status attr. - # it is assumed that only the location's TZ/DST info can change - # so no ?includeTemperatureControlSystems=True + Returns the raw JSON of the latest state. + """ - config: EvoLocConfigResponseT = await self._auth.get( - f"location/{self._id}/installationInfo" # TODO: add schema + status: EvoLocStatusResponseT = await self._auth.get( + f"{self._TYPE}/{self.id}/status?includeTemperatureControlSystems=True", + schema=self.SCH_STATUS, ) # type: ignore[assignment] - self._config = config[SZ_LOCATION_INFO] - - # new TzInfo object, or update the existing one? - self._tzinfo = _create_tzinfo(self.time_zone_info, dst_enabled=self.dst_enabled) - # lf._tzinfo._update(time_zone_info=time_zone_info, use_dst_switching=use_dst_switching) + self._update_status(status) + return status def _update_status(self, status: EvoLocStatusResponseT) -> None: - """Update the location's latest status (and its gateways and their TCSs).""" + """Update the LOC's status and cascade to its descendants.""" + # No ActiveFaults in location node of status self._status = status diff --git a/src/evohomeasync2/main.py b/src/evohomeasync2/main.py index 754f455..85414f7 100644 --- a/src/evohomeasync2/main.py +++ b/src/evohomeasync2/main.py @@ -71,16 +71,16 @@ def __init__( def __str__(self) -> str: return f"{self.__class__.__name__}(auth='{self.auth}')" - async def update( # noqa: C901 + async def update( self, /, *, _reset_config: bool = False, _dont_update_status: bool = False, ) -> list[EvoLocConfigResponseT]: - """Retrieve the latest state of the installation and it's locations. + """Retrieve the latest state of the user's' locations. - If required, or when `_reset_config` is true, first retrieves the user + If required, or whenever `_reset_config` is true, first retrieves the user information & installation configuration. If `_disable_status_update` is True, does not update the status of each location @@ -89,7 +89,27 @@ async def update( # noqa: C901 if _reset_config: self._user_info = None - self._user_locs = None + self._user_locs = None # and thus self._locations, etc. + + await self._get_config(_dont_update_status=_dont_update_status) + + assert self._locations is not None # mypy (internal hint) + + if not _dont_update_status: # see warning, above + for loc in self._locations: + await loc.update() + + assert self._user_locs is not None # mypy (internal hint) + + return self._user_locs + + async def _get_config( + self, /, *, _dont_update_status: bool = False + ) -> list[EvoLocConfigResponseT]: + """Ensures the config of the user and their locations. + + If required, first retrieves the user information & installation configuration. + """ if self._user_info is None: # will handle a bad access_token url = "userAccount" @@ -137,12 +157,6 @@ async def update( # noqa: C901 "limits by individually updating only the necessary locations." ) - assert self._locations is not None # mypy (internal hint) - - if not _dont_update_status: # see warning, above - for loc in self._locations: - await loc.update() - return self._user_locs @property diff --git a/src/evohomeasync2/system.py b/src/evohomeasync2/system.py index ecf6f3f..a93b24f 100644 --- a/src/evohomeasync2/system.py +++ b/src/evohomeasync2/system.py @@ -5,7 +5,7 @@ from __future__ import annotations import json -from typing import TYPE_CHECKING, Final +from typing import TYPE_CHECKING, Final, NoReturn from evohome.helpers import camel_to_snake @@ -104,30 +104,6 @@ def __init__(self, gateway: Gateway, config: EvoTcsConfigResponseT) -> None: if dhw_entry := config.get(SZ_DHW): self.hotwater = HotWater(self, dhw_entry) - def _update_status(self, status: EvoTcsStatusResponseT) -> None: - self._update_faults(status["active_faults"]) - self._status = status - - for zon_status in self._status[SZ_ZONES]: - if zone := self.zones_by_id.get(zon_status[SZ_ZONE_ID]): - zone._update_status(zon_status) - - else: - self._logger.warning( - f"{self}: zone_id='{zon_status[SZ_ZONE_ID]}' not known" - ", (has the system configuration been changed?)" - ) - - if dhw_status := self._status.get(SZ_DHW): - if self.hotwater and self.hotwater.id == dhw_status[SZ_DHW_ID]: - self.hotwater._update_status(dhw_status) - - else: - self._logger.warning( - f"{self}: dhw_id='{dhw_status[SZ_DHW_ID]}' not known" - ", (has the system configuration been changed?)" - ) - @property # TODO: deprecate in favour of .id attr def systemId(self) -> str: # noqa: N802 return self._id @@ -136,11 +112,6 @@ def systemId(self) -> str: # noqa: N802 def model(self) -> TcsModelType: return self._config[SZ_MODEL_TYPE] - @property - def zones_by_name(self) -> dict[str, Zone]: - """Return the zones by name (names are not fixed attrs).""" - return {zone.name: zone for zone in self.zones} - @property def allowed_system_modes(self) -> tuple[EvoAllowedSystemModeResponseT, ...]: """ @@ -161,6 +132,46 @@ def allowed_system_modes(self) -> tuple[EvoAllowedSystemModeResponseT, ...]: def modes(self) -> tuple[SystemMode, ...]: return tuple(d[SZ_SYSTEM_MODE] for d in self.allowed_system_modes) + @property + def zones_by_name(self) -> dict[str, Zone]: + """Return the zones by name (names are not fixed attrs).""" + return {zone.name: zone for zone in self.zones} + + async def _get_status(self) -> NoReturn: + """Get the latest state of the control system and update its status attrs. + + It is more efficient to call Location.update() as all descendants are updated + with a single GET. Returns the raw JSON of the latest state. + """ + + raise NotImplementedError + + def _update_status(self, status: EvoTcsStatusResponseT) -> None: + """Update the TCS's status and cascade to its descendants.""" + + self._update_faults(status["active_faults"]) + self._status = status + + for zon_status in self._status[SZ_ZONES]: + if zone := self.zones_by_id.get(zon_status[SZ_ZONE_ID]): + zone._update_status(zon_status) + + else: + self._logger.warning( + f"{self}: zone_id='{zon_status[SZ_ZONE_ID]}' not known" + ", (has the system configuration been changed?)" + ) + + if dhw_status := self._status.get(SZ_DHW): + if self.hotwater and self.hotwater.id == dhw_status[SZ_DHW_ID]: + self.hotwater._update_status(dhw_status) + + else: + self._logger.warning( + f"{self}: dhw_id='{dhw_status[SZ_DHW_ID]}' not known" + ", (has the system configuration been changed?)" + ) + @property def system_mode_status(self) -> EvoSystemModeStatusResponseT: """ diff --git a/src/evohomeasync2/zone.py b/src/evohomeasync2/zone.py index ce15091..8b4c2d2 100644 --- a/src/evohomeasync2/zone.py +++ b/src/evohomeasync2/zone.py @@ -391,13 +391,11 @@ def __init__(self, entity_id: str, tcs: ControlSystem) -> None: self.location = tcs.location self.tcs = tcs - async def _update(self) -> EvoDhwStatusResponseT | EvoZonStatusResponseT: - """Get the latest state of the DHW/zone and update its status. + async def _get_status(self) -> EvoDhwStatusResponseT | EvoZonStatusResponseT: + """Get the latest state of the DHW/zone and update its status attrs. - It is more efficient to call Location.update() as all zones are updated - with a single GET. - - Returns the raw JSON of the latest state. + It is more efficient to call Location.update() as all descendants are updated + with a single GET. Returns the raw JSON of the latest state. """ status: EvoDhwStatusResponseT | EvoZonStatusResponseT = await self._auth.get( # type: ignore[assignment] @@ -410,6 +408,8 @@ async def _update(self) -> EvoDhwStatusResponseT | EvoZonStatusResponseT: def _update_status( self, status: EvoDhwStatusResponseT | EvoZonStatusResponseT ) -> None: + """Update the DHW/ZON's status.""" + self._update_faults(status["active_faults"]) self._status = status diff --git a/tests/tests/test_v2_auth.py b/tests/tests/test_v2_auth.py index ee8ee10..af3bce8 100644 --- a/tests/tests/test_v2_auth.py +++ b/tests/tests/test_v2_auth.py @@ -199,10 +199,10 @@ async def test_token_manager( assert await cache_manager.get_access_token() == "new_access_token..." assert caplog.records[0].message == ( - "Null/Expired/Invalid access_token, re-authenticating." + "Null/Expired/Invalid access_token, will re-authenticate..." ) assert caplog.records[1].message == ( - "Authenticating with the refresh_token..." + "Authenticating with the refresh_token" ) assert caplog.records[2].message.startswith( "POST https://tccna.resideo.com/Auth/OAuth/Token" diff --git a/tests/tests_rf/test_v2_apis.py b/tests/tests_rf/test_v2_apis.py index dc89ebb..eff80d4 100644 --- a/tests/tests_rf/test_v2_apis.py +++ b/tests/tests_rf/test_v2_apis.py @@ -7,7 +7,8 @@ import pytest import evohomeasync2 as evo2 -from evohomeasync2.schemas import TCC_GET_DHW_SCHEDULE, TCC_GET_ZON_SCHEDULE, SystemMode +from evohome.helpers import camel_to_snake +from evohomeasync2.schemas import SystemMode, factory_dhw_schedule, factory_zon_schedule from evohomeasync2.schemas.const import S2_MODE from tests.const import _DBG_USE_REAL_AIOHTTP @@ -15,124 +16,162 @@ from .common import skipif_auth_failed if TYPE_CHECKING: - from evohomeasync2.zone import Zone from tests.conftest import EvohomeClientv2 +def _get_dhw(evo: EvohomeClientv2) -> evo2.HotWater | None: + """Return the DHW object of a TCS.""" + for loc in evo.locations: + for gwy in loc.gateways: + for tcs in gwy.systems: + if tcs.hotwater: + return tcs.hotwater + return None + + +def _get_zon(evo: EvohomeClientv2) -> evo2.Zone | None: + """Return the Zone object of a TCS.""" + for loc in evo.locations: + for gwy in loc.gateways: + for tcs in gwy.systems: + if tcs.zones: + return tcs.zones[0] + return None + + ####################################################################################### -async def _test_basics_apis(evo: EvohomeClientv2) -> None: - """Test authentication, `user_account()` and `installation()`.""" +async def _test_usr_apis(evo: EvohomeClientv2) -> None: + """Test User and Location methods. + + Includes: evo.user_account(), evo.installation() and loc.update() methods. + """ - # STEP 1: retrieve base data + # STEP 1: retrieve config only: evo.user_account(), evo.installation() await evo.update(_dont_update_status=True) assert evo2.main.SCH_USER_ACCOUNT(evo.user_account) assert evo2.main.SCH_USER_LOCATIONS(evo.user_installation) - # STEP 4: Status, GET /location/{loc.id}/status + # STEP 2: GET /location/{loc.id}/status for loc in evo.locations: loc_status = await loc.update() assert evo2.Location.SCH_STATUS(loc_status) -async def _test_sched__apis(evo: EvohomeClientv2) -> None: - """Test `get_schedule()` and `get_schedule()`.""" +async def _test_tcs_apis(evo: EvohomeClientv2) -> None: + """Test ControlSystem methods. - # STEP 1: retrieve base data - await evo.update() + Includes tcs.update() and tcs.set_mode(). + Does not include tcs.get_schedules(), tcs.set_schedules(). + """ - # STEP 2: GET & PUT /{x._TYPE}/{x.id}/schedule - if dhw := evo._get_single_tcs().hotwater: - sched_hw = await dhw.get_schedule() - assert TCC_GET_DHW_SCHEDULE(sched_hw) - await dhw.set_schedule(sched_hw) + # STEP 1: retrieve config only + await evo.update(_dont_update_status=False) - zone: Zone | None + # STEP 2: GET /temperatureControlSystem/{tcs.id}/status + tcs = evo.locations[0].gateways[0].systems[0] - if (zone := evo._get_single_tcs().zones[0]) and zone.id != faked.GHOST_ZONE_ID: - schedule = await zone.get_schedule() - assert TCC_GET_ZON_SCHEDULE(schedule) - await zone.set_schedule(schedule) + # tcs_status = await tcs._update() + # assert evo2.ControlSystem.SCH_STATUS(tcs_status) - if zone := evo._get_single_tcs().zones_by_id.get(faked.GHOST_ZONE_ID): - try: - schedule = await zone.get_schedule() - except evo2.InvalidScheduleError: - pass - else: - pytest.fail("Did not raise expected exception") + assert tcs.system_mode_status is not None + mode = tcs.system_mode_status[S2_MODE] + assert mode in SystemMode -async def _test_update_apis(evo: EvohomeClientv2) -> None: - """Test `_update()` for DHW/zone.""" + # STEP 3: PUT /temperatureControlSystem/{tcs.id}/mode + await tcs.set_mode(SystemMode.AWAY) + await evo.update() - # STEP 1: retrieve config - await evo.update(_dont_update_status=True) + await tcs.set_mode(mode) - # STEP 2: GET /{x._TYPE}/{x.id}/status - if dhw := evo._get_single_tcs().hotwater: - dhw_status = await dhw._update() - assert evo2.HotWater.SCH_STATUS(dhw_status) - if zone := evo._get_single_tcs().zones[0]: - zone_status = await zone._update() - assert evo2.Zone.SCH_STATUS(zone_status) +async def _test_dhw_apis(evo: EvohomeClientv2) -> None: + """Test Hotwater methods. + Includes dhw._update() and dhw.get_schedule(). + """ -async def _test_system_apis(evo: EvohomeClientv2) -> None: - """Test `set_mode()` for TCS.""" + # STEP 1: retrieve config only + await evo.update(_dont_update_status=True) - # STEP 1: retrieve base data - await evo.update() + if not (dhw := _get_dhw(evo)): + pytest.skip("No DHW found in TCS") - # STEP 2: GET /{x._TYPE}/{x.id}/status - try: - tcs = evo._get_single_tcs() - except evo2.NoSingleTcsError: - tcs = evo.locations[0].gateways[0].systems[0] + # STEP 2: GET /domesticHotWater/{dhw.id}/??? + dhw_status = await dhw._get_status() + assert evo2.HotWater.SCH_STATUS(dhw_status) - assert tcs.system_mode_status is not None - mode = tcs.system_mode_status[S2_MODE] + # STEP 2: GET /domesticHotWater/{dhw.id}/get_schedule + schedule = await dhw.get_schedule() + assert factory_dhw_schedule(camel_to_snake)({"daily_schedules": schedule}) - assert mode in SystemMode + await dhw.set_schedule(schedule) - await tcs.set_mode(SystemMode.AWAY) - await evo.update() - await tcs.set_mode(mode) +async def _test_zon_apis(evo: EvohomeClientv2) -> None: + """Test Zone methods. + Includes zon._update() and zon.get_schedule(). + """ -####################################################################################### + # STEP 1: retrieve config only + await evo.update(_dont_update_status=True) + if not (zone := _get_zon(evo)): + pytest.skip("No zones found in TCS") -@skipif_auth_failed -async def test_basics(evohome_v2: EvohomeClientv2) -> None: - """Test authentication, `user_account()` and `installation()`.""" - await _test_basics_apis(evohome_v2) + # STEP 2: GET /temperatureZone/{zon.id}/status + zon_status = await zone._get_status() + assert evo2.Zone.SCH_STATUS(zon_status) + # STEP 2: GET /temperatureZone/{zon.id}/get_schedule + if zone.id != faked.GHOST_ZONE_ID: + schedule = await zone.get_schedule() + assert factory_zon_schedule(camel_to_snake)({"daily_schedules": schedule}) -@skipif_auth_failed -async def _test_sched_(evohome_v2: EvohomeClientv2) -> None: - """Test `get_schedule()` and `get_schedule()`.""" - await _test_sched__apis(evohome_v2) + await zone.set_schedule(schedule) + + if zone := zone.tcs.zones_by_id.get(faked.GHOST_ZONE_ID): + try: + schedule = await zone.get_schedule() + except evo2.InvalidScheduleError: + pass + else: + pytest.fail("Did not raise expected exception") + + +####################################################################################### @skipif_auth_failed -async def test_status(evohome_v2: EvohomeClientv2) -> None: - """Test `_update()` for DHW/zone.""" - await _test_update_apis(evohome_v2) +async def test_usr_apis(evohome_v2: EvohomeClientv2) -> None: + """Test user_account() and installation().""" + await _test_usr_apis(evohome_v2) @skipif_auth_failed -async def test_system(evohome_v2: EvohomeClientv2) -> None: - """Test `set_mode()` for TCS""" +async def test_tcs(evohome_v2: EvohomeClientv2) -> None: + """Test set_mode() for TCS""" try: - await _test_system_apis(evohome_v2) + await _test_tcs_apis(evohome_v2) except NotImplementedError: # TODO: implement if _DBG_USE_REAL_AIOHTTP: raise pytest.skip("Mocked server API not implemented") + + +@skipif_auth_failed +async def test_dhw_apis(evohome_v2: EvohomeClientv2) -> None: + """Test get_schedule() and get_schedule().""" + await _test_dhw_apis(evohome_v2) + + +@skipif_auth_failed +async def test_zon_apis(evohome_v2: EvohomeClientv2) -> None: + """Test _update() for DHW/zone.""" + await _test_zon_apis(evohome_v2)