FabledSteward/tests/plugins/homeassistant/test_service.py

"""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 = {}
    homeassistant_plugin.register_mcp_tools(lambda fn, role, read_only=False: seen.update({fn.__name__: (role, read_only)}))
    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