"""Test config entries API.""" from collections import OrderedDict from collections.abc import Generator from http import HTTPStatus from typing import Any from unittest.mock import ANY, AsyncMock, patch from aiohttp.test_utils import TestClient from freezegun.api import FrozenDateTimeFactory import pytest import voluptuous as vol from homeassistant import config_entries as core_ce, data_entry_flow, loader from homeassistant.components.config import config_entries from homeassistant.config_entries import HANDLERS, ConfigFlow, ConfigFlowResult from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_RADIUS from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResultType from homeassistant.helpers import config_entry_flow, config_validation as cv from homeassistant.helpers.discovery_flow import DiscoveryKey from homeassistant.helpers.service_info.hassio import HassioServiceInfo from homeassistant.loader import IntegrationNotFound from homeassistant.setup import async_setup_component from homeassistant.util.dt import utcnow from tests.common import ( MockConfigEntry, MockModule, MockUser, mock_config_flow, mock_integration, mock_platform, ) from tests.typing import ClientSessionGenerator, WebSocketGenerator @pytest.fixture def clear_handlers() -> Generator[None]: """Clear config entry handlers.""" with patch.dict(HANDLERS, clear=True): yield @pytest.fixture(autouse=True) def mock_test_component(hass: HomeAssistant) -> None: """Ensure a component called 'test' exists.""" mock_integration(hass, MockModule("test")) @pytest.fixture async def client( hass: HomeAssistant, hass_client: ClientSessionGenerator ) -> TestClient: """Fixture that can interact with the config manager API.""" await async_setup_component(hass, "http", {}) config_entries.async_setup(hass) return await hass_client() @pytest.fixture def mock_flow() -> Generator[None]: """Mock a config flow.""" class Comp1ConfigFlow(ConfigFlow): """Config flow with options flow.""" @staticmethod @callback def async_get_options_flow(config_entry): """Get options flow.""" with mock_config_flow("comp1", Comp1ConfigFlow): yield @pytest.mark.usefixtures("freezer") @pytest.mark.usefixtures("clear_handlers", "mock_flow") async def test_get_entries(hass: HomeAssistant, client: TestClient) -> None: """Test get entries.""" mock_integration(hass, MockModule("comp1")) mock_integration( hass, MockModule("comp2", partial_manifest={"integration_type": "helper"}) ) mock_integration( hass, MockModule("comp3", partial_manifest={"integration_type": "hub"}) ) mock_integration( hass, MockModule("comp4", partial_manifest={"integration_type": "device"}) ) mock_integration( hass, MockModule("comp5", partial_manifest={"integration_type": "service"}) ) config_entry_flow.register_discovery_flow("comp2", "Comp 2", lambda: None) entry = MockConfigEntry( domain="comp1", title="Test 1", source="bla", ) entry.supports_unload = True entry.add_to_hass(hass) MockConfigEntry( domain="comp2", title="Test 2", source="bla2", state=core_ce.ConfigEntryState.SETUP_ERROR, reason="Unsupported API", ).add_to_hass(hass) MockConfigEntry( domain="comp3", title="Test 3", source="bla3", disabled_by=core_ce.ConfigEntryDisabler.USER, ).add_to_hass(hass) MockConfigEntry( domain="comp4", title="Test 4", source="bla4", ).add_to_hass(hass) MockConfigEntry( domain="comp5", title="Test 5", source="bla5", ).add_to_hass(hass) resp = await client.get("/api/config/config_entries/entry") assert resp.status == HTTPStatus.OK data = await resp.json() for entry in data: entry.pop("entry_id") timestamp = utcnow().timestamp() assert data == [ { "created_at": timestamp, "disabled_by": None, "domain": "comp1", "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": core_ce.ConfigEntryState.NOT_LOADED.value, "supports_options": True, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": True, "title": "Test 1", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp2", "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": "Unsupported API", "source": "bla2", "state": core_ce.ConfigEntryState.SETUP_ERROR.value, "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 2", }, { "created_at": timestamp, "disabled_by": core_ce.ConfigEntryDisabler.USER, "domain": "comp3", "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla3", "state": core_ce.ConfigEntryState.NOT_LOADED.value, "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 3", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp4", "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla4", "state": core_ce.ConfigEntryState.NOT_LOADED.value, "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 4", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp5", "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla5", "state": core_ce.ConfigEntryState.NOT_LOADED.value, "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 5", }, ] resp = await client.get("/api/config/config_entries/entry?domain=comp3") assert resp.status == HTTPStatus.OK data = await resp.json() assert len(data) == 1 assert data[0]["domain"] == "comp3" resp = await client.get("/api/config/config_entries/entry?domain=comp3&type=helper") assert resp.status == HTTPStatus.OK data = await resp.json() assert len(data) == 0 resp = await client.get("/api/config/config_entries/entry?type=hub") assert resp.status == HTTPStatus.OK data = await resp.json() assert len(data) == 2 assert data[0]["domain"] == "comp1" assert data[1]["domain"] == "comp3" resp = await client.get("/api/config/config_entries/entry?type=device") assert resp.status == HTTPStatus.OK data = await resp.json() assert len(data) == 1 assert data[0]["domain"] == "comp4" resp = await client.get("/api/config/config_entries/entry?type=service") assert resp.status == HTTPStatus.OK data = await resp.json() assert len(data) == 1 assert data[0]["domain"] == "comp5" async def test_remove_entry(hass: HomeAssistant, client: TestClient) -> None: """Test removing an entry via the API.""" entry = MockConfigEntry(domain="test", state=core_ce.ConfigEntryState.LOADED) entry.add_to_hass(hass) resp = await client.delete(f"/api/config/config_entries/entry/{entry.entry_id}") assert resp.status == HTTPStatus.OK data = await resp.json() assert data == {"require_restart": True} assert len(hass.config_entries.async_entries()) == 0 async def test_reload_entry(hass: HomeAssistant, client: TestClient) -> None: """Test reloading an entry via the API.""" entry = MockConfigEntry(domain="test", state=core_ce.ConfigEntryState.LOADED) entry.add_to_hass(hass) hass.config.components.add("test") resp = await client.post( f"/api/config/config_entries/entry/{entry.entry_id}/reload" ) assert resp.status == HTTPStatus.OK data = await resp.json() assert data == {"require_restart": True} assert len(hass.config_entries.async_entries()) == 1 async def test_reload_invalid_entry(hass: HomeAssistant, client: TestClient) -> None: """Test reloading an invalid entry via the API.""" resp = await client.post("/api/config/config_entries/entry/invalid/reload") assert resp.status == HTTPStatus.NOT_FOUND async def test_remove_entry_unauth( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser ) -> None: """Test removing an entry via the API.""" hass_admin_user.groups = [] entry = MockConfigEntry(domain="demo", state=core_ce.ConfigEntryState.LOADED) entry.add_to_hass(hass) resp = await client.delete(f"/api/config/config_entries/entry/{entry.entry_id}") assert resp.status == HTTPStatus.UNAUTHORIZED assert len(hass.config_entries.async_entries()) == 1 async def test_reload_entry_unauth( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser ) -> None: """Test reloading an entry via the API.""" hass_admin_user.groups = [] entry = MockConfigEntry(domain="demo", state=core_ce.ConfigEntryState.LOADED) entry.add_to_hass(hass) resp = await client.post( f"/api/config/config_entries/entry/{entry.entry_id}/reload" ) assert resp.status == HTTPStatus.UNAUTHORIZED assert len(hass.config_entries.async_entries()) == 1 async def test_reload_entry_in_failed_state( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser ) -> None: """Test reloading an entry via the API that has already failed to unload.""" entry = MockConfigEntry(domain="demo", state=core_ce.ConfigEntryState.FAILED_UNLOAD) entry.add_to_hass(hass) hass.config.components.add("demo") resp = await client.post( f"/api/config/config_entries/entry/{entry.entry_id}/reload" ) assert resp.status == HTTPStatus.FORBIDDEN assert len(hass.config_entries.async_entries()) == 1 async def test_reload_entry_in_setup_retry( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser ) -> None: """Test reloading an entry via the API that is in setup retry.""" mock_setup_entry = AsyncMock(return_value=True) mock_unload_entry = AsyncMock(return_value=True) mock_migrate_entry = AsyncMock(return_value=True) mock_integration( hass, MockModule( "comp", async_setup_entry=mock_setup_entry, async_unload_entry=mock_unload_entry, async_migrate_entry=mock_migrate_entry, ), ) mock_platform(hass, "comp.config_flow", None) entry = MockConfigEntry(domain="comp", state=core_ce.ConfigEntryState.SETUP_RETRY) entry.supports_unload = True entry.add_to_hass(hass) hass.config.components.add("comp") with patch.dict(HANDLERS, {"comp": ConfigFlow, "test": ConfigFlow}): resp = await client.post( f"/api/config/config_entries/entry/{entry.entry_id}/reload" ) await hass.async_block_till_done() assert resp.status == HTTPStatus.OK data = await resp.json() assert data == {"require_restart": False} assert len(hass.config_entries.async_entries()) == 1 @pytest.mark.parametrize( ("type_filter", "result"), [ (None, {"hello", "another", "world"}), ("integration", {"hello", "another"}), ("helper", {"world"}), ], ) async def test_available_flows( hass: HomeAssistant, client: TestClient, type_filter: str | None, result: set[str] ) -> None: """Test querying the available flows.""" with patch.object( loader, "FLOWS", {"integration": ["hello", "another"], "helper": ["world"]}, ): resp = await client.get( "/api/config/config_entries/flow_handlers", params={"type": type_filter} if type_filter else {}, ) assert resp.status == HTTPStatus.OK data = await resp.json() assert set(data) == result ############################ # FLOW MANAGER API TESTS # ############################ @pytest.mark.parametrize( "ignore_translations", ["component.test.config.error.Should be unique."], ) async def test_initialize_flow(hass: HomeAssistant, client: TestClient) -> None: """Test we can initialize a flow.""" mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): schema = OrderedDict() schema[vol.Required("username")] = str schema[vol.Required("password")] = str return self.async_show_form( step_id="user", data_schema=vol.Schema(schema), description_placeholders={ "url": "https://example.com", "show_advanced_options": self.show_advanced_options, }, errors={"username": "Should be unique."}, ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test", "show_advanced_options": True}, ) assert resp.status == HTTPStatus.OK data = await resp.json() data.pop("flow_id") assert data == { "type": "form", "handler": "test", "step_id": "user", "data_schema": [ {"name": "username", "required": True, "type": "string"}, {"name": "password", "required": True, "type": "string"}, ], "description_placeholders": { "url": "https://example.com", "show_advanced_options": True, }, "errors": {"username": "Should be unique."}, "last_step": None, "preview": None, } async def test_initialize_flow_unmet_dependency( hass: HomeAssistant, client: TestClient ) -> None: """Test unmet dependencies are listed.""" mock_platform(hass, "test.config_flow", None) config_schema = vol.Schema({"comp_conf": {"hello": str}}, required=True) mock_integration( hass, MockModule(domain="dependency_1", config_schema=config_schema) ) # The test2 config flow should fail because dependency_1 can't be automatically setup mock_integration( hass, MockModule(domain="test2", partial_manifest={"dependencies": ["dependency_1"]}), ) class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): pass with patch.dict(HANDLERS, {"test2": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test2", "show_advanced_options": True}, ) assert resp.status == HTTPStatus.BAD_REQUEST data = await resp.text() assert data == "Failed dependencies dependency_1" async def test_initialize_flow_unauth( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser ) -> None: """Test we can initialize a flow.""" hass_admin_user.groups = [] class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): schema = OrderedDict() schema[vol.Required("username")] = str schema[vol.Required("password")] = str return self.async_show_form( step_id="user", data_schema=schema, description_placeholders={"url": "https://example.com"}, errors={"username": "Should be unique."}, ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.UNAUTHORIZED @pytest.mark.parametrize( "ignore_translations", ["component.test.config.abort.bla"], ) async def test_abort(hass: HomeAssistant, client: TestClient) -> None: """Test a flow that aborts.""" mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): return self.async_abort(reason="bla") with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK data = await resp.json() data.pop("flow_id") assert data == { "description_placeholders": None, "handler": "test", "reason": "bla", "type": "abort", } @pytest.mark.usefixtures("enable_custom_integrations", "freezer") async def test_create_account(hass: HomeAssistant, client: TestClient) -> None: """Test a flow that creates an account.""" mock_platform(hass, "test.config_flow", None) mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) class TestFlow(core_ce.ConfigFlow): VERSION = 1 async def async_step_user(self, user_input=None): return self.async_create_entry( title="Test Entry", data={"secret": "account_token"} ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK entries = hass.config_entries.async_entries("test") assert len(entries) == 1 timestamp = utcnow().timestamp() data = await resp.json() data.pop("flow_id") assert data == { "handler": "test", "title": "Test Entry", "type": "create_entry", "version": 1, "result": { "created_at": timestamp, "disabled_by": None, "domain": "test", "entry_id": entries[0].entry_id, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": core_ce.SOURCE_USER, "state": core_ce.ConfigEntryState.LOADED.value, "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test Entry", }, "description": None, "description_placeholders": None, "options": {}, "minor_version": 1, } @pytest.mark.usefixtures("enable_custom_integrations", "freezer") async def test_two_step_flow(hass: HomeAssistant, client: TestClient) -> None: """Test we can finish a two step flow.""" mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): VERSION = 1 async def async_step_user(self, user_input=None): return self.async_show_form( step_id="account", data_schema=vol.Schema({"user_title": str}) ) async def async_step_account(self, user_input=None): return self.async_create_entry( title=user_input["user_title"], data={"secret": "account_token"} ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data.pop("flow_id") assert data == { "type": "form", "handler": "test", "step_id": "account", "data_schema": [{"name": "user_title", "type": "string"}], "description_placeholders": None, "errors": None, "last_step": None, "preview": None, } with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( f"/api/config/config_entries/flow/{flow_id}", json={"user_title": "user-title"}, ) assert resp.status == HTTPStatus.OK entries = hass.config_entries.async_entries("test") assert len(entries) == 1 timestamp = utcnow().timestamp() data = await resp.json() data.pop("flow_id") assert data == { "handler": "test", "type": "create_entry", "title": "user-title", "version": 1, "result": { "created_at": timestamp, "disabled_by": None, "domain": "test", "entry_id": entries[0].entry_id, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": core_ce.SOURCE_USER, "state": core_ce.ConfigEntryState.LOADED.value, "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "user-title", }, "description": None, "description_placeholders": None, "options": {}, "minor_version": 1, } async def test_continue_flow_unauth( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser ) -> None: """Test we can't finish a two step flow.""" mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): VERSION = 1 async def async_step_user(self, user_input=None): return self.async_show_form( step_id="account", data_schema=vol.Schema({"user_title": str}) ) async def async_step_account(self, user_input=None): return self.async_create_entry( title=user_input["user_title"], data={"secret": "account_token"} ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data.pop("flow_id") assert data == { "type": "form", "handler": "test", "step_id": "account", "data_schema": [{"name": "user_title", "type": "string"}], "description_placeholders": None, "errors": None, "last_step": None, "preview": None, } hass_admin_user.groups = [] resp = await client.post( f"/api/config/config_entries/flow/{flow_id}", json={"user_title": "user-title"}, ) assert resp.status == HTTPStatus.UNAUTHORIZED async def test_get_progress_index( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test querying for the flows that are in progress.""" assert await async_setup_component(hass, "config", {}) mock_platform(hass, "test.config_flow", None) ws_client = await hass_ws_client(hass) mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) entry = MockConfigEntry(domain="test", title="Test", entry_id="1234") entry.add_to_hass(hass) class TestFlow(core_ce.ConfigFlow): VERSION = 5 async def async_step_hassio( self, discovery_info: HassioServiceInfo ) -> ConfigFlowResult: """Handle a Hass.io discovery.""" return await self.async_step_account() async def async_step_account(self, user_input: dict[str, Any] | None = None): """Show a form to the user.""" return self.async_show_form(step_id="account") async def async_step_user(self, user_input: dict[str, Any] | None = None): """Handle a config flow initialized by the user.""" return await self.async_step_account() async def async_step_reconfigure( self, user_input: dict[str, Any] | None = None ): """Handle a reconfiguration flow initialized by the user.""" nonlocal entry assert self._get_reconfigure_entry() is entry return await self.async_step_account() with patch.dict(HANDLERS, {"test": TestFlow}): form_hassio = await hass.config_entries.flow.async_init( "test", context={"source": core_ce.SOURCE_HASSIO} ) form_user = await hass.config_entries.flow.async_init( "test", context={"source": core_ce.SOURCE_USER} ) form_reconfigure = await hass.config_entries.flow.async_init( "test", context={"source": core_ce.SOURCE_RECONFIGURE, "entry_id": "1234"} ) for form in (form_hassio, form_user, form_reconfigure): assert form["type"] == data_entry_flow.FlowResultType.FORM assert form["step_id"] == "account" await ws_client.send_json({"id": 5, "type": "config_entries/flow/progress"}) response = await ws_client.receive_json() assert response["success"] # Active flows with SOURCE_USER and SOURCE_RECONFIGURE should be filtered out assert response["result"] == [ { "flow_id": form_hassio["flow_id"], "handler": "test", "step_id": "account", "context": {"source": core_ce.SOURCE_HASSIO}, } ] async def test_get_progress_index_unauth( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, hass_admin_user: MockUser ) -> None: """Test we can't get flows that are in progress.""" assert await async_setup_component(hass, "config", {}) hass_admin_user.groups = [] ws_client = await hass_ws_client(hass) await ws_client.send_json({"id": 5, "type": "config_entries/flow/progress"}) response = await ws_client.receive_json() assert not response["success"] assert response["error"]["code"] == "unauthorized" @pytest.mark.parametrize( "ignore_translations", ["component.test.config.error.Should be unique."], ) async def test_get_progress_flow(hass: HomeAssistant, client: TestClient) -> None: """Test we can query the API for same result as we get from init a flow.""" mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): schema = OrderedDict() schema[vol.Required("username")] = str schema[vol.Required("password")] = str return self.async_show_form( step_id="user", data_schema=vol.Schema(schema), errors={"username": "Should be unique."}, ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK data = await resp.json() resp2 = await client.get(f"/api/config/config_entries/flow/{data['flow_id']}") assert resp2.status == HTTPStatus.OK data2 = await resp2.json() assert data == data2 @pytest.mark.parametrize( "ignore_translations", ["component.test.config.error.Should be unique."], ) async def test_get_progress_flow_unauth( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser ) -> None: """Test we can can't query the API for result of flow.""" mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): schema = OrderedDict() schema[vol.Required("username")] = str schema[vol.Required("password")] = str return self.async_show_form( step_id="user", data_schema=vol.Schema(schema), errors={"username": "Should be unique."}, ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK data = await resp.json() hass_admin_user.groups = [] resp2 = await client.get(f"/api/config/config_entries/flow/{data['flow_id']}") assert resp2.status == HTTPStatus.UNAUTHORIZED async def test_options_flow(hass: HomeAssistant, client: TestClient) -> None: """Test we can change options.""" class TestFlow(core_ce.ConfigFlow): @staticmethod @callback def async_get_options_flow(config_entry): class OptionsFlowHandler(data_entry_flow.FlowHandler): async def async_step_init(self, user_input=None): schema = OrderedDict() schema[vol.Required("enabled")] = bool return self.async_show_form( step_id="user", data_schema=vol.Schema(schema), description_placeholders={"enabled": "Set to true to be true"}, ) async def async_step_user(self, user_input=None): raise NotImplementedError return OptionsFlowHandler() mock_integration(hass, MockModule("test")) mock_platform(hass, "test.config_flow", None) MockConfigEntry( domain="test", entry_id="test1", source="bla", ).add_to_hass(hass) entry = hass.config_entries.async_entries()[0] with patch.dict(HANDLERS, {"test": TestFlow}): url = "/api/config/config_entries/options/flow" resp = await client.post(url, json={"handler": entry.entry_id}) assert resp.status == HTTPStatus.OK data = await resp.json() data.pop("flow_id") assert data == { "type": "form", "handler": "test1", "step_id": "user", "data_schema": [{"name": "enabled", "required": True, "type": "boolean"}], "description_placeholders": {"enabled": "Set to true to be true"}, "errors": None, "last_step": None, "preview": None, } @pytest.mark.parametrize( ("endpoint", "method"), [ ("/api/config/config_entries/options/flow", "post"), ("/api/config/config_entries/options/flow/1", "get"), ("/api/config/config_entries/options/flow/1", "post"), ], ) async def test_options_flow_unauth( hass: HomeAssistant, client: TestClient, hass_admin_user: MockUser, endpoint: str, method: str, ) -> None: """Test unauthorized on options flow.""" class TestFlow(core_ce.ConfigFlow): @staticmethod @callback def async_get_options_flow(config_entry): class OptionsFlowHandler(data_entry_flow.FlowHandler): async def async_step_init(self, user_input=None): schema = OrderedDict() schema[vol.Required("enabled")] = bool return self.async_show_form( step_id="user", data_schema=schema, description_placeholders={"enabled": "Set to true to be true"}, ) return OptionsFlowHandler() mock_integration(hass, MockModule("test")) mock_platform(hass, "test.config_flow", None) MockConfigEntry( domain="test", entry_id="test1", source="bla", ).add_to_hass(hass) entry = hass.config_entries.async_entries()[0] hass_admin_user.groups = [] with patch.dict(HANDLERS, {"test": TestFlow}): resp = await getattr(client, method)(endpoint, json={"handler": entry.entry_id}) assert resp.status == HTTPStatus.UNAUTHORIZED async def test_two_step_options_flow(hass: HomeAssistant, client: TestClient) -> None: """Test we can finish a two step options flow.""" mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): @staticmethod @callback def async_get_options_flow(config_entry): class OptionsFlowHandler(data_entry_flow.FlowHandler): async def async_step_init(self, user_input=None): return self.async_show_form( step_id="finish", data_schema=vol.Schema({"enabled": bool}) ) async def async_step_finish(self, user_input=None): return self.async_create_entry( title="Enable disable", data=user_input ) return OptionsFlowHandler() MockConfigEntry( domain="test", entry_id="test1", source="bla", ).add_to_hass(hass) entry = hass.config_entries.async_entries()[0] with patch.dict(HANDLERS, {"test": TestFlow}): url = "/api/config/config_entries/options/flow" resp = await client.post(url, json={"handler": entry.entry_id}) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data.pop("flow_id") assert data == { "type": "form", "handler": "test1", "step_id": "finish", "data_schema": [{"name": "enabled", "type": "boolean"}], "description_placeholders": None, "errors": None, "last_step": None, "preview": None, } with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( f"/api/config/config_entries/options/flow/{flow_id}", json={"enabled": True}, ) assert resp.status == HTTPStatus.OK data = await resp.json() data.pop("flow_id") assert data == { "handler": "test1", "type": "create_entry", "title": "Enable disable", "description": None, "description_placeholders": None, } async def test_options_flow_with_invalid_data( hass: HomeAssistant, client: TestClient ) -> None: """Test an options flow with invalid_data.""" mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): @staticmethod @callback def async_get_options_flow(config_entry): class OptionsFlowHandler(data_entry_flow.FlowHandler): async def async_step_init(self, user_input=None): return self.async_show_form( step_id="finish", data_schema=vol.Schema( { vol.Required( "choices", default=["invalid", "valid"] ): cv.multi_select({"valid": "Valid"}) } ), ) async def async_step_finish(self, user_input=None): return self.async_create_entry( title="Enable disable", data=user_input ) return OptionsFlowHandler() MockConfigEntry( domain="test", entry_id="test1", source="bla", ).add_to_hass(hass) entry = hass.config_entries.async_entries()[0] with patch.dict(HANDLERS, {"test": TestFlow}): url = "/api/config/config_entries/options/flow" resp = await client.post(url, json={"handler": entry.entry_id}) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data.pop("flow_id") assert data == { "type": "form", "handler": "test1", "step_id": "finish", "data_schema": [ { "default": ["invalid", "valid"], "name": "choices", "options": {"valid": "Valid"}, "required": True, "type": "multi_select", } ], "description_placeholders": None, "errors": None, "last_step": None, "preview": None, } with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( f"/api/config/config_entries/options/flow/{flow_id}", json={"choices": ["valid", "invalid"]}, ) assert resp.status == HTTPStatus.BAD_REQUEST data = await resp.json() assert data == {"errors": {"choices": "invalid is not a valid option"}} @pytest.mark.usefixtures("freezer") async def test_get_single( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test that we can get a config entry.""" assert await async_setup_component(hass, "config", {}) ws_client = await hass_ws_client(hass) entry = MockConfigEntry(domain="test", state=core_ce.ConfigEntryState.LOADED) entry.add_to_hass(hass) assert entry.pref_disable_new_entities is False assert entry.pref_disable_polling is False await ws_client.send_json_auto_id( { "type": "config_entries/get_single", "entry_id": entry.entry_id, } ) response = await ws_client.receive_json() timestamp = utcnow().timestamp() assert response["success"] assert response["result"]["config_entry"] == { "created_at": timestamp, "disabled_by": None, "domain": "test", "entry_id": entry.entry_id, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "user", "state": "loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Mock Title", } await ws_client.send_json_auto_id( { "type": "config_entries/get_single", "entry_id": "blah", } ) response = await ws_client.receive_json() assert not response["success"] assert response["error"] == { "code": "not_found", "message": "Config entry not found", } async def test_update_prefrences( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test that we can update system options.""" assert await async_setup_component(hass, "config", {}) ws_client = await hass_ws_client(hass) entry = MockConfigEntry(domain="test", state=core_ce.ConfigEntryState.LOADED) entry.add_to_hass(hass) hass.config.components.add("test") assert entry.pref_disable_new_entities is False assert entry.pref_disable_polling is False await ws_client.send_json( { "id": 6, "type": "config_entries/update", "entry_id": entry.entry_id, "pref_disable_new_entities": True, } ) response = await ws_client.receive_json() assert response["success"] assert response["result"]["require_restart"] is False assert response["result"]["config_entry"]["pref_disable_new_entities"] is True assert response["result"]["config_entry"]["pref_disable_polling"] is False assert entry.pref_disable_new_entities is True assert entry.pref_disable_polling is False await ws_client.send_json( { "id": 7, "type": "config_entries/update", "entry_id": entry.entry_id, "pref_disable_new_entities": False, "pref_disable_polling": True, } ) response = await ws_client.receive_json() assert response["success"] assert response["result"]["require_restart"] is True assert response["result"]["config_entry"]["pref_disable_new_entities"] is False assert response["result"]["config_entry"]["pref_disable_polling"] is True assert entry.pref_disable_new_entities is False assert entry.pref_disable_polling is True async def test_update_entry( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test that we can update entry.""" assert await async_setup_component(hass, "config", {}) ws_client = await hass_ws_client(hass) entry = MockConfigEntry(domain="demo", title="Initial Title") entry.add_to_hass(hass) await ws_client.send_json( { "id": 5, "type": "config_entries/update", "entry_id": entry.entry_id, "title": "Updated Title", } ) response = await ws_client.receive_json() assert response["success"] assert response["result"]["config_entry"]["title"] == "Updated Title" assert entry.title == "Updated Title" async def test_update_entry_nonexisting( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test that we can update entry.""" assert await async_setup_component(hass, "config", {}) ws_client = await hass_ws_client(hass) await ws_client.send_json( { "id": 5, "type": "config_entries/update", "entry_id": "non_existing", "title": "Updated Title", } ) response = await ws_client.receive_json() assert not response["success"] assert response["error"]["code"] == "not_found" async def test_disable_entry( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test that we can disable entry.""" assert await async_setup_component(hass, "config", {}) ws_client = await hass_ws_client(hass) entry = MockConfigEntry(domain="test", state=core_ce.ConfigEntryState.LOADED) entry.add_to_hass(hass) assert entry.disabled_by is None hass.config.components.add("test") # Disable await ws_client.send_json( { "id": 5, "type": "config_entries/disable", "entry_id": entry.entry_id, "disabled_by": core_ce.ConfigEntryDisabler.USER, } ) response = await ws_client.receive_json() assert response["success"] assert response["result"] == {"require_restart": True} assert entry.disabled_by is core_ce.ConfigEntryDisabler.USER assert entry.state is core_ce.ConfigEntryState.FAILED_UNLOAD # Enable await ws_client.send_json( { "id": 6, "type": "config_entries/disable", "entry_id": entry.entry_id, "disabled_by": None, } ) response = await ws_client.receive_json() assert response["success"] assert response["result"] == {"require_restart": True} assert entry.disabled_by is None assert entry.state == core_ce.ConfigEntryState.FAILED_UNLOAD # Enable again -> no op await ws_client.send_json( { "id": 7, "type": "config_entries/disable", "entry_id": entry.entry_id, "disabled_by": None, } ) response = await ws_client.receive_json() assert response["success"] assert response["result"] == {"require_restart": False} assert entry.disabled_by is None assert entry.state == core_ce.ConfigEntryState.FAILED_UNLOAD async def test_disable_entry_nonexisting( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test that we can disable entry.""" assert await async_setup_component(hass, "config", {}) ws_client = await hass_ws_client(hass) await ws_client.send_json( { "id": 5, "type": "config_entries/disable", "entry_id": "non_existing", "disabled_by": core_ce.ConfigEntryDisabler.USER, } ) response = await ws_client.receive_json() assert not response["success"] assert response["error"]["code"] == "not_found" @pytest.mark.parametrize( ( "flow_context", "entry_discovery_keys", ), [ ( {}, {}, ), ( {"discovery_key": DiscoveryKey(domain="test", key="blah", version=1)}, {"test": (DiscoveryKey(domain="test", key="blah", version=1),)}, ), ], ) async def test_ignore_flow( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, flow_context: dict, entry_discovery_keys: dict[str, tuple[DiscoveryKey, ...]], ) -> None: """Test we can ignore a flow.""" assert await async_setup_component(hass, "config", {}) mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): VERSION = 1 async def async_step_user(self, user_input=None): await self.async_set_unique_id("mock-unique-id") return self.async_show_form(step_id="account") async def async_step_account(self, user_input=None): raise NotImplementedError ws_client = await hass_ws_client(hass) with patch.dict(HANDLERS, {"test": TestFlow}): result = await hass.config_entries.flow.async_init( "test", context={"source": core_ce.SOURCE_USER} | flow_context ) assert result["type"] is FlowResultType.FORM await ws_client.send_json( { "id": 5, "type": "config_entries/ignore_flow", "flow_id": result["flow_id"], "title": "Test Integration", } ) response = await ws_client.receive_json() assert response["success"] assert len(hass.config_entries.flow.async_progress()) == 0 entry = hass.config_entries.async_entries("test")[0] assert entry.source == "ignore" assert entry.unique_id == "mock-unique-id" assert entry.title == "Test Integration" assert entry.data == {} assert entry.discovery_keys == entry_discovery_keys async def test_ignore_flow_nonexisting( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test we can ignore a flow.""" assert await async_setup_component(hass, "config", {}) ws_client = await hass_ws_client(hass) await ws_client.send_json( { "id": 5, "type": "config_entries/ignore_flow", "flow_id": "non_existing", "title": "Test Integration", } ) response = await ws_client.receive_json() assert not response["success"] assert response["error"]["code"] == "not_found" @pytest.mark.usefixtures("clear_handlers", "freezer") async def test_get_matching_entries_ws( hass: HomeAssistant, hass_ws_client: WebSocketGenerator ) -> None: """Test get entries with the websocket api.""" assert await async_setup_component(hass, "config", {}) mock_integration(hass, MockModule("comp1")) mock_integration( hass, MockModule("comp2", partial_manifest={"integration_type": "helper"}) ) mock_integration( hass, MockModule("comp3", partial_manifest={"integration_type": "hub"}) ) mock_integration( hass, MockModule("comp4", partial_manifest={"integration_type": "device"}) ) mock_integration( hass, MockModule("comp5", partial_manifest={"integration_type": "service"}) ) entry = MockConfigEntry( domain="comp1", title="Test 1", source="bla", ) entry.add_to_hass(hass) MockConfigEntry( domain="comp2", title="Test 2", source="bla2", state=core_ce.ConfigEntryState.SETUP_ERROR, reason="Unsupported API", ).add_to_hass(hass) MockConfigEntry( domain="comp3", title="Test 3", source="bla3", disabled_by=core_ce.ConfigEntryDisabler.USER, ).add_to_hass(hass) MockConfigEntry( domain="comp4", title="Test 4", source="bla4", ).add_to_hass(hass) MockConfigEntry( domain="comp5", title="Test 5", source="bla5", ).add_to_hass(hass) ws_client = await hass_ws_client(hass) await ws_client.send_json_auto_id({"type": "config_entries/get"}) response = await ws_client.receive_json() timestamp = utcnow().timestamp() assert response["result"] == [ { "created_at": timestamp, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 1", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp2", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": "Unsupported API", "source": "bla2", "state": "setup_error", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 2", }, { "created_at": timestamp, "disabled_by": "user", "domain": "comp3", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla3", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 3", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp4", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla4", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 4", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp5", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla5", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 5", }, ] await ws_client.send_json_auto_id( { "type": "config_entries/get", "domain": "comp1", "type_filter": "hub", } ) response = await ws_client.receive_json() assert response["result"] == [ { "created_at": timestamp, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 1", } ] await ws_client.send_json_auto_id( { "type": "config_entries/get", "type_filter": ["service", "device"], } ) response = await ws_client.receive_json() assert response["result"] == [ { "created_at": timestamp, "disabled_by": None, "domain": "comp4", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla4", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 4", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp5", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla5", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 5", }, ] await ws_client.send_json_auto_id( { "type": "config_entries/get", "type_filter": "hub", } ) response = await ws_client.receive_json() assert response["result"] == [ { "created_at": timestamp, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 1", }, { "created_at": timestamp, "disabled_by": "user", "domain": "comp3", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla3", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 3", }, ] # Verify we skip broken integrations with patch( "homeassistant.components.config.config_entries.async_get_integrations", return_value={"any": IntegrationNotFound("any")}, ): await ws_client.send_json_auto_id( { "type": "config_entries/get", "type_filter": "hub", } ) response = await ws_client.receive_json() assert response["result"] == [ { "created_at": timestamp, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 1", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp2", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": "Unsupported API", "source": "bla2", "state": "setup_error", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 2", }, { "created_at": timestamp, "disabled_by": "user", "domain": "comp3", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla3", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 3", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp4", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla4", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 4", }, { "created_at": timestamp, "disabled_by": None, "domain": "comp5", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": timestamp, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla5", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 5", }, ] # Verify we don't send config entries when only helpers are requested with patch( "homeassistant.components.config.config_entries.async_get_integrations", return_value={"any": IntegrationNotFound("any")}, ): await ws_client.send_json_auto_id( { "type": "config_entries/get", "type_filter": ["helper"], } ) response = await ws_client.receive_json() assert response["result"] == [] # Verify we raise if something really goes wrong with patch( "homeassistant.components.config.config_entries.async_get_integrations", return_value={"any": Exception()}, ): await ws_client.send_json_auto_id( { "type": "config_entries/get", "type_filter": ["device", "hub", "service"], } ) response = await ws_client.receive_json() assert response["success"] is False @pytest.mark.usefixtures("clear_handlers") async def test_subscribe_entries_ws( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, ) -> None: """Test subscribe entries with the websocket api.""" assert await async_setup_component(hass, "config", {}) mock_integration(hass, MockModule("comp1")) mock_integration( hass, MockModule("comp2", partial_manifest={"integration_type": "helper"}) ) mock_integration( hass, MockModule("comp3", partial_manifest={"integration_type": "device"}) ) entry = MockConfigEntry( domain="comp1", title="Test 1", source="bla", ) entry.add_to_hass(hass) MockConfigEntry( domain="comp2", title="Test 2", source="bla2", state=core_ce.ConfigEntryState.SETUP_ERROR, reason="Unsupported API", ).add_to_hass(hass) MockConfigEntry( domain="comp3", title="Test 3", source="bla3", disabled_by=core_ce.ConfigEntryDisabler.USER, ).add_to_hass(hass) ws_client = await hass_ws_client(hass) await ws_client.send_json( { "id": 5, "type": "config_entries/subscribe", } ) response = await ws_client.receive_json() assert response["id"] == 5 assert response["result"] is None assert response["success"] is True assert response["type"] == "result" response = await ws_client.receive_json() assert response["id"] == 5 created = utcnow().timestamp() assert response["event"] == [ { "type": None, "entry": { "created_at": created, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": created, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 1", }, }, { "type": None, "entry": { "created_at": created, "disabled_by": None, "domain": "comp2", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": created, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": "Unsupported API", "source": "bla2", "state": "setup_error", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 2", }, }, { "type": None, "entry": { "created_at": created, "disabled_by": "user", "domain": "comp3", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": created, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla3", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 3", }, }, ] freezer.tick() modified = utcnow().timestamp() assert hass.config_entries.async_update_entry(entry, title="changed") response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "entry": { "created_at": created, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": modified, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "changed", }, "type": "updated", } ] freezer.tick() modified = utcnow().timestamp() await hass.config_entries.async_remove(entry.entry_id) response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "entry": { "created_at": created, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": modified, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "changed", }, "type": "removed", } ] freezer.tick() await hass.config_entries.async_add(entry) response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "entry": { "created_at": entry.created_at.timestamp(), "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": entry.modified_at.timestamp(), "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "changed", }, "type": "added", } ] @pytest.mark.usefixtures("clear_handlers") async def test_subscribe_entries_ws_filtered( hass: HomeAssistant, hass_ws_client: WebSocketGenerator, freezer: FrozenDateTimeFactory, ) -> None: """Test subscribe entries with the websocket api with a type filter.""" created = utcnow().timestamp() assert await async_setup_component(hass, "config", {}) mock_integration(hass, MockModule("comp1")) mock_integration( hass, MockModule("comp2", partial_manifest={"integration_type": "helper"}) ) mock_integration( hass, MockModule("comp3", partial_manifest={"integration_type": "device"}) ) mock_integration( hass, MockModule("comp4", partial_manifest={"integration_type": "service"}) ) entry = MockConfigEntry( domain="comp1", title="Test 1", source="bla", ) entry.add_to_hass(hass) entry2 = MockConfigEntry( domain="comp2", title="Test 2", source="bla2", state=core_ce.ConfigEntryState.SETUP_ERROR, reason="Unsupported API", ) entry2.add_to_hass(hass) entry3 = MockConfigEntry( domain="comp3", title="Test 3", source="bla3", disabled_by=core_ce.ConfigEntryDisabler.USER, ) entry3.add_to_hass(hass) entry4 = MockConfigEntry( domain="comp4", title="Test 4", source="bla4", ) entry4.add_to_hass(hass) ws_client = await hass_ws_client(hass) await ws_client.send_json( { "id": 5, "type": "config_entries/subscribe", "type_filter": ["hub", "device"], } ) response = await ws_client.receive_json() assert response["id"] == 5 assert response["result"] is None assert response["success"] is True assert response["type"] == "result" response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "type": None, "entry": { "created_at": created, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": created, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 1", }, }, { "type": None, "entry": { "created_at": created, "disabled_by": "user", "domain": "comp3", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": created, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla3", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "Test 3", }, }, ] freezer.tick() modified = utcnow().timestamp() assert hass.config_entries.async_update_entry(entry, title="changed") assert hass.config_entries.async_update_entry(entry3, title="changed too") assert hass.config_entries.async_update_entry(entry4, title="changed but ignored") response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "entry": { "created_at": created, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": modified, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "changed", }, "type": "updated", } ] response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "entry": { "created_at": created, "disabled_by": "user", "domain": "comp3", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": modified, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla3", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "changed too", }, "type": "updated", } ] freezer.tick() modified = utcnow().timestamp() await hass.config_entries.async_remove(entry.entry_id) await hass.config_entries.async_remove(entry2.entry_id) response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "entry": { "created_at": created, "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": modified, "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "changed", }, "type": "removed", } ] freezer.tick() await hass.config_entries.async_add(entry) response = await ws_client.receive_json() assert response["id"] == 5 assert response["event"] == [ { "entry": { "created_at": entry.created_at.timestamp(), "disabled_by": None, "domain": "comp1", "entry_id": ANY, "error_reason_translation_key": None, "error_reason_translation_placeholders": None, "modified_at": entry.modified_at.timestamp(), "pref_disable_new_entities": False, "pref_disable_polling": False, "reason": None, "source": "bla", "state": "not_loaded", "supports_options": False, "supports_reconfigure": False, "supports_remove_device": False, "supports_unload": False, "title": "changed", }, "type": "added", } ] async def test_flow_with_multiple_schema_errors( hass: HomeAssistant, client: TestClient ) -> None: """Test an config flow with multiple schema errors.""" mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): return self.async_show_form( step_id="user", data_schema=vol.Schema( { vol.Required(CONF_LATITUDE): cv.latitude, vol.Required(CONF_LONGITUDE): cv.longitude, vol.Required(CONF_RADIUS): vol.All(int, vol.Range(min=5)), } ), ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK flow_id = (await resp.json())["flow_id"] resp = await client.post( f"/api/config/config_entries/flow/{flow_id}", json={"latitude": 30000, "longitude": 30000, "radius": 1}, ) assert resp.status == HTTPStatus.BAD_REQUEST data = await resp.json() assert data == { "errors": { "latitude": "invalid latitude", "longitude": "invalid longitude", "radius": "value must be at least 5", } } async def test_flow_with_multiple_schema_errors_base( hass: HomeAssistant, client: TestClient ) -> None: """Test an config flow with multiple schema errors where fields are not in the schema.""" mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) mock_platform(hass, "test.config_flow", None) class TestFlow(core_ce.ConfigFlow): async def async_step_user(self, user_input=None): return self.async_show_form( step_id="user", data_schema=vol.Schema( { vol.Required(CONF_LATITUDE): cv.latitude, } ), ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test"} ) assert resp.status == HTTPStatus.OK flow_id = (await resp.json())["flow_id"] resp = await client.post( f"/api/config/config_entries/flow/{flow_id}", json={"invalid": 30000, "invalid_2": 30000}, ) assert resp.status == HTTPStatus.BAD_REQUEST data = await resp.json() assert data == { "errors": { "base": [ "extra keys not allowed @ data['invalid']", "extra keys not allowed @ data['invalid_2']", ], "latitude": "required key not provided", } } @pytest.mark.parametrize( "ignore_translations", ["component.test.config.abort.reconfigure_successful"], ) @pytest.mark.usefixtures("enable_custom_integrations", "freezer") async def test_supports_reconfigure( hass: HomeAssistant, client: TestClient, ) -> None: """Test a flow that support reconfigure step.""" mock_platform(hass, "test.config_flow", None) mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) entry = MockConfigEntry(domain="test", title="Test", entry_id="1") entry.add_to_hass(hass) class TestFlow(core_ce.ConfigFlow): VERSION = 1 async def async_step_user(self, user_input=None): return self.async_create_entry( title="Test Entry", data={"secret": "account_token"} ) async def async_step_reconfigure(self, user_input=None): if user_input is None: return self.async_show_form( step_id="reconfigure", data_schema=vol.Schema({}) ) return self.async_update_reload_and_abort( self._get_reconfigure_entry(), title="Test Entry", data={"secret": "account_token"}, ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test", "entry_id": "1"}, ) assert resp.status == HTTPStatus.OK data = await resp.json() flow_id = data.pop("flow_id") assert data == { "type": "form", "handler": "test", "step_id": "reconfigure", "data_schema": [], "last_step": None, "preview": None, "description_placeholders": None, "errors": None, } with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( f"/api/config/config_entries/flow/{flow_id}", json={}, ) assert resp.status == HTTPStatus.OK entries = hass.config_entries.async_entries("test") assert len(entries) == 1 data = await resp.json() data.pop("flow_id") assert data == { "handler": "test", "reason": "reconfigure_successful", "type": "abort", "description_placeholders": None, } @pytest.mark.usefixtures("enable_custom_integrations") async def test_does_not_support_reconfigure( hass: HomeAssistant, client: TestClient ) -> None: """Test a flow that does not support reconfigure step.""" mock_platform(hass, "test.config_flow", None) mock_integration( hass, MockModule("test", async_setup_entry=AsyncMock(return_value=True)) ) class TestFlow(core_ce.ConfigFlow): VERSION = 1 async def async_step_user(self, user_input=None): return self.async_create_entry( title="Test Entry", data={"secret": "account_token"} ) with patch.dict(HANDLERS, {"test": TestFlow}): resp = await client.post( "/api/config/config_entries/flow", json={"handler": "test", "entry_id": "1"}, ) assert resp.status == HTTPStatus.BAD_REQUEST response = await resp.text() assert ( response == '{"message":"Handler ConfigEntriesFlowManager doesn\'t support step reconfigure"}' )