Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions src/api/hive_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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")
Expand Down
36 changes: 36 additions & 0 deletions src/api/hive_async_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.")
Expand Down
62 changes: 61 additions & 1 deletion src/devices/hub.py
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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
54 changes: 54 additions & 0 deletions tests/module/test_hub.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -116,6 +117,59 @@ 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."""

Expand Down
63 changes: 63 additions & 0 deletions tests/unit/test_hive_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -615,6 +622,62 @@ 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 = 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,
"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
# ---------------------------------------------------------------------------
Expand Down
55 changes: 53 additions & 2 deletions tests/unit/test_hive_async_api.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Unit tests for HiveApiAsync."""

import asyncio
import json
from unittest.mock import AsyncMock, MagicMock, patch

import aiohttp
Expand Down Expand Up @@ -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
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -514,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
Expand Down
Loading