bvandeusenandClaude Opus 5.5
9c9e14e79c
Without it the plugin's hook raised at boot and left it with no tools. The test's stand-in for register gave read_only a default, so now both arguments are required there too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
193 lines
10 KiB
Python
193 lines
10 KiB
Python
"""Reading what Home Assistant reports: which updates Steward may install, how
|
|
its backups stand, the automation to add, and what the pages and MCP show."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace as NS
|
|
|
|
import yaml
|
|
|
|
import plugins.homeassistant as homeassistant_plugin
|
|
from plugins.homeassistant import mcp_tools, scheduler, service as s
|
|
from steward.models.users import UserRole
|
|
from tests._templates import render_plugin_template
|
|
from tests.test_style_tokens import hex_colours
|
|
|
|
NOW = datetime(2026, 10, 31, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _entity(entity_id, state="on", **attrs):
|
|
return {"entity_id": entity_id, "state": state, "attributes": attrs}
|
|
|
|
|
|
def _addon(slug="core_example", state="on"):
|
|
return _entity(f"update.{slug}_update", state, title="Example add-on", installed_version="1.0",
|
|
latest_version="1.1", entity_picture=f"/api/hassio/addons/{slug}/icon")
|
|
|
|
|
|
def test_only_core_os_and_add_ons_can_be_installed():
|
|
assert s.classify(_entity(s.CORE)) == "core"
|
|
assert s.classify(_entity(s.OS)) == "os"
|
|
assert s.classify(_addon()) == "addon"
|
|
assert s.classify(_entity(s.SUPERVISOR, entity_picture="/api/brands/integration/hassio/icon.png")) is None
|
|
assert s.classify(_entity("update.switch_firmware", device_class="firmware",
|
|
entity_picture="/api/brands/integration/unifi/icon.png")) is None
|
|
assert s.classify(_entity("update.custom_locks_update",
|
|
entity_picture="https://brands.home-assistant.io/_/locks/icon.png")) is None
|
|
|
|
|
|
def test_a_snapshot_keeps_waiting_updates_first_and_finds_the_automation_by_its_id():
|
|
snapshot = s.read_snapshot([
|
|
_addon("core_quiet", "off"), _entity("update.switch_firmware", title="Switch"),
|
|
_addon(), _entity(s.CORE, title="Home Assistant Core"),
|
|
_entity("update.gone", "unavailable"),
|
|
_entity(s.LAST_BACKUP, "2026-10-31T03:00:00+00:00"), _entity(s.NEXT_BACKUP, "unknown"),
|
|
_entity(s.BACKUP_EVENT, "2026-10-31T03:00:00+00:00", event_type="completed"),
|
|
_entity("automation.renamed_by_someone", "on", id=s.AUTOMATION_ID, last_triggered=None),
|
|
_entity("light.kitchen", "on"),
|
|
])
|
|
assert [u["entity_id"] for u in snapshot["updates"]] == [
|
|
s.CORE, "update.core_example_update", "update.switch_firmware", "update.core_quiet_update"]
|
|
assert snapshot["automation"] == "on"
|
|
backup = snapshot["backup"]
|
|
assert backup["last_ok"] == "2026-10-31T03:00:00+00:00" and backup["next"] is None
|
|
assert backup["last_result"] == "completed"
|
|
assert s.automation_entity_id(NS(backup=backup)) == "automation.renamed_by_someone"
|
|
assert s.read_snapshot([])["automation"] is None
|
|
|
|
|
|
def test_a_release_link_is_only_kept_when_it_is_a_web_address():
|
|
assert s.read_snapshot([_entity(s.CORE, release_url="javascript:alert(1)")])["updates"][0]["release_url"] is None
|
|
assert s.read_snapshot([_entity(s.CORE, release_url="https://example.com/r")])["updates"][0]["release_url"]
|
|
|
|
|
|
def test_newer_is_judged_on_the_readings_alone():
|
|
assert s.is_newer("2026-10-31T03:00:01+00:00", "2026-10-31T03:00:00+00:00")
|
|
assert not s.is_newer("2026-10-31T03:00:00+00:00", "2026-10-31T03:00:00+00:00")
|
|
assert not s.is_newer("2026-10-30T03:00:00+00:00", "2026-10-31T03:00:00+00:00")
|
|
assert s.is_newer("2026-10-31T03:00:00+00:00", "unknown")
|
|
assert not s.is_newer("unavailable", "2026-10-31T03:00:00+00:00")
|
|
|
|
|
|
def test_how_the_backups_read():
|
|
ok = {"known": True, "last_ok": (NOW - timedelta(hours=25)).isoformat(), "last_result": "completed"}
|
|
assert s.backup_verdict(ok, NOW)[0] == "ok"
|
|
assert abs(s.backup_age_hours(ok, NOW) - 25) < 0.01
|
|
stale = {**ok, "last_ok": (NOW - timedelta(hours=27)).isoformat()}
|
|
assert s.backup_verdict(stale, NOW) == ("warn", "The last good backup is 27 hours old, past the 26 hours expected.")
|
|
failed = {**ok, "last_result": "failed", "failed_reason": "upload_failed"}
|
|
assert "failed (upload_failed)" in s.backup_verdict(failed, NOW)[1]
|
|
assert "no successful" in s.backup_verdict({"known": True, "last_ok": None}, NOW)[1]
|
|
assert "reports no automatic backups" in s.backup_verdict({}, NOW)[1]
|
|
|
|
|
|
def test_the_automation_to_add_only_takes_a_backup_or_installs_what_it_allows():
|
|
text = s.automation_yaml("steward_abc")
|
|
config = yaml.safe_load(text)
|
|
assert config["id"] == s.AUTOMATION_ID and config["mode"] == "single"
|
|
[trigger] = config["triggers"]
|
|
assert trigger == {"trigger": "webhook", "webhook_id": "steward_abc", "allowed_methods": ["POST"], "local_only": True}
|
|
assert "integration_entities('hassio')" in config["variables"]["allowed"]
|
|
assert "'update\\\\.'" in config["variables"]["allowed"]
|
|
assert s.SUPERVISOR in config["variables"]["allowed"]
|
|
branches = config["actions"][0]["choose"]
|
|
assert [b["sequence"][0]["action"] for b in branches] == ["backup.create_automatic", "update.install"]
|
|
assert "in allowed" in branches[1]["conditions"][0]["value_template"]
|
|
|
|
|
|
def test_a_pasted_secret_loses_a_terminals_wrapping():
|
|
assert s.no_spaces(" eyJabc\n def ghi\n") == "eyJabcdefghi"
|
|
assert s.no_spaces(None) == ""
|
|
|
|
|
|
def test_a_refused_token_stops_polling_until_someone_acts():
|
|
fresh = NS(refused=False, last_polled_at=NOW - timedelta(seconds=60))
|
|
assert not scheduler.is_due(fresh, NOW)
|
|
assert scheduler.is_due(NS(refused=False, last_polled_at=NOW - timedelta(seconds=300)), NOW)
|
|
assert scheduler.is_due(NS(refused=False, last_polled_at=None), NOW)
|
|
assert not scheduler.is_due(NS(refused=True, last_polled_at=None), NOW)
|
|
|
|
|
|
def _connection(**over):
|
|
return NS(**{"url": "http://ha.example:8123", "token": "tok-s3cret", "webhook_id": "hook-s3cret",
|
|
"refused": False, "last_error": None, "last_polled_at": NOW, "last_ok_at": NOW,
|
|
"automation": "on", "backup": {"known": True, "last_ok": NOW.isoformat(), "last_result": "completed"},
|
|
"updates": [], **over})
|
|
|
|
|
|
def test_the_overview_names_what_needs_a_look_and_no_more():
|
|
assert mcp_tools.overview_findings(None, None, NOW) == []
|
|
assert mcp_tools.overview_findings(_connection(), None, NOW) == []
|
|
waiting = _connection(updates=[{"waiting": True, "kind": "core", "title": "Core"}])
|
|
assert mcp_tools.overview_findings(waiting, None, NOW) == []
|
|
down = mcp_tools.overview_findings(_connection(last_error="Couldn't reach Home Assistant."), None, NOW)
|
|
assert [f["kind"] for f in down] == ["home_assistant_down"]
|
|
stale = _connection(backup={"known": True, "last_ok": (NOW - timedelta(days=2)).isoformat()})
|
|
assert [f["kind"] for f in mcp_tools.overview_findings(stale, None, NOW)] == ["home_assistant_backup"]
|
|
failed = NS(state="failed", title="Core", message="No new backup.", finished_at=NOW)
|
|
[f] = mcp_tools.overview_findings(_connection(), failed, NOW)
|
|
assert f["kind"] == "home_assistant_update_failed" and "No new backup." in f["summary"]
|
|
|
|
|
|
def test_reading_is_open_to_a_viewer_and_installing_is_for_an_admin():
|
|
seen = {}
|
|
|
|
def register(fn, *, role, read_only): # both required, as on the server
|
|
seen[fn.__name__] = (role, read_only)
|
|
homeassistant_plugin.register_mcp_tools(register)
|
|
assert seen == {"get_home_assistant": (UserRole.viewer, True), "update_home_assistant": (UserRole.admin, False)}
|
|
|
|
|
|
def _page_values(**over):
|
|
update = {"entity_id": s.CORE, "title": "Core", "kind": "core", "kind_word": "Core", "installed": "1.0",
|
|
"latest": "1.1", "waiting": True, "in_progress": False, "auto_update": False,
|
|
"release_url": "https://example.com/r"}
|
|
return {"connection": _connection(), "reach": "up", "says": "Answering.", "installable": [update], "others": [],
|
|
"current": [], "runs": [], "under_way": None, "backup": {}, "backup_tone": "ok",
|
|
"backup_says": "Backups are up to date.", "backup_at": NOW, "backup_next": None, "backup_age": 1.0,
|
|
"stale_hours": 26, **over}
|
|
|
|
|
|
async def _render(app, role, **values):
|
|
from quart import session
|
|
async with app.test_request_context("/plugins/homeassistant/"):
|
|
session["user_role"] = role
|
|
page = await render_plugin_template(app, "homeassistant", "homeassistant/index.html", **values)
|
|
return " ".join(page.split())
|
|
|
|
|
|
async def test_an_admin_is_asked_before_an_install_and_a_viewer_cant_install(app):
|
|
page = await _render(app, "admin", **_page_values())
|
|
assert "1 update waiting" in page and 'action="/plugins/homeassistant/install"' in page
|
|
assert "Steward takes a backup first" in page and "Home Assistant restarts." in page
|
|
assert "tok-s3cret" not in page and "hook-s3cret" not in page
|
|
main = page[page.index("<main"):]
|
|
assert not hex_colours(main), hex_colours(main)
|
|
assert 'action="/plugins/homeassistant/install"' not in await _render(app, "viewer", **_page_values())
|
|
|
|
|
|
async def test_install_is_held_back_while_it_couldnt_work_and_the_page_says_why(app):
|
|
run = NS(id="r", title="Core", state="backing_up", from_version="1.0", to_version="1.1", started_by="someone",
|
|
started_at=NOW, message=None)
|
|
busy = await _render(app, "admin", **_page_values(under_way=run, runs=[run]))
|
|
assert 'action="/plugins/homeassistant/install"' not in busy and "taking a backup first" in busy
|
|
assert 'hx-trigger="every 15s"' in busy
|
|
gone = await _render(app, "admin", **_page_values(connection=_connection(automation=None)))
|
|
assert 'action="/plugins/homeassistant/install"' not in gone
|
|
assert "maintenance automation isn't in Home Assistant" in gone
|
|
|
|
|
|
async def test_an_unconnected_page_offers_to_connect(app):
|
|
page = await _render(app, "admin", connection=None)
|
|
assert 'href="/plugins/homeassistant/connection"' in page
|
|
|
|
|
|
async def test_the_form_shows_the_automation_but_never_a_saved_secret_in_a_field(app):
|
|
from quart import session
|
|
async with app.test_request_context("/plugins/homeassistant/connection"):
|
|
session["user_role"] = "admin"
|
|
page = await render_plugin_template(
|
|
app, "homeassistant", "homeassistant/form.html", connection=_connection(),
|
|
v={"url": "http://ha.example:8123", "verify_tls": True, "has_token": True, "has_webhook": True},
|
|
error=None, automation=s.automation_yaml("generated-id"))
|
|
assert "generated-id" in page and "tok-s3cret" not in page
|
|
assert "kept; type to change" in page
|