From 3d313760e5edb15c5cb489be0e13664acfaec8fe Mon Sep 17 00:00:00 2001 From: Neil Sleightholm Date: Sat, 11 Jul 2026 13:38:59 +0100 Subject: [PATCH 1/6] feat: add holiday mode support (get/set/cancel) Adds get_holiday_mode/set_holiday_mode/cancel_holiday_mode to the async and sync API layers and a HiveHub wrapper accepting datetime start/end. Payload shape (GET/POST/DELETE on /holiday-mode, epoch-ms start/end + temperature) was confirmed via a HAR capture of Hive's own website, since this endpoint was previously unimplemented. Includes the diagnostic scripts used to reverse-engineer it. Co-Authored-By: Claude Opus 5 --- scripts/diagnose_heat_on_demand.py | 173 ++++++++++++++++++++++++ scripts/diagnose_holiday_mode.py | 203 +++++++++++++++++++++++++++++ src/api/hive_api.py | 26 ++++ src/api/hive_async_api.py | 36 +++++ src/devices/hub.py | 62 ++++++++- tests/module/test_hub.py | 46 +++++++ tests/unit/test_hive_api.py | 62 +++++++++ tests/unit/test_hive_async_api.py | 53 ++++++++ 8 files changed, 660 insertions(+), 1 deletion(-) create mode 100644 scripts/diagnose_heat_on_demand.py create mode 100644 scripts/diagnose_holiday_mode.py diff --git a/scripts/diagnose_heat_on_demand.py b/scripts/diagnose_heat_on_demand.py new file mode 100644 index 0000000..1ec33c1 --- /dev/null +++ b/scripts/diagnose_heat_on_demand.py @@ -0,0 +1,173 @@ +"""Diagnose the Heat on Demand MALFORMED_REQUEST error (issue #204). + +WARNING: this talks to your REAL Hive account and REAL thermostat, not a +sandbox. It stops at the first payload shape that gets a verified state +change (HTTP 2xx AND a re-fetch confirming autoBoost actually changed -- +Hive's backend has been observed returning 200 for shapes it silently +ignores). A verified success will actually flip Heat on Demand on your real +device. + +Credentials are never hard-coded or logged: set HIVE_USERNAME / HIVE_PASSWORD +as environment variables before running, or leave them unset and you'll be +prompted (password via getpass, hidden input). + +Prerequisite: `pip install -e .` from the repo root so `apyhiveapi` resolves +to this local source tree, then run: + + python scripts/diagnose_heat_on_demand.py +""" + +import asyncio +import getpass +import json +import logging +import os + +from apyhiveapi import Hive +from apyhiveapi.helper.hive_exceptions import HiveApiError, HiveReauthRequired + +logging.basicConfig(level=logging.DEBUG, format="%(name)s: %(message)s") + +# Ordered from "reproduce the known bug" to increasingly different shapes. +# None of these are confirmed against Hive's backend -- they're guesses. A +# 200 response is NOT enough to call a shape correct: Hive's backend was +# observed accepting {"auto_boost": "ENABLED"} with HTTP 200 while silently +# not applying it. So every 2xx response gets re-verified against a re-fetch +# of the device before being trusted. +CANDIDATE_BODIES = [ + ("current code (expected to fail with MALFORMED_REQUEST)", {"autoBoost": "ENABLED"}), + ("boolean value", {"autoBoost": True}), + ("lowercase value", {"autoBoost": "enabled"}), + ( + "snake_case field name (known false positive: 200 but no effect)", + {"auto_boost": "ENABLED"}, + ), + ("nested object matching read-side shape", {"autoBoost": {"active": True}}), + ( + "nested object, full read-side shape echoed back", + None, # filled in once we know the device's current props + ), + ("snake_case field, nested object", {"auto_boost": {"active": True}}), + ("nested under props", {"props": {"autoBoost": {"active": True}}}), + ("wrapped in products array", None), # filled in once we know node_id +] + + +async def attempt(hive: Hive, node_type: str, node_id: str, label: str, body: dict) -> bool: + """POST *body* to the node's state endpoint and report the outcome.""" + url = hive.api.urls["nodes"].format(node_type, node_id) + print(f"\n=== {label} ===\nPOST {url}\nBody: {json.dumps(body)}") + try: + resp = await hive.api._call_endpoint( # noqa: SLF001 -- deliberate direct call for diagnostics + "post", url, data=json.dumps(body) + ) + except HiveApiError: + print("-> Raised HiveApiError (see log line above for HTTP status + response body)") + return False + status = resp.get("original") + print(f"-> HTTP {status} — parsed: {resp.get('parsed')}") + return str(status).startswith("20") + + +async def verify_state_changed(hive: Hive, node_id: str, baseline: dict) -> dict | None: + """Re-fetch the device and return the new autoBoost value if it differs from *baseline*.""" + await asyncio.sleep(2) # give Hive's backend a moment to settle + await hive.get_devices("No_ID") + new_boost = hive.data.products[node_id].get("props", {}).get("autoBoost") + print(f"Re-fetched autoBoost: {new_boost!r} (was: {baseline!r})") + return new_boost if new_boost != baseline else None + + +async def login_with_sms_fallback(hive: Hive) -> None: + """Log in, prompting for an SMS code if challenged. + + If Hive's "remembered device" flow rejects us (HiveReauthRequired -- + device is known but Cognito wants it re-verified via SMS, a path + hive.login() doesn't surface directly), drop any stale device + credentials and retry once as a plain fresh login so Cognito issues a + top-level SMS_MFA challenge instead. + """ + try: + login_result = await hive.login() + except HiveReauthRequired: + print( + "Device re-authentication required — retrying as a fresh login " + "to force an SMS code prompt..." + ) + hive.auth.device_group_key = None + hive.auth.device_key = None + hive.auth.device_password = None + login_result = await hive.login() + + if login_result and login_result.get("ChallengeName") == hive.auth.SMS_MFA_CHALLENGE: + code = input("Enter the SMS 2FA code sent to your phone: ") + await hive.sms2fa(code, login_result) + + +async def main() -> None: + username = os.environ.get("HIVE_USERNAME") or input("Hive username (email): ") + password = os.environ.get("HIVE_PASSWORD") or getpass.getpass("Hive password: ") + + async with Hive(username=username, password=password) as hive: + await login_with_sms_fallback(hive) + + await hive.start_session() + + heating_products = [ + p for p in hive.data.products.values() if p.get("type") == "heating" + ] + if not heating_products: + print("No 'heating' product found on this account — nothing to test.") + return + + product = heating_products[0] + node_id = product["id"] + current_boost = product.get("props", {}).get("autoBoost") + print( + f"\nTarget device: {node_id} (current autoBoost value: {current_boost!r})" + ) + input( + "This will send several test requests to your REAL thermostat and may " + "change its Heat on Demand setting. Press Enter to continue, Ctrl+C to abort..." + ) + + full_shape_echo = dict(current_boost) if isinstance(current_boost, dict) else {} + full_shape_echo["active"] = True + for i, (label, body) in enumerate(CANDIDATE_BODIES): + if body is None and "full read-side shape" in label: + CANDIDATE_BODIES[i] = (label, {"autoBoost": full_shape_echo}) + elif body is None and "products array" in label: + CANDIDATE_BODIES[i] = ( + label, + {"products": [{"id": node_id, "props": {"autoBoost": full_shape_echo}}]}, + ) + + baseline = current_boost + for label, body in CANDIDATE_BODIES: + got_2xx = await attempt(hive, "heating", node_id, label, body) + if not got_2xx: + await asyncio.sleep(1) + continue + + new_boost = await verify_state_changed(hive, node_id, baseline) + if new_boost is not None: + print(f"\n*** VERIFIED SUCCESS: {label} -> {json.dumps(body)} ***") + print(f"autoBoost actually changed: {baseline!r} -> {new_boost!r}") + print("This is the payload shape to report on issue #204.") + return + + print( + "HTTP 2xx but the device did NOT actually change — false positive, " + "trying next shape." + ) + await asyncio.sleep(1) + + print( + "\nNone of the candidate shapes produced a verified change. Copy the " + "HTTP status + response bodies above into issue #204 -- the response " + "text often names the exact field Hive's backend is rejecting." + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/diagnose_holiday_mode.py b/scripts/diagnose_holiday_mode.py new file mode 100644 index 0000000..91258fc --- /dev/null +++ b/scripts/diagnose_holiday_mode.py @@ -0,0 +1,203 @@ +"""Diagnose Hive's undocumented /holiday-mode endpoint. + +WARNING: this talks to your REAL Hive account and affects REAL heating +behaviour while active. /holiday-mode has never been implemented in this +library (checked back to 2021) -- only the URL was ever recorded, so we are +starting from nothing. + +Safety approach: + 1. GET first (read-only) to see the current shape, if any. + 2. Only if GET doesn't reveal enough, try a short-window candidate PUT + (default: 5 minutes from now), verified by re-fetching afterwards. + 3. Immediately attempt to cancel/revert after a verified success, and + verify the cancellation too. + 4. Every write step requires an explicit confirmation prompt. + +Credentials are never hard-coded or logged: set HIVE_USERNAME / HIVE_PASSWORD +as environment variables before running, or leave them unset and you'll be +prompted (password via getpass, hidden input). + +Prerequisite: `pip install -e .` from the repo root, then run: + + python scripts/diagnose_holiday_mode.py +""" + +import asyncio +import getpass +import json +import logging +import os +from datetime import datetime, timedelta, timezone + +from apyhiveapi import Hive +from apyhiveapi.helper.hive_exceptions import HiveApiError, HiveReauthRequired + +logging.basicConfig(level=logging.DEBUG, format="%(name)s: %(message)s") + +TEST_WINDOW_MINUTES = 5 + +# Guesses only -- informed by whatever GET reveals, tried in order. Each is +# verified by re-fetching /holiday-mode (and the target product's autoBoost) +# afterwards; a 2xx alone is not trusted (autoBoost diagnostics showed Hive +# returns 200 for shapes it silently ignores). +def build_candidate_bodies(start: datetime, end: datetime) -> list[tuple[str, dict]]: + start_iso = start.strftime("%Y-%m-%dT%H:%M:%S.000Z") + end_iso = end.strftime("%Y-%m-%dT%H:%M:%S.000Z") + return [ + ("camelCase start/end", {"startDate": start_iso, "endDate": end_iso}), + ("snake_case start/end", {"start_date": start_iso, "end_date": end_iso}), + ("short keys", {"start": start_iso, "end": end_iso}), + ( + "camelCase with enabled flag", + {"enabled": True, "startDate": start_iso, "endDate": end_iso}, + ), + ] + + +async def get_holiday_mode(hive: Hive) -> dict: + """GET the current holiday-mode resource. Read-only, always safe.""" + url = hive.api.urls["holiday_mode"] + print(f"\nGET {url}") + resp = await hive.api._call_endpoint("get", url) # noqa: SLF001 + print(f"-> HTTP {resp.get('original')} — parsed: {resp.get('parsed')}") + return resp + + +async def attempt(hive: Hive, label: str, method: str, body: dict | None) -> tuple[bool, dict]: + url = hive.api.urls["holiday_mode"] + print(f"\n=== {label} ({method.upper()}) ===\nBody: {json.dumps(body)}") + try: + resp = await hive.api._call_endpoint( # noqa: SLF001 + method, url, data=json.dumps(body) if body is not None else None + ) + except HiveApiError: + print("-> Raised HiveApiError (see log line above for HTTP status + response body)") + return False, {} + status = resp.get("original") + print(f"-> HTTP {status} — parsed: {resp.get('parsed')}") + return str(status).startswith("20"), resp + + +async def login_with_sms_fallback(hive: Hive) -> None: + """Log in, prompting for an SMS code if challenged (see diagnose_heat_on_demand.py).""" + try: + login_result = await hive.login() + except HiveReauthRequired: + print( + "Device re-authentication required — retrying as a fresh login " + "to force an SMS code prompt..." + ) + hive.auth.device_group_key = None + hive.auth.device_key = None + hive.auth.device_password = None + login_result = await hive.login() + + if login_result and login_result.get("ChallengeName") == hive.auth.SMS_MFA_CHALLENGE: + code = input("Enter the SMS 2FA code sent to your phone: ") + await hive.sms2fa(code, login_result) + + +async def main() -> None: + username = os.environ.get("HIVE_USERNAME") or input("Hive username (email): ") + password = os.environ.get("HIVE_PASSWORD") or getpass.getpass("Hive password: ") + + async with Hive(username=username, password=password) as hive: + await login_with_sms_fallback(hive) + await hive.start_session() + + print("\nStep 1: read-only GET of /holiday-mode (safe, no changes made).") + await get_holiday_mode(hive) + + proceed = input( + "\nStep 2 will send a WRITE request that may activate holiday mode " + f"on your REAL account for up to {TEST_WINDOW_MINUTES} minutes, and will " + "immediately try to cancel it afterwards. Type 'yes' to continue, " + "anything else to stop here: " + ) + if proceed.strip().lower() != "yes": + print("Stopping after the read-only GET, as requested.") + return + + now = datetime.now(timezone.utc) + end = now + timedelta(minutes=TEST_WINDOW_MINUTES) + candidates = build_candidate_bodies(now, end) + + heating_products = [ + p for p in hive.data.products.values() if p.get("type") == "heating" + ] + node_id = heating_products[0]["id"] if heating_products else None + baseline_boost = ( + heating_products[0].get("props", {}).get("autoBoost") + if heating_products + else None + ) + + for label, body in candidates: + ok, _ = await attempt(hive, label, "put", body) + if not ok: + await asyncio.sleep(1) + continue + + print("Got 2xx — verifying via re-fetch (GET /holiday-mode + device autoBoost)...") + await asyncio.sleep(2) + hm_resp = await get_holiday_mode(hive) + new_boost = None + if node_id: + await hive.get_devices("No_ID") + new_boost = hive.data.products[node_id].get("props", {}).get("autoBoost") + print(f"autoBoost now: {new_boost!r} (was: {baseline_boost!r})") + + verified = new_boost is not None and new_boost != baseline_boost + if not verified: + print("No verified change — treating as a false positive, trying next shape.") + await asyncio.sleep(1) + continue + + print(f"\n*** VERIFIED SUCCESS: {label} -> {json.dumps(body)} ***") + print(f"holiday-mode GET now shows: {hm_resp.get('parsed')}") + print("Attempting immediate cancellation...") + + cancelled = await try_cancel(hive) + if cancelled: + print("Cancellation verified — holiday mode is back off.") + else: + print( + "!! Could not verify cancellation. The test window was only " + f"{TEST_WINDOW_MINUTES} minutes, so it will lapse on its own, " + "but check the Hive app now to be sure." + ) + return + + print( + "\nNo candidate body produced a verified holiday-mode activation. " + "See the GET response above for whatever shape Hive did return -- " + "that's the most useful lead for a follow-up attempt." + ) + + +async def try_cancel(hive: Hive) -> bool: + """Best-effort cancellation attempt, verified via re-fetch.""" + url = hive.api.urls["holiday_mode"] + for label, method, body in [ + ("DELETE", "delete", None), + ("PUT enabled=false", "put", {"enabled": False}), + ]: + print(f"\nCancel attempt: {label}") + try: + resp = await hive.api._call_endpoint( # noqa: SLF001 + method, url, data=json.dumps(body) if body is not None else None + ) + except HiveApiError: + print("-> Raised HiveApiError") + continue + print(f"-> HTTP {resp.get('original')} — parsed: {resp.get('parsed')}") + await asyncio.sleep(2) + check = await get_holiday_mode(hive) + parsed = check.get("parsed") + if isinstance(parsed, dict) and parsed.get("enabled") is False: + return True + return False + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/api/hive_api.py b/src/api/hive_api.py index c4ae70c..199ed44 100644 --- a/src/api/hive_api.py +++ b/src/api/hive_api.py @@ -76,6 +76,10 @@ def request(self, http_method, url, jsc=None): return requests.post( url=url, headers=self.headers, data=jsc, timeout=self.timeout ) + if http_method == "DELETE": + return requests.delete( + url=url, headers=self.headers, data=jsc, timeout=self.timeout + ) raise ValueError(f"Unsupported request type: {http_method}") except Exception as e: _LOGGER.error("Request failed: %s", e) @@ -187,6 +191,28 @@ def set_action(self, n_id, data): url = self.urls["base"] + self.urls["actions"] + "/" + n_id return self._call_endpoint("POST", url, jsc) + def get_holiday_mode(self): + """Get the current holiday mode configuration.""" + url = self.urls["base"] + self.urls["holiday_mode"] + return self._call_endpoint("GET", url) + + def set_holiday_mode(self, start, end, temperature): + """Schedule holiday mode. + + Args: + start: Start time as epoch milliseconds. + end: End time as epoch milliseconds. + temperature: Frost-protection temperature to hold during holiday mode. + """ + jsc = json.dumps({"start": start, "end": end, "temperature": temperature}) + url = self.urls["base"] + self.urls["holiday_mode"] + return self._call_endpoint("POST", url, jsc) + + def cancel_holiday_mode(self): + """Cancel any scheduled or active holiday mode.""" + url = self.urls["base"] + self.urls["holiday_mode"] + return self._call_endpoint("DELETE", url, json.dumps({})) + def error(self): """An error has occurred interacting with the Hive API.""" _LOGGER.error("API error occurred - returning error response") diff --git a/src/api/hive_async_api.py b/src/api/hive_async_api.py index e68d833..9747a68 100644 --- a/src/api/hive_async_api.py +++ b/src/api/hive_async_api.py @@ -192,6 +192,42 @@ async def set_action(self, n_id, data): return {"original": "file"} return await self._call_endpoint("put", url, data=data) + async def get_holiday_mode(self): + """Get the current holiday mode configuration.""" + return await self._call_endpoint("get", self.urls["holiday_mode"]) + + async def set_holiday_mode(self, start: int, end: int, temperature: float): + """Schedule holiday mode. + + Args: + start: Start time as epoch milliseconds. + end: End time as epoch milliseconds. + temperature: Frost-protection temperature to hold during holiday mode. + """ + _LOGGER.debug( + "set_holiday_mode - Scheduling holiday mode from %s to %s at %s°.", + start, + end, + temperature, + ) + jsc = json.dumps({"start": start, "end": end, "temperature": temperature}) + try: + await self.is_file_being_used() + except FileInUse: + return {"original": "file"} + return await self._call_endpoint("post", self.urls["holiday_mode"], data=jsc) + + async def cancel_holiday_mode(self): + """Cancel any scheduled or active holiday mode.""" + _LOGGER.debug("cancel_holiday_mode - Cancelling holiday mode.") + try: + await self.is_file_being_used() + except FileInUse: + return {"original": "file"} + return await self._call_endpoint( + "delete", self.urls["holiday_mode"], data=json.dumps({}) + ) + async def error(self): """An error has occurred interacting with the Hive API.""" _LOGGER.error("HTTP error occurred during Hive API interaction.") diff --git a/src/devices/hub.py b/src/devices/hub.py index c8cfc58..52da410 100644 --- a/src/devices/hub.py +++ b/src/devices/hub.py @@ -1,9 +1,10 @@ """Hive Hub Module.""" import logging +from datetime import datetime from typing import Any -from ..helper.const import HIVETOHA +from ..helper.const import HIVETOHA, HTTP_OK from ..helper.device_handler_base import BaseDeviceHandler from ..helper.hivedataclasses import Device @@ -102,3 +103,62 @@ async def get_glass_break_status(self, device: Device): "get_glass_break_status - %s glass break status: %s", device.hive_id, result ) return result + + async def get_holiday_mode(self) -> dict | None: + """Get the current holiday mode configuration. + + Returns: + dict: Keys are active (bool), enabled (bool), start (epoch ms), + end (epoch ms) and temperature. None on failure. + """ + await self.session.hive_refresh_tokens() + resp = await self.session.api.get_holiday_mode() + if resp["original"] == HTTP_OK: + return resp["parsed"] + _LOGGER.error( + "get_holiday_mode - Failed to fetch holiday mode: HTTP %s", + resp["original"], + ) + return None + + async def set_holiday_mode( + self, start: datetime, end: datetime, temperature: float + ) -> bool: + """Schedule holiday mode. + + Args: + start: Start date/time. Naive datetimes are treated as local time. + end: End date/time. Naive datetimes are treated as local time. + temperature: Frost-protection temperature to hold during holiday mode. + + Returns: + bool: True if successful. + """ + start_ms = int(start.timestamp() * 1000) + end_ms = int(end.timestamp() * 1000) + _LOGGER.debug( + "set_holiday_mode - Scheduling holiday mode from %s to %s at %s°.", + start, + end, + temperature, + ) + await self.session.hive_refresh_tokens() + resp = await self.session.api.set_holiday_mode(start_ms, end_ms, temperature) + if resp["original"] == HTTP_OK: + return True + _LOGGER.error("set_holiday_mode - Failed: HTTP %s", resp["original"]) + return False + + async def cancel_holiday_mode(self) -> bool: + """Cancel any scheduled or active holiday mode. + + Returns: + bool: True if successful. + """ + _LOGGER.debug("cancel_holiday_mode - Cancelling holiday mode.") + await self.session.hive_refresh_tokens() + resp = await self.session.api.cancel_holiday_mode() + if resp["original"] == HTTP_OK: + return True + _LOGGER.error("cancel_holiday_mode - Failed: HTTP %s", resp["original"]) + return False diff --git a/tests/module/test_hub.py b/tests/module/test_hub.py index 06b61d7..410e1f8 100644 --- a/tests/module/test_hub.py +++ b/tests/module/test_hub.py @@ -1,6 +1,7 @@ """Tests for session polling behaviour, HiveHub sensor status, and Hive lifecycle.""" # pylint: disable=protected-access +from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock from apyhiveapi import Hive @@ -116,6 +117,51 @@ async def test_glass_break_missing_returns_none(self): assert await hub.get_glass_break_status(_make_hub_device()) is None +class TestHiveHubHolidayMode: + """Tests for HiveHub holiday mode get/set/cancel methods.""" + + def _make_hub(self, resp): + session = MagicMock() + session.hive_refresh_tokens = AsyncMock() + session.api = MagicMock() + session.api.get_holiday_mode = AsyncMock(return_value=resp) + session.api.set_holiday_mode = AsyncMock(return_value=resp) + session.api.cancel_holiday_mode = AsyncMock(return_value=resp) + return HiveHub(session=session) + + async def test_get_holiday_mode_returns_parsed_on_200(self): + payload = {"active": False, "enabled": False, "start": 1, "end": 2, "temperature": 12} + hub = self._make_hub({"original": 200, "parsed": payload}) + assert await hub.get_holiday_mode() == payload + + async def test_get_holiday_mode_returns_none_on_failure(self): + hub = self._make_hub({"original": 400, "parsed": {"error": "MALFORMED_REQUEST"}}) + assert await hub.get_holiday_mode() is None + + async def test_set_holiday_mode_returns_true_on_200(self): + hub = self._make_hub({"original": 200, "parsed": {}}) + start = datetime(2026, 8, 1, 12, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 8, 8, 12, 0, 0, tzinfo=timezone.utc) + assert await hub.set_holiday_mode(start, end, 12) is True + hub.session.api.set_holiday_mode.assert_awaited_once_with( + int(start.timestamp() * 1000), int(end.timestamp() * 1000), 12 + ) + + async def test_set_holiday_mode_returns_false_on_failure(self): + hub = self._make_hub({"original": 400, "parsed": {}}) + start = datetime(2026, 8, 1, 12, 0, 0, tzinfo=timezone.utc) + end = datetime(2026, 8, 8, 12, 0, 0, tzinfo=timezone.utc) + assert await hub.set_holiday_mode(start, end, 12) is False + + async def test_cancel_holiday_mode_returns_true_on_200(self): + hub = self._make_hub({"original": 200, "parsed": {"set": True}}) + assert await hub.cancel_holiday_mode() is True + + async def test_cancel_holiday_mode_returns_false_on_failure(self): + hub = self._make_hub({"original": 400, "parsed": {}}) + assert await hub.cancel_holiday_mode() is False + + class TestHiveLifecycle: """Tests for Hive context manager.""" diff --git a/tests/unit/test_hive_api.py b/tests/unit/test_hive_api.py index 6e272bd..9d52845 100644 --- a/tests/unit/test_hive_api.py +++ b/tests/unit/test_hive_api.py @@ -118,7 +118,14 @@ def test_post_method_calls_requests_post(self): def test_unsupported_method_raises_value_error(self): api = _make_api() with pytest.raises(ValueError, match="Unsupported request type"): + api.request("PATCH", "https://example.com/") + + def test_delete_method_calls_requests_delete(self): + api = _make_api() + with patch("apyhiveapi.api.hive_api.requests.delete") as mock_delete: + mock_delete.return_value = _make_mock_response(200) api.request("DELETE", "https://example.com/") + mock_delete.assert_called_once() def test_exception_is_reraised(self): api = _make_api() @@ -615,6 +622,61 @@ def test_runtime_error_calls_error(self): assert api.json_return["original"] == "Error making API call" +# --------------------------------------------------------------------------- +# Tests: HiveApi holiday mode +# --------------------------------------------------------------------------- + + +class TestHolidayMode: + def test_get_holiday_mode_returns_parsed_json(self): + api = _make_api() + payload = { + "active": False, + "enabled": False, + "start": 1783006500000, + "end": 1783296000000, + "temperature": 12, + "status": "OK", + } + mock_resp = _make_mock_response(200, json_data=payload) + + with patch.object(api, "request", return_value=mock_resp): + result = api.get_holiday_mode() + + assert result["original"] == 200 + assert result["parsed"] == payload + + def test_set_holiday_mode_posts_epoch_ms_body(self): + api = _make_api() + payload = {"start": 1783767707701, "end": 1784372507701, "temperature": 12} + mock_resp = _make_mock_response(200, json_data=payload) + + with patch.object(api, "request", return_value=mock_resp) as mock_req: + result = api.set_holiday_mode(1783767707701, 1784372507701, 12) + + assert result["original"] == 200 + assert result["parsed"] == payload + method_arg, url_arg, jsc_arg = mock_req.call_args[0] + assert method_arg == "POST" + assert json.loads(jsc_arg) == { + "start": 1783767707701, + "end": 1784372507701, + "temperature": 12, + } + + def test_cancel_holiday_mode_sends_delete(self): + api = _make_api() + mock_resp = _make_mock_response(200, json_data={"set": True}) + + with patch.object(api, "request", return_value=mock_resp) as mock_req: + result = api.cancel_holiday_mode() + + assert result["original"] == 200 + assert result["parsed"] == {"set": True} + method_arg = mock_req.call_args[0][0] + assert method_arg == "DELETE" + + # --------------------------------------------------------------------------- # Tests: result isolation between calls # --------------------------------------------------------------------------- diff --git a/tests/unit/test_hive_async_api.py b/tests/unit/test_hive_async_api.py index 92d1027..f676e1a 100644 --- a/tests/unit/test_hive_async_api.py +++ b/tests/unit/test_hive_async_api.py @@ -1,6 +1,7 @@ """Unit tests for HiveApiAsync.""" import asyncio +import json from unittest.mock import AsyncMock, MagicMock, patch import aiohttp @@ -248,6 +249,58 @@ async def test_os_error_calls_error_method(self): await api.set_action("action-1", "{}") +# --------------------------------------------------------------------------- +# Tests: HiveApiAsync holiday mode +# --------------------------------------------------------------------------- + + +class TestHolidayMode: + async def test_get_holiday_mode_returns_parsed_json(self): + payload = { + "active": False, + "enabled": False, + "start": 1783006500000, + "end": 1783296000000, + "temperature": 12, + "status": "OK", + } + api = _make_api(status=200, json_data=payload) + result = await api.get_holiday_mode() + assert result["original"] == 200 + assert result["parsed"] == payload + + async def test_set_holiday_mode_posts_epoch_ms_body(self): + payload = {"start": 1783767707701, "end": 1784372507701, "temperature": 12} + api = _make_api(status=200, json_data=payload) + result = await api.set_holiday_mode(1783767707701, 1784372507701, 12) + assert result["original"] == 200 + assert result["parsed"] == payload + _, kwargs = api.websession.request.call_args + assert json.loads(kwargs["data"]) == { + "start": 1783767707701, + "end": 1784372507701, + "temperature": 12, + } + + async def test_set_holiday_mode_file_in_use_returns_file_response(self): + api = _make_api(status=200, file_mode=True) + result = await api.set_holiday_mode(1, 2, 12) + assert result == {"original": "file"} + + async def test_cancel_holiday_mode_returns_set_true(self): + api = _make_api(status=200, json_data={"set": True}) + result = await api.cancel_holiday_mode() + assert result["original"] == 200 + assert result["parsed"] == {"set": True} + args, _ = api.websession.request.call_args + assert args[0] == "delete" + + async def test_cancel_holiday_mode_file_in_use_returns_file_response(self): + api = _make_api(status=200, file_mode=True) + result = await api.cancel_holiday_mode() + assert result == {"original": "file"} + + # --------------------------------------------------------------------------- # Tests: HiveApiAsync.motion_sensor # --------------------------------------------------------------------------- From a77ee3c575862511a6fc397556ef3707251c7e01 Mon Sep 17 00:00:00 2001 From: Neil Sleightholm Date: Sat, 11 Jul 2026 14:07:33 +0100 Subject: [PATCH 2/6] test: add live end-to-end script for holiday mode Exercises HiveHub.get_holiday_mode/set_holiday_mode/cancel_holiday_mode against a real Hive account: schedules a 7-day window starting in an hour (verified via GET readback, then cancelled before it can go active), and a second window starting an hour in the past to check whether the backend rejects it (it doesn't -- only the app's UI enforces that, not the API). Prints both UTC and local time so results are directly comparable against the Hive app/website. Co-Authored-By: Claude Opus 5 --- scripts/test_holiday_mode_live.py | 174 ++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 scripts/test_holiday_mode_live.py diff --git a/scripts/test_holiday_mode_live.py b/scripts/test_holiday_mode_live.py new file mode 100644 index 0000000..c428a41 --- /dev/null +++ b/scripts/test_holiday_mode_live.py @@ -0,0 +1,174 @@ +"""Live end-to-end test of the new HiveHub holiday mode support. + +WARNING: this talks to your REAL Hive account. Test 1 schedules a real +Holiday Mode window starting 1 hour from now -- since "enabled" (scheduled) +and "active" (currently in effect) are separate, this never actually goes +active as long as you cancel it before the hour is up (the script does this +for you). Test 2 tries a start time in the past, which is expected to be +rejected the same way the Hive app/website reject it. + +Credentials are never hard-coded or logged: set HIVE_USERNAME / HIVE_PASSWORD +as environment variables before running, or leave them unset and you'll be +prompted (password via getpass, hidden input). + +Prerequisite: `pip install -e .` from the repo root, then run: + + python scripts/test_holiday_mode_live.py +""" + +import asyncio +import getpass +import logging +import os +from datetime import datetime, timedelta, timezone + +from apyhiveapi import Hive +from apyhiveapi.helper.hive_exceptions import HiveApiError, HiveReauthRequired + +logging.basicConfig(level=logging.DEBUG, format="%(name)s: %(message)s") + + +def fmt(dt: datetime) -> str: + """Render a datetime as both UTC and local time, for comparing against the Hive app.""" + local = dt.astimezone() + return f"{dt.isoformat()} (local: {local.isoformat()})" + + +def fmt_epoch_ms(epoch_ms: int | None) -> str: + """Render epoch milliseconds from a Hive API response as UTC + local time.""" + if epoch_ms is None: + return "None" + return fmt(datetime.fromtimestamp(epoch_ms / 1000, tz=timezone.utc)) + + +def fmt_state(state: dict | None) -> str: + """Render a get_holiday_mode() response with human-readable start/end times.""" + if not state: + return repr(state) + rendered = dict(state) + if "start" in rendered: + rendered["start"] = fmt_epoch_ms(rendered["start"]) + if "end" in rendered: + rendered["end"] = fmt_epoch_ms(rendered["end"]) + return repr(rendered) + + +async def login_with_sms_fallback(hive: Hive) -> None: + """Log in, prompting for an SMS code if challenged (see diagnose_heat_on_demand.py).""" + try: + login_result = await hive.login() + except HiveReauthRequired: + print( + "Device re-authentication required — retrying as a fresh login " + "to force an SMS code prompt..." + ) + hive.auth.device_group_key = None + hive.auth.device_key = None + hive.auth.device_password = None + login_result = await hive.login() + + if login_result and login_result.get("ChallengeName") == hive.auth.SMS_MFA_CHALLENGE: + code = input("Enter the SMS 2FA code sent to your phone: ") + await hive.sms2fa(code, login_result) + + +async def test_valid_future_schedule(hive: Hive) -> None: + print("\n" + "=" * 80) + print("TEST 1: schedule holiday mode starting 1 hour from now") + print("=" * 80) + + now = datetime.now(timezone.utc) + start = now + timedelta(hours=1) + end = start + timedelta(days=7) + print(f"start: {fmt(start)}") + print(f"end: {fmt(end)}") + + confirm = input( + "\nThis will schedule Holiday Mode on your REAL account (won't go " + "active for an hour, and this script cancels it before then). " + "Type 'yes' to proceed, anything else to skip: " + ) + if confirm.strip().lower() != "yes": + print("Skipped.") + return + + ok = await hive.hub.set_holiday_mode(start, end, 12) + print(f"set_holiday_mode() -> {ok}") + + input("Check the Hive app/website now, then press Enter to continue...") + + state = await hive.hub.get_holiday_mode() + print(f"get_holiday_mode() -> {fmt_state(state)}") + + if state: + expected_start_ms = int(start.timestamp() * 1000) + expected_end_ms = int(end.timestamp() * 1000) + # Hive rounds to the nearest minute in practice; allow slack. + start_close = abs(state.get("start", 0) - expected_start_ms) < 60_000 + end_close = abs(state.get("end", 0) - expected_end_ms) < 60_000 + verified = bool(state.get("enabled")) and start_close and end_close + print(f"Verified scheduled with matching start/end: {verified}") + else: + print("Could not read back holiday mode state.") + + print("\nCancelling now (before it can go active)...") + cancelled = await hive.hub.cancel_holiday_mode() + print(f"cancel_holiday_mode() -> {cancelled}") + final_state = await hive.hub.get_holiday_mode() + print(f"get_holiday_mode() after cancel -> {fmt_state(final_state)}") + if final_state and final_state.get("enabled") is False: + print("Confirmed cancelled.") + else: + print("!! Could not confirm cancellation — check the Hive app.") + + +async def test_start_in_the_past(hive: Hive) -> None: + print("\n" + "=" * 80) + print("TEST 2: start time in the past (expected to be rejected)") + print("=" * 80) + + now = datetime.now(timezone.utc) + start = now - timedelta(hours=1) + end = start + timedelta(days=7) + print(f"start: {fmt(start)} (1 hour ago)") + print(f"end: {fmt(end)}") + + try: + ok = await hive.hub.set_holiday_mode(start, end, 12) + except HiveApiError: + print( + "-> Raised HiveApiError (see log line above for HTTP status + " + "response body). This matches the expectation that a past start " + "time is rejected, same as the Hive app/website." + ) + return + + print(f"set_holiday_mode() -> {ok} (did NOT raise — Hive accepted a past start time)") + + input("Check the Hive app/website now, then press Enter to continue...") + + state = await hive.hub.get_holiday_mode() + print(f"get_holiday_mode() -> {fmt_state(state)}") + if ok and state and state.get("enabled"): + print("It was accepted and applied. Cancelling now to be safe...") + cancelled = await hive.hub.cancel_holiday_mode() + print(f"cancel_holiday_mode() -> {cancelled}") + + +async def main() -> None: + username = os.environ.get("HIVE_USERNAME") or input("Hive username (email): ") + password = os.environ.get("HIVE_PASSWORD") or getpass.getpass("Hive password: ") + + async with Hive(username=username, password=password) as hive: + await login_with_sms_fallback(hive) + await hive.start_session() + + print("\nCurrent holiday mode state:") + print(fmt_state(await hive.hub.get_holiday_mode())) + + await test_valid_future_schedule(hive) + await test_start_in_the_past(hive) + + +if __name__ == "__main__": + asyncio.run(main()) From 3f41e0a4b2fdbc555322934d6a4dd24d4ce68832 Mon Sep 17 00:00:00 2001 From: Neil Sleightholm Date: Sat, 11 Jul 2026 14:09:19 +0100 Subject: [PATCH 3/6] chore: remove live holiday mode test script from tracking Not intended to be committed to the repo -- kept locally only. Co-Authored-By: Claude Opus 5 --- scripts/test_holiday_mode_live.py | 174 ------------------------------ 1 file changed, 174 deletions(-) delete mode 100644 scripts/test_holiday_mode_live.py diff --git a/scripts/test_holiday_mode_live.py b/scripts/test_holiday_mode_live.py deleted file mode 100644 index c428a41..0000000 --- a/scripts/test_holiday_mode_live.py +++ /dev/null @@ -1,174 +0,0 @@ -"""Live end-to-end test of the new HiveHub holiday mode support. - -WARNING: this talks to your REAL Hive account. Test 1 schedules a real -Holiday Mode window starting 1 hour from now -- since "enabled" (scheduled) -and "active" (currently in effect) are separate, this never actually goes -active as long as you cancel it before the hour is up (the script does this -for you). Test 2 tries a start time in the past, which is expected to be -rejected the same way the Hive app/website reject it. - -Credentials are never hard-coded or logged: set HIVE_USERNAME / HIVE_PASSWORD -as environment variables before running, or leave them unset and you'll be -prompted (password via getpass, hidden input). - -Prerequisite: `pip install -e .` from the repo root, then run: - - python scripts/test_holiday_mode_live.py -""" - -import asyncio -import getpass -import logging -import os -from datetime import datetime, timedelta, timezone - -from apyhiveapi import Hive -from apyhiveapi.helper.hive_exceptions import HiveApiError, HiveReauthRequired - -logging.basicConfig(level=logging.DEBUG, format="%(name)s: %(message)s") - - -def fmt(dt: datetime) -> str: - """Render a datetime as both UTC and local time, for comparing against the Hive app.""" - local = dt.astimezone() - return f"{dt.isoformat()} (local: {local.isoformat()})" - - -def fmt_epoch_ms(epoch_ms: int | None) -> str: - """Render epoch milliseconds from a Hive API response as UTC + local time.""" - if epoch_ms is None: - return "None" - return fmt(datetime.fromtimestamp(epoch_ms / 1000, tz=timezone.utc)) - - -def fmt_state(state: dict | None) -> str: - """Render a get_holiday_mode() response with human-readable start/end times.""" - if not state: - return repr(state) - rendered = dict(state) - if "start" in rendered: - rendered["start"] = fmt_epoch_ms(rendered["start"]) - if "end" in rendered: - rendered["end"] = fmt_epoch_ms(rendered["end"]) - return repr(rendered) - - -async def login_with_sms_fallback(hive: Hive) -> None: - """Log in, prompting for an SMS code if challenged (see diagnose_heat_on_demand.py).""" - try: - login_result = await hive.login() - except HiveReauthRequired: - print( - "Device re-authentication required — retrying as a fresh login " - "to force an SMS code prompt..." - ) - hive.auth.device_group_key = None - hive.auth.device_key = None - hive.auth.device_password = None - login_result = await hive.login() - - if login_result and login_result.get("ChallengeName") == hive.auth.SMS_MFA_CHALLENGE: - code = input("Enter the SMS 2FA code sent to your phone: ") - await hive.sms2fa(code, login_result) - - -async def test_valid_future_schedule(hive: Hive) -> None: - print("\n" + "=" * 80) - print("TEST 1: schedule holiday mode starting 1 hour from now") - print("=" * 80) - - now = datetime.now(timezone.utc) - start = now + timedelta(hours=1) - end = start + timedelta(days=7) - print(f"start: {fmt(start)}") - print(f"end: {fmt(end)}") - - confirm = input( - "\nThis will schedule Holiday Mode on your REAL account (won't go " - "active for an hour, and this script cancels it before then). " - "Type 'yes' to proceed, anything else to skip: " - ) - if confirm.strip().lower() != "yes": - print("Skipped.") - return - - ok = await hive.hub.set_holiday_mode(start, end, 12) - print(f"set_holiday_mode() -> {ok}") - - input("Check the Hive app/website now, then press Enter to continue...") - - state = await hive.hub.get_holiday_mode() - print(f"get_holiday_mode() -> {fmt_state(state)}") - - if state: - expected_start_ms = int(start.timestamp() * 1000) - expected_end_ms = int(end.timestamp() * 1000) - # Hive rounds to the nearest minute in practice; allow slack. - start_close = abs(state.get("start", 0) - expected_start_ms) < 60_000 - end_close = abs(state.get("end", 0) - expected_end_ms) < 60_000 - verified = bool(state.get("enabled")) and start_close and end_close - print(f"Verified scheduled with matching start/end: {verified}") - else: - print("Could not read back holiday mode state.") - - print("\nCancelling now (before it can go active)...") - cancelled = await hive.hub.cancel_holiday_mode() - print(f"cancel_holiday_mode() -> {cancelled}") - final_state = await hive.hub.get_holiday_mode() - print(f"get_holiday_mode() after cancel -> {fmt_state(final_state)}") - if final_state and final_state.get("enabled") is False: - print("Confirmed cancelled.") - else: - print("!! Could not confirm cancellation — check the Hive app.") - - -async def test_start_in_the_past(hive: Hive) -> None: - print("\n" + "=" * 80) - print("TEST 2: start time in the past (expected to be rejected)") - print("=" * 80) - - now = datetime.now(timezone.utc) - start = now - timedelta(hours=1) - end = start + timedelta(days=7) - print(f"start: {fmt(start)} (1 hour ago)") - print(f"end: {fmt(end)}") - - try: - ok = await hive.hub.set_holiday_mode(start, end, 12) - except HiveApiError: - print( - "-> Raised HiveApiError (see log line above for HTTP status + " - "response body). This matches the expectation that a past start " - "time is rejected, same as the Hive app/website." - ) - return - - print(f"set_holiday_mode() -> {ok} (did NOT raise — Hive accepted a past start time)") - - input("Check the Hive app/website now, then press Enter to continue...") - - state = await hive.hub.get_holiday_mode() - print(f"get_holiday_mode() -> {fmt_state(state)}") - if ok and state and state.get("enabled"): - print("It was accepted and applied. Cancelling now to be safe...") - cancelled = await hive.hub.cancel_holiday_mode() - print(f"cancel_holiday_mode() -> {cancelled}") - - -async def main() -> None: - username = os.environ.get("HIVE_USERNAME") or input("Hive username (email): ") - password = os.environ.get("HIVE_PASSWORD") or getpass.getpass("Hive password: ") - - async with Hive(username=username, password=password) as hive: - await login_with_sms_fallback(hive) - await hive.start_session() - - print("\nCurrent holiday mode state:") - print(fmt_state(await hive.hub.get_holiday_mode())) - - await test_valid_future_schedule(hive) - await test_start_in_the_past(hive) - - -if __name__ == "__main__": - asyncio.run(main()) From e1bd7311ccabe77700af9017857cfac2281ca161 Mon Sep 17 00:00:00 2001 From: Neil Sleightholm Date: Sat, 11 Jul 2026 14:11:05 +0100 Subject: [PATCH 4/6] chore: remove diagnostic scripts from tracking Not intended to be committed to the repo -- kept locally only. Co-Authored-By: Claude Opus 5 --- scripts/diagnose_heat_on_demand.py | 173 ------------------------ scripts/diagnose_holiday_mode.py | 203 ----------------------------- 2 files changed, 376 deletions(-) delete mode 100644 scripts/diagnose_heat_on_demand.py delete mode 100644 scripts/diagnose_holiday_mode.py diff --git a/scripts/diagnose_heat_on_demand.py b/scripts/diagnose_heat_on_demand.py deleted file mode 100644 index 1ec33c1..0000000 --- a/scripts/diagnose_heat_on_demand.py +++ /dev/null @@ -1,173 +0,0 @@ -"""Diagnose the Heat on Demand MALFORMED_REQUEST error (issue #204). - -WARNING: this talks to your REAL Hive account and REAL thermostat, not a -sandbox. It stops at the first payload shape that gets a verified state -change (HTTP 2xx AND a re-fetch confirming autoBoost actually changed -- -Hive's backend has been observed returning 200 for shapes it silently -ignores). A verified success will actually flip Heat on Demand on your real -device. - -Credentials are never hard-coded or logged: set HIVE_USERNAME / HIVE_PASSWORD -as environment variables before running, or leave them unset and you'll be -prompted (password via getpass, hidden input). - -Prerequisite: `pip install -e .` from the repo root so `apyhiveapi` resolves -to this local source tree, then run: - - python scripts/diagnose_heat_on_demand.py -""" - -import asyncio -import getpass -import json -import logging -import os - -from apyhiveapi import Hive -from apyhiveapi.helper.hive_exceptions import HiveApiError, HiveReauthRequired - -logging.basicConfig(level=logging.DEBUG, format="%(name)s: %(message)s") - -# Ordered from "reproduce the known bug" to increasingly different shapes. -# None of these are confirmed against Hive's backend -- they're guesses. A -# 200 response is NOT enough to call a shape correct: Hive's backend was -# observed accepting {"auto_boost": "ENABLED"} with HTTP 200 while silently -# not applying it. So every 2xx response gets re-verified against a re-fetch -# of the device before being trusted. -CANDIDATE_BODIES = [ - ("current code (expected to fail with MALFORMED_REQUEST)", {"autoBoost": "ENABLED"}), - ("boolean value", {"autoBoost": True}), - ("lowercase value", {"autoBoost": "enabled"}), - ( - "snake_case field name (known false positive: 200 but no effect)", - {"auto_boost": "ENABLED"}, - ), - ("nested object matching read-side shape", {"autoBoost": {"active": True}}), - ( - "nested object, full read-side shape echoed back", - None, # filled in once we know the device's current props - ), - ("snake_case field, nested object", {"auto_boost": {"active": True}}), - ("nested under props", {"props": {"autoBoost": {"active": True}}}), - ("wrapped in products array", None), # filled in once we know node_id -] - - -async def attempt(hive: Hive, node_type: str, node_id: str, label: str, body: dict) -> bool: - """POST *body* to the node's state endpoint and report the outcome.""" - url = hive.api.urls["nodes"].format(node_type, node_id) - print(f"\n=== {label} ===\nPOST {url}\nBody: {json.dumps(body)}") - try: - resp = await hive.api._call_endpoint( # noqa: SLF001 -- deliberate direct call for diagnostics - "post", url, data=json.dumps(body) - ) - except HiveApiError: - print("-> Raised HiveApiError (see log line above for HTTP status + response body)") - return False - status = resp.get("original") - print(f"-> HTTP {status} — parsed: {resp.get('parsed')}") - return str(status).startswith("20") - - -async def verify_state_changed(hive: Hive, node_id: str, baseline: dict) -> dict | None: - """Re-fetch the device and return the new autoBoost value if it differs from *baseline*.""" - await asyncio.sleep(2) # give Hive's backend a moment to settle - await hive.get_devices("No_ID") - new_boost = hive.data.products[node_id].get("props", {}).get("autoBoost") - print(f"Re-fetched autoBoost: {new_boost!r} (was: {baseline!r})") - return new_boost if new_boost != baseline else None - - -async def login_with_sms_fallback(hive: Hive) -> None: - """Log in, prompting for an SMS code if challenged. - - If Hive's "remembered device" flow rejects us (HiveReauthRequired -- - device is known but Cognito wants it re-verified via SMS, a path - hive.login() doesn't surface directly), drop any stale device - credentials and retry once as a plain fresh login so Cognito issues a - top-level SMS_MFA challenge instead. - """ - try: - login_result = await hive.login() - except HiveReauthRequired: - print( - "Device re-authentication required — retrying as a fresh login " - "to force an SMS code prompt..." - ) - hive.auth.device_group_key = None - hive.auth.device_key = None - hive.auth.device_password = None - login_result = await hive.login() - - if login_result and login_result.get("ChallengeName") == hive.auth.SMS_MFA_CHALLENGE: - code = input("Enter the SMS 2FA code sent to your phone: ") - await hive.sms2fa(code, login_result) - - -async def main() -> None: - username = os.environ.get("HIVE_USERNAME") or input("Hive username (email): ") - password = os.environ.get("HIVE_PASSWORD") or getpass.getpass("Hive password: ") - - async with Hive(username=username, password=password) as hive: - await login_with_sms_fallback(hive) - - await hive.start_session() - - heating_products = [ - p for p in hive.data.products.values() if p.get("type") == "heating" - ] - if not heating_products: - print("No 'heating' product found on this account — nothing to test.") - return - - product = heating_products[0] - node_id = product["id"] - current_boost = product.get("props", {}).get("autoBoost") - print( - f"\nTarget device: {node_id} (current autoBoost value: {current_boost!r})" - ) - input( - "This will send several test requests to your REAL thermostat and may " - "change its Heat on Demand setting. Press Enter to continue, Ctrl+C to abort..." - ) - - full_shape_echo = dict(current_boost) if isinstance(current_boost, dict) else {} - full_shape_echo["active"] = True - for i, (label, body) in enumerate(CANDIDATE_BODIES): - if body is None and "full read-side shape" in label: - CANDIDATE_BODIES[i] = (label, {"autoBoost": full_shape_echo}) - elif body is None and "products array" in label: - CANDIDATE_BODIES[i] = ( - label, - {"products": [{"id": node_id, "props": {"autoBoost": full_shape_echo}}]}, - ) - - baseline = current_boost - for label, body in CANDIDATE_BODIES: - got_2xx = await attempt(hive, "heating", node_id, label, body) - if not got_2xx: - await asyncio.sleep(1) - continue - - new_boost = await verify_state_changed(hive, node_id, baseline) - if new_boost is not None: - print(f"\n*** VERIFIED SUCCESS: {label} -> {json.dumps(body)} ***") - print(f"autoBoost actually changed: {baseline!r} -> {new_boost!r}") - print("This is the payload shape to report on issue #204.") - return - - print( - "HTTP 2xx but the device did NOT actually change — false positive, " - "trying next shape." - ) - await asyncio.sleep(1) - - print( - "\nNone of the candidate shapes produced a verified change. Copy the " - "HTTP status + response bodies above into issue #204 -- the response " - "text often names the exact field Hive's backend is rejecting." - ) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/scripts/diagnose_holiday_mode.py b/scripts/diagnose_holiday_mode.py deleted file mode 100644 index 91258fc..0000000 --- a/scripts/diagnose_holiday_mode.py +++ /dev/null @@ -1,203 +0,0 @@ -"""Diagnose Hive's undocumented /holiday-mode endpoint. - -WARNING: this talks to your REAL Hive account and affects REAL heating -behaviour while active. /holiday-mode has never been implemented in this -library (checked back to 2021) -- only the URL was ever recorded, so we are -starting from nothing. - -Safety approach: - 1. GET first (read-only) to see the current shape, if any. - 2. Only if GET doesn't reveal enough, try a short-window candidate PUT - (default: 5 minutes from now), verified by re-fetching afterwards. - 3. Immediately attempt to cancel/revert after a verified success, and - verify the cancellation too. - 4. Every write step requires an explicit confirmation prompt. - -Credentials are never hard-coded or logged: set HIVE_USERNAME / HIVE_PASSWORD -as environment variables before running, or leave them unset and you'll be -prompted (password via getpass, hidden input). - -Prerequisite: `pip install -e .` from the repo root, then run: - - python scripts/diagnose_holiday_mode.py -""" - -import asyncio -import getpass -import json -import logging -import os -from datetime import datetime, timedelta, timezone - -from apyhiveapi import Hive -from apyhiveapi.helper.hive_exceptions import HiveApiError, HiveReauthRequired - -logging.basicConfig(level=logging.DEBUG, format="%(name)s: %(message)s") - -TEST_WINDOW_MINUTES = 5 - -# Guesses only -- informed by whatever GET reveals, tried in order. Each is -# verified by re-fetching /holiday-mode (and the target product's autoBoost) -# afterwards; a 2xx alone is not trusted (autoBoost diagnostics showed Hive -# returns 200 for shapes it silently ignores). -def build_candidate_bodies(start: datetime, end: datetime) -> list[tuple[str, dict]]: - start_iso = start.strftime("%Y-%m-%dT%H:%M:%S.000Z") - end_iso = end.strftime("%Y-%m-%dT%H:%M:%S.000Z") - return [ - ("camelCase start/end", {"startDate": start_iso, "endDate": end_iso}), - ("snake_case start/end", {"start_date": start_iso, "end_date": end_iso}), - ("short keys", {"start": start_iso, "end": end_iso}), - ( - "camelCase with enabled flag", - {"enabled": True, "startDate": start_iso, "endDate": end_iso}, - ), - ] - - -async def get_holiday_mode(hive: Hive) -> dict: - """GET the current holiday-mode resource. Read-only, always safe.""" - url = hive.api.urls["holiday_mode"] - print(f"\nGET {url}") - resp = await hive.api._call_endpoint("get", url) # noqa: SLF001 - print(f"-> HTTP {resp.get('original')} — parsed: {resp.get('parsed')}") - return resp - - -async def attempt(hive: Hive, label: str, method: str, body: dict | None) -> tuple[bool, dict]: - url = hive.api.urls["holiday_mode"] - print(f"\n=== {label} ({method.upper()}) ===\nBody: {json.dumps(body)}") - try: - resp = await hive.api._call_endpoint( # noqa: SLF001 - method, url, data=json.dumps(body) if body is not None else None - ) - except HiveApiError: - print("-> Raised HiveApiError (see log line above for HTTP status + response body)") - return False, {} - status = resp.get("original") - print(f"-> HTTP {status} — parsed: {resp.get('parsed')}") - return str(status).startswith("20"), resp - - -async def login_with_sms_fallback(hive: Hive) -> None: - """Log in, prompting for an SMS code if challenged (see diagnose_heat_on_demand.py).""" - try: - login_result = await hive.login() - except HiveReauthRequired: - print( - "Device re-authentication required — retrying as a fresh login " - "to force an SMS code prompt..." - ) - hive.auth.device_group_key = None - hive.auth.device_key = None - hive.auth.device_password = None - login_result = await hive.login() - - if login_result and login_result.get("ChallengeName") == hive.auth.SMS_MFA_CHALLENGE: - code = input("Enter the SMS 2FA code sent to your phone: ") - await hive.sms2fa(code, login_result) - - -async def main() -> None: - username = os.environ.get("HIVE_USERNAME") or input("Hive username (email): ") - password = os.environ.get("HIVE_PASSWORD") or getpass.getpass("Hive password: ") - - async with Hive(username=username, password=password) as hive: - await login_with_sms_fallback(hive) - await hive.start_session() - - print("\nStep 1: read-only GET of /holiday-mode (safe, no changes made).") - await get_holiday_mode(hive) - - proceed = input( - "\nStep 2 will send a WRITE request that may activate holiday mode " - f"on your REAL account for up to {TEST_WINDOW_MINUTES} minutes, and will " - "immediately try to cancel it afterwards. Type 'yes' to continue, " - "anything else to stop here: " - ) - if proceed.strip().lower() != "yes": - print("Stopping after the read-only GET, as requested.") - return - - now = datetime.now(timezone.utc) - end = now + timedelta(minutes=TEST_WINDOW_MINUTES) - candidates = build_candidate_bodies(now, end) - - heating_products = [ - p for p in hive.data.products.values() if p.get("type") == "heating" - ] - node_id = heating_products[0]["id"] if heating_products else None - baseline_boost = ( - heating_products[0].get("props", {}).get("autoBoost") - if heating_products - else None - ) - - for label, body in candidates: - ok, _ = await attempt(hive, label, "put", body) - if not ok: - await asyncio.sleep(1) - continue - - print("Got 2xx — verifying via re-fetch (GET /holiday-mode + device autoBoost)...") - await asyncio.sleep(2) - hm_resp = await get_holiday_mode(hive) - new_boost = None - if node_id: - await hive.get_devices("No_ID") - new_boost = hive.data.products[node_id].get("props", {}).get("autoBoost") - print(f"autoBoost now: {new_boost!r} (was: {baseline_boost!r})") - - verified = new_boost is not None and new_boost != baseline_boost - if not verified: - print("No verified change — treating as a false positive, trying next shape.") - await asyncio.sleep(1) - continue - - print(f"\n*** VERIFIED SUCCESS: {label} -> {json.dumps(body)} ***") - print(f"holiday-mode GET now shows: {hm_resp.get('parsed')}") - print("Attempting immediate cancellation...") - - cancelled = await try_cancel(hive) - if cancelled: - print("Cancellation verified — holiday mode is back off.") - else: - print( - "!! Could not verify cancellation. The test window was only " - f"{TEST_WINDOW_MINUTES} minutes, so it will lapse on its own, " - "but check the Hive app now to be sure." - ) - return - - print( - "\nNo candidate body produced a verified holiday-mode activation. " - "See the GET response above for whatever shape Hive did return -- " - "that's the most useful lead for a follow-up attempt." - ) - - -async def try_cancel(hive: Hive) -> bool: - """Best-effort cancellation attempt, verified via re-fetch.""" - url = hive.api.urls["holiday_mode"] - for label, method, body in [ - ("DELETE", "delete", None), - ("PUT enabled=false", "put", {"enabled": False}), - ]: - print(f"\nCancel attempt: {label}") - try: - resp = await hive.api._call_endpoint( # noqa: SLF001 - method, url, data=json.dumps(body) if body is not None else None - ) - except HiveApiError: - print("-> Raised HiveApiError") - continue - print(f"-> HTTP {resp.get('original')} — parsed: {resp.get('parsed')}") - await asyncio.sleep(2) - check = await get_holiday_mode(hive) - parsed = check.get("parsed") - if isinstance(parsed, dict) and parsed.get("enabled") is False: - return True - return False - - -if __name__ == "__main__": - asyncio.run(main()) From 9ccc4d31e22292f20b5df94a628f2d0b8b6e8629 Mon Sep 17 00:00:00 2001 From: Neil Sleightholm Date: Sun, 12 Jul 2026 14:49:26 +0000 Subject: [PATCH 5/6] style: fix ruff-format violation in holiday mode tests Two dict literals in test_hub.py exceeded ruff format's line-wrap threshold, failing the Lint CI check (ruff check itself was clean, since E501 is ignored, but ruff format still wants the wrap). --- tests/module/test_hub.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/tests/module/test_hub.py b/tests/module/test_hub.py index 410e1f8..8a790b0 100644 --- a/tests/module/test_hub.py +++ b/tests/module/test_hub.py @@ -130,12 +130,20 @@ def _make_hub(self, resp): return HiveHub(session=session) async def test_get_holiday_mode_returns_parsed_on_200(self): - payload = {"active": False, "enabled": False, "start": 1, "end": 2, "temperature": 12} + payload = { + "active": False, + "enabled": False, + "start": 1, + "end": 2, + "temperature": 12, + } hub = self._make_hub({"original": 200, "parsed": payload}) assert await hub.get_holiday_mode() == payload async def test_get_holiday_mode_returns_none_on_failure(self): - hub = self._make_hub({"original": 400, "parsed": {"error": "MALFORMED_REQUEST"}}) + hub = self._make_hub( + {"original": 400, "parsed": {"error": "MALFORMED_REQUEST"}} + ) assert await hub.get_holiday_mode() is None async def test_set_holiday_mode_returns_true_on_200(self): From 412d32caee07cd6ec317659c2a99f75b50a54b5c Mon Sep 17 00:00:00 2001 From: Neil Sleightholm Date: Mon, 13 Jul 2026 08:01:55 +0000 Subject: [PATCH 6/6] fix: resolve remaining pylint warnings in holiday mode tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_hive_api.py: drop unused url_arg from tuple-unpack, index the two needed values individually (matches existing pattern elsewhere in this file) — fixes W0612 unused-variable - test_hive_async_api.py: remove redundant local 'import json', module already imports it at the top — fixes W0404 reimported --- tests/unit/test_hive_api.py | 3 ++- tests/unit/test_hive_async_api.py | 2 -- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_hive_api.py b/tests/unit/test_hive_api.py index 9d52845..de2b172 100644 --- a/tests/unit/test_hive_api.py +++ b/tests/unit/test_hive_api.py @@ -656,7 +656,8 @@ def test_set_holiday_mode_posts_epoch_ms_body(self): assert result["original"] == 200 assert result["parsed"] == payload - method_arg, url_arg, jsc_arg = mock_req.call_args[0] + method_arg = mock_req.call_args[0][0] + jsc_arg = mock_req.call_args[0][2] assert method_arg == "POST" assert json.loads(jsc_arg) == { "start": 1783767707701, diff --git a/tests/unit/test_hive_async_api.py b/tests/unit/test_hive_async_api.py index f676e1a..5dc878f 100644 --- a/tests/unit/test_hive_async_api.py +++ b/tests/unit/test_hive_async_api.py @@ -567,8 +567,6 @@ class TestSetStateJsonEncoding: async def test_set_state_escapes_quotes_in_value(self): """A value containing double-quotes must produce valid, parseable JSON.""" - import json # noqa: PLC0415 - session = MagicMock() session.tokens.token_data = {"token": "tok"} session.config.file = False