"""Config flow fglair: меню, ручной ввод, импорт, облако (discover замокан).""" from __future__ import annotations import json from unittest.mock import AsyncMock, patch from helpers import DSN, KEY_ID, LANIP_KEY from homeassistant import config_entries from homeassistant.core import HomeAssistant from homeassistant.setup import async_setup_component from pytest_homeassistant_custom_component.common import MockConfigEntry import pyfglair from custom_components.fglair.const import DOMAIN from custom_components.fglair.trial import TrialResult from pyfglair import Prop, Value, ValueKind from pyfglair.provision import Device MANUAL_INPUT = { "host": "192.168.1.50", "device_port": 80, "dsn": DSN, "lanip_key": LANIP_KEY, "lanip_key_id": KEY_ID, "name": "Bedroom", "model": "AP-WC1E", } PREVIEW_INPUT = { "template": "A", "temp_min": 16.0, "temp_max": 30.0, "temp_step": 0.5, "temp_num": 1, "temp_den": 1, "temp_offset": 0, } TRIAL_OK = TrialResult(ok=True) TRIAL_NETWORK_FAIL = TrialResult(ok=False, reason="unreachable") async def _menu(hass: HomeAssistant): return await hass.config_entries.flow.async_init( DOMAIN, context={"source": config_entries.SOURCE_USER} ) async def _select(hass: HomeAssistant, flow_id: str, option: str): return await hass.config_entries.flow.async_configure( flow_id, {"next_step_id": option} ) async def _configure(hass: HomeAssistant, flow_id: str, user_input: dict): return await hass.config_entries.flow.async_configure(flow_id, user_input) def _suggested(schema, field: str): for key in schema.schema: if getattr(key, "schema", None) == field: description = key.description if isinstance(description, dict): return description.get("suggested_value") return None async def test_user_menu(hass: HomeAssistant): result = await _menu(hass) assert result["type"] == "menu" assert set(result["menu_options"]) == {"cloud", "manual", "import_json"} async def test_manual_success(hass: HomeAssistant): with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ) as trial: result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") assert result["type"] == "form" result = await _configure(hass, result["flow_id"], MANUAL_INPUT) assert result["type"] == "form" assert result["step_id"] == "preview" result = await _configure(hass, result["flow_id"], PREVIEW_INPUT) assert result["type"] == "create_entry" assert result["title"] == "Bedroom" assert result["result"].unique_id == DSN data = result["data"] assert data["lanip_key_id"] == KEY_ID assert data["template"] == "A" assert data["temp_step"] == 5 assert data["overrides"]["adjust_temperature"] == { "num": 1, "den": 1, "offset": 0, "min": 160, "max": 300, } trial.assert_called_once() assert trial.call_args.args[0]["host"] == "192.168.1.50" async def test_preview_template_change_and_conversion(hass: HomeAssistant): with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") result = await _configure(hass, result["flow_id"], MANUAL_INPUT) preview_input = { **PREVIEW_INPUT, "template": "B", "temp_min": 17.0, "temp_max": 28.0, "temp_step": 1.0, "temp_num": 2, "temp_den": 4, "temp_offset": 5, } result = await _configure(hass, result["flow_id"], preview_input) assert result["type"] == "create_entry" assert result["data"]["template"] == "B" assert result["data"]["temp_step"] == 10 assert result["data"]["overrides"]["adjust_temperature"] == { "num": 2, "den": 4, "offset": 5, "min": 170, "max": 280, } async def test_preview_invalid_range(hass: HomeAssistant): with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") result = await _configure(hass, result["flow_id"], MANUAL_INPUT) result = await _configure( hass, result["flow_id"], {**PREVIEW_INPUT, "temp_min": 30.0, "temp_max": 16.0} ) assert result["type"] == "form" assert result["errors"] == {"base": "invalid_range"} result = await _configure(hass, result["flow_id"], PREVIEW_INPUT) assert result["type"] == "create_entry" async def test_preview_contains_trial_values(hass: HomeAssistant): props = { Prop.DISPLAY_TEMPERATURE: Value(ValueKind.INT, 200), Prop.DEVICE_CAPABILITIES: Value(ValueKind.INT, 0b11011), } with patch( "custom_components.fglair.config_flow._run_trial", return_value=TrialResult(ok=True, properties=props), ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") result = await _configure(hass, result["flow_id"], MANUAL_INPUT) preview = result["description_placeholders"]["preview"] assert "20.0" in preview assert "0x0000001B" in preview assert "cool" in preview and "heat" in preview async def test_manual_connect_error(hass: HomeAssistant): with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_NETWORK_FAIL, ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") result = await _configure(hass, result["flow_id"], MANUAL_INPUT) assert result["type"] == "form" assert result["errors"] == {"base": "cannot_connect"} async def test_manual_key_mismatch_error(hass: HomeAssistant): with patch( "custom_components.fglair.config_flow._run_trial", return_value=TrialResult(ok=False, reason="key_mismatch"), ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") result = await _configure(hass, result["flow_id"], MANUAL_INPUT) assert result["type"] == "form" assert result["errors"] == {"base": "key_mismatch"} async def test_manual_already_configured(hass: HomeAssistant): existing = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DSN) existing.add_to_hass(hass) with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") result = await _configure(hass, result["flow_id"], MANUAL_INPUT) assert result["type"] == "abort" assert result["reason"] == "already_configured" async def test_manual_error_keeps_input(hass: HomeAssistant): with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_NETWORK_FAIL, ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "manual") result = await _configure(hass, result["flow_id"], MANUAL_INPUT) assert result["errors"] == {"base": "cannot_connect"} assert _suggested(result["data_schema"], "host") == MANUAL_INPUT["host"] assert _suggested(result["data_schema"], "dsn") == DSN assert _suggested(result["data_schema"], "lanip_key") is None async def test_import_json_success(hass: HomeAssistant): payload = json.dumps( { "ip_address": "192.168.1.60", "dsn": DSN, "lanip_key": LANIP_KEY, "lanip_key_id": KEY_ID, "model": "AP-WB2E", "name": "Kitchen", } ) with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "import_json") assert result["type"] == "form" result = await _configure( hass, result["flow_id"], {"config_json": payload} ) assert result["step_id"] == "preview" assert _suggested(result["data_schema"], "template") == "B" result = await _configure( hass, result["flow_id"], {**PREVIEW_INPUT, "template": "B"} ) assert result["type"] == "create_entry" assert result["data"]["host"] == "192.168.1.60" assert result["data"]["template"] == "B" async def test_import_json_invalid(hass: HomeAssistant): result = await _menu(hass) result = await _select(hass, result["flow_id"], "import_json") result = await _configure( hass, result["flow_id"], {"config_json": "not json"} ) assert result["type"] == "form" assert result["errors"] == {"base": "invalid_import"} async def test_cloud_flow_success(hass: HomeAssistant): device = Device( name="Bedroom", model="AP-WC1E", dsn=DSN, lan_ip="192.168.1.50", lanip_key=LANIP_KEY, lanip_key_id=KEY_ID, region="eu", ) with patch( "pyfglair.discover", AsyncMock(return_value=[device]) ), patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "cloud") assert result["type"] == "form" result = await _configure( hass, result["flow_id"], {"email": "user@example.com", "password": "secret", "region": "eu"}, ) assert result["step_id"] == "cloud_device" result = await _configure(hass, result["flow_id"], {"dsn": DSN}) assert result["step_id"] == "preview" result = await _configure(hass, result["flow_id"], PREVIEW_INPUT) assert result["type"] == "create_entry" assert result["data"]["host"] == "192.168.1.50" assert result["data"]["lanip_key_id"] == KEY_ID async def test_cloud_no_devices(hass: HomeAssistant): with patch( "pyfglair.discover", AsyncMock( side_effect=pyfglair.ProvisionError( "У аккаунта нет устройств", "no_devices" ) ), ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "cloud") result = await _configure( hass, result["flow_id"], {"email": "user@example.com", "password": "secret", "region": "eu"}, ) assert result["type"] == "form" assert result["errors"] == {"base": "no_devices"} async def test_cloud_auth_error(hass: HomeAssistant): with patch( "pyfglair.discover", AsyncMock( side_effect=pyfglair.ProvisionError("401 unauthorized", "auth") ), ): result = await _menu(hass) result = await _select(hass, result["flow_id"], "cloud") result = await _configure( hass, result["flow_id"], {"email": "user@example.com", "password": "bad", "region": "eu"}, ) assert result["type"] == "form" assert result["errors"] == {"base": "invalid_auth"} async def test_yaml_import(hass: HomeAssistant): payload = { "ip_address": "192.168.1.70", "dsn": DSN, "lanip_key": LANIP_KEY, "lanip_key_id": KEY_ID, "model": "AP-WF1E", } with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ): assert await async_setup_component(hass, DOMAIN, {DOMAIN: payload}) await hass.async_block_till_done() entries = hass.config_entries.async_entries(DOMAIN) assert len(entries) == 1 assert entries[0].data["template"] == "F" assert entries[0].data["host"] == "192.168.1.70" async def test_yaml_import_invalid(hass: HomeAssistant): assert await async_setup_component(hass, DOMAIN, {DOMAIN: "not-a-dict"}) await hass.async_block_till_done() assert hass.config_entries.async_entries(DOMAIN) == [] assert not hass.config_entries.flow.async_progress() async def test_import_json_null_device_port(hass: HomeAssistant): payload = json.dumps( { "ip_address": "192.168.1.71", "dsn": DSN, "lanip_key": LANIP_KEY, "lanip_key_id": str(KEY_ID), "device_port": None, } ) with patch( "custom_components.fglair.config_flow._run_trial", return_value=TRIAL_OK, ) as trial: result = await _menu(hass) result = await _select(hass, result["flow_id"], "import_json") result = await _configure( hass, result["flow_id"], {"config_json": payload} ) result = await _configure(hass, result["flow_id"], PREVIEW_INPUT) assert result["type"] == "create_entry" assert result["data"]["device_port"] == 80 assert trial.call_args.args[0]["device_port"] == 80