mirror of
https://github.com/home-assistant/core.git
synced 2025-11-09 19:09:32 +00:00
1587 lines
53 KiB
Python
1587 lines
53 KiB
Python
"""Test the switchbot config flow."""
|
|
|
|
from collections.abc import Generator
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
from switchbot import SwitchbotAccountConnectionError, SwitchbotAuthenticationError
|
|
|
|
from homeassistant.components.bluetooth import BluetoothScanningMode
|
|
from homeassistant.components.switchbot.const import (
|
|
CONF_ENCRYPTION_KEY,
|
|
CONF_KEY_ID,
|
|
CONF_LOCK_NIGHTLATCH,
|
|
CONF_RETRY_COUNT,
|
|
)
|
|
from homeassistant.config_entries import SOURCE_BLUETOOTH, SOURCE_IGNORE, SOURCE_USER
|
|
from homeassistant.const import (
|
|
CONF_ADDRESS,
|
|
CONF_NAME,
|
|
CONF_PASSWORD,
|
|
CONF_SENSOR_TYPE,
|
|
CONF_USERNAME,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.data_entry_flow import FlowResultType
|
|
|
|
from . import (
|
|
NOT_SWITCHBOT_INFO,
|
|
USER_INPUT,
|
|
WOCURTAIN_SERVICE_INFO,
|
|
WOHAND_ENCRYPTED_SERVICE_INFO,
|
|
WOHAND_SERVICE_ALT_ADDRESS_INFO,
|
|
WOHAND_SERVICE_INFO,
|
|
WOHAND_SERVICE_INFO_NOT_CONNECTABLE,
|
|
WOLOCK_SERVICE_INFO,
|
|
WORELAY_SWITCH_1PM_SERVICE_INFO,
|
|
WOSENSORTH_SERVICE_INFO,
|
|
init_integration,
|
|
patch_async_setup_entry,
|
|
)
|
|
|
|
from tests.common import MockConfigEntry
|
|
|
|
DOMAIN = "switchbot"
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_scanners_all_active() -> Generator[None]:
|
|
"""Mock all scanners as active mode."""
|
|
mock_scanner = Mock()
|
|
mock_scanner.current_mode = BluetoothScanningMode.ACTIVE
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_current_scanners",
|
|
return_value=[mock_scanner],
|
|
):
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_scanners_all_passive() -> Generator[None]:
|
|
"""Mock all scanners as passive mode."""
|
|
mock_scanner = Mock()
|
|
mock_scanner.current_mode = BluetoothScanningMode.PASSIVE
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_current_scanners",
|
|
return_value=[mock_scanner],
|
|
):
|
|
yield
|
|
|
|
|
|
async def test_bluetooth_discovery(hass: HomeAssistant) -> None:
|
|
"""Test discovery via bluetooth with a valid device."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=WOHAND_SERVICE_INFO,
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_bluetooth_discovery_requires_password(hass: HomeAssistant) -> None:
|
|
"""Test discovery via bluetooth with a valid device that needs a password."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=WOHAND_ENCRYPTED_SERVICE_INFO,
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "password"
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{CONF_PASSWORD: "abc123"},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot 923B"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
CONF_PASSWORD: "abc123",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_bluetooth_discovery_encrypted_key(hass: HomeAssistant) -> None:
|
|
"""Test discovery via bluetooth with a lock."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=WOLOCK_SERVICE_INFO,
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_key"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_key"
|
|
assert result["errors"] == {}
|
|
|
|
with patch(
|
|
"switchbot.SwitchbotLock.verify_encryption_key",
|
|
return_value=False,
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_KEY_ID: "",
|
|
CONF_ENCRYPTION_KEY: "",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_key"
|
|
assert result["errors"] == {"base": "encryption_key_invalid"}
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotLock.verify_encryption_key",
|
|
return_value=True,
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Lock EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "lock",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_bluetooth_discovery_key(hass: HomeAssistant) -> None:
|
|
"""Test discovery via bluetooth with a encrypted device."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=WORELAY_SWITCH_1PM_SERVICE_INFO,
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_key"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_key"
|
|
assert result["errors"] == {}
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotRelaySwitch.verify_encryption_key", return_value=True
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Relay Switch 1PM EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "relay_switch_1pm",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_bluetooth_discovery_already_setup(hass: HomeAssistant) -> None:
|
|
"""Test discovery via bluetooth with a valid device when already setup."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_NAME: "test-name",
|
|
CONF_PASSWORD: "test-password",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
},
|
|
unique_id="aabbccddeeff",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=WOHAND_SERVICE_INFO,
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "already_configured"
|
|
|
|
|
|
async def test_async_step_bluetooth_not_switchbot(hass: HomeAssistant) -> None:
|
|
"""Test discovery via bluetooth not switchbot."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=NOT_SWITCHBOT_INFO,
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "not_supported"
|
|
|
|
|
|
async def test_async_step_bluetooth_not_connectable(hass: HomeAssistant) -> None:
|
|
"""Test discovery via bluetooth and its not connectable switchbot."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=WOHAND_SERVICE_INFO_NOT_CONNECTABLE,
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "not_supported"
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wohand(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form with password and valid mac."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
assert result["errors"] is None
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wohand_already_configured(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form with password and valid mac."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_NAME: "test-name",
|
|
CONF_PASSWORD: "test-password",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
},
|
|
unique_id="aabbccddeeff",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "no_devices_found"
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wohand_replaces_ignored(hass: HomeAssistant) -> None:
|
|
"""Test setting up a switchbot replaces an ignored entry."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN, data={}, unique_id="aabbccddeeff", source=SOURCE_IGNORE
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wocurtain(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form with password and valid mac."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOCURTAIN_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
assert result["errors"] is None
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Curtain EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_SENSOR_TYPE: "curtain",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wocurtain_or_bot(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form with valid address."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[
|
|
NOT_SWITCHBOT_INFO,
|
|
WOCURTAIN_SERVICE_INFO,
|
|
WOHAND_SERVICE_ALT_ADDRESS_INFO,
|
|
WOHAND_SERVICE_INFO_NOT_CONNECTABLE,
|
|
],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "select_device"
|
|
assert result["errors"] == {}
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
USER_INPUT,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Curtain EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_SENSOR_TYPE: "curtain",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wocurtain_or_bot_with_password(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form and valid address and a bot with a password."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[
|
|
WOCURTAIN_SERVICE_INFO,
|
|
WOHAND_ENCRYPTED_SERVICE_INFO,
|
|
WOHAND_SERVICE_INFO_NOT_CONNECTABLE,
|
|
],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "select_device"
|
|
assert result["errors"] == {}
|
|
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B"},
|
|
)
|
|
assert result2["type"] is FlowResultType.FORM
|
|
assert result2["step_id"] == "password"
|
|
assert result2["errors"] is None
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result3 = await hass.config_entries.flow.async_configure(
|
|
result2["flow_id"],
|
|
{CONF_PASSWORD: "abc123"},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result3["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result3["title"] == "Bot 923B"
|
|
assert result3["data"] == {
|
|
CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B",
|
|
CONF_PASSWORD: "abc123",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_single_bot_with_password(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form for a bot with a password."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_ENCRYPTED_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "password"
|
|
assert result["errors"] is None
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{CONF_PASSWORD: "abc123"},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["title"] == "Bot 923B"
|
|
assert result2["data"] == {
|
|
CONF_ADDRESS: "798A8547-2A3D-C609-55FF-73FA824B923B",
|
|
CONF_PASSWORD: "abc123",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_woencrypted_key(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form for a lock."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOLOCK_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_key"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_key"
|
|
assert result["errors"] == {}
|
|
|
|
with patch(
|
|
"switchbot.SwitchbotLock.verify_encryption_key",
|
|
return_value=False,
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_KEY_ID: "",
|
|
CONF_ENCRYPTION_KEY: "",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_key"
|
|
assert result["errors"] == {"base": "encryption_key_invalid"}
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotLock.verify_encryption_key",
|
|
return_value=True,
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Lock EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "lock",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_woencrypted_auth(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form for a lock."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOLOCK_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_auth"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
assert result["errors"] == {}
|
|
|
|
with patch(
|
|
"switchbot.SwitchbotLock.async_retrieve_encryption_key",
|
|
side_effect=SwitchbotAuthenticationError("error from api"),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "",
|
|
CONF_PASSWORD: "",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
assert result["errors"] == {"base": "auth_failed"}
|
|
assert "error from api" in result["description_placeholders"]["error_detail"]
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotLock.verify_encryption_key",
|
|
return_value=True,
|
|
),
|
|
patch(
|
|
"switchbot.SwitchbotLock.async_retrieve_encryption_key",
|
|
return_value={
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "username",
|
|
CONF_PASSWORD: "password",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Lock EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "lock",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_woencrypted_auth_switchbot_api_down(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test the user initiated form for a lock when the switchbot api is down."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOLOCK_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_auth"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
assert result["errors"] == {}
|
|
|
|
with patch(
|
|
"switchbot.SwitchbotLock.async_retrieve_encryption_key",
|
|
side_effect=SwitchbotAccountConnectionError("Switchbot API down"),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "",
|
|
CONF_PASSWORD: "",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "api_error"
|
|
assert result["description_placeholders"] == {"error_detail": "Switchbot API down"}
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wolock_or_bot(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form for a lock."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[
|
|
WOLOCK_SERVICE_INFO,
|
|
WOHAND_SERVICE_ALT_ADDRESS_INFO,
|
|
],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "select_device"
|
|
assert result["errors"] == {}
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
USER_INPUT,
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_key"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_key"
|
|
assert result["errors"] == {}
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotLock.verify_encryption_key",
|
|
return_value=True,
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Lock EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "lock",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_wosensor(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form with password and valid mac."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOSENSORTH_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
assert result["errors"] is None
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Meter EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_SENSOR_TYPE: "hygrometer",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_cloud_login(hass: HomeAssistant) -> None:
|
|
"""Test the cloud login flow."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "cloud_login"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "cloud_login"
|
|
|
|
# Test successful cloud login
|
|
with (
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.fetch_cloud_devices",
|
|
return_value=None,
|
|
),
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_SERVICE_INFO],
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "test@example.com",
|
|
CONF_PASSWORD: "testpass",
|
|
},
|
|
)
|
|
|
|
# Should proceed to device selection with single device, so go to confirm
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
|
|
# Confirm device setup
|
|
with patch_async_setup_entry():
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_cloud_login_auth_failed(hass: HomeAssistant) -> None:
|
|
"""Test the cloud login flow with authentication failure."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "cloud_login"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "cloud_login"
|
|
|
|
# Test authentication failure
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.fetch_cloud_devices",
|
|
side_effect=SwitchbotAuthenticationError("Invalid credentials"),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "test@example.com",
|
|
CONF_PASSWORD: "wrongpass",
|
|
},
|
|
)
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "cloud_login"
|
|
assert result["errors"] == {"base": "auth_failed"}
|
|
assert "Invalid credentials" in result["description_placeholders"]["error_detail"]
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_cloud_login_api_error(hass: HomeAssistant) -> None:
|
|
"""Test the cloud login flow with API error."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "cloud_login"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "cloud_login"
|
|
|
|
# Test API connection error
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.fetch_cloud_devices",
|
|
side_effect=SwitchbotAccountConnectionError("API is down"),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "test@example.com",
|
|
CONF_PASSWORD: "testpass",
|
|
},
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "api_error"
|
|
assert result["description_placeholders"] == {"error_detail": "API is down"}
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_cloud_login_then_encrypted_device(hass: HomeAssistant) -> None:
|
|
"""Test cloud login followed by encrypted device setup using saved credentials."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "cloud_login"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "cloud_login"
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.fetch_cloud_devices",
|
|
return_value=None,
|
|
),
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOLOCK_SERVICE_INFO],
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "test@example.com",
|
|
CONF_PASSWORD: "testpass",
|
|
},
|
|
)
|
|
|
|
# Should go to encrypted device choice menu
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
# Choose encrypted auth
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_auth"}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
None,
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotLock.async_retrieve_encryption_key",
|
|
return_value={
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
),
|
|
patch("switchbot.SwitchbotLock.verify_encryption_key", return_value=True),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "test@example.com",
|
|
CONF_PASSWORD: "testpass",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Lock EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "lock",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_no_devices(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form with password and valid mac."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "no_devices_found"
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_async_step_user_takes_precedence_over_discovery(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test manual setup takes precedence over discovery."""
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_BLUETOOTH},
|
|
data=WOCURTAIN_SERVICE_INFO,
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN,
|
|
context={"source": SOURCE_USER},
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOCURTAIN_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result2 = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
user_input={},
|
|
)
|
|
|
|
assert result2["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result2["title"] == "Curtain EEFF"
|
|
assert result2["data"] == {
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_SENSOR_TYPE: "curtain",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
# Verify the original one was aborted
|
|
assert not hass.config_entries.flow.async_progress(DOMAIN)
|
|
|
|
|
|
async def test_options_flow(hass: HomeAssistant) -> None:
|
|
"""Test updating options."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_NAME: "test-name",
|
|
CONF_PASSWORD: "test-password",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
},
|
|
options={
|
|
CONF_RETRY_COUNT: 10,
|
|
},
|
|
unique_id="aabbccddeeff",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
entry = await init_integration(hass)
|
|
|
|
result = await hass.config_entries.options.async_init(entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "init"
|
|
assert result["errors"] is None
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
CONF_RETRY_COUNT: 3,
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"][CONF_RETRY_COUNT] == 3
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 2
|
|
|
|
# Test changing of entry options.
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
entry = await init_integration(hass)
|
|
|
|
result = await hass.config_entries.options.async_init(entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "init"
|
|
assert result["errors"] is None
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
CONF_RETRY_COUNT: 6,
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"][CONF_RETRY_COUNT] == 6
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
assert entry.options[CONF_RETRY_COUNT] == 6
|
|
|
|
|
|
async def test_options_flow_lock_pro(hass: HomeAssistant) -> None:
|
|
"""Test updating options."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN,
|
|
data={
|
|
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
|
|
CONF_NAME: "test-name",
|
|
CONF_PASSWORD: "test-password",
|
|
CONF_SENSOR_TYPE: "lock_pro",
|
|
},
|
|
options={CONF_RETRY_COUNT: 10},
|
|
unique_id="aabbccddeeff",
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
# Test Force night_latch should be disabled by default.
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
await hass.config_entries.async_setup(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
result = await hass.config_entries.options.async_init(entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "init"
|
|
assert result["errors"] is None
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
CONF_RETRY_COUNT: 3,
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"][CONF_LOCK_NIGHTLATCH] is False
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
# Test Set force night_latch to be enabled.
|
|
|
|
with patch_async_setup_entry() as mock_setup_entry:
|
|
result = await hass.config_entries.options.async_init(entry.entry_id)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "init"
|
|
assert result["errors"] is None
|
|
|
|
result = await hass.config_entries.options.async_configure(
|
|
result["flow_id"],
|
|
user_input={
|
|
CONF_LOCK_NIGHTLATCH: True,
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["data"][CONF_LOCK_NIGHTLATCH] is True
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 0
|
|
|
|
assert entry.options[CONF_LOCK_NIGHTLATCH] is True
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_worelay_switch_1pm_key(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form for a relay switch 1pm."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WORELAY_SWITCH_1PM_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_key"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_key"
|
|
assert result["errors"] == {}
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotRelaySwitch.verify_encryption_key", return_value=True
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Relay Switch 1PM EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "relay_switch_1pm",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_worelay_switch_1pm_auth(hass: HomeAssistant) -> None:
|
|
"""Test the user initiated form for a relay switch 1pm."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WORELAY_SWITCH_1PM_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_auth"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
assert result["errors"] == {}
|
|
|
|
with patch(
|
|
"switchbot.SwitchbotRelaySwitch.async_retrieve_encryption_key",
|
|
side_effect=SwitchbotAuthenticationError("error from api"),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "",
|
|
CONF_PASSWORD: "",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
assert result["errors"] == {"base": "auth_failed"}
|
|
assert "error from api" in result["description_placeholders"]["error_detail"]
|
|
|
|
with (
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
patch(
|
|
"switchbot.SwitchbotRelaySwitch.async_retrieve_encryption_key",
|
|
return_value={
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
},
|
|
),
|
|
patch(
|
|
"switchbot.SwitchbotRelaySwitch.verify_encryption_key", return_value=True
|
|
),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "username",
|
|
CONF_PASSWORD: "password",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Relay Switch 1PM EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_KEY_ID: "ff",
|
|
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
|
|
CONF_SENSOR_TYPE: "relay_switch_1pm",
|
|
}
|
|
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_passive")
|
|
async def test_user_setup_worelay_switch_1pm_auth_switchbot_api_down(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test the user initiated form for a relay switch 1pm when the switchbot api is down."""
|
|
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
|
|
with patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WORELAY_SWITCH_1PM_SERVICE_INFO],
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{"next_step_id": "select_device"},
|
|
)
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "encrypted_choose_method"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={"next_step_id": "encrypted_auth"}
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "encrypted_auth"
|
|
assert result["errors"] == {}
|
|
|
|
with patch(
|
|
"switchbot.SwitchbotRelaySwitch.async_retrieve_encryption_key",
|
|
side_effect=SwitchbotAccountConnectionError("Switchbot API down"),
|
|
):
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"],
|
|
{
|
|
CONF_USERNAME: "",
|
|
CONF_PASSWORD: "",
|
|
},
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert result["type"] is FlowResultType.ABORT
|
|
assert result["reason"] == "api_error"
|
|
assert result["description_placeholders"] == {"error_detail": "Switchbot API down"}
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_scanners_all_active")
|
|
async def test_user_skip_menu_when_all_scanners_active(hass: HomeAssistant) -> None:
|
|
"""Test that menu is skipped when all scanners are in active mode."""
|
|
with (
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_SERVICE_INFO],
|
|
),
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
|
|
# Should skip menu and go directly to select_device -> confirm
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={}
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_user_show_menu_when_passive_scanner_present(hass: HomeAssistant) -> None:
|
|
"""Test that menu is shown when any scanner is in passive mode."""
|
|
mock_scanner_active = Mock()
|
|
mock_scanner_active.current_mode = BluetoothScanningMode.ACTIVE
|
|
mock_scanner_passive = Mock()
|
|
mock_scanner_passive.current_mode = BluetoothScanningMode.PASSIVE
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.async_current_scanners",
|
|
return_value=[mock_scanner_active, mock_scanner_passive],
|
|
),
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_SERVICE_INFO],
|
|
),
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
|
|
# Should show menu since not all scanners are active
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
assert set(result["menu_options"]) == {"cloud_login", "select_device"}
|
|
|
|
# Choose select_device from menu
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {"next_step_id": "select_device"}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
|
|
# Confirm the device
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={}
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
assert len(mock_setup_entry.mock_calls) == 1
|
|
|
|
|
|
async def test_user_show_menu_when_no_scanners(hass: HomeAssistant) -> None:
|
|
"""Test that menu is shown when no scanners are available."""
|
|
with (
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.async_current_scanners",
|
|
return_value=[],
|
|
),
|
|
patch(
|
|
"homeassistant.components.switchbot.config_flow.async_discovered_service_info",
|
|
return_value=[WOHAND_SERVICE_INFO],
|
|
),
|
|
patch_async_setup_entry() as mock_setup_entry,
|
|
):
|
|
result = await hass.config_entries.flow.async_init(
|
|
DOMAIN, context={"source": SOURCE_USER}
|
|
)
|
|
|
|
# Should show menu when no scanners are available
|
|
assert result["type"] is FlowResultType.MENU
|
|
assert result["step_id"] == "user"
|
|
assert set(result["menu_options"]) == {"cloud_login", "select_device"}
|
|
|
|
# Choose select_device from menu
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], {"next_step_id": "select_device"}
|
|
)
|
|
assert result["type"] is FlowResultType.FORM
|
|
assert result["step_id"] == "confirm"
|
|
|
|
# Confirm the device
|
|
result = await hass.config_entries.flow.async_configure(
|
|
result["flow_id"], user_input={}
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert result["type"] is FlowResultType.CREATE_ENTRY
|
|
assert result["title"] == "Bot EEFF"
|
|
assert result["data"] == {
|
|
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
|
|
CONF_SENSOR_TYPE: "bot",
|
|
}
|
|
assert len(mock_setup_entry.mock_calls) == 1
|