mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-07-27 17:26:48 +00:00
db5d172fb8
* fix(services): preserve service call data through coordinator data fetch
In `_handle_find_block` and `_handle_find_hours`, the local `data`
variable holding the resolved service call data was rebound to the
coordinator data dict returned by `get_entry_and_data()`. As a result,
the subsequent calls to `validate_search_params(data)`,
`apply_must_finish_by(data, ...)` and `resolve_search_range(...)` read
search-range parameters from coordinator data instead of from the
service call, silently ignoring:
- must_finish_by
- search_scope
- search_start, search_end
- search_start_time, search_end_time
- search_start_day_offset, search_end_day_offset
- search_start_offset_minutes, search_end_offset_minutes
- include_current_interval
The functions fell back to the default range ("now → end of tomorrow")
for every call that depended on these parameters.
Rename the third return value of `get_entry_and_data()` to
`coordinator_data` so the service call `data` survives, restoring
deadline and search-scope semantics. `find_cheapest_schedule.py`
already uses `data_dict` for the same purpose and was not affected.
Verified locally against v0.31.0: a call with
`must_finish_by: 2026-06-01T20:00:00+02:00` now correctly produces
`search_end: 2026-06-01T20:00:00+02:00` (was end-of-tomorrow before).
* refactor(services): update data handling in find_cheapest_schedule service
Refactor the data retrieval process to use coordinator data instead of entry data for improved clarity and consistency.
Impact: Enhances maintainability of the service code without altering user-facing functionality.
---------
Co-authored-by: “WouterK” <kwaken.geringd0w@icloud.com”git config --global user.name “WouterK”git config --global user.email kwaken.geringd0w@icloud.com”>
Co-authored-by: Julian Pawlowski <jpawlowski@users.noreply.github.com>
403 lines
16 KiB
Python
403 lines
16 KiB
Python
"""Tests for find service response contracts: reason codes and schedule comparison details."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from typing import TYPE_CHECKING, Any, cast
|
|
|
|
if TYPE_CHECKING:
|
|
from homeassistant.core import ServiceCall
|
|
|
|
import pytest
|
|
|
|
from custom_components.tibber_prices.services import (
|
|
find_cheapest_block as block_module,
|
|
find_cheapest_hours as hours_module,
|
|
find_cheapest_schedule as schedule_module,
|
|
)
|
|
from custom_components.tibber_prices.services.find_cheapest_block import (
|
|
_determine_no_window_reason,
|
|
handle_find_cheapest_block,
|
|
)
|
|
from custom_components.tibber_prices.services.find_cheapest_hours import (
|
|
_determine_no_intervals_reason,
|
|
handle_find_cheapest_hours,
|
|
)
|
|
from custom_components.tibber_prices.services.find_cheapest_schedule import (
|
|
FIND_CHEAPEST_SCHEDULE_SERVICE_SCHEMA,
|
|
_compute_task_price_comparison,
|
|
_determine_schedule_reason,
|
|
)
|
|
|
|
|
|
def _make_intervals(prices: list[float], start: datetime | None = None) -> list[dict]:
|
|
"""Create quarter-hour intervals for tests."""
|
|
base = start or datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
|
|
return [
|
|
{
|
|
"startsAt": (base + timedelta(minutes=15 * i)).isoformat(),
|
|
"total": price,
|
|
"level": "NORMAL",
|
|
}
|
|
for i, price in enumerate(prices)
|
|
]
|
|
|
|
|
|
class TestBlockNoResultReasons:
|
|
"""Reason classification for contiguous block service."""
|
|
|
|
def test_reason_no_data(self) -> None:
|
|
"""Return no_data_in_range when no intervals exist."""
|
|
reason = _determine_no_window_reason([], [], 4, level_filter_active=False)
|
|
assert reason == "no_data_in_range"
|
|
|
|
def test_reason_level_filter_eliminated_all(self) -> None:
|
|
"""Return level-filter reason when filters remove all intervals."""
|
|
reason = _determine_no_window_reason(
|
|
_make_intervals([10.0, 12.0]),
|
|
[],
|
|
4,
|
|
level_filter_active=True,
|
|
)
|
|
assert reason == "no_intervals_matching_level_filter"
|
|
|
|
def test_reason_not_enough_intervals_after_filter(self) -> None:
|
|
"""Return insufficient_intervals_after_filter for short filtered pool."""
|
|
reason = _determine_no_window_reason(
|
|
_make_intervals([10.0, 12.0, 13.0]),
|
|
_make_intervals([10.0, 12.0]),
|
|
4,
|
|
level_filter_active=False,
|
|
)
|
|
assert reason == "insufficient_intervals_after_filter"
|
|
|
|
|
|
class TestHoursNoResultReasons:
|
|
"""Reason classification for cheapest/most-expensive hours service."""
|
|
|
|
def test_reason_no_data(self) -> None:
|
|
"""Return no_data_in_range when interval pool is empty."""
|
|
reason = _determine_no_intervals_reason([], [], 6, level_filter_active=False)
|
|
assert reason == "no_data_in_range"
|
|
|
|
def test_reason_level_filter_eliminated_all(self) -> None:
|
|
"""Return level-filter reason when all intervals are filtered out."""
|
|
reason = _determine_no_intervals_reason(
|
|
_make_intervals([10.0, 11.0]),
|
|
[],
|
|
4,
|
|
level_filter_active=True,
|
|
)
|
|
assert reason == "no_intervals_matching_level_filter"
|
|
|
|
def test_reason_not_enough_intervals_after_filter(self) -> None:
|
|
"""Return insufficient_intervals_after_filter when pool is too short."""
|
|
reason = _determine_no_intervals_reason(
|
|
_make_intervals([10.0, 11.0, 12.0]),
|
|
_make_intervals([10.0, 11.0]),
|
|
4,
|
|
level_filter_active=False,
|
|
)
|
|
assert reason == "insufficient_intervals_after_filter"
|
|
|
|
|
|
class TestScheduleReasonAndComparison:
|
|
"""Schedule service reason codes and comparison details behavior."""
|
|
|
|
def test_schedule_reason_no_data(self) -> None:
|
|
"""Return no_data_in_range when schedule has no source intervals."""
|
|
reason = _determine_schedule_reason(
|
|
all_tasks_scheduled=False,
|
|
assignments_count=0,
|
|
price_info=[],
|
|
filtered_price_info=[],
|
|
level_filter_active=False,
|
|
)
|
|
assert reason == "no_data_in_range"
|
|
|
|
def test_schedule_reason_level_filter(self) -> None:
|
|
"""Return level-filter reason when filter removes all schedule candidates."""
|
|
reason = _determine_schedule_reason(
|
|
all_tasks_scheduled=False,
|
|
assignments_count=0,
|
|
price_info=_make_intervals([10.0, 20.0]),
|
|
filtered_price_info=[],
|
|
level_filter_active=True,
|
|
)
|
|
assert reason == "no_intervals_matching_level_filter"
|
|
|
|
def test_schedule_reason_partial(self) -> None:
|
|
"""Return partial-schedule reason when some tasks remain unscheduled."""
|
|
reason = _determine_schedule_reason(
|
|
all_tasks_scheduled=False,
|
|
assignments_count=1,
|
|
price_info=_make_intervals([10.0, 20.0, 30.0]),
|
|
filtered_price_info=_make_intervals([10.0, 20.0, 30.0]),
|
|
level_filter_active=False,
|
|
)
|
|
assert reason == "insufficient_contiguous_window_for_some_tasks"
|
|
|
|
def test_schedule_schema_accepts_include_comparison_details(self) -> None:
|
|
"""Schedule schema should accept include_comparison_details flag."""
|
|
result = cast(
|
|
"dict[str, Any]",
|
|
FIND_CHEAPEST_SCHEDULE_SERVICE_SCHEMA(
|
|
{
|
|
"tasks": [{"name": "dishwasher", "duration": timedelta(hours=2)}],
|
|
"include_comparison_details": True,
|
|
}
|
|
),
|
|
)
|
|
assert result["include_comparison_details"] is True
|
|
|
|
def test_task_comparison_includes_details(self) -> None:
|
|
"""Task comparison helper should emit detail fields when enabled."""
|
|
full_intervals = _make_intervals([5.0, 10.0, 50.0, 60.0])
|
|
task_intervals = full_intervals[:2]
|
|
|
|
comparison = _compute_task_price_comparison(
|
|
task_intervals,
|
|
full_intervals,
|
|
1,
|
|
include_details=True,
|
|
)
|
|
|
|
assert comparison is not None
|
|
assert "comparison_price_mean" in comparison
|
|
assert "comparison_price_min" in comparison
|
|
assert "comparison_price_max" in comparison
|
|
assert "comparison_window_start" in comparison
|
|
assert "comparison_window_end" in comparison
|
|
|
|
|
|
class _FakePool:
|
|
"""Minimal async interval pool for service handler tests."""
|
|
|
|
def __init__(self, intervals: list[dict]) -> None:
|
|
"""Store static interval list returned by get_intervals."""
|
|
self._intervals = intervals
|
|
|
|
async def get_intervals(self, **_kwargs: object) -> tuple[list[dict], bool]:
|
|
"""Return predefined intervals and no API-call marker."""
|
|
return self._intervals, False
|
|
|
|
|
|
def _build_fake_entry_and_coordinator(intervals: list[dict]) -> tuple[SimpleNamespace, SimpleNamespace, dict]:
|
|
"""Build a minimal entry/coordinator/data tuple used by service handlers."""
|
|
pool = _FakePool(intervals)
|
|
entry = SimpleNamespace(
|
|
data={"home_id": "home_1", "currency": "EUR"},
|
|
runtime_data=SimpleNamespace(interval_pool=pool),
|
|
)
|
|
coordinator = SimpleNamespace(
|
|
api=object(),
|
|
_cached_user_data={"viewer": {"homes": [{"id": "home_1", "timeZone": "UTC"}]}},
|
|
)
|
|
data = {"priceInfo": intervals}
|
|
return entry, coordinator, data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_block_handler_returns_level_filter_reason(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Block handler should return reason when level filter eliminates all intervals."""
|
|
intervals = _make_intervals([10.0, 11.0, 12.0, 13.0])
|
|
fake_tuple = _build_fake_entry_and_coordinator(intervals)
|
|
|
|
monkeypatch.setattr(block_module, "get_entry_and_data", lambda _hass, _entry_id: fake_tuple)
|
|
monkeypatch.setattr(block_module, "resolve_home_timezone", lambda _coord, _home_id: "UTC")
|
|
monkeypatch.setattr(
|
|
block_module,
|
|
"resolve_search_range",
|
|
lambda _call_data, _now, _home_tz: (
|
|
datetime(2026, 1, 1, 0, 0, tzinfo=UTC),
|
|
datetime(2026, 1, 1, 2, 0, tzinfo=UTC),
|
|
),
|
|
)
|
|
|
|
call = SimpleNamespace(
|
|
hass=object(),
|
|
data={
|
|
"duration": timedelta(hours=1),
|
|
"max_price_level": "very_cheap",
|
|
"use_base_unit": True,
|
|
"allow_relaxation": False,
|
|
},
|
|
)
|
|
response = cast("dict[str, Any]", await handle_find_cheapest_block(cast("ServiceCall", call)))
|
|
|
|
assert response["window_found"] is False
|
|
assert response["reason"] == "no_intervals_matching_level_filter"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hours_handler_returns_insufficient_intervals_reason(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Hours handler should return insufficient_intervals_after_filter when pool is too short."""
|
|
intervals = _make_intervals([10.0, 11.0, 12.0]) # 3 intervals only
|
|
fake_tuple = _build_fake_entry_and_coordinator(intervals)
|
|
|
|
monkeypatch.setattr(hours_module, "get_entry_and_data", lambda _hass, _entry_id: fake_tuple)
|
|
monkeypatch.setattr(hours_module, "resolve_home_timezone", lambda _coord, _home_id: "UTC")
|
|
monkeypatch.setattr(
|
|
hours_module,
|
|
"resolve_search_range",
|
|
lambda _call_data, _now, _home_tz: (
|
|
datetime(2026, 1, 1, 0, 0, tzinfo=UTC),
|
|
datetime(2026, 1, 1, 2, 0, tzinfo=UTC),
|
|
),
|
|
)
|
|
|
|
call = SimpleNamespace(
|
|
hass=object(),
|
|
data={
|
|
"duration": timedelta(hours=1), # needs 4 intervals
|
|
"use_base_unit": True,
|
|
"allow_relaxation": False,
|
|
},
|
|
)
|
|
response = cast("dict[str, Any]", await handle_find_cheapest_hours(cast("ServiceCall", call)))
|
|
|
|
assert response["intervals_found"] is False
|
|
assert response["reason"] == "insufficient_intervals_after_filter"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_handler_adds_per_task_comparison_details(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Schedule handler should include per-task comparison details when requested."""
|
|
intervals = _make_intervals([5.0, 6.0, 50.0, 60.0])
|
|
fake_tuple = _build_fake_entry_and_coordinator(intervals)
|
|
|
|
monkeypatch.setattr(schedule_module, "get_entry_and_data", lambda _hass, _entry_id: fake_tuple)
|
|
monkeypatch.setattr(schedule_module, "resolve_home_timezone", lambda _coord, _home_id: "UTC")
|
|
monkeypatch.setattr(
|
|
schedule_module,
|
|
"resolve_search_range",
|
|
lambda _call_data, _now, _home_tz: (
|
|
datetime(2026, 1, 1, 0, 0, tzinfo=UTC),
|
|
datetime(2026, 1, 1, 2, 0, tzinfo=UTC),
|
|
),
|
|
)
|
|
|
|
call = SimpleNamespace(
|
|
hass=object(),
|
|
data={
|
|
"tasks": [{"name": "dishwasher", "duration": timedelta(minutes=30)}],
|
|
"include_comparison_details": True,
|
|
"use_base_unit": True,
|
|
},
|
|
)
|
|
response = cast("dict[str, Any]", await schedule_module.handle_find_cheapest_schedule(cast("ServiceCall", call)))
|
|
|
|
assert response["all_tasks_scheduled"] is True
|
|
assert response["reason"] is None
|
|
tasks = cast("list[dict[str, Any]]", response["tasks"])
|
|
assert len(tasks) == 1
|
|
comparison = cast("dict[str, Any] | None", tasks[0]["price_comparison"])
|
|
assert comparison is not None
|
|
assert "comparison_price_min" in comparison
|
|
assert "comparison_price_max" in comparison
|
|
assert "comparison_window_end" in comparison
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_block_handler_preserves_service_search_data(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Block handler must pass resolved call data (not coordinator data) into search helpers."""
|
|
intervals = _make_intervals([10.0, 11.0, 12.0, 13.0])
|
|
fake_tuple = _build_fake_entry_and_coordinator(intervals)
|
|
deadline = datetime(2026, 1, 1, 8, 0, tzinfo=UTC)
|
|
fixed_start = datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
|
|
|
|
monkeypatch.setattr(block_module, "get_entry_and_data", lambda _hass, _entry_id: fake_tuple)
|
|
monkeypatch.setattr(block_module, "resolve_home_timezone", lambda _coord, _home_id: "UTC")
|
|
|
|
def _validate_search_params(call_data: dict[str, Any]) -> None:
|
|
assert call_data["must_finish_by"] == deadline
|
|
assert call_data["include_current_interval"] is False
|
|
|
|
def _apply_must_finish_by(call_data: dict[str, Any], _home_tz: Any) -> tuple[dict[str, Any], datetime]:
|
|
assert call_data["must_finish_by"] == deadline
|
|
modified = dict(call_data)
|
|
modified["search_end"] = deadline
|
|
modified.pop("must_finish_by", None)
|
|
return modified, deadline
|
|
|
|
def _resolve_search_range(call_data: dict[str, Any], _now: datetime, _home_tz: Any) -> tuple[datetime, datetime]:
|
|
assert call_data["include_current_interval"] is False
|
|
assert call_data["search_end"] == deadline
|
|
return fixed_start, deadline
|
|
|
|
async def _fetch_intervals(*_args: Any, **_kwargs: Any) -> tuple[list[dict[str, Any]], bool]:
|
|
return [], False
|
|
|
|
monkeypatch.setattr(block_module, "validate_search_params", _validate_search_params)
|
|
monkeypatch.setattr(block_module, "apply_must_finish_by", _apply_must_finish_by)
|
|
monkeypatch.setattr(block_module, "resolve_search_range", _resolve_search_range)
|
|
monkeypatch.setattr(block_module, "async_fetch_service_intervals", _fetch_intervals)
|
|
|
|
call = SimpleNamespace(
|
|
hass=object(),
|
|
data={
|
|
"duration": timedelta(hours=1),
|
|
"use_base_unit": True,
|
|
"must_finish_by": deadline,
|
|
"include_current_interval": False,
|
|
},
|
|
)
|
|
|
|
response = cast("dict[str, Any]", await handle_find_cheapest_block(cast("ServiceCall", call)))
|
|
assert response["success"] is False
|
|
assert response["search_start"] == fixed_start.isoformat()
|
|
assert response["search_end"] == deadline.isoformat()
|
|
assert response["must_finish_by"] == deadline.isoformat()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_hours_handler_preserves_service_search_data(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Hours handler must pass resolved call data (not coordinator data) into search helpers."""
|
|
intervals = _make_intervals([10.0, 11.0, 12.0, 13.0])
|
|
fake_tuple = _build_fake_entry_and_coordinator(intervals)
|
|
deadline = datetime(2026, 1, 1, 8, 0, tzinfo=UTC)
|
|
fixed_start = datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
|
|
|
|
monkeypatch.setattr(hours_module, "get_entry_and_data", lambda _hass, _entry_id: fake_tuple)
|
|
monkeypatch.setattr(hours_module, "resolve_home_timezone", lambda _coord, _home_id: "UTC")
|
|
|
|
def _validate_search_params(call_data: dict[str, Any]) -> None:
|
|
assert call_data["must_finish_by"] == deadline
|
|
assert call_data["include_current_interval"] is False
|
|
|
|
def _apply_must_finish_by(call_data: dict[str, Any], _home_tz: Any) -> tuple[dict[str, Any], datetime]:
|
|
assert call_data["must_finish_by"] == deadline
|
|
modified = dict(call_data)
|
|
modified["search_end"] = deadline
|
|
modified.pop("must_finish_by", None)
|
|
return modified, deadline
|
|
|
|
def _resolve_search_range(call_data: dict[str, Any], _now: datetime, _home_tz: Any) -> tuple[datetime, datetime]:
|
|
assert call_data["include_current_interval"] is False
|
|
assert call_data["search_end"] == deadline
|
|
return fixed_start, deadline
|
|
|
|
async def _fetch_intervals(*_args: Any, **_kwargs: Any) -> tuple[list[dict[str, Any]], bool]:
|
|
return [], False
|
|
|
|
monkeypatch.setattr(hours_module, "validate_search_params", _validate_search_params)
|
|
monkeypatch.setattr(hours_module, "apply_must_finish_by", _apply_must_finish_by)
|
|
monkeypatch.setattr(hours_module, "resolve_search_range", _resolve_search_range)
|
|
monkeypatch.setattr(hours_module, "async_fetch_service_intervals", _fetch_intervals)
|
|
|
|
call = SimpleNamespace(
|
|
hass=object(),
|
|
data={
|
|
"duration": timedelta(hours=1),
|
|
"use_base_unit": True,
|
|
"must_finish_by": deadline,
|
|
"include_current_interval": False,
|
|
},
|
|
)
|
|
|
|
response = cast("dict[str, Any]", await handle_find_cheapest_hours(cast("ServiceCall", call)))
|
|
assert response["success"] is False
|
|
assert response["search_start"] == fixed_start.isoformat()
|
|
assert response["search_end"] == deadline.isoformat()
|
|
assert response["must_finish_by"] == deadline.isoformat()
|