bvandeusenandClaude Opus 5.5
4c57735dd9
The tile shows Home Assistant's own used figure beside the total, so it matches Home Assistant's storage page. An OS update doesn't reboot the machine: Supervisor writes the image, reports the new version and raises a repair asking for a reboot. The confirmation, schedule warning, MCP docs and plugin docs now say so. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
253 lines
15 KiB
Python
253 lines
15 KiB
Python
"""Reading what Home Assistant reports: which updates Steward may install, how
|
|
its backups stand, and what the pages and MCP show."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace as NS
|
|
|
|
import plugins.homeassistant as homeassistant_plugin
|
|
from plugins.homeassistant import mcp_tools, schedule, 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():
|
|
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("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"]
|
|
backup = snapshot["backup"]
|
|
assert backup["last_ok"] == "2026-10-31T03:00:00+00:00" and backup["next"] is None
|
|
assert backup["last_result"] == "completed"
|
|
|
|
|
|
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_a_refused_sign_in_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", "refresh_token": "tok-s3cret", "can_install": True,
|
|
"verify_tls": True, "signed_in_by": "someone", "signed_in_at": NOW,
|
|
"refused": False, "last_error": None, "last_polled_at": NOW, "last_ok_at": NOW,
|
|
"backup": {"known": True, "last_ok": NOW.isoformat(), "last_result": "completed"},
|
|
"updates": [], "health": {}, **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"]
|
|
two = {"problems": [{"title": "Disk trouble"}, {"title": "Another"}]}
|
|
[f] = mcp_tools.overview_findings(_connection(health=two), None, NOW)
|
|
assert f["kind"] == "home_assistant_problem" and f["summary"] == "Disk trouble (and 1 more)"
|
|
plan = NS(stopped="Core: it failed.", last_slot_at=None)
|
|
[f] = mcp_tools.overview_findings(_connection(), None, NOW, plan)
|
|
assert f["kind"] == "home_assistant_schedule_stopped" and "Core: it failed." in f["summary"]
|
|
|
|
|
|
def test_system_problems_are_read_from_what_home_assistant_answers():
|
|
healthy = s.read_health({"config": {"recovery_mode": False}, "issues": [], "resolution": {"unhealthy": [], "unsupported": []},
|
|
"host": {"disk_free": 99.8, "disk_used": 12.0, "disk_total": 116.5},
|
|
"addons": [{"slug": "a", "state": "started"}]})
|
|
assert healthy["problems"] == [] and healthy["repairs_read"] and healthy["supervised"]
|
|
assert healthy["disk_free_pct"] == 85.7 and healthy["disk_used_gb"] == 12.0 and healthy["disk_used_pct"] == 10.3
|
|
bad = s.read_health({
|
|
"config": {"recovery_mode": True},
|
|
"issues": [{"domain": "hassio", "translation_key": "issue_system_reboot_required", "severity": "warning"},
|
|
{"domain": "demo", "translation_key": "named", "translation_placeholders": {"who": "X"}, "severity": "critical"},
|
|
{"domain": "demo", "translation_key": "gone", "ignored": True},
|
|
{"domain": "demo", "translation_key": "old", "dismissed_version": "1.0"}],
|
|
"titles": {"component.demo.issues.named.title": "{who} broke"},
|
|
"resolution": {"unhealthy": ["docker_gateway"], "unsupported": []},
|
|
"host": {"disk_free": 5.0, "disk_total": 100.0},
|
|
"addons": [{"slug": "should", "name": "Should run", "state": "stopped"}, {"slug": "once", "state": "stopped"}],
|
|
"boots": {"should": "auto", "once": "manual"}})
|
|
assert [p["kind"] for p in bad["problems"]] == ["recovery", "repair", "repair", "supervisor", "disk", "addon"]
|
|
titles = [p["title"] for p in bad["problems"]]
|
|
assert titles[1] == "System reboot required" and titles[2] == "X broke" and "docker gateway" in titles[3]
|
|
assert "5 GB of 100 GB" in titles[4] and "Should run" in titles[5]
|
|
|
|
|
|
def test_what_home_assistant_wouldnt_answer_is_not_read_as_fine():
|
|
found = s.read_health({"config": {}, "issues": None, "resolution": None, "host": None, "addons": None})
|
|
assert not found["repairs_read"] and not found["supervised"]
|
|
assert found["disk_free_pct"] is None and found["disk_used_pct"] is None
|
|
|
|
|
|
def test_a_schedule_installs_add_ons_then_core_then_the_os():
|
|
def update(entity_id, kind, in_progress=False):
|
|
return {"entity_id": entity_id, "kind": kind, "in_progress": in_progress}
|
|
waiting = [update("os", "os"), update("core", "core"), update("addon", "addon")]
|
|
assert schedule.next_to_install(waiting, set())["entity_id"] == "addon"
|
|
assert schedule.next_to_install(waiting, {"addon"})["entity_id"] == "core"
|
|
assert schedule.next_to_install(waiting, {"addon", "core"})["entity_id"] == "os"
|
|
assert schedule.next_to_install(waiting, {"addon", "core", "os"}) is None
|
|
assert schedule.next_to_install([update("addon", "addon", in_progress=True)], set()) is None
|
|
|
|
|
|
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),
|
|
"set_home_assistant_schedule": (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, "health": {"repairs_read": True, "supervised": True}, "problems": [], "plan": None,
|
|
"plan_when": None, "plan_next": None, "weekdays": ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday",
|
|
"Saturday", "Sunday"), "default_zone": "UTC", **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
|
|
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
|
|
status_only = await _render(app, "admin", **_page_values(connection=_connection(can_install=False)))
|
|
assert 'action="/plugins/homeassistant/install"' not in status_only
|
|
assert "isn't an administrator" in status_only
|
|
|
|
|
|
async def test_the_page_shows_system_problems_and_what_it_couldnt_read(app):
|
|
clear = await _render(app, "viewer", **_page_values())
|
|
assert "No repairs open" in clear
|
|
problem = {"kind": "repair", "severity": "warning", "title": "System reboot required"}
|
|
shown = await _render(app, "viewer", **_page_values(problems=[problem]))
|
|
assert "System reboot required" in shown and "Settings → System → Repairs" in shown
|
|
hidden = await _render(app, "viewer", **_page_values(health={"repairs_read": False, "supervised": False}))
|
|
assert "wouldn't share its repairs" in hidden
|
|
bare = await _render(app, "viewer", **_page_values(health={"repairs_read": True, "supervised": False}))
|
|
assert "No Supervisor answered" in bare
|
|
|
|
|
|
async def test_an_admin_sets_the_schedule_warned_of_what_it_does_and_a_viewer_only_reads_it(app):
|
|
off = await _render(app, "admin", **_page_values())
|
|
assert 'action="/plugins/homeassistant/schedule"' in off and "Off. Updates are only installed" in off
|
|
assert "with nobody there to watch" in off
|
|
assert '<option value="5" selected>Saturday</option>' in off and 'value="05:00"' in off
|
|
plan = NS(enabled=True, weekday=5, at_time="05:00", time_zone="Europe/London", stopped="Core: it failed.")
|
|
values = _page_values(plan=plan, plan_when="Saturdays at 05:00 (Europe/London), up to 4 h", plan_next=NOW)
|
|
on = await _render(app, "viewer", **values)
|
|
assert "Saturdays at 05:00" in on and "Core: it failed." in on
|
|
assert 'action="/plugins/homeassistant/schedule"' not in on
|
|
|
|
|
|
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 _form(app, **values):
|
|
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", error=None,
|
|
url="http://ha.example:8123", verify_tls=True, **values)
|
|
return " ".join(page.split())
|
|
|
|
|
|
async def test_the_form_warns_what_the_credential_can_do_and_recommends_a_service_account(app):
|
|
page = await _form(app, connection=None)
|
|
assert "same access as the user who signs in" in page and "including controlling every device" in page
|
|
assert "We recommend a service account" in page and "in Home Assistant's logs" in page
|
|
assert 'action="/plugins/homeassistant/connection/sign-in"' in page and "Continue to Home Assistant" in page
|
|
assert "connection/delete" not in page
|
|
|
|
|
|
async def test_a_connected_form_says_who_signed_in_and_never_shows_the_credential(app):
|
|
page = await _form(app, connection=_connection())
|
|
assert "tok-s3cret" not in page and "by someone" in page and "an administrator" in page
|
|
assert "Sign in again" in page and "connection/delete" in page
|