Compare commits

...
16 changed files with 1577 additions and 3 deletions
+6
View File
@@ -0,0 +1,6 @@
__pycache__/
*.py[cod]
*$py.class
.venv/
.pytest_cache/
*.metadata.json
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 NestQuest Contributors / Maniacs Mansion
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+68
View File
@@ -0,0 +1,68 @@
# NestQuest - House Chaos Coordination
Home Assistant custom integration for managing chores, custody schedules, and house coordination.
## Overview
NestQuest helps households coordinate daily life, chore tracking, custody schedules, and shared tasks directly within Home Assistant. It provides sensors, calendar events, and service calls designed for busy households and modern co-parenting setups.
## Installation
### HACS Installation (Recommended)
1. Ensure [HACS (Home Assistant Community Store)](https://hacs.xyz/) is installed.
2. In Home Assistant, open **HACS** > **Integrations**.
3. Select the three-dots menu in the top right and click **Custom repositories**.
4. Enter the repository URL: `https://github.com/talon2king/NestQuest`
5. Select category **Integration** and click **Add**.
6. Search for **NestQuest** and click **Download**.
7. Restart Home Assistant.
> **Note:** Development happens in the private Gitea repository
> (`https://code.cubecraftlabs.com/Maniacs_Mansion/NestQuest`). The GitHub
> repository is a public mirror maintained for HACS distribution.
### Manual Installation
1. Download the latest release from the repository.
2. Copy the `custom_components/nestquest` directory into your Home Assistant installation's `config/custom_components/` folder:
```bash
cp -R custom_components/nestquest <config_dir>/custom_components/
```
3. Restart Home Assistant.
## Configuration
After installation and restart:
1. Navigate to **Settings** > **Devices & Services** in Home Assistant.
2. Click **Add Integration** and search for **NestQuest**.
3. Follow the on-screen configuration flow.
## Developer Setup and Testing
This project uses [`uv`](https://github.com/astral-sh/uv) for fast Python package and dependency management.
### Prerequisites
- Python 3.11 or newer
- [uv](https://docs.astral.sh/uv/) installed on your machine
### Setup Environment
To set up the development environment and install dependencies:
```bash
uv sync
```
### Running Tests
Run the test suite with `pytest`:
```bash
uv run pytest
```
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
+39 -3
View File
@@ -1,15 +1,26 @@
"""The NestQuest integration."""
from __future__ import annotations
import logging
import inspect
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .const import DOMAIN, LOGGER
_LOGGER = logging.getLogger(__name__)
_LOGGER = LOGGER
@dataclass
class NestQuestRuntimeData:
"""Runtime data stored on a NestQuest config entry."""
entry_id: str
options: dict[str, Any]
remove_update_listener: Callable[[], Any]
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
@@ -20,9 +31,34 @@ async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up NestQuest from a config entry."""
hass.data.setdefault(DOMAIN, {})
async def _async_update_listener(
listener_hass: HomeAssistant, listener_entry: ConfigEntry
) -> None:
"""Reload the entry when its options change."""
await listener_hass.config_entries.async_reload(listener_entry.entry_id)
runtime_data = NestQuestRuntimeData(
entry_id=entry.entry_id,
options=dict(entry.options),
remove_update_listener=entry.add_update_listener(_async_update_listener),
)
entry.runtime_data = runtime_data
hass.data[DOMAIN][entry.entry_id] = runtime_data
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
runtime_data = hass.data.setdefault(DOMAIN, {}).pop(entry.entry_id, None)
if runtime_data is None:
runtime_data = getattr(entry, "runtime_data", None)
if runtime_data is not None:
remove_update_listener = getattr(runtime_data, "remove_update_listener", None)
if remove_update_listener is not None:
result = remove_update_listener()
if inspect.isawaitable(result):
await result
if getattr(entry, "runtime_data", None) is not None:
entry.runtime_data = None
return True
@@ -0,0 +1,39 @@
"""Config flow for NestQuest."""
from __future__ import annotations
from typing import Any
from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from .const import DOMAIN
class NestQuestConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for NestQuest."""
VERSION = 1
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> Any:
"""Create the options flow handler for NestQuest."""
from .options_flow import NestQuestOptionsFlow
return NestQuestOptionsFlow(config_entry=config_entry)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the user step of the config flow."""
await self.async_set_unique_id(DOMAIN, raise_on_progress=False)
if self._async_in_progress() or self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
if user_input is None:
return self.async_show_form(step_id="user")
return self.async_create_entry(title="NestQuest", data=user_input)
+20
View File
@@ -1,2 +1,22 @@
"""Constants for the NestQuest integration."""
from __future__ import annotations
import logging
DOMAIN = "nestquest"
LOGGER_NAME = "custom_components.nestquest"
LOGGER = logging.getLogger(LOGGER_NAME)
PLATFORMS: list[str] = []
SQLITE_DB_FILENAME = "nestquest.db"
# Configuration keys and default values
CONF_HORIZON_DAYS = "horizon_days"
DEFAULT_HORIZON_DAYS = 14
CONF_DAY_ROLLOVER_TIME = "day_rollover_time"
DEFAULT_DAY_ROLLOVER_TIME = "00:00"
CONF_PANEL_IDLE_TIMEOUT = "panel_idle_timeout"
DEFAULT_PANEL_IDLE_TIMEOUT = 300
+120
View File
@@ -0,0 +1,120 @@
"""Options flow for NestQuest post-setup settings."""
from __future__ import annotations
import re
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from .const import (
CONF_DAY_ROLLOVER_TIME,
CONF_HORIZON_DAYS,
CONF_PANEL_IDLE_TIMEOUT,
DEFAULT_DAY_ROLLOVER_TIME,
DEFAULT_HORIZON_DAYS,
DEFAULT_PANEL_IDLE_TIMEOUT,
)
MIN_HORIZON_DAYS = 1
MIN_PANEL_IDLE_TIMEOUT = 30
_TIME_PATTERN = re.compile(r"(?:[01]\d|2[0-3]):[0-5]\d")
def _parse_time(value: str) -> bool:
"""Return True when value is a valid 24-hour HH:MM string."""
return _TIME_PATTERN.fullmatch(value) is not None
def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
"""Validate user input and return field-level errors instead of raising."""
errors: dict[str, str] = {}
horizon_days = user_input.get(CONF_HORIZON_DAYS)
if (
isinstance(horizon_days, bool)
or not isinstance(horizon_days, int)
or horizon_days < MIN_HORIZON_DAYS
):
errors[CONF_HORIZON_DAYS] = "invalid"
day_rollover_time = user_input.get(CONF_DAY_ROLLOVER_TIME)
if not isinstance(day_rollover_time, str) or not _parse_time(day_rollover_time):
errors[CONF_DAY_ROLLOVER_TIME] = "invalid_time"
panel_idle_timeout = user_input.get(CONF_PANEL_IDLE_TIMEOUT)
if (
isinstance(panel_idle_timeout, bool)
or not isinstance(panel_idle_timeout, int)
or panel_idle_timeout < MIN_PANEL_IDLE_TIMEOUT
):
errors[CONF_PANEL_IDLE_TIMEOUT] = "invalid"
return errors
def _build_schema(current: dict[str, Any]) -> vol.Schema:
"""Build the form schema, pre-filled with the current values."""
return vol.Schema(
{
vol.Required(
CONF_HORIZON_DAYS, default=current[CONF_HORIZON_DAYS]
): int,
vol.Required(
CONF_DAY_ROLLOVER_TIME, default=current[CONF_DAY_ROLLOVER_TIME]
): str,
vol.Required(
CONF_PANEL_IDLE_TIMEOUT, default=current[CONF_PANEL_IDLE_TIMEOUT]
): int,
}
)
class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
"""Handle the options flow for NestQuest."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the NestQuest options."""
if user_input is None:
return self.async_show_form(
step_id="init", data_schema=_build_schema(self._current_values())
)
errors = _async_validate(user_input)
if errors:
return self.async_show_form(
step_id="init",
data_schema=_build_schema(self._current_values()),
errors=errors,
)
return self.async_create_entry(
title="NestQuest",
data={
CONF_HORIZON_DAYS: user_input[CONF_HORIZON_DAYS],
CONF_DAY_ROLLOVER_TIME: user_input[CONF_DAY_ROLLOVER_TIME],
CONF_PANEL_IDLE_TIMEOUT: user_input[CONF_PANEL_IDLE_TIMEOUT],
},
)
def _current_values(self) -> dict[str, Any]:
"""Resolve current values from entry options, then entry data, then defaults."""
options = self.options
entry_data = getattr(self.config_entry, "data", None) or {}
values: dict[str, Any] = {}
for key, default in (
(CONF_HORIZON_DAYS, DEFAULT_HORIZON_DAYS),
(CONF_DAY_ROLLOVER_TIME, DEFAULT_DAY_ROLLOVER_TIME),
(CONF_PANEL_IDLE_TIMEOUT, DEFAULT_PANEL_IDLE_TIMEOUT),
):
if key in options:
values[key] = options[key]
elif key in entry_data:
values[key] = entry_data[key]
else:
values[key] = default
return values
+34
View File
@@ -0,0 +1,34 @@
{
"title": "NestQuest",
"config": {
"step": {
"user": {
"title": "Add NestQuest",
"description": "Set up NestQuest to start coordinating the household. No additional configuration is required."
}
},
"abort": {
"single_instance_allowed": "NestQuest is already configured. Only a single instance is allowed."
},
"create_entry": {
"default": "NestQuest was successfully configured."
}
},
"options": {
"step": {
"init": {
"title": "NestQuest Settings",
"description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
"data": {
"horizon_days": "Generation horizon (days)",
"day_rollover_time": "Day rollover time (HH:MM)",
"panel_idle_timeout": "Panel idle timeout (seconds)"
}
}
},
"errors": {
"invalid": "Invalid value.",
"invalid_time": "Time must use 24-hour HH:MM format."
}
}
}
@@ -0,0 +1,34 @@
{
"title": "NestQuest",
"config": {
"step": {
"user": {
"title": "Add NestQuest",
"description": "Set up NestQuest to start coordinating the household. No additional configuration is required."
}
},
"abort": {
"single_instance_allowed": "NestQuest is already configured. Only a single instance is allowed."
},
"create_entry": {
"default": "NestQuest was successfully configured."
}
},
"options": {
"step": {
"init": {
"title": "NestQuest Settings",
"description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
"data": {
"horizon_days": "Generation horizon (days)",
"day_rollover_time": "Day rollover time (HH:MM)",
"panel_idle_timeout": "Panel idle timeout (seconds)"
}
}
},
"errors": {
"invalid": "Invalid value.",
"invalid_time": "Time must use 24-hour HH:MM format."
}
}
}
+5
View File
@@ -0,0 +1,5 @@
{
"name": "NestQuest",
"homeassistant": "2024.6.0",
"render_readme": true
}
+1
View File
@@ -9,6 +9,7 @@ dependencies = []
[dependency-groups]
dev = [
"pytest>=8.0.0",
"voluptuous>=0.14",
]
[tool.pytest.ini_options]
+225
View File
@@ -0,0 +1,225 @@
"""Tests for the config flow of NestQuest."""
from __future__ import annotations
import ast
import json
from pathlib import Path
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock
# Mock homeassistant modules if not installed in environment
for _mod in (
"homeassistant",
"homeassistant.core",
"homeassistant.config_entries",
"homeassistant.data_entry_flow",
):
sys.modules.setdefault(_mod, MagicMock())
# Wire parent package attributes to the actual module mocks so that
# "from homeassistant import config_entries" resolves to the mocked module.
sys.modules["homeassistant"].config_entries = sys.modules["homeassistant.config_entries"]
sys.modules["homeassistant"].data_entry_flow = sys.modules["homeassistant.data_entry_flow"]
class _ConfigFlowBase:
"""Minimal stand-in for homeassistant ConfigFlow."""
VERSION = 1
def __init_subclass__(cls, domain=None, **kwargs):
super().__init_subclass__(**kwargs)
cls._domain = domain
def __init__(self):
self.context = {}
def async_show_form(self, *, step_id, **kwargs):
return {"type": "form", "step_id": step_id}
def async_abort(self, *, reason, **kwargs):
return {"type": "abort", "reason": reason}
def async_create_entry(self, *, title, data, **kwargs):
return {"type": "create_entry", "title": title, "data": data}
async def async_set_unique_id(self, unique_id, raise_on_progress=True):
self.context["unique_id"] = unique_id
return None
def _async_current_entries(self):
return self.hass.config_entries.async_entries(self._domain)
def _async_in_progress(self):
return []
sys.modules["homeassistant.config_entries"].ConfigFlow = _ConfigFlowBase
import pytest
from custom_components.nestquest import config_flow
from custom_components.nestquest.const import DOMAIN
PKG_DIR = Path(config_flow.__file__).parent
STRINGS_PATH = PKG_DIR / "strings.json"
TRANSLATIONS_EN_PATH = PKG_DIR / "translations" / "en.json"
def _run(coro):
import asyncio
return asyncio.new_event_loop().run_until_complete(coro)
def _make_flow(existing_entries: list | None = None, in_progress: list | None = None):
"""Create a config flow instance with a stubbed hass."""
flow = config_flow.NestQuestConfigFlow()
flow.hass = SimpleNamespace(
config_entries=SimpleNamespace(
async_entries=lambda domain: [
entry
for entry in (existing_entries or [])
if entry.domain == domain
]
)
)
if in_progress:
flow._async_in_progress = lambda: in_progress
return flow
def _make_entry(domain: str):
"""Create a stub config entry with the given domain."""
return SimpleNamespace(domain=domain)
def test_flow_shows_form_initially() -> None:
"""The initial user step shows a form."""
flow = _make_flow()
result = _run(flow.async_step_user(None))
assert result["type"] == "form"
assert result["step_id"] == "user"
def test_flow_creates_entry_on_submit() -> None:
"""Submitting the user step creates a config entry."""
flow = _make_flow()
result = _run(flow.async_step_user({}))
assert result["type"] == "create_entry"
assert result["title"] == "NestQuest"
assert result["data"] == {}
def test_second_flow_aborts_single_instance() -> None:
"""A second flow aborts with single_instance_allowed."""
flow = _make_flow([_make_entry(DOMAIN)])
result = _run(flow.async_step_user(None))
assert result["type"] == "abort"
assert result["reason"] == "single_instance_allowed"
def test_flow_aborts_when_in_progress() -> None:
"""A concurrent flow in progress aborts with single_instance_allowed."""
flow = _make_flow(in_progress=[{"flow_id": "other"}])
result = _run(flow.async_step_user(None))
assert result["type"] == "abort"
assert result["reason"] == "single_instance_allowed"
def test_flow_sets_stable_unique_id() -> None:
"""The flow assigns a stable unique id equal to the domain."""
flow = _make_flow()
assert flow.context.get("unique_id") is None
_run(flow.async_step_user(None))
assert flow.context["unique_id"] == DOMAIN
def test_flow_ignores_other_domain_entries() -> None:
"""Entries of other domains do not trigger the abort."""
flow = _make_flow([_make_entry("other_domain")])
result = _run(flow.async_step_user(None))
assert result["type"] == "form"
assert result["step_id"] == "user"
@pytest.fixture
def strings() -> dict:
"""Load and parse strings.json."""
assert STRINGS_PATH.exists(), f"strings.json does not exist at {STRINGS_PATH}"
with open(STRINGS_PATH, encoding="utf-8") as f:
return json.load(f)
@pytest.fixture
def translations_en() -> dict:
"""Load and parse translations/en.json."""
assert TRANSLATIONS_EN_PATH.exists(), (
f"translations/en.json does not exist at {TRANSLATIONS_EN_PATH}"
)
with open(TRANSLATIONS_EN_PATH, encoding="utf-8") as f:
return json.load(f)
@pytest.mark.parametrize("fixture_name", ["strings", "translations_en"])
def test_flow_strings_valid_json(fixture_name: str, request: pytest.FixtureRequest) -> None:
"""strings.json and translations/en.json are valid JSON objects."""
content = request.getfixturevalue(fixture_name)
assert isinstance(content, dict)
@pytest.mark.parametrize("fixture_name", ["strings", "translations_en"])
def test_flow_strings_structure(fixture_name: str, request: pytest.FixtureRequest) -> None:
"""Config flow strings contain title, user step, and abort reasons."""
content = request.getfixturevalue(fixture_name)
assert content.get("title")
config = content.get("config", {})
assert config.get("step", {}).get("user", {}).get("title")
assert config.get("step", {}).get("user", {}).get("description")
abort = config.get("abort", {})
assert "single_instance_allowed" in abort
assert abort["single_instance_allowed"]
@pytest.mark.parametrize("fixture_name", ["strings", "translations_en"])
def test_flow_strings_no_raw_keys(fixture_name: str, request: pytest.FixtureRequest) -> None:
"""No placeholder/raw keys appear in the flow strings."""
content = request.getfixturevalue(fixture_name)
def check(value) -> None:
if isinstance(value, dict):
for key, val in value.items():
assert val, f"Empty string value for key '{key}'"
check(val)
elif isinstance(value, str):
assert value.strip() != ""
check(content)
def test_flow_strings_match_translations(
strings: dict, translations_en: dict
) -> None:
"""strings.json and translations/en.json carry identical flow strings."""
assert strings == translations_en
def test_no_hardcoded_domain_or_db_filename_in_new_modules() -> None:
"""Ensure no module other than const.py hardcodes the domain or db filename."""
pkg_dir = Path(config_flow.__file__).parent
py_files = [f for f in pkg_dir.glob("*.py") if f.name not in ("const.py",)]
assert len(py_files) > 0, "No python modules found in package to check"
for py_file in py_files:
tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
if node.value == "nestquest":
raise AssertionError(
f"Found hard-coded domain string in {py_file.name}:{node.lineno}"
)
assert node.value != "nestquest.db", (
f"Found hard-coded database filename in {py_file.name}:{node.lineno}"
)
+101
View File
@@ -0,0 +1,101 @@
"""Tests for const.py of NestQuest."""
from __future__ import annotations
import ast
from datetime import datetime
import logging
from pathlib import Path
import sys
from unittest.mock import MagicMock
# Mock homeassistant modules if not installed in environment
for _mod in ("homeassistant", "homeassistant.core", "homeassistant.config_entries"):
sys.modules.setdefault(_mod, MagicMock())
from custom_components.nestquest import const
def test_domain() -> None:
"""Verify DOMAIN is defined and equals 'nestquest'."""
assert hasattr(const, "DOMAIN")
assert const.DOMAIN == "nestquest"
assert isinstance(const.DOMAIN, str)
def test_logger() -> None:
"""Verify LOGGER_NAME and LOGGER are defined with correct types and values."""
assert hasattr(const, "LOGGER_NAME")
assert const.LOGGER_NAME == "custom_components.nestquest"
assert isinstance(const.LOGGER_NAME, str)
assert hasattr(const, "LOGGER")
assert isinstance(const.LOGGER, logging.Logger)
assert const.LOGGER.name == "custom_components.nestquest"
def test_platforms() -> None:
"""Verify PLATFORMS is defined as an empty list."""
assert hasattr(const, "PLATFORMS")
assert isinstance(const.PLATFORMS, list)
assert const.PLATFORMS == []
def test_sqlite_db_filename() -> None:
"""Verify SQLITE_DB_FILENAME is defined and equals 'nestquest.db'."""
assert hasattr(const, "SQLITE_DB_FILENAME")
assert const.SQLITE_DB_FILENAME == "nestquest.db"
assert isinstance(const.SQLITE_DB_FILENAME, str)
def test_horizon_days_config() -> None:
"""Verify CONF_HORIZON_DAYS and DEFAULT_HORIZON_DAYS are valid."""
assert hasattr(const, "CONF_HORIZON_DAYS")
assert const.CONF_HORIZON_DAYS == "horizon_days"
assert hasattr(const, "DEFAULT_HORIZON_DAYS")
assert isinstance(const.DEFAULT_HORIZON_DAYS, int)
assert const.DEFAULT_HORIZON_DAYS == 14
assert const.DEFAULT_HORIZON_DAYS > 0
def test_day_rollover_time_config() -> None:
"""Verify CONF_DAY_ROLLOVER_TIME and DEFAULT_DAY_ROLLOVER_TIME are valid."""
assert hasattr(const, "CONF_DAY_ROLLOVER_TIME")
assert const.CONF_DAY_ROLLOVER_TIME == "day_rollover_time"
assert hasattr(const, "DEFAULT_DAY_ROLLOVER_TIME")
assert isinstance(const.DEFAULT_DAY_ROLLOVER_TIME, str)
assert const.DEFAULT_DAY_ROLLOVER_TIME == "00:00"
# Validate time format HH:MM
parsed = datetime.strptime(const.DEFAULT_DAY_ROLLOVER_TIME, "%H:%M")
assert parsed.hour == 0 and parsed.minute == 0
def test_panel_idle_timeout_config() -> None:
"""Verify CONF_PANEL_IDLE_TIMEOUT and DEFAULT_PANEL_IDLE_TIMEOUT are valid."""
assert hasattr(const, "CONF_PANEL_IDLE_TIMEOUT")
assert const.CONF_PANEL_IDLE_TIMEOUT == "panel_idle_timeout"
assert hasattr(const, "DEFAULT_PANEL_IDLE_TIMEOUT")
assert isinstance(const.DEFAULT_PANEL_IDLE_TIMEOUT, int)
assert const.DEFAULT_PANEL_IDLE_TIMEOUT == 300
assert const.DEFAULT_PANEL_IDLE_TIMEOUT > 0
def test_no_hardcoded_domain_or_db_filename_in_modules() -> None:
"""Ensure no module other than const.py hardcodes DOMAIN string or SQLITE_DB_FILENAME."""
pkg_dir = Path(const.__file__).parent
py_files = [f for f in pkg_dir.glob("*.py") if f.name != "const.py"]
assert len(py_files) > 0, "No python modules found in package to check"
for py_file in py_files:
content = py_file.read_text(encoding="utf-8")
tree = ast.parse(content, filename=str(py_file))
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
assert node.value != "nestquest.db", (
f"Found hard-coded database filename '{node.value}' in {py_file.name}:{node.lineno}"
)
if node.value == "nestquest":
raise AssertionError(
f"Found hard-coded domain string '{node.value}' in {py_file.name}:{node.lineno}"
)
+82
View File
@@ -0,0 +1,82 @@
"""Tests verifying HACS repository metadata, documentation, and licensing."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).parent.parent
HACS_JSON_PATH = REPO_ROOT / "hacs.json"
LICENSE_PATH = REPO_ROOT / "LICENSE"
README_PATH = REPO_ROOT / "README.md"
MANIFEST_PATH = REPO_ROOT / "custom_components" / "nestquest" / "manifest.json"
@pytest.fixture
def hacs_config() -> dict:
"""Load and parse hacs.json."""
assert HACS_JSON_PATH.exists(), f"hacs.json does not exist at {HACS_JSON_PATH}"
with open(HACS_JSON_PATH, encoding="utf-8") as f:
data = json.load(f)
return data
def test_hacs_json_exists_and_is_valid(hacs_config: dict) -> None:
"""Verify hacs.json exists and is valid JSON."""
assert isinstance(hacs_config, dict)
assert hacs_config.get("name") == "NestQuest"
assert hacs_config.get("homeassistant") == "2024.6.0"
assert hacs_config.get("render_readme") is True
def test_hacs_json_minimum_version_supports_runtime_data(hacs_config: dict) -> None:
"""Advertised HA floor must support ConfigEntry.runtime_data (added in 2024.6)."""
advertised = tuple(int(part) for part in hacs_config.get("homeassistant", "").split("."))
assert advertised >= (2024, 6, 0)
def test_license_exists_and_non_empty() -> None:
"""Verify LICENSE file exists and is non-empty."""
assert LICENSE_PATH.exists(), f"LICENSE file does not exist at {LICENSE_PATH}"
content = LICENSE_PATH.read_text(encoding="utf-8").strip()
assert len(content) > 0
assert "MIT License" in content
assert "NestQuest Contributors / Maniacs Mansion" in content
def test_readme_exists_and_contains_required_sections() -> None:
"""Verify README.md exists and documents HACS installation, manual installation, and test commands."""
assert README_PATH.exists(), f"README.md does not exist at {README_PATH}"
content = README_PATH.read_text(encoding="utf-8")
assert len(content.strip()) > 0
# Title and overview
assert "NestQuest - House Chaos Coordination" in content
assert "Home Assistant custom integration" in content
# Installation methods
assert "HACS" in content
assert "Manual Installation" in content or "manual installation" in content.lower()
assert "custom_components/nestquest" in content
# Developer Setup and Testing commands
assert "uv sync" in content
assert "uv run pytest" in content
def test_readme_hacs_points_at_public_github_mirror() -> None:
"""Verify README directs HACS users to the public GitHub mirror, not the private Gitea repo."""
content = README_PATH.read_text(encoding="utf-8")
assert "https://github.com/talon2king/NestQuest" in content
assert "https://code.cubecraftlabs.com/Maniacs_Mansion/NestQuest" in content
assert content.index("https://github.com/talon2king/NestQuest") < content.index(
"https://code.cubecraftlabs.com/Maniacs_Mansion/NestQuest"
)
def test_hacs_directory_structure() -> None:
"""Verify directory structure matches HACS integration expectations."""
assert MANIFEST_PATH.exists(), f"Integration manifest does not exist at {MANIFEST_PATH}"
with open(MANIFEST_PATH, encoding="utf-8") as f:
manifest = json.load(f)
assert manifest.get("domain") == "nestquest"
+250
View File
@@ -0,0 +1,250 @@
"""Tests for the setup/unload lifecycle of NestQuest."""
from __future__ import annotations
import asyncio
import inspect
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock
# Mock homeassistant modules if not installed in environment
for _mod in (
"homeassistant",
"homeassistant.core",
"homeassistant.config_entries",
):
sys.modules.setdefault(_mod, MagicMock())
# Wire parent package attributes to the actual module mocks so that
# "from homeassistant import config_entries" resolves to the mocked module.
sys.modules["homeassistant"].config_entries = sys.modules["homeassistant.config_entries"]
import pytest
from custom_components.nestquest import DOMAIN, async_setup_entry, async_unload_entry
from custom_components.nestquest.const import DOMAIN as DOMAIN_CONST
def _run(coro):
return asyncio.new_event_loop().run_until_complete(coro)
class _ListenerRegistry:
"""Models HA's update-listener bookkeeping: appends to a list per entry, with per-listener removers."""
def __init__(self, hass=None):
self._hass = hass
self._listeners: dict[str, list[object]] = {}
self.reloaded: list[str] = []
self._loop = asyncio.new_event_loop()
def add(self, entry_id: str, listener) -> object:
def _remove():
self._listeners.get(entry_id, []).remove(listener)
self._listeners.setdefault(entry_id, []).append(listener)
return _remove
def dispatch_options_update(self, entry) -> None:
for listener in list(self._listeners.get(entry.entry_id, [])):
result = listener(self._hass, entry)
if inspect.isawaitable(result):
self._loop.run_until_complete(result)
@property
def size(self) -> int:
return sum(len(listeners) for listeners in self._listeners.values())
def _make_hass():
hass = SimpleNamespace(
data={},
config_entries=SimpleNamespace(
async_reload=None,
),
)
registry = _ListenerRegistry(hass)
async def _async_reload(entry_id: str) -> None:
registry.reloaded.append(entry_id)
hass.config_entries.async_reload = MagicMock(side_effect=_async_reload)
return hass, registry
def _make_entry(entry_id: str = "test_entry", options: dict | None = None):
entry = SimpleNamespace(entry_id=entry_id, options=dict(options or {}))
entry.add_update_listener = None
return entry
def _wire_entry(entry, registry: _ListenerRegistry):
entry.add_update_listener = lambda listener: registry.add(entry.entry_id, listener)
return entry
def test_setup_stores_runtime_data_in_hass_data() -> None:
hass, _ = _make_hass()
entry = _wire_entry(_make_entry(), _ListenerRegistry())
assert _run(async_setup_entry(hass, entry)) is True
assert entry.entry_id in hass.data[DOMAIN]
assert hass.data[DOMAIN][entry.entry_id] is entry.runtime_data
def test_setup_runtime_data_has_entry_id_and_options() -> None:
hass, _ = _make_hass()
entry = _wire_entry(_make_entry(options={"horizon_days": 7}), _ListenerRegistry())
_run(async_setup_entry(hass, entry))
runtime = entry.runtime_data
assert runtime.entry_id == entry.entry_id
assert runtime.options == {"horizon_days": 7}
assert callable(runtime.remove_update_listener)
def test_unload_returns_true_and_removes_hass_data() -> None:
hass, _ = _make_hass()
registry = _ListenerRegistry()
entry = _wire_entry(_make_entry(), registry)
_run(async_setup_entry(hass, entry))
assert _run(async_unload_entry(hass, entry)) is True
assert entry.entry_id not in hass.data[DOMAIN]
def test_unload_is_idempotent() -> None:
hass, _ = _make_hass()
registry = _ListenerRegistry()
entry = _wire_entry(_make_entry(), registry)
_run(async_setup_entry(hass, entry))
assert _run(async_unload_entry(hass, entry)) is True
assert _run(async_unload_entry(hass, entry)) is True
assert entry.entry_id not in hass.data[DOMAIN]
def test_unload_without_setup_returns_true() -> None:
hass, _ = _make_hass()
entry = _make_entry()
assert _run(async_unload_entry(hass, entry)) is True
assert DOMAIN not in hass.data or entry.entry_id not in hass.data[DOMAIN]
def test_unload_clears_entry_runtime_data() -> None:
hass, _ = _make_hass()
registry = _ListenerRegistry()
entry = _wire_entry(_make_entry(), registry)
_run(async_setup_entry(hass, entry))
_run(async_unload_entry(hass, entry))
assert entry.runtime_data is None
def test_unload_removes_update_listener() -> None:
hass, _ = _make_hass()
registry = _ListenerRegistry()
entry = _wire_entry(_make_entry(), registry)
_run(async_setup_entry(hass, entry))
assert registry.size == 1
_run(async_unload_entry(hass, entry))
assert registry.size == 0
def test_options_update_triggers_reload_once() -> None:
hass, registry = _make_hass()
entry = _wire_entry(_make_entry(), registry)
_run(async_setup_entry(hass, entry))
entry.options = {"horizon_days": 30}
registry.dispatch_options_update(entry)
assert hass.config_entries.async_reload.call_count == 1
assert registry.reloaded == [entry.entry_id]
def test_no_reload_after_unload() -> None:
hass, registry = _make_hass()
entry = _wire_entry(_make_entry(), registry)
_run(async_setup_entry(hass, entry))
_run(async_unload_entry(hass, entry))
registry.dispatch_options_update(entry)
assert hass.config_entries.async_reload.call_count == 0
def test_three_reload_cycles_keep_single_listener() -> None:
hass, registry = _make_hass()
entry = _wire_entry(_make_entry(), registry)
for _ in range(3):
assert _run(async_setup_entry(hass, entry)) is True
assert registry.size == 1
assert _run(async_unload_entry(hass, entry)) is True
assert registry.size == 0
assert hass.config_entries.async_reload.call_count == 0
def test_three_setup_unload_cycles_leave_no_listeners() -> None:
"""Listeners accumulate per entry (HA semantics); unload must remove them all."""
hass, registry = _make_hass()
entry = _wire_entry(_make_entry(), registry)
for _ in range(3):
_run(async_setup_entry(hass, entry))
_run(async_unload_entry(hass, entry))
assert registry._listeners.get(entry.entry_id, []) == []
assert registry.size == 0
def test_options_update_after_each_setup_invokes_exactly_one_listener() -> None:
"""After each setup, an options update dispatches to exactly one listener."""
hass, registry = _make_hass()
entry = _wire_entry(_make_entry(), registry)
for _ in range(3):
_run(async_setup_entry(hass, entry))
registry.dispatch_options_update(entry)
assert hass.config_entries.async_reload.call_count == 1
assert registry.reloaded == [entry.entry_id]
_run(async_unload_entry(hass, entry))
hass.config_entries.async_reload.reset_mock()
registry.reloaded.clear()
def test_reload_setup_cycles_dispatch_reload_each_time() -> None:
hass, registry = _make_hass()
entry = _wire_entry(_make_entry(), registry)
for _ in range(3):
_run(async_setup_entry(hass, entry))
registry.dispatch_options_update(entry)
_run(async_unload_entry(hass, entry))
assert hass.config_entries.async_reload.call_count == 3
assert registry.size == 0
def test_no_orphaned_tasks_after_unload() -> None:
async def _scenario():
hass, registry = _make_hass()
entry = _wire_entry(_make_entry(), registry)
await async_setup_entry(hass, entry)
await async_unload_entry(hass, entry)
pending = [task for task in asyncio.all_tasks() if task is not asyncio.current_task()]
assert pending == []
_run(_scenario())
def test_setup_entry_uses_domain_constant() -> None:
hass, _ = _make_hass()
entry = _wire_entry(_make_entry(), _ListenerRegistry())
_run(async_setup_entry(hass, entry))
assert set(hass.data) == {DOMAIN_CONST}
def test_setup_awaitable_listener_removal_supported() -> None:
"""A coroutine-returning remover is awaited during unload."""
async def _scenario():
removed = []
async def _remover():
removed.append(True)
entry = SimpleNamespace(entry_id="awaitable", options={})
entry.add_update_listener = lambda listener: _remover
hass = SimpleNamespace(data={}, config_entries=SimpleNamespace(async_reload=None))
await async_setup_entry(hass, entry)
result = await async_unload_entry(hass, entry)
assert result is True
assert removed == [True]
_run(_scenario())
+532
View File
@@ -0,0 +1,532 @@
"""Tests for the options flow of NestQuest."""
from __future__ import annotations
import ast
import asyncio
import importlib
import inspect
import json
from pathlib import Path
import sys
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock
# Mock homeassistant modules if not installed in environment.
# The mock mirrors the REAL Home Assistant 2024.6 contract:
# config_entries.OptionsFlowWithConfigEntry accepts a config_entry kwarg and
# exposes .options / .config_entry; homeassistant.core.callback exists as a
# decorator; data_entry_flow.RESULT_TYPE_* constants and FlowResult exist.
for _mod in (
"homeassistant",
"homeassistant.core",
"homeassistant.config_entries",
"homeassistant.data_entry_flow",
):
sys.modules.setdefault(_mod, MagicMock())
# Wire parent package attributes to the actual module mocks so that
# "from homeassistant import config_entries" resolves to the mocked module.
sys.modules["homeassistant"].config_entries = sys.modules["homeassistant.config_entries"]
sys.modules["homeassistant"].data_entry_flow = sys.modules["homeassistant.data_entry_flow"]
class _OptionsFlowWithConfigEntryBase:
"""Stand-in mirroring HA 2024.6 config_entries.OptionsFlowWithConfigEntry."""
def __init__(self, config_entry=None):
self.config_entry = config_entry
self.hass = None
self.options = dict(config_entry.options) if config_entry is not None else {}
def async_show_form(self, *, step_id, **kwargs):
return {"type": "form", "step_id": step_id, **kwargs}
def async_create_entry(self, *, title, data, **kwargs):
return {"type": "create_entry", "title": title, "data": data}
class _ConfigFlowBase:
"""Stand-in for homeassistant ConfigFlow; a superset of test_config_flow's base.
test_config_flow.py may be imported after this file, so the concrete
NestQuestConfigFlow can end up bound to this base. It must therefore also
satisfy that file's tests.
"""
VERSION = 1
def __init_subclass__(cls, domain=None, **kwargs):
super().__init_subclass__(**kwargs)
cls._domain = domain
def __init__(self):
self.context = {}
def async_show_form(self, *, step_id, **kwargs):
return {"type": "form", "step_id": step_id, **kwargs}
def async_abort(self, *, reason, **kwargs):
return {"type": "abort", "reason": reason}
def async_create_entry(self, *, title, data, **kwargs):
return {"type": "create_entry", "title": title, "data": data}
async def async_set_unique_id(self, unique_id, raise_on_progress=True):
self.context["unique_id"] = unique_id
return None
def _async_current_entries(self):
return self.hass.config_entries.async_entries(self._domain)
def _async_in_progress(self):
return []
@staticmethod
def async_get_options_flow(config_entry):
raise NotImplementedError
def _callback_decorator(fn):
"""Mirror homeassistant.core.callback: flags the fn and returns it unchanged."""
fn.__ha_callback__ = True
return fn
_config_entries_mock = sys.modules["homeassistant.config_entries"]
_core_mock = sys.modules["homeassistant.core"]
_data_entry_flow_mock = sys.modules["homeassistant.data_entry_flow"]
# The concrete 2024.6-shaped surface the integration is written against:
# plain OptionsFlow has no config_entry constructor (NQ-001); only
# OptionsFlowWithConfigEntry does.
_config_entries_mock.OptionsFlowWithConfigEntry = _OptionsFlowWithConfigEntryBase
_config_entries_mock.ConfigFlow = _ConfigFlowBase
_core_mock.callback = _callback_decorator
# data_entry_flow constants/annotations referenced by the production modules.
_data_entry_flow_mock.RESULT_TYPE_FORM = "form"
_data_entry_flow_mock.RESULT_TYPE_CREATE_ENTRY = "create_entry"
_data_entry_flow_mock.RESULT_TYPE_ABORT = "abort"
_data_entry_flow_mock.FlowResult = dict
# test_config_flow.py may run first and import the integration modules against a
# bare MagicMock surface. Force a clean re-import now that the 2024.6-shaped
# mocks are installed, so the concrete classes bind to them.
for _name in [
name
for name in sys.modules
if name.startswith("custom_components.nestquest")
]:
del sys.modules[_name]
importlib.import_module("custom_components.nestquest")
importlib.import_module("custom_components.nestquest.config_flow")
importlib.import_module("custom_components.nestquest.options_flow")
import pytest
from custom_components.nestquest import config_flow
from custom_components.nestquest import options_flow
from custom_components.nestquest.const import (
CONF_DAY_ROLLOVER_TIME,
CONF_HORIZON_DAYS,
CONF_PANEL_IDLE_TIMEOUT,
DEFAULT_DAY_ROLLOVER_TIME,
DEFAULT_HORIZON_DAYS,
DEFAULT_PANEL_IDLE_TIMEOUT,
DOMAIN,
)
PKG_DIR = Path(options_flow.__file__).parent
STRINGS_PATH = PKG_DIR / "strings.json"
TRANSLATIONS_EN_PATH = PKG_DIR / "translations" / "en.json"
VALID_INPUT = {
CONF_HORIZON_DAYS: 7,
CONF_DAY_ROLLOVER_TIME: "03:30",
CONF_PANEL_IDLE_TIMEOUT: 60,
}
def _run(coro):
return asyncio.new_event_loop().run_until_complete(coro)
class _ListenerRegistry:
"""Models HA's update-listener bookkeeping with reload tracking."""
def __init__(self, hass=None):
self._hass = hass
self._listeners: dict[str, list[object]] = {}
self.reloaded: list[str] = []
self._loop = asyncio.new_event_loop()
def add(self, entry_id: str, listener) -> object:
def _remove():
self._listeners.get(entry_id, []).remove(listener)
self._listeners.setdefault(entry_id, []).append(listener)
return _remove
def dispatch_options_update(self, entry) -> None:
for listener in list(self._listeners.get(entry.entry_id, [])):
result = listener(self._hass, entry)
if inspect.isawaitable(result):
self._loop.run_until_complete(result)
@property
def size(self) -> int:
return sum(len(listeners) for listeners in self._listeners.values())
def _make_hass():
hass = SimpleNamespace(
data={},
config_entries=SimpleNamespace(async_reload=None),
)
registry = _ListenerRegistry(hass)
async def _async_reload(entry_id: str) -> None:
registry.reloaded.append(entry_id)
hass.config_entries.async_reload = MagicMock(side_effect=_async_reload)
return hass, registry
def _make_entry(entry_id: str = "test_entry", options: dict | None = None, data: dict | None = None):
entry = SimpleNamespace(
entry_id=entry_id, options=dict(options or {}), data=dict(data or {})
)
entry.add_update_listener = None
return entry
def _make_flow(entry) -> options_flow.NestQuestOptionsFlow:
"""Create an options flow instance bound to a stubbed config entry."""
flow = options_flow.NestQuestOptionsFlow(config_entry=entry)
flow.hass = SimpleNamespace()
return flow
def _schema_values(schema) -> dict:
"""Extract the per-field default values from a real voluptuous schema."""
return {
marker.schema: marker.default()
for marker in schema.schema
if hasattr(marker, "schema")
}
def test_options_flow_reachable_via_configure_button() -> None:
"""The config flow exposes async_get_options_flow returning the options flow."""
entry = _make_entry()
handler = config_flow.NestQuestConfigFlow.async_get_options_flow(entry)
assert isinstance(handler, options_flow.NestQuestOptionsFlow)
assert handler.config_entry is entry
def test_options_flow_get_options_flow_is_ha_callback() -> None:
"""async_get_options_flow is wrapped by the homeassistant.core.callback decorator."""
assert getattr(
config_flow.NestQuestConfigFlow.async_get_options_flow, "__ha_callback__", False
) is True
def test_options_flow_constructs_through_2024_6_surface() -> None:
"""The handler constructs against the mocked 2024.6 base and inherits its contract."""
entry = _make_entry(options={CONF_HORIZON_DAYS: 21})
flow = options_flow.NestQuestOptionsFlow(config_entry=entry)
assert isinstance(flow, sys.modules["homeassistant.config_entries"].OptionsFlowWithConfigEntry)
assert flow.config_entry is entry
assert flow.options == {CONF_HORIZON_DAYS: 21}
def test_options_flow_submits_through_2024_6_surface() -> None:
"""The handler can be submitted through the mocked 2024.6-shaped surface."""
entry = _make_entry()
handler = config_flow.NestQuestConfigFlow.async_get_options_flow(entry)
result = _run(handler.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
assert result["title"] == "NestQuest"
assert result["data"] == VALID_INPUT
def test_options_flow_shows_form_initially() -> None:
"""The init step shows a form."""
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(None))
assert result["type"] == "form"
assert result["step_id"] == "init"
def test_options_flow_schema_is_voluptuous() -> None:
"""The form's data_schema is a real voluptuous Schema (no dict fallback)."""
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(None))
import voluptuous as vol
assert isinstance(result["data_schema"], vol.Schema)
def test_options_flow_prefills_from_defaults() -> None:
"""Without stored options, the form pre-fills with the constant defaults."""
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(None))
values = _schema_values(result["data_schema"])
assert values[CONF_HORIZON_DAYS] == DEFAULT_HORIZON_DAYS
assert values[CONF_DAY_ROLLOVER_TIME] == DEFAULT_DAY_ROLLOVER_TIME
assert values[CONF_PANEL_IDLE_TIMEOUT] == DEFAULT_PANEL_IDLE_TIMEOUT
def test_options_flow_prefills_from_entry_options() -> None:
"""Stored entry options win over entry data and defaults."""
entry = _make_entry(
options={
CONF_HORIZON_DAYS: 21,
CONF_DAY_ROLLOVER_TIME: "05:45",
CONF_PANEL_IDLE_TIMEOUT: 120,
},
data={
CONF_HORIZON_DAYS: 99,
CONF_DAY_ROLLOVER_TIME: "23:59",
CONF_PANEL_IDLE_TIMEOUT: 999,
},
)
result = _run(_make_flow(entry).async_step_init(None))
values = _schema_values(result["data_schema"])
assert values[CONF_HORIZON_DAYS] == 21
assert values[CONF_DAY_ROLLOVER_TIME] == "05:45"
assert values[CONF_PANEL_IDLE_TIMEOUT] == 120
def test_options_flow_falls_back_to_entry_data() -> None:
"""Pre-upgrade entries without options fall back to entry.data."""
entry = _make_entry(
data={
CONF_HORIZON_DAYS: 10,
CONF_DAY_ROLLOVER_TIME: "01:15",
CONF_PANEL_IDLE_TIMEOUT: 90,
}
)
result = _run(_make_flow(entry).async_step_init(None))
values = _schema_values(result["data_schema"])
assert values[CONF_HORIZON_DAYS] == 10
assert values[CONF_DAY_ROLLOVER_TIME] == "01:15"
assert values[CONF_PANEL_IDLE_TIMEOUT] == 90
def test_options_flow_schema_applies_stored_defaults() -> None:
"""A real voluptuous validate({}) fills every field from the stored defaults."""
entry = _make_entry(
options={
CONF_HORIZON_DAYS: 21,
CONF_DAY_ROLLOVER_TIME: "05:45",
CONF_PANEL_IDLE_TIMEOUT: 120,
}
)
result = _run(_make_flow(entry).async_step_init(None))
assert result["data_schema"]({}) == {
CONF_HORIZON_DAYS: 21,
CONF_DAY_ROLLOVER_TIME: "05:45",
CONF_PANEL_IDLE_TIMEOUT: 120,
}
def test_options_flow_schema_coerces_field_types() -> None:
"""The schema validates the expected field types via voluptuous."""
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(None))
schema = result["data_schema"]
validators = {marker.schema: validator for marker, validator in schema.schema.items()}
assert validators[CONF_HORIZON_DAYS] is int
assert validators[CONF_DAY_ROLLOVER_TIME] is str
assert validators[CONF_PANEL_IDLE_TIMEOUT] is int
@pytest.mark.parametrize("bad_horizon", [0, -5])
def test_options_flow_rejects_non_positive_horizon(bad_horizon: int) -> None:
"""Zero or negative horizon values yield a field-level error, no exception."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_HORIZON_DAYS: bad_horizon})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_HORIZON_DAYS: "invalid"}
def test_options_flow_rejects_malformed_time_not_a_time() -> None:
"""Non-HH:MM strings are rejected with invalid_time."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_DAY_ROLLOVER_TIME: "noon"})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_DAY_ROLLOVER_TIME: "invalid_time"}
def test_options_flow_rejects_malformed_time_out_of_range() -> None:
"""Out-of-range HH:MM values are rejected with invalid_time."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_DAY_ROLLOVER_TIME: "25:99"})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_DAY_ROLLOVER_TIME: "invalid_time"}
def test_options_flow_rejects_pasted_unicode_superscript_time() -> None:
"""Pasted Unicode digits like '²²:²²' yield a field-level error, no exception."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_DAY_ROLLOVER_TIME: "²²:²²"})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_DAY_ROLLOVER_TIME: "invalid_time"}
@pytest.mark.parametrize("bad_timeout", ["abc", 0, -10, 10])
def test_options_flow_rejects_invalid_idle_timeout(bad_timeout) -> None:
"""Non-numeric or too-low idle timeouts yield a field-level error."""
flow = _make_flow(_make_entry())
result = _run(
flow.async_step_init({**VALID_INPUT, CONF_PANEL_IDLE_TIMEOUT: bad_timeout})
)
assert result["type"] == "form"
assert result["errors"] == {CONF_PANEL_IDLE_TIMEOUT: "invalid"}
def test_options_flow_creates_entry_with_three_keys() -> None:
"""A valid submit creates an entry with exactly the three validated keys."""
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
assert result["title"] == "NestQuest"
assert set(result["data"]) == {
CONF_HORIZON_DAYS,
CONF_DAY_ROLLOVER_TIME,
CONF_PANEL_IDLE_TIMEOUT,
}
assert result["data"] == VALID_INPUT
def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once() -> None:
"""The flow's created data is written to entry.options and reloads exactly once."""
hass, registry = _make_hass()
entry = _make_entry()
entry.add_update_listener = lambda listener: registry.add(entry.entry_id, listener)
# Simulate setup registering the merged lifecycle update listener.
from custom_components.nestquest import async_setup_entry
_run(async_setup_entry(hass, entry))
# Run the options flow and apply its created data to entry.options like HA does.
flow = _make_flow(entry)
result = _run(flow.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
entry.options = dict(result["data"])
registry.dispatch_options_update(entry)
assert hass.config_entries.async_reload.call_count == 1
assert registry.reloaded == [entry.entry_id]
assert entry.options == VALID_INPUT
def test_options_flow_validation_never_raises() -> None:
"""Validation returns errors instead of raising for arbitrary junk input."""
junk = {CONF_HORIZON_DAYS: None, CONF_DAY_ROLLOVER_TIME: None, CONF_PANEL_IDLE_TIMEOUT: None}
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(junk))
assert result["type"] == "form"
assert result["errors"] == {
CONF_HORIZON_DAYS: "invalid",
CONF_DAY_ROLLOVER_TIME: "invalid_time",
CONF_PANEL_IDLE_TIMEOUT: "invalid",
}
@pytest.fixture
def strings() -> dict:
"""Load and parse strings.json."""
assert STRINGS_PATH.exists(), f"strings.json does not exist at {STRINGS_PATH}"
with open(STRINGS_PATH, encoding="utf-8") as f:
return json.load(f)
@pytest.fixture
def translations_en() -> dict:
"""Load and parse translations/en.json."""
assert TRANSLATIONS_EN_PATH.exists(), (
f"translations/en.json does not exist at {TRANSLATIONS_EN_PATH}"
)
with open(TRANSLATIONS_EN_PATH, encoding="utf-8") as f:
return json.load(f)
def test_options_strings_present_and_matching(strings: dict, translations_en: dict) -> None:
"""Both string files carry identical options step/error strings."""
assert strings.get("options") == translations_en.get("options")
options = strings["options"]
init = options["step"]["init"]
assert init["title"]
assert init["description"]
data = init["data"]
for key in (CONF_HORIZON_DAYS, CONF_DAY_ROLLOVER_TIME, CONF_PANEL_IDLE_TIMEOUT):
assert data.get(key), f"Missing field name for '{key}'"
errors = options["errors"]
for key in ("invalid", "invalid_time"):
assert errors.get(key), f"Missing error string for '{key}'"
def test_options_strings_no_raw_keys(strings: dict, translations_en: dict) -> None:
"""No placeholder/raw keys appear in the options strings."""
def check(value) -> None:
if isinstance(value, dict):
for key, val in value.items():
assert val, f"Empty string value for key '{key}'"
check(val)
elif isinstance(value, str):
assert value.strip() != ""
check(strings["options"])
check(translations_en["options"])
def test_options_flow_no_plain_dict_schema_fallback() -> None:
"""No try/except ImportError fallback for voluptuous in the options flow module."""
py_file = Path(options_flow.__file__)
tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
for node in ast.walk(tree):
if isinstance(node, ast.Try):
handlers = node.handlers or []
has_import_error_handler = any(
getattr(handler.type, "id", getattr(handler.type, "attr", None))
== "ImportError"
for handler in handlers
)
assert not has_import_error_handler, (
f"Plain-dict schema fallback (try/except ImportError) found in {py_file.name}:{node.lineno}"
)
def test_options_flow_subclasses_options_flow_with_config_entry() -> None:
"""NestQuestOptionsFlow subclasses config_entries.OptionsFlowWithConfigEntry."""
base = sys.modules["homeassistant.config_entries"].OptionsFlowWithConfigEntry
assert issubclass(options_flow.NestQuestOptionsFlow, base)
assert options_flow.NestQuestOptionsFlow.__bases__ == (base,)
def test_options_flow_no_hardcoded_domain_or_db_filename() -> None:
"""The options flow module does not hardcode the domain or db filename."""
py_file = Path(options_flow.__file__)
tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
for node in ast.walk(tree):
if isinstance(node, ast.Constant) and isinstance(node.value, str):
assert node.value != DOMAIN, (
f"Hard-coded domain string in {py_file.name}:{node.lineno}"
)
assert node.value != "nestquest.db", (
f"Hard-coded database filename in {py_file.name}:{node.lineno}"
)