"""Offline tests for the seven read-only tools.""" import pytest from ha_mcp import backend as backend_mod from ha_mcp import tools from ha_mcp.backend import BackendError SALON_TEMP = "sensor.thsalon_temperature" MDWEJSCIE = "binary_sensor.0x00124b00251472ef_occupancy" KLIMA_OFF_ID = "1784804668795" # --- list_entities ------------------------------------------------------- def test_list_entities_filters_by_domain(fake_backend): result = tools.list_entities(fake_backend, domain="sensor") ids = [e["entity_id"] for e in result["entities"]] assert ids == [ "sensor.thsalon_humidity", SALON_TEMP, "sensor.thsypialnia_temperature", ] assert result["count"] == 3 # select.thsalon_temperature_unit shares the prefix but not the domain assert "select.thsalon_temperature_unit" not in ids def test_list_entities_filters_by_area(fake_backend): result = tools.list_entities(fake_backend, area="Salon") ids = {e["entity_id"] for e in result["entities"]} # three via the device registry, one via its own area_id assert ids == { SALON_TEMP, "sensor.thsalon_humidity", "select.thsalon_temperature_unit", "light.salon_lampa", } assert all(e["area"] == "Salon" for e in result["entities"]) def test_list_entities_area_accepts_id_alias_and_any_casing(fake_backend): assert tools.list_entities(fake_backend, area="salon")["count"] == 4 # area_id / lowercase by_alias = tools.list_entities(fake_backend, area="wejście") # registry alias for Hall assert by_alias["count"] == 1 assert by_alias["filters"]["area"] == "Hall" # reported under its canonical name assert tools.list_entities(fake_backend, area="wejscie")["count"] == 1 # no diacritics def test_list_entities_domain_and_area_combined(fake_backend): result = tools.list_entities(fake_backend, domain="sensor", area="Salon") assert [e["entity_id"] for e in result["entities"]] == [ "sensor.thsalon_humidity", SALON_TEMP, ] def test_list_entities_unknown_area_is_an_error_not_an_empty_list(fake_backend): with pytest.raises(BackendError) as exc: tools.list_entities(fake_backend, area="Piwnica") assert "unknown area" in str(exc.value) assert "Salon" in str(exc.value) def test_list_entities_marks_unavailable(fake_backend): result = tools.list_entities(fake_backend) by_id = {e["entity_id"]: e for e in result["entities"]} dead = by_id[MDWEJSCIE] assert dead["unavailable"] is True # how long it has been dead is the actionable part (audit 1.2) assert dead["unavailable_since"] == "2026-07-17T15:07:12.000000+00:00" alive = by_id[SALON_TEMP] assert "unavailable" not in alive assert result["unavailable_count"] == 2 # mdwejscie + thsypialnia def test_list_entities_truncates_loudly(fake_backend): result = tools.list_entities(fake_backend, limit=2) assert len(result["entities"]) == 2 assert result["truncated"] is True assert result["count"] == 10 # the full match count is still reported def test_list_entities_reports_where_areas_came_from(fake_backend): assert tools.list_entities(fake_backend)["area_source"] == "websocket" # --- get_state ----------------------------------------------------------- def test_get_state_returns_full_attributes_and_area(fake_backend): result = tools.get_state(fake_backend, SALON_TEMP) assert result["state"] == "24.3" assert result["area"] == "Salon" assert result["attributes"]["unit_of_measurement"] == "°C" assert result["attributes"]["device_class"] == "temperature" assert result["last_changed"] == "2026-07-27T09:12:03.101000+00:00" def test_get_state_flags_unavailable(fake_backend): result = tools.get_state(fake_backend, MDWEJSCIE) assert result["unavailable"] is True assert result["state"] == "unavailable" def test_get_state_unknown_entity_suggests_candidates(fake_backend): with pytest.raises(BackendError) as exc: tools.get_state(fake_backend, "sensor.thsalon_temperatur") message = str(exc.value) assert "no entity" in message assert SALON_TEMP in message # closest match offered, not just a flat error # --- get_areas ----------------------------------------------------------- def test_get_areas_counts_entities_and_dead_ones(fake_backend): result = tools.get_areas(fake_backend) by_name = {a["name"]: a for a in result["areas"]} assert by_name["Salon"]["entity_count"] == 4 assert by_name["Salon"]["unavailable_count"] == 0 assert by_name["Hall"]["entity_count"] == 1 assert by_name["Hall"]["unavailable_count"] == 1 assert by_name["Sypialnia"]["unavailable_count"] == 1 assert by_name["Hall"]["aliases"] == ["Wejście"] # switch.tasmota_14 + the three automations have no area assert result["entities_without_area"] == 4 # --- find_entities_by_description --------------------------------------- def test_find_entities_by_description_finds_the_salon_thermometer(fake_backend): result = tools.find_entities_by_description(fake_backend, "czujnik temperatury salon") top = result["matches"][0] assert top["entity_id"] == SALON_TEMP assert top["matched_tokens"] == "3/3" # the reason is human-readable and names the PL->EN bridge it crossed assert "'czujnik'->'sensor'" in top["why"] assert "'temperatury'->'temperature'" in top["why"] def test_find_entities_ranks_full_token_matches_first(fake_backend): result = tools.find_entities_by_description(fake_backend, "czujnik temperatury salon") ids = [m["entity_id"] for m in result["matches"]] assert ids.index(SALON_TEMP) < ids.index("sensor.thsalon_humidity") assert ids.index(SALON_TEMP) < ids.index("light.salon_lampa") def test_find_entities_matches_by_area(fake_backend): result = tools.find_entities_by_description(fake_backend, "sypialnia") assert result["matches"][0]["entity_id"] == "sensor.thsypialnia_temperature" assert result["matches"][0]["unavailable"] is True assert result["unavailable_count"] == 1 def test_find_entities_caps_at_20(fake_backend): states = [ {"entity_id": f"sensor.salon_{i}", "state": "1", "attributes": {"friendly_name": f"Salon {i}"}} for i in range(40) ] backend = type(fake_backend)(states=states) result = tools.find_entities_by_description(backend, "salon") assert result["total_matches"] == 40 assert result["returned"] == 20 def test_find_entities_empty_query_errors(fake_backend): with pytest.raises(BackendError): tools.find_entities_by_description(fake_backend, " ") def test_find_entities_ranks_first_against_the_real_1600_entity_snapshot(real_ken_backend): """Ten hits differently at scale: 672 of ~1650 entities match at least one token.""" result = tools.find_entities_by_description(real_ken_backend, "czujnik temperatury salon") assert result["total_matches"] > 100 assert result["matches"][0]["entity_id"] == SALON_TEMP assert result["matches"][0]["state"] # a real reading, not a placeholder # --- read_automation (repo-backed) -------------------------------------- def test_read_automation_by_alias_returns_the_klima_file(fake_backend): result = tools.read_automation(fake_backend, "Klima salon: wyłącz") assert result["id"] == KLIMA_OFF_ID assert result["path"] == ( f"services/home-assistant/config/ken/automations/{KLIMA_OFF_ID}.yaml" ) assert result["alias"] == "Klima salon: wyłącz chłodzenie i osusz parownik" assert "script.klima_salon_dry_off" in result["yaml"] assert result["automation"]["mode"] == "single" assert result["source"].startswith("repo") def test_read_automation_annotates_with_live_state(fake_backend): result = tools.read_automation(fake_backend, KLIMA_OFF_ID) assert result["live"]["state"] == "on" assert result["live"]["last_triggered"] == "2026-07-27T19:31:00.000000+00:00" assert result["live"]["entity_id"] == "automation.klima_salon_off" def test_read_automation_works_with_the_instance_down(unreachable_backend): result = tools.read_automation(unreachable_backend, KLIMA_OFF_ID) assert "script.klima_salon_dry_off" in result["yaml"] # repo read still succeeds assert "unreachable" in result["live_error"] def test_read_automation_ambiguous_alias_lists_candidates(fake_backend): with pytest.raises(BackendError) as exc: tools.read_automation(fake_backend, "Klima salon") message = str(exc.value) assert "matches 2 automations" in message assert KLIMA_OFF_ID in message def test_read_automation_unknown_errors_cleanly(fake_backend): with pytest.raises(BackendError) as exc: tools.read_automation(fake_backend, "nie ma takiej automatyzacji") assert "no automation matching" in str(exc.value) def test_read_automation_does_not_escape_the_automations_dir(fake_backend): with pytest.raises(BackendError): tools.read_automation(fake_backend, "../../../etc/passwd") # --- list_automations ---------------------------------------------------- def test_list_automations_joins_repo_and_live_state(fake_backend): rows = {r["id"]: r for r in tools.list_automations(fake_backend)["automations"]} klima = rows[KLIMA_OFF_ID] assert klima["alias"] == "Klima salon: wyłącz chłodzenie i osusz parownik" assert klima["state"] == "on" assert klima["last_triggered"] == "2026-07-27T19:31:00.000000+00:00" assert klima["in_repo"] and klima["on_instance"] # in the repo but not loaded on the instance repo_only = [r for r in rows.values() if not r["on_instance"]] assert repo_only and all(r["state"] is None for r in repo_only) # on the instance but never imported into the repo ui_only = rows["9999999999999"] assert ui_only["in_repo"] is False assert ui_only["alias"] == "Created in the UI, never imported" def test_list_automations_filter_is_substring_and_diacritic_insensitive(fake_backend): result = tools.list_automations(fake_backend, filter="klima salon") assert result["count"] == 2 assert all("Klima salon" in r["alias"] for r in result["automations"]) assert tools.list_automations(fake_backend, filter="wylacz chlodzenie")["count"] == 1 def test_list_automations_degrades_when_instance_is_down(unreachable_backend): result = tools.list_automations(unreachable_backend) assert result["count"] > 100 # repo listing still works assert "unreachable" in result["live_error"] assert all(r["state"] is None for r in result["automations"]) # --- instance_status ----------------------------------------------------- def test_instance_status_reports_version_and_dead_entities(fake_backend): result = tools.instance_status(fake_backend) assert result["reachable"] is True assert result["version"] == "2026.7.2" assert result["entity_count"] == 10 assert result["unavailable_count"] == 2 assert result["automation_count"] == 3 assert result["area_count"] == 3 assert "audyt" in result["unavailable_note"] def test_instance_status_unreachable_is_reported_not_raised(unreachable_backend): result = tools.instance_status(unreachable_backend) assert result["reachable"] is False assert "unreachable" in result["error"] def test_instance_status_offline_instance_short_circuits(repo_root): """chelsty-ha: answer from instances.yaml, never open a socket.""" from ha_mcp.config import get_instance name, cfg = get_instance("chelsty-ha", repo_root) backend = backend_mod.LiveBackend(name, cfg, repo_root) def explode(*_args, **_kwargs): # pragma: no cover - must never run raise AssertionError("network call attempted against an offline instance") backend._get = explode result = tools.instance_status(backend) assert result["reachable"] is False assert result["status"] == "offline" assert "instances.yaml" in result["reason"] assert "error" not in result