"""Backend/config tests, plus the structural read-only guarantees. No network here either: the token/unreachable paths are exercised with a bogus token_path and a base_url nothing listens on, and the WebSocket allowlist is asserted on the constant, not by dialing an instance. """ import sys from pathlib import Path import pytest from ha_mcp import backend as backend_mod from ha_mcp import config as cfg from ha_mcp.backend import BackendError, LiveBackend from ha_mcp.config import ConfigError, get_instance SERVICE_DIR = Path(__file__).resolve().parents[1] # --- read-only by construction ------------------------------------------ def test_rest_client_has_no_mutating_method(repo_root): """scripts/ha/lib/ha_api.Client is GET-only — nothing to call by accident.""" sys.path.insert(0, str(cfg.ha_lib_dir(repo_root))) import ha_api public = {n for n in dir(ha_api.Client) if not n.startswith("_")} assert public == {"get", "get_raw_text"} for forbidden in ("post", "put", "patch", "delete", "post_config", "call_service"): assert not hasattr(ha_api.Client, forbidden) def test_ws_allowlist_contains_only_list_reads(): for command in backend_mod.READ_ONLY_WS_COMMANDS: assert command.endswith("/list"), command for mutating in ( "config/area_registry/create", "config/entity_registry/update", "config/device_registry/update", "call_service", ): assert mutating not in backend_mod.READ_ONLY_WS_COMMANDS def test_no_write_verbs_anywhere_in_the_package(): """A grep-level guard: this package must never learn to POST.""" forbidden = ("requests.post", "requests.put", "requests.delete", "call_service", "post_config") for path in (SERVICE_DIR / "src" / "ha_mcp").glob("*.py"): text = path.read_text(encoding="utf-8") for token in forbidden: # backend.py names the mutating commands only inside comments code = "\n".join( line for line in text.splitlines() if not line.strip().startswith("#") ) assert token not in code, f"{path.name} references {token}" # --- instances.yaml ------------------------------------------------------ def test_default_instance_is_ken(repo_root): name, instance = get_instance(None, repo_root) assert name == "ken" assert instance["adapter"] == "api" assert instance["status"] == "active" def test_unknown_instance_lists_the_known_ones(repo_root): with pytest.raises(ConfigError) as exc: get_instance("nope", repo_root) assert "ken" in str(exc.value) and "chelsty-ha" in str(exc.value) # --- failure paths: reportable, never a crash --------------------------- def test_missing_token_is_a_readable_error(repo_root, tmp_path): backend = LiveBackend( "ken", {"base_url": "http://127.0.0.1:1", "status": "active", "token_path": str(tmp_path / "absent")}, repo_root, ) with pytest.raises(BackendError) as exc: backend.states() message = str(exc.value) assert "no HA token" in message assert "absent" in message # says which path it looked at def test_offline_instance_never_opens_a_socket(repo_root): name, instance = get_instance("chelsty-ha", repo_root) backend = LiveBackend(name, instance, repo_root) with pytest.raises(BackendError) as exc: backend.states() assert "status: offline" in str(exc.value) def test_timeout_default_is_short(): assert backend_mod.DEFAULT_TIMEOUT == 5 # --- area index ---------------------------------------------------------- def test_area_index_prefers_entity_area_over_device_area(): index = backend_mod._index_from_registries( [{"area_id": "salon", "name": "Salon"}, {"area_id": "hall", "name": "Hall"}], [{"entity_id": "light.x", "area_id": "hall", "device_id": "d1"}], [{"id": "d1", "area_id": "salon"}], "websocket", ) assert index.area_of("light.x") == "Hall" def test_offline_area_index_falls_back_to_storage_export(repo_root): index = backend_mod.load_offline_area_index("ken", repo_root) assert index.source == "storage-export" assert {a["name"] for a in index.areas} >= {"Salon", "Kuchnia", "Sypialnia"} # and it says out loud what the offline export cannot resolve assert "device" in index.note def test_offline_area_index_for_an_instance_without_an_export(repo_root): index = backend_mod.load_offline_area_index("chelsty-ha", repo_root) assert index.source == "none" assert index.entity_area == {} def test_automation_files_are_repo_relative(repo_root): files = backend_mod.automation_files("ken", repo_root) assert len(files) > 100 rel, absolute = files[0] assert rel.startswith("services/home-assistant/config/ken/automations/") assert absolute.is_file()