Add config flow for Ping (#103743)

This commit is contained in:
Jan-Philipp Benecke 2023-11-17 20:30:30 +01:00 committed by GitHub
parent 2d891c77ef
commit e5bc25523e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 473 additions and 211 deletions

View file

@ -6,16 +6,18 @@ import logging
from icmplib import SocketPermissionError, async_ping
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.reload import async_setup_reload_service
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, PLATFORMS
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.platform_only_config_schema(DOMAIN)
PLATFORMS = [Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER]
@dataclass(slots=True)
@ -27,7 +29,6 @@ class PingDomainData:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the ping integration."""
await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
hass.data[DOMAIN] = PingDomainData(
privileged=await _can_use_icmp_lib_with_privilege(),
@ -36,6 +37,25 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Ping (ICMP) from a config entry."""
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(async_reload_entry))
return True
async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle an options update."""
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def _can_use_icmp_lib_with_privilege() -> None | bool:
"""Verify we can create a raw socket."""
try:

View file

@ -12,30 +12,26 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_NAME, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import PingDomainData
from .const import DOMAIN
from .const import CONF_IMPORTED_BY, CONF_PING_COUNT, DEFAULT_PING_COUNT, DOMAIN
from .helpers import PingDataICMPLib, PingDataSubProcess
_LOGGER = logging.getLogger(__name__)
ATTR_ROUND_TRIP_TIME_AVG = "round_trip_time_avg"
ATTR_ROUND_TRIP_TIME_MAX = "round_trip_time_max"
ATTR_ROUND_TRIP_TIME_MDEV = "round_trip_time_mdev"
ATTR_ROUND_TRIP_TIME_MIN = "round_trip_time_min"
CONF_PING_COUNT = "count"
DEFAULT_NAME = "Ping"
DEFAULT_PING_COUNT = 5
SCAN_INTERVAL = timedelta(minutes=5)
PARALLEL_UPDATES = 50
@ -57,22 +53,49 @@ async def async_setup_platform(
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Ping Binary sensor."""
"""YAML init: import via config flow."""
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={CONF_IMPORTED_BY: "binary_sensor", **config},
)
)
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2024.6.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Ping",
},
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a Ping config entry."""
data: PingDomainData = hass.data[DOMAIN]
host: str = config[CONF_HOST]
count: int = config[CONF_PING_COUNT]
name: str = config.get(CONF_NAME, f"{DEFAULT_NAME} {host}")
privileged: bool | None = data.privileged
host: str = entry.options[CONF_HOST]
count: int = int(entry.options[CONF_PING_COUNT])
ping_cls: type[PingDataSubProcess | PingDataICMPLib]
if privileged is None:
if data.privileged is None:
ping_cls = PingDataSubProcess
else:
ping_cls = PingDataICMPLib
async_add_entities(
[PingBinarySensor(name, ping_cls(hass, host, count, privileged))]
[PingBinarySensor(entry, ping_cls(hass, host, count, data.privileged))]
)
@ -80,12 +103,24 @@ class PingBinarySensor(RestoreEntity, BinarySensorEntity):
"""Representation of a Ping Binary sensor."""
_attr_device_class = BinarySensorDeviceClass.CONNECTIVITY
_attr_available = False
def __init__(self, name: str, ping: PingDataSubProcess | PingDataICMPLib) -> None:
def __init__(
self,
config_entry: ConfigEntry,
ping_cls: PingDataSubProcess | PingDataICMPLib,
) -> None:
"""Initialize the Ping Binary sensor."""
self._attr_available = False
self._attr_name = name
self._ping = ping
self._attr_name = config_entry.title
self._attr_unique_id = config_entry.entry_id
# if this was imported just enable it when it was enabled before
if CONF_IMPORTED_BY in config_entry.data:
self._attr_entity_registry_enabled_default = bool(
config_entry.data[CONF_IMPORTED_BY] == "binary_sensor"
)
self._ping = ping_cls
@property
def is_on(self) -> bool:

View file

@ -0,0 +1,107 @@
"""Config flow for Ping (ICMP) integration."""
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import selector
from homeassistant.util.network import is_ip_address
from .const import CONF_IMPORTED_BY, CONF_PING_COUNT, DEFAULT_PING_COUNT, DOMAIN
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Ping."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
}
),
)
if not is_ip_address(user_input[CONF_HOST]):
self.async_abort(reason="invalid_ip_address")
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
return self.async_create_entry(
title=user_input[CONF_HOST],
data={},
options={**user_input, CONF_PING_COUNT: DEFAULT_PING_COUNT},
)
async def async_step_import(self, import_info: Mapping[str, Any]) -> FlowResult:
"""Import an entry."""
to_import = {
CONF_HOST: import_info[CONF_HOST],
CONF_PING_COUNT: import_info[CONF_PING_COUNT],
}
title = import_info.get(CONF_NAME, import_info[CONF_HOST])
self._async_abort_entries_match({CONF_HOST: to_import[CONF_HOST]})
return self.async_create_entry(
title=title,
data={CONF_IMPORTED_BY: import_info[CONF_IMPORTED_BY]},
options=to_import,
)
@staticmethod
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> config_entries.OptionsFlow:
"""Create the options flow."""
return OptionsFlowHandler(config_entry)
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle an options flow for Ping."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
return self.async_show_form(
step_id="init",
data_schema=vol.Schema(
{
vol.Required(
CONF_HOST, default=self.config_entry.options[CONF_HOST]
): str,
vol.Optional(
CONF_PING_COUNT,
default=self.config_entry.options[CONF_PING_COUNT],
): selector.NumberSelector(
selector.NumberSelectorConfig(
min=1, max=100, mode=selector.NumberSelectorMode.BOX
)
),
}
),
)

View file

@ -1,6 +1,5 @@
"""Tracks devices by sending a ICMP echo request (ping)."""
from homeassistant.const import Platform
# The ping binary and icmplib timeouts are not the same
# timeout. ping is an overall timeout, icmplib is the
@ -15,4 +14,7 @@ ICMP_TIMEOUT = 1
PING_ATTEMPTS_COUNT = 3
DOMAIN = "ping"
PLATFORMS = [Platform.BINARY_SENSOR]
CONF_PING_COUNT = "count"
CONF_IMPORTED_BY = "imported_by"
DEFAULT_PING_COUNT = 5

View file

@ -1,38 +1,33 @@
"""Tracks devices by sending a ICMP echo request (ping)."""
from __future__ import annotations
import asyncio
from datetime import datetime, timedelta
from datetime import timedelta
import logging
import subprocess
from icmplib import async_multiping
import voluptuous as vol
from homeassistant.components.device_tracker import (
CONF_SCAN_INTERVAL,
PLATFORM_SCHEMA as BASE_PLATFORM_SCHEMA,
SCAN_INTERVAL,
AsyncSeeCallback,
ScannerEntity,
SourceType,
)
from homeassistant.const import CONF_HOSTS
from homeassistant.core import HomeAssistant
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST, CONF_HOSTS, CONF_NAME
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util
from homeassistant.util.async_ import gather_with_limited_concurrency
from homeassistant.util.process import kill_subprocess
from . import PingDomainData
from .const import DOMAIN, ICMP_TIMEOUT, PING_ATTEMPTS_COUNT, PING_TIMEOUT
from .const import CONF_IMPORTED_BY, CONF_PING_COUNT, DOMAIN
from .helpers import PingDataICMPLib, PingDataSubProcess
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
CONF_PING_COUNT = "count"
CONCURRENT_PING_LIMIT = 6
SCAN_INTERVAL = timedelta(minutes=5)
PLATFORM_SCHEMA = BASE_PLATFORM_SCHEMA.extend(
{
@ -42,123 +37,110 @@ PLATFORM_SCHEMA = BASE_PLATFORM_SCHEMA.extend(
)
class HostSubProcess:
"""Host object with ping detection."""
def __init__(
self,
ip_address: str,
dev_id: str,
hass: HomeAssistant,
config: ConfigType,
privileged: bool | None,
) -> None:
"""Initialize the Host pinger."""
self.hass = hass
self.ip_address = ip_address
self.dev_id = dev_id
self._count = config[CONF_PING_COUNT]
self._ping_cmd = ["ping", "-n", "-q", "-c1", "-W1", ip_address]
def ping(self) -> bool | None:
"""Send an ICMP echo request and return True if success."""
with subprocess.Popen(
self._ping_cmd,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
close_fds=False, # required for posix_spawn
) as pinger:
try:
pinger.communicate(timeout=1 + PING_TIMEOUT)
return pinger.returncode == 0
except subprocess.TimeoutExpired:
kill_subprocess(pinger)
return False
except subprocess.CalledProcessError:
return False
def update(self) -> bool:
"""Update device state by sending one or more ping messages."""
failed = 0
while failed < self._count: # check more times if host is unreachable
if self.ping():
return True
failed += 1
_LOGGER.debug("No response from %s failed=%d", self.ip_address, failed)
return False
async def async_setup_scanner(
hass: HomeAssistant,
config: ConfigType,
async_see: AsyncSeeCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> bool:
"""Set up the Host objects and return the update function."""
"""Legacy init: import via config flow."""
for dev_name, dev_host in config[CONF_HOSTS].items():
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
CONF_IMPORTED_BY: "device_tracker",
CONF_NAME: dev_name,
CONF_HOST: dev_host,
CONF_PING_COUNT: config[CONF_PING_COUNT],
},
)
)
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2024.6.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Ping",
},
)
return True
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up a Ping config entry."""
data: PingDomainData = hass.data[DOMAIN]
privileged = data.privileged
ip_to_dev_id = {ip: dev_id for (dev_id, ip) in config[CONF_HOSTS].items()}
interval = config.get(
CONF_SCAN_INTERVAL,
timedelta(seconds=len(ip_to_dev_id) * config[CONF_PING_COUNT]) + SCAN_INTERVAL,
)
_LOGGER.debug(
"Started ping tracker with interval=%s on hosts: %s",
interval,
",".join(ip_to_dev_id.keys()),
)
if privileged is None:
hosts = [
HostSubProcess(ip, dev_id, hass, config, privileged)
for (dev_id, ip) in config[CONF_HOSTS].items()
]
async def async_update(now: datetime) -> None:
"""Update all the hosts on every interval time."""
results = await gather_with_limited_concurrency(
CONCURRENT_PING_LIMIT,
*(hass.async_add_executor_job(host.update) for host in hosts),
)
await asyncio.gather(
*(
async_see(dev_id=host.dev_id, source_type=SourceType.ROUTER)
for idx, host in enumerate(hosts)
if results[idx]
)
)
host: str = entry.options[CONF_HOST]
count: int = int(entry.options[CONF_PING_COUNT])
ping_cls: type[PingDataSubProcess | PingDataICMPLib]
if data.privileged is None:
ping_cls = PingDataSubProcess
else:
ping_cls = PingDataICMPLib
async def async_update(now: datetime) -> None:
"""Update all the hosts on every interval time."""
responses = await async_multiping(
list(ip_to_dev_id),
count=PING_ATTEMPTS_COUNT,
timeout=ICMP_TIMEOUT,
privileged=privileged,
)
_LOGGER.debug("Multiping responses: %s", responses)
await asyncio.gather(
*(
async_see(dev_id=dev_id, source_type=SourceType.ROUTER)
for idx, dev_id in enumerate(ip_to_dev_id.values())
if responses[idx].is_alive
)
)
async_add_entities(
[PingDeviceTracker(entry, ping_cls(hass, host, count, data.privileged))]
)
async def _async_update_interval(now: datetime) -> None:
try:
await async_update(now)
finally:
if not hass.is_stopping:
async_track_point_in_utc_time(
hass, _async_update_interval, now + interval
)
await _async_update_interval(dt_util.now())
return True
class PingDeviceTracker(ScannerEntity):
"""Representation of a Ping device tracker."""
ping: PingDataSubProcess | PingDataICMPLib
def __init__(
self,
config_entry: ConfigEntry,
ping_cls: PingDataSubProcess | PingDataICMPLib,
) -> None:
"""Initialize the Ping device tracker."""
super().__init__()
self._attr_name = config_entry.title
self.ping = ping_cls
self.config_entry = config_entry
@property
def ip_address(self) -> str:
"""Return the primary ip address of the device."""
return self.ping.ip_address
@property
def unique_id(self) -> str:
"""Return a unique ID."""
return self.config_entry.entry_id
@property
def source_type(self) -> SourceType:
"""Return the source type which is router."""
return SourceType.ROUTER
@property
def is_connected(self) -> bool:
"""Return true if ping returns is_alive."""
return self.ping.is_alive
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if entity is enabled by default."""
if CONF_IMPORTED_BY in self.config_entry.data:
return bool(self.config_entry.data[CONF_IMPORTED_BY] == "device_tracker")
return False
async def async_update(self) -> None:
"""Update the sensor."""
await self.ping.async_update()

View file

@ -33,7 +33,7 @@ class PingData:
def __init__(self, hass: HomeAssistant, host: str, count: int) -> None:
"""Initialize the data object."""
self.hass = hass
self._ip_address = host
self.ip_address = host
self._count = count
@ -49,10 +49,10 @@ class PingDataICMPLib(PingData):
async def async_update(self) -> None:
"""Retrieve the latest details from the host."""
_LOGGER.debug("ping address: %s", self._ip_address)
_LOGGER.debug("ping address: %s", self.ip_address)
try:
data = await async_ping(
self._ip_address,
self.ip_address,
count=self._count,
timeout=ICMP_TIMEOUT,
privileged=self._privileged,
@ -89,7 +89,7 @@ class PingDataSubProcess(PingData):
"-c",
str(self._count),
"-W1",
self._ip_address,
self.ip_address,
]
async def async_ping(self) -> dict[str, Any] | None:

View file

@ -2,6 +2,7 @@
"domain": "ping",
"name": "Ping (ICMP)",
"codeowners": ["@jpbede"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/ping",
"iot_class": "local_polling",
"loggers": ["icmplib"],

View file

@ -1 +0,0 @@
reload:

View file

@ -1,8 +1,31 @@
{
"services": {
"reload": {
"name": "[%key:common::action::reload%]",
"description": "Reloads ping sensors from the YAML-configuration."
"config": {
"step": {
"user": {
"title": "Add Ping",
"description": "Ping allows you to check the availability of a host.",
"data": {
"host": "[%key:common::config_flow::data::host%]",
"count": "Ping count"
}
}
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"invalid_ip_address": "Invalid IP address."
}
},
"options": {
"step": {
"init": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"count": "[%key:component::ping::config::step::user::data::count%]"
}
}
},
"abort": {
"invalid_ip_address": "[%key:component::ping::config::abort::invalid_ip_address%]"
}
}
}

View file

@ -355,6 +355,7 @@ FLOWS = {
"philips_js",
"pi_hole",
"picnic",
"ping",
"plaato",
"plex",
"plugwise",

View file

@ -4291,7 +4291,7 @@
"ping": {
"name": "Ping (ICMP)",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "local_polling"
},
"pioneer": {

View file

@ -0,0 +1,14 @@
"""Test configuration for ping."""
from unittest.mock import patch
import pytest
@pytest.fixture
def patch_setup(*args, **kwargs):
"""Patch setup methods."""
with patch(
"homeassistant.components.ping.async_setup_entry",
return_value=True,
), patch("homeassistant.components.ping.async_setup", return_value=True):
yield

View file

@ -0,0 +1,11 @@
"""Constants for tests."""
from icmplib import Host
BINARY_SENSOR_IMPORT_DATA = {
"name": "test2",
"host": "127.0.0.1",
"count": 1,
"scan_interval": 50,
}
NON_AVAILABLE_HOST_PING = Host("192.168.178.1", 10, [])

View file

@ -1,55 +0,0 @@
"""The test for the ping binary_sensor platform."""
from unittest.mock import patch
import pytest
from homeassistant import config as hass_config, setup
from homeassistant.components.ping import DOMAIN
from homeassistant.const import SERVICE_RELOAD
from homeassistant.core import HomeAssistant
from tests.common import get_fixture_path
@pytest.fixture
def mock_ping() -> None:
"""Mock icmplib.ping."""
with patch("homeassistant.components.ping.async_ping"):
yield
async def test_reload(hass: HomeAssistant, mock_ping: None) -> None:
"""Verify we can reload trend sensors."""
await setup.async_setup_component(
hass,
"binary_sensor",
{
"binary_sensor": {
"platform": "ping",
"name": "test",
"host": "127.0.0.1",
"count": 1,
}
},
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
assert hass.states.get("binary_sensor.test")
yaml_path = get_fixture_path("configuration.yaml", "ping")
with patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path):
await hass.services.async_call(
DOMAIN,
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
assert hass.states.get("binary_sensor.test") is None
assert hass.states.get("binary_sensor.test2")

View file

@ -0,0 +1,122 @@
"""Test the Ping (ICMP) config flow."""
from __future__ import annotations
import pytest
from homeassistant import config_entries
from homeassistant.components.ping import DOMAIN
from homeassistant.components.ping.const import CONF_IMPORTED_BY
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import BINARY_SENSOR_IMPORT_DATA
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("host", "expected_title"),
(("192.618.178.1", "192.618.178.1"),),
)
@pytest.mark.usefixtures("patch_setup")
async def test_form(hass: HomeAssistant, host, expected_title) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["step_id"] == "user"
assert result["type"] == FlowResultType.FORM
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"host": host,
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == expected_title
assert result["data"] == {}
assert result["options"] == {
"count": 5,
"host": host,
}
@pytest.mark.parametrize(
("host", "count", "expected_title"),
(("192.618.178.1", 10, "192.618.178.1"),),
)
@pytest.mark.usefixtures("patch_setup")
async def test_options(hass: HomeAssistant, host, count, expected_title) -> None:
"""Test options flow."""
config_entry = MockConfigEntry(
version=1,
source=config_entries.SOURCE_USER,
data={},
domain=DOMAIN,
options={"count": count, "host": host},
title=expected_title,
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
{
"host": "10.10.10.1",
"count": count,
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["data"] == {
"count": count,
"host": "10.10.10.1",
}
@pytest.mark.usefixtures("patch_setup")
async def test_step_import(hass: HomeAssistant) -> None:
"""Test for import step."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_IMPORTED_BY: "binary_sensor", **BINARY_SENSOR_IMPORT_DATA},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "test2"
assert result["data"] == {CONF_IMPORTED_BY: "binary_sensor"}
assert result["options"] == {
"host": "127.0.0.1",
"count": 1,
}
# test import without name
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_IMPORTED_BY: "binary_sensor", "host": "10.10.10.10", "count": 5},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["title"] == "10.10.10.10"
assert result["data"] == {CONF_IMPORTED_BY: "binary_sensor"}
assert result["options"] == {
"host": "10.10.10.10",
"count": 5,
}