Add Universal Powerline Bus (#34692)

* Initial version.

* Tests.

* Refactored tests.

* Update requirements_all

* Increase test coverage. Catch exception.

* Update .coveragerc

* Fix lint msg.

* Tweak test (more to force CI build).

* Update based on PR comments.

* Change unique_id to use stable string.

* Add Universal Powerline Bus "link" support.

* Fix missed call.

* Revert botched merge.

* Update homeassistant/components/upb/light.py

Co-authored-by: J. Nick Koston <nick@koston.org>

* Three changes.

Update service schema to require one of brightness/brightness_pct.
Fix bug in setting brightness to zero.
Replace async_update_status and replace with async_update.

Co-authored-by: J. Nick Koston <nick@koston.org>

.coveragerc

@@ -793,6 +793,9 @@ omit =
     homeassistant/components/ubus/device_tracker.py
     homeassistant/components/ue_smart_radio/media_player.py
     homeassistant/components/unifiled/*
+    homeassistant/components/upb/__init__.py
+    homeassistant/components/upb/const.py
+    homeassistant/components/upb/light.py
     homeassistant/components/upcloud/*
     homeassistant/components/upnp/*
     homeassistant/components/upc_connect/*

CODEOWNERS

@@ -420,6 +420,7 @@ homeassistant/components/twilio_sms/* @robbiet480
 homeassistant/components/ubee/* @mzdrale
 homeassistant/components/unifi/* @Kane610
 homeassistant/components/unifiled/* @florisvdk
+homeassistant/components/upb/* @gwww
 homeassistant/components/upc_connect/* @pvizeli
 homeassistant/components/upcloud/* @scop
 homeassistant/components/updater/* @home-assistant/core

homeassistant/components/upb/.translations/en.json

@@ -0,0 +1,23 @@
+{
+    "config": {
+        "abort": {
+            "address_already_configured": "An UPB PIM with this address is already configured."
+        },
+        "error": {
+            "cannot_connect": "Failed to connect to UPB PIM, please try again.",
+            "invalid_upb_file": "Missing or invalid UPB UPStart export file, check the name and path of the file.",
+            "unknown": "Unexpected error."
+        },
+        "step": {
+            "user": {
+                "data": {
+                    "address": "Address (see description above)",
+                    "file_path": "Path and name of the UPStart UPB export file.",
+                    "protocol": "Protocol"
+                },
+                "description": "Connect a Universal Powerline Bus Powerline Interface Module (UPB PIM). The address string must be in the form 'address[:port]' for 'tcp'. The port is optional and defaults to 2101. Example: '192.168.1.42'. For the serial protocol, the address must be in the form 'tty[:baud]'. The baud is optional and defaults to 4800. Example: '/dev/ttyS1'.",
+                "title": "Connect to UPB PIM"
+            }
+        }
+    }
+}

homeassistant/components/upb/__init__.py

@@ -0,0 +1,122 @@
+"""Support the UPB PIM."""
+import asyncio
+
+import upb_lib
+
+from homeassistant.const import CONF_FILE_PATH, CONF_HOST
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.typing import ConfigType
+
+from .const import DOMAIN
+
+UPB_PLATFORMS = ["light"]
+
+
+async def async_setup(hass: HomeAssistant, hass_config: ConfigType) -> bool:
+    """Set up the UPB platform."""
+    return True
+
+
+async def async_setup_entry(hass, config_entry):
+    """Set up a new config_entry for UPB PIM."""
+
+    url = config_entry.data[CONF_HOST]
+    file = config_entry.data[CONF_FILE_PATH]
+
+    upb = upb_lib.UpbPim({"url": url, "UPStartExportFile": file})
+    upb.connect()
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][config_entry.entry_id] = {"upb": upb}
+
+    for component in UPB_PLATFORMS:
+        hass.async_create_task(
+            hass.config_entries.async_forward_entry_setup(config_entry, component)
+        )
+
+    return True
+
+
+async def async_unload_entry(hass, config_entry):
+    """Unload the config_entry."""
+
+    unload_ok = all(
+        await asyncio.gather(
+            *[
+                hass.config_entries.async_forward_entry_unload(config_entry, component)
+                for component in UPB_PLATFORMS
+            ]
+        )
+    )
+
+    if unload_ok:
+        upb = hass.data[DOMAIN][config_entry.entry_id]["upb"]
+        upb.disconnect()
+        hass.data[DOMAIN].pop(config_entry.entry_id)
+
+    return unload_ok
+
+
+class UpbEntity(Entity):
+    """Base class for all UPB entities."""
+
+    def __init__(self, element, unique_id, upb):
+        """Initialize the base of all UPB devices."""
+        self._upb = upb
+        self._element = element
+        element_type = "link" if element.addr.is_link else "device"
+        self._unique_id = f"{unique_id}_{element_type}_{element.addr}"
+
+    @property
+    def name(self):
+        """Name of the element."""
+        return self._element.name
+
+    @property
+    def unique_id(self):
+        """Return unique id of the element."""
+        return self._unique_id
+
+    @property
+    def should_poll(self) -> bool:
+        """Don't poll this device."""
+        return False
+
+    @property
+    def device_state_attributes(self):
+        """Return the default attributes of the element."""
+        return self._element.as_dict()
+
+    @property
+    def available(self):
+        """Is the entity available to be updated."""
+        return self._upb.is_connected()
+
+    def _element_changed(self, element, changeset):
+        pass
+
+    @callback
+    def _element_callback(self, element, changeset):
+        """Handle callback from an UPB element that has changed."""
+        self._element_changed(element, changeset)
+        self.async_write_ha_state()
+
+    async def async_added_to_hass(self):
+        """Register callback for UPB changes and update entity state."""
+        self._element.add_callback(self._element_callback)
+        self._element_callback(self._element, {})
+
+
+class UpbAttachedEntity(UpbEntity):
+    """Base class for UPB attached entities."""
+
+    @property
+    def device_info(self):
+        """Device info for the entity."""
+        return {
+            "name": self._element.name,
+            "identifiers": {(DOMAIN, self._element.index)},
+            "sw_version": self._element.version,
+            "manufacturer": self._element.manufacturer,
+            "model": self._element.product,
+        }

homeassistant/components/upb/config_flow.py

@@ -0,0 +1,140 @@
+"""Config flow for UPB PIM integration."""
+import asyncio
+import logging
+from urllib.parse import urlparse
+
+import async_timeout
+import upb_lib
+import voluptuous as vol
+
+from homeassistant import config_entries, exceptions
+from homeassistant.const import CONF_ADDRESS, CONF_FILE_PATH, CONF_HOST, CONF_PROTOCOL
+
+from .const import DOMAIN  # pylint:disable=unused-import
+
+_LOGGER = logging.getLogger(__name__)
+PROTOCOL_MAP = {"TCP": "tcp://", "Serial port": "serial://"}
+DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_PROTOCOL, default="Serial port"): vol.In(
+            ["TCP", "Serial port"]
+        ),
+        vol.Required(CONF_ADDRESS): str,
+        vol.Required(CONF_FILE_PATH, default=""): str,
+    }
+)
+VALIDATE_TIMEOUT = 15
+
+
+async def _validate_input(data):
+    """Validate the user input allows us to connect."""
+
+    def _connected_callback():
+        connected_event.set()
+
+    connected_event = asyncio.Event()
+    file_path = data.get(CONF_FILE_PATH)
+    url = _make_url_from_data(data)
+
+    upb = upb_lib.UpbPim({"url": url, "UPStartExportFile": file_path})
+    if not upb.config_ok:
+        _LOGGER.error("Missing or invalid UPB file: %s", file_path)
+        raise InvalidUpbFile
+
+    upb.connect(_connected_callback)
+
+    try:
+        with async_timeout.timeout(VALIDATE_TIMEOUT):
+            await connected_event.wait()
+    except asyncio.TimeoutError:
+        pass
+
+    upb.disconnect()
+
+    if not connected_event.is_set():
+        _LOGGER.error(
+            "Timed out after %d seconds trying to connect with UPB PIM at %s",
+            VALIDATE_TIMEOUT,
+            url,
+        )
+        raise CannotConnect
+
+    # Return info that you want to store in the config entry.
+    return (upb.network_id, {"title": "UPB", CONF_HOST: url, CONF_FILE_PATH: file_path})
+
+
+def _make_url_from_data(data):
+    host = data.get(CONF_HOST)
+    if host:
+        return host
+
+    protocol = PROTOCOL_MAP[data[CONF_PROTOCOL]]
+    address = data[CONF_ADDRESS]
+    return f"{protocol}{address}"
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for UPB PIM."""
+
+    VERSION = 1
+    CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
+
+    def __init__(self):
+        """Initialize the UPB config flow."""
+        self.importing = False
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial step."""
+        errors = {}
+        if user_input is not None:
+            try:
+                if self._url_already_configured(_make_url_from_data(user_input)):
+                    return self.async_abort(reason="address_already_configured")
+                network_id, info = await _validate_input(user_input)
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidUpbFile:
+                errors["base"] = "invalid_upb_file"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+            if "base" not in errors:
+                await self.async_set_unique_id(network_id)
+                self._abort_if_unique_id_configured()
+
+                if self.importing:
+                    return self.async_create_entry(title=info["title"], data=user_input)
+
+                return self.async_create_entry(
+                    title=info["title"],
+                    data={
+                        CONF_HOST: info[CONF_HOST],
+                        CONF_FILE_PATH: user_input[CONF_FILE_PATH],
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_import(self, user_input):
+        """Handle import."""
+        self.importing = True
+        return await self.async_step_user(user_input)
+
+    def _url_already_configured(self, url):
+        """See if we already have a UPB PIM matching user input configured."""
+        existing_hosts = {
+            urlparse(entry.data[CONF_HOST]).hostname
+            for entry in self._async_current_entries()
+        }
+        return urlparse(url).hostname in existing_hosts
+
+
+class CannotConnect(exceptions.HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidUpbFile(exceptions.HomeAssistantError):
+    """Error to indicate there is invalid or missing UPB config file."""

homeassistant/components/upb/const.py

@@ -0,0 +1,33 @@
+"""Support the UPB PIM."""
+
+import voluptuous as vol
+
+import homeassistant.helpers.config_validation as cv
+
+CONF_NETWORK = "network"
+DOMAIN = "upb"
+
+ATTR_BLINK_RATE = "blink_rate"
+ATTR_BRIGHTNESS = "brightness"
+ATTR_BRIGHTNESS_PCT = "brightness_pct"
+ATTR_RATE = "rate"
+VALID_BRIGHTNESS = vol.All(vol.Coerce(int), vol.Clamp(min=0, max=255))
+VALID_BRIGHTNESS_PCT = vol.All(vol.Coerce(float), vol.Range(min=0, max=100))
+VALID_RATE = vol.All(vol.Coerce(float), vol.Clamp(min=-1, max=3600))
+
+UPB_BRIGHTNESS_RATE_SCHEMA = vol.All(
+    cv.has_at_least_one_key(ATTR_BRIGHTNESS, ATTR_BRIGHTNESS_PCT),
+    cv.make_entity_service_schema(
+        {
+            vol.Exclusive(ATTR_BRIGHTNESS, ATTR_BRIGHTNESS): VALID_BRIGHTNESS,
+            vol.Exclusive(ATTR_BRIGHTNESS_PCT, ATTR_BRIGHTNESS): VALID_BRIGHTNESS_PCT,
+            vol.Optional(ATTR_RATE, default=-1): VALID_RATE,
+        }
+    ),
+)
+
+UPB_BLINK_RATE_SCHEMA = {
+    vol.Required(ATTR_BLINK_RATE, default=0.5): vol.All(
+        vol.Coerce(float), vol.Range(min=0, max=4.25)
+    )
+}

homeassistant/components/upb/light.py

@@ -0,0 +1,104 @@
+"""Platform for UPB light integration."""
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    ATTR_FLASH,
+    ATTR_TRANSITION,
+    SUPPORT_BRIGHTNESS,
+    SUPPORT_FLASH,
+    SUPPORT_TRANSITION,
+    Light,
+)
+from homeassistant.helpers import entity_platform
+
+from . import UpbAttachedEntity
+from .const import DOMAIN, UPB_BLINK_RATE_SCHEMA, UPB_BRIGHTNESS_RATE_SCHEMA
+
+SERVICE_LIGHT_FADE_START = "light_fade_start"
+SERVICE_LIGHT_FADE_STOP = "light_fade_stop"
+SERVICE_LIGHT_BLINK = "light_blink"
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up the UPB light based on a config entry."""
+
+    upb = hass.data[DOMAIN][config_entry.entry_id]["upb"]
+    unique_id = config_entry.entry_id
+    async_add_entities(
+        UpbLight(upb.devices[dev], unique_id, upb) for dev in upb.devices
+    )
+
+    platform = entity_platform.current_platform.get()
+
+    platform.async_register_entity_service(
+        SERVICE_LIGHT_FADE_START, UPB_BRIGHTNESS_RATE_SCHEMA, "async_light_fade_start"
+    )
+    platform.async_register_entity_service(
+        SERVICE_LIGHT_FADE_STOP, {}, "async_light_fade_stop"
+    )
+    platform.async_register_entity_service(
+        SERVICE_LIGHT_BLINK, UPB_BLINK_RATE_SCHEMA, "async_light_blink"
+    )
+
+
+class UpbLight(UpbAttachedEntity, Light):
+    """Representation of an UPB Light."""
+
+    def __init__(self, element, unique_id, upb):
+        """Initialize an UpbLight."""
+        super().__init__(element, unique_id, upb)
+        self._brightness = self._element.status
+
+    @property
+    def supported_features(self):
+        """Flag supported features."""
+        if self._element.dimmable:
+            return SUPPORT_BRIGHTNESS | SUPPORT_TRANSITION | SUPPORT_FLASH
+        return SUPPORT_FLASH
+
+    @property
+    def brightness(self):
+        """Get the brightness."""
+        return self._brightness
+
+    @property
+    def is_on(self) -> bool:
+        """Get the current brightness."""
+        return self._brightness != 0
+
+    async def async_turn_on(self, **kwargs):
+        """Turn on the light."""
+        flash = kwargs.get(ATTR_FLASH)
+        if flash:
+            await self.async_light_blink(0.5 if flash == "short" else 1.5)
+        else:
+            rate = kwargs.get(ATTR_TRANSITION, -1)
+            brightness = kwargs.get(ATTR_BRIGHTNESS, 255) / 2.55
+            self._element.turn_on(brightness, rate)
+
+    async def async_turn_off(self, **kwargs):
+        """Turn off the device."""
+        rate = kwargs.get(ATTR_TRANSITION, -1)
+        self._element.turn_off(rate)
+
+    async def async_light_fade_start(self, rate, brightness=None, brightness_pct=None):
+        """Start dimming of device."""
+        if brightness is not None:
+            brightness_pct = brightness / 2.55
+        self._element.fade_start(brightness_pct, rate)
+
+    async def async_light_fade_stop(self):
+        """Stop dimming of device."""
+        self._element.fade_stop()
+
+    async def async_light_blink(self, blink_rate):
+        """Request device to blink."""
+        blink_rate = int(blink_rate * 60)  # Convert seconds to 60 hz pulses
+        self._element.blink(blink_rate)
+
+    async def async_update(self):
+        """Request the device to update its status."""
+        self._element.update_status()
+
+    def _element_changed(self, element, changeset):
+        status = self._element.status
+        self._brightness = round(status * 2.55) if status else 0

homeassistant/components/upb/manifest.json

@@ -0,0 +1,8 @@
+{
+  "domain": "upb",
+  "name": "Universal Powerline Bus (UPB)",
+  "documentation": "https://www.home-assistant.io/integrations/upb",
+  "requirements": ["upb_lib==0.4.10"],
+  "codeowners": ["@gwww"],
+  "config_flow": true
+}

homeassistant/components/upb/services.yaml

@@ -0,0 +1,32 @@
+light_fade_start:
+  description: Start fading a light either up or down from current brightness.
+  fields:
+    entity_id:
+      description: Name(s) of lights to start fading
+      example: "light.kitchen"
+    brightness:
+      description: Number between 0 and 255 indicating brightness, where 0 turns the light off, 1 is the minimum brightness and 255 is the maximum brightness.
+      example: 142
+    brightness_pct:
+      description: Number between 0 and 100 indicating percentage of full brightness, where 0 turns the light off, 1 is the minimum brightness and 100 is the maximum brightness.
+      example: 42
+    rate:
+      description: Rate for light to transition to new brightness
+      example: 3
+
+light_fade_stop:
+  description: Stop a light fade.
+  fields:
+    entity_id:
+      description: Name(s) of lights to stop fadding
+      example: "light.kitchen, light.family_room"
+
+light_blink:
+  description: Blink a light
+  fields:
+    entity_id:
+      description: Name(s) of lights to start fading
+      example: "light.kitchen"
+    rate:
+      description: Number of seconds between 0 and 4.25 that the link flashes on.
+      example: 4.2

homeassistant/components/upb/strings.json

@@ -0,0 +1,23 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "title": "Connect to UPB PIM",
+        "description": "Connect a Universal Powerline Bus Powerline Interface Module (UPB PIM). The address string must be in the form 'address[:port]' for 'tcp'. The port is optional and defaults to 2101. Example: '192.168.1.42'. For the serial protocol, the address must be in the form 'tty[:baud]'. The baud is optional and defaults to 4800. Example: '/dev/ttyS1'.",
+        "data": {
+          "protocol": "Protocol",
+          "address": "Address (see description above)",
+          "file_path": "Path and name of the UPStart UPB export file."
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "Failed to connect to UPB PIM, please try again.",
+      "invalid_upb_file": "Missing or invalid UPB UPStart export file, check the name and path of the file.",
+      "unknown": "Unexpected error."
+    },
+    "abort": {
+      "address_already_configured": "An UPB PIM with this address is already configured."
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -141,6 +141,7 @@ FLOWS = [
     "twentemilieu",
     "twilio",
     "unifi",
+    "upb",
     "upnp",
     "velbus",
     "vera",

requirements_all.txt

@@ -2101,6 +2101,9 @@ uEagle==0.0.1
 # homeassistant.components.unifiled
 unifiled==0.11
 
+# homeassistant.components.upb
+upb_lib==0.4.10
+
 # homeassistant.components.upcloud
 upcloud-api==0.4.5
 

requirements_test_all.txt

@@ -822,6 +822,9 @@ twentemilieu==0.3.0
 # homeassistant.components.twilio
 twilio==6.32.0
 
+# homeassistant.components.upb
+upb_lib==0.4.10
+
 # homeassistant.components.huawei_lte
 url-normalize==1.4.1
 

tests/components/upb/__init__.py

@@ -0,0 +1 @@
+"""Tests for the UPB integration."""

tests/components/upb/test_config_flow.py

@@ -0,0 +1,152 @@
+"""Test the UPB Control config flow."""
+
+from asynctest import MagicMock, PropertyMock, patch
+
+from homeassistant import config_entries, setup
+from homeassistant.components.upb.const import DOMAIN
+
+
+def mocked_upb(sync_complete=True, config_ok=True):
+    """Mock UPB lib."""
+
+    def _upb_lib_connect(callback):
+        callback()
+
+    upb_mock = MagicMock()
+    type(upb_mock).network_id = PropertyMock(return_value="42")
+    type(upb_mock).config_ok = PropertyMock(return_value=config_ok)
+    if sync_complete:
+        upb_mock.connect.side_effect = _upb_lib_connect
+    return patch(
+        "homeassistant.components.upb.config_flow.upb_lib.UpbPim", return_value=upb_mock
+    )
+
+
+async def valid_tcp_flow(hass, sync_complete=True, config_ok=True):
+    """Get result dict that are standard for most tests."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+    with mocked_upb(sync_complete, config_ok), patch(
+        "homeassistant.components.upb.async_setup_entry", return_value=True
+    ):
+        flow = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+        result = await hass.config_entries.flow.async_configure(
+            flow["flow_id"],
+            {"protocol": "TCP", "address": "1.2.3.4", "file_path": "upb.upe"},
+        )
+    return result
+
+
+async def test_full_upb_flow_with_serial_port(hass):
+    """Test a full UPB config flow with serial port."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    with mocked_upb(), patch(
+        "homeassistant.components.upb.async_setup", return_value=True
+    ) as mock_setup, patch(
+        "homeassistant.components.upb.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        flow = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+
+        result = await hass.config_entries.flow.async_configure(
+            flow["flow_id"],
+            {
+                "protocol": "Serial port",
+                "address": "/dev/ttyS0:115200",
+                "file_path": "upb.upe",
+            },
+        )
+
+    assert flow["type"] == "form"
+    assert flow["errors"] == {}
+    assert result["type"] == "create_entry"
+    assert result["title"] == "UPB"
+    assert result["data"] == {
+        "host": "serial:///dev/ttyS0:115200",
+        "file_path": "upb.upe",
+    }
+    await hass.async_block_till_done()
+    assert len(mock_setup.mock_calls) == 1
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_form_user_with_tcp_upb(hass):
+    """Test we can setup a serial upb."""
+    result = await valid_tcp_flow(hass)
+    assert result["type"] == "create_entry"
+    assert result["data"] == {"host": "tcp://1.2.3.4", "file_path": "upb.upe"}
+    await hass.async_block_till_done()
+
+
+async def test_form_cannot_connect(hass):
+    """Test we handle cannot connect error."""
+    from asyncio import TimeoutError
+
+    with patch(
+        "homeassistant.components.upb.config_flow.async_timeout.timeout",
+        side_effect=TimeoutError,
+    ):
+        result = await valid_tcp_flow(hass, sync_complete=False)
+
+    assert result["type"] == "form"
+    assert result["errors"] == {"base": "cannot_connect"}
+
+
+async def test_form_missing_upb_file(hass):
+    """Test we handle cannot connect error."""
+    result = await valid_tcp_flow(hass, config_ok=False)
+    assert result["type"] == "form"
+    assert result["errors"] == {"base": "invalid_upb_file"}
+
+
+async def test_form_user_with_already_configured(hass):
+    """Test we can setup a TCP upb."""
+    _ = await valid_tcp_flow(hass)
+    result2 = await valid_tcp_flow(hass)
+    assert result2["type"] == "abort"
+    assert result2["reason"] == "address_already_configured"
+    await hass.async_block_till_done()
+
+
+async def test_form_import(hass):
+    """Test we get the form with import source."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    with mocked_upb(), patch(
+        "homeassistant.components.upb.async_setup", return_value=True
+    ) as mock_setup, patch(
+        "homeassistant.components.upb.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_IMPORT},
+            data={"host": "tcp://42.4.2.42", "file_path": "upb.upe"},
+        )
+
+    assert result["type"] == "create_entry"
+    assert result["title"] == "UPB"
+
+    assert result["data"] == {"host": "tcp://42.4.2.42", "file_path": "upb.upe"}
+    await hass.async_block_till_done()
+    assert len(mock_setup.mock_calls) == 1
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+async def test_form_junk_input(hass):
+    """Test we get the form with import source."""
+    await setup.async_setup_component(hass, "persistent_notification", {})
+
+    with mocked_upb():
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN,
+            context={"source": config_entries.SOURCE_IMPORT},
+            data={"foo": "goo", "goo": "foo"},
+        )
+
+    assert result["type"] == "form"
+    assert result["errors"] == {"base": "unknown"}
+
+    await hass.async_block_till_done()