Add Internet Printing Protocol (IPP) integration (#32859)

* Create __init__.py

* Create manifest.json

* Update zeroconf.py

* more work on integration

* more work on integration.

* add more sensor tests.

* Update const.py

* Update sensor.py

* more work on ipp.

* Update test_config_flow.py

* more work on ipp.

* Update config_flow.py

* Update config_flow.py

CODEOWNERS

@@ -187,6 +187,7 @@ homeassistant/components/intesishome/* @jnimmo
 homeassistant/components/ios/* @robbiet480
 homeassistant/components/iperf3/* @rohankapoorcom
 homeassistant/components/ipma/* @dgomes @abmantis
+homeassistant/components/ipp/* @ctalkington
 homeassistant/components/iqvia/* @bachya
 homeassistant/components/irish_rail_transport/* @ttroy50
 homeassistant/components/izone/* @Swamp-Ig

homeassistant/components/ipp/.translations/en.json

@@ -0,0 +1,32 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "This printer is already configured.",
+            "connection_error": "Failed to connect to printer.",
+            "connection_upgrade": "Failed to connect to printer due to connection upgrade being required."
+        },
+        "error": {
+            "connection_error": "Failed to connect to printer.",
+            "connection_upgrade": "Failed to connect to printer. Please try again with SSL/TLS option checked."
+        },
+        "flow_title": "Printer: {name}",
+        "step": {
+            "user": {
+                "data": {
+                    "base_path": "Relative path to the printer",
+                    "host": "Host or IP address",
+                    "port": "Port",
+                    "ssl": "Printer supports communication over SSL/TLS",
+                    "verify_ssl": "Printer uses a proper SSL certificate"
+                },
+                "description": "Set up your printer via Internet Printing Protocol (IPP) to integrate with Home Assistant.",
+                "title": "Link your printer"
+            },
+            "zeroconf_confirm": {
+                "description": "Do you want to add the printer named `{name}` to Home Assistant?",
+                "title": "Discovered printer"
+            }
+        },
+        "title": "Internet Printing Protocol (IPP)"
+    }
+}

homeassistant/components/ipp/__init__.py

@@ -0,0 +1,190 @@
+"""The Internet Printing Protocol (IPP) integration."""
+import asyncio
+from datetime import timedelta
+import logging
+from typing import Any, Dict
+
+from pyipp import IPP, IPPError, Printer as IPPPrinter
+
+from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    ATTR_NAME,
+    CONF_HOST,
+    CONF_PORT,
+    CONF_SSL,
+    CONF_VERIFY_SSL,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+    ATTR_IDENTIFIERS,
+    ATTR_MANUFACTURER,
+    ATTR_MODEL,
+    ATTR_SOFTWARE_VERSION,
+    CONF_BASE_PATH,
+    DOMAIN,
+)
+
+PLATFORMS = [SENSOR_DOMAIN]
+SCAN_INTERVAL = timedelta(seconds=60)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup(hass: HomeAssistant, config: Dict) -> bool:
+    """Set up the IPP component."""
+    hass.data.setdefault(DOMAIN, {})
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up IPP from a config entry."""
+
+    # Create IPP instance for this entry
+    coordinator = IPPDataUpdateCoordinator(
+        hass,
+        host=entry.data[CONF_HOST],
+        port=entry.data[CONF_PORT],
+        base_path=entry.data[CONF_BASE_PATH],
+        tls=entry.data[CONF_SSL],
+        verify_ssl=entry.data[CONF_VERIFY_SSL],
+    )
+    await coordinator.async_refresh()
+
+    if not coordinator.last_update_success:
+        raise ConfigEntryNotReady
+
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+
+    for component in PLATFORMS:
+        hass.async_create_task(
+            hass.config_entries.async_forward_entry_setup(entry, component)
+        )
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = all(
+        await asyncio.gather(
+            *[
+                hass.config_entries.async_forward_entry_unload(entry, component)
+                for component in PLATFORMS
+            ]
+        )
+    )
+
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok
+
+
+class IPPDataUpdateCoordinator(DataUpdateCoordinator):
+    """Class to manage fetching IPP data from single endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        *,
+        host: str,
+        port: int,
+        base_path: str,
+        tls: bool,
+        verify_ssl: bool,
+    ):
+        """Initialize global IPP data updater."""
+        self.ipp = IPP(
+            host=host,
+            port=port,
+            base_path=base_path,
+            tls=tls,
+            verify_ssl=verify_ssl,
+            session=async_get_clientsession(hass, verify_ssl),
+        )
+
+        super().__init__(
+            hass, _LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL,
+        )
+
+    async def _async_update_data(self) -> IPPPrinter:
+        """Fetch data from IPP."""
+        try:
+            return await self.ipp.printer()
+        except IPPError as error:
+            raise UpdateFailed(f"Invalid response from API: {error}")
+
+
+class IPPEntity(Entity):
+    """Defines a base IPP entity."""
+
+    def __init__(
+        self,
+        *,
+        entry_id: str,
+        coordinator: IPPDataUpdateCoordinator,
+        name: str,
+        icon: str,
+        enabled_default: bool = True,
+    ) -> None:
+        """Initialize the IPP entity."""
+        self._enabled_default = enabled_default
+        self._entry_id = entry_id
+        self._icon = icon
+        self._name = name
+        self._unsub_dispatcher = None
+        self.coordinator = coordinator
+
+    @property
+    def name(self) -> str:
+        """Return the name of the entity."""
+        return self._name
+
+    @property
+    def icon(self) -> str:
+        """Return the mdi icon of the entity."""
+        return self._icon
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return self.coordinator.last_update_success
+
+    @property
+    def entity_registry_enabled_default(self) -> bool:
+        """Return if the entity should be enabled when first added to the entity registry."""
+        return self._enabled_default
+
+    @property
+    def should_poll(self) -> bool:
+        """Return the polling requirement of the entity."""
+        return False
+
+    async def async_added_to_hass(self) -> None:
+        """Connect to dispatcher listening for entity data notifications."""
+        self.coordinator.async_add_listener(self.async_write_ha_state)
+
+    async def async_will_remove_from_hass(self) -> None:
+        """Disconnect from update signal."""
+        self.coordinator.async_remove_listener(self.async_write_ha_state)
+
+    async def async_update(self) -> None:
+        """Update an IPP entity."""
+        await self.coordinator.async_request_refresh()
+
+    @property
+    def device_info(self) -> Dict[str, Any]:
+        """Return device information about this IPP device."""
+        return {
+            ATTR_IDENTIFIERS: {(DOMAIN, self.coordinator.data.info.uuid)},
+            ATTR_NAME: self.coordinator.data.info.name,
+            ATTR_MANUFACTURER: self.coordinator.data.info.manufacturer,
+            ATTR_MODEL: self.coordinator.data.info.model,
+            ATTR_SOFTWARE_VERSION: self.coordinator.data.info.version,
+        }

homeassistant/components/ipp/config_flow.py

@@ -0,0 +1,144 @@
+"""Config flow to configure the IPP integration."""
+import logging
+from typing import Any, Dict, Optional
+
+from pyipp import IPP, IPPConnectionError, IPPConnectionUpgradeRequired
+import voluptuous as vol
+
+from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_NAME,
+    CONF_PORT,
+    CONF_SSL,
+    CONF_VERIFY_SSL,
+)
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.typing import ConfigType, HomeAssistantType
+
+from .const import CONF_BASE_PATH, CONF_UUID
+from .const import DOMAIN  # pylint: disable=unused-import
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def validate_input(hass: HomeAssistantType, data: dict) -> Dict[str, Any]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from DATA_SCHEMA with values provided by the user.
+    """
+    session = async_get_clientsession(hass)
+    ipp = IPP(
+        host=data[CONF_HOST],
+        port=data[CONF_PORT],
+        base_path=data[CONF_BASE_PATH],
+        tls=data[CONF_SSL],
+        verify_ssl=data[CONF_VERIFY_SSL],
+        session=session,
+    )
+
+    printer = await ipp.printer()
+
+    return {CONF_UUID: printer.info.uuid}
+
+
+class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle an IPP config flow."""
+
+    VERSION = 1
+    CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL
+
+    def __init__(self):
+        """Set up the instance."""
+        self.discovery_info = {}
+
+    async def async_step_user(
+        self, user_input: Optional[ConfigType] = None
+    ) -> Dict[str, Any]:
+        """Handle a flow initiated by the user."""
+        if user_input is None:
+            return self._show_setup_form()
+
+        try:
+            info = await validate_input(self.hass, user_input)
+        except IPPConnectionUpgradeRequired:
+            return self._show_setup_form({"base": "connection_upgrade"})
+        except IPPConnectionError:
+            return self._show_setup_form({"base": "connection_error"})
+        user_input[CONF_UUID] = info[CONF_UUID]
+
+        await self.async_set_unique_id(user_input[CONF_UUID])
+        self._abort_if_unique_id_configured(updates={CONF_HOST: user_input[CONF_HOST]})
+
+        return self.async_create_entry(title=user_input[CONF_HOST], data=user_input)
+
+    async def async_step_zeroconf(self, discovery_info: ConfigType) -> Dict[str, Any]:
+        """Handle zeroconf discovery."""
+        # Hostname is format: EPSON123456.local.
+        host = discovery_info["hostname"].rstrip(".")
+        port = discovery_info["port"]
+        name, _ = host.rsplit(".")
+        tls = discovery_info["type"] == "_ipps._tcp.local."
+
+        # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+        self.context.update({"title_placeholders": {"name": name}})
+
+        self.discovery_info.update(
+            {
+                CONF_HOST: host,
+                CONF_PORT: port,
+                CONF_SSL: tls,
+                CONF_VERIFY_SSL: False,
+                CONF_BASE_PATH: "/"
+                + discovery_info["properties"].get("rp", "ipp/print"),
+                CONF_NAME: name,
+                CONF_UUID: discovery_info["properties"].get("UUID"),
+            }
+        )
+
+        try:
+            info = await validate_input(self.hass, self.discovery_info)
+        except IPPConnectionUpgradeRequired:
+            return self.async_abort(reason="connection_upgrade")
+        except IPPConnectionError:
+            return self.async_abort(reason="connection_error")
+
+        self.discovery_info[CONF_UUID] = info[CONF_UUID]
+
+        await self.async_set_unique_id(self.discovery_info[CONF_UUID])
+        self._abort_if_unique_id_configured(
+            updates={CONF_HOST: self.discovery_info[CONF_HOST]}
+        )
+
+        return await self.async_step_zeroconf_confirm()
+
+    async def async_step_zeroconf_confirm(
+        self, user_input: ConfigType = None
+    ) -> Dict[str, Any]:
+        """Handle a confirmation flow initiated by zeroconf."""
+        if user_input is None:
+            return self.async_show_form(
+                step_id="zeroconf_confirm",
+                description_placeholders={"name": self.discovery_info[CONF_NAME]},
+                errors={},
+            )
+
+        return self.async_create_entry(
+            title=self.discovery_info[CONF_NAME], data=self.discovery_info,
+        )
+
+    def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
+        """Show the setup form to the user."""
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_HOST): str,
+                    vol.Required(CONF_PORT, default=631): int,
+                    vol.Required(CONF_BASE_PATH, default="/ipp/print"): str,
+                    vol.Required(CONF_SSL, default=False): bool,
+                    vol.Required(CONF_VERIFY_SSL, default=False): bool,
+                }
+            ),
+            errors=errors or {},
+        )

homeassistant/components/ipp/const.py

@@ -0,0 +1,25 @@
+"""Constants for the IPP integration."""
+
+# Integration domain
+DOMAIN = "ipp"
+
+# Attributes
+ATTR_COMMAND_SET = "command_set"
+ATTR_IDENTIFIERS = "identifiers"
+ATTR_INFO = "info"
+ATTR_LOCATION = "location"
+ATTR_MANUFACTURER = "manufacturer"
+ATTR_MARKER_TYPE = "marker_type"
+ATTR_MARKER_LOW_LEVEL = "marker_low_level"
+ATTR_MARKER_HIGH_LEVEL = "marker_high_level"
+ATTR_MODEL = "model"
+ATTR_SERIAL = "serial"
+ATTR_SOFTWARE_VERSION = "sw_version"
+ATTR_STATE_MESSAGE = "state_message"
+ATTR_STATE_REASON = "state_reason"
+ATTR_URI_SUPPORTED = "uri_supported"
+
+# Config Keys
+CONF_BASE_PATH = "base_path"
+CONF_TLS = "tls"
+CONF_UUID = "uuid"

homeassistant/components/ipp/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "ipp",
+  "name": "Internet Printing Protocol (IPP)",
+  "documentation": "https://www.home-assistant.io/integrations/ipp",
+  "requirements": ["pyipp==0.8.1"],
+  "dependencies": [],
+  "codeowners": ["@ctalkington"],
+  "config_flow": true,
+  "quality_scale": "platinum",
+  "zeroconf": ["_ipps._tcp.local.", "_ipp._tcp.local."]
+}

homeassistant/components/ipp/sensor.py

@@ -0,0 +1,178 @@
+"""Support for IPP sensors."""
+from datetime import timedelta
+from typing import Any, Callable, Dict, List, Optional, Union
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import DEVICE_CLASS_TIMESTAMP, UNIT_PERCENTAGE
+from homeassistant.helpers.entity import Entity
+from homeassistant.helpers.typing import HomeAssistantType
+from homeassistant.util.dt import utcnow
+
+from . import IPPDataUpdateCoordinator, IPPEntity
+from .const import (
+    ATTR_COMMAND_SET,
+    ATTR_INFO,
+    ATTR_LOCATION,
+    ATTR_MARKER_HIGH_LEVEL,
+    ATTR_MARKER_LOW_LEVEL,
+    ATTR_MARKER_TYPE,
+    ATTR_SERIAL,
+    ATTR_STATE_MESSAGE,
+    ATTR_STATE_REASON,
+    ATTR_URI_SUPPORTED,
+    DOMAIN,
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistantType,
+    entry: ConfigEntry,
+    async_add_entities: Callable[[List[Entity], bool], None],
+) -> None:
+    """Set up IPP sensor based on a config entry."""
+    coordinator: IPPDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+
+    sensors = []
+
+    sensors.append(IPPPrinterSensor(entry.entry_id, coordinator))
+    sensors.append(IPPUptimeSensor(entry.entry_id, coordinator))
+
+    for marker_index in range(len(coordinator.data.markers)):
+        sensors.append(IPPMarkerSensor(entry.entry_id, coordinator, marker_index))
+
+    async_add_entities(sensors, True)
+
+
+class IPPSensor(IPPEntity):
+    """Defines an IPP sensor."""
+
+    def __init__(
+        self,
+        *,
+        coordinator: IPPDataUpdateCoordinator,
+        enabled_default: bool = True,
+        entry_id: str,
+        icon: str,
+        key: str,
+        name: str,
+        unit_of_measurement: Optional[str] = None,
+    ) -> None:
+        """Initialize IPP sensor."""
+        self._unit_of_measurement = unit_of_measurement
+        self._key = key
+
+        super().__init__(
+            entry_id=entry_id,
+            coordinator=coordinator,
+            name=name,
+            icon=icon,
+            enabled_default=enabled_default,
+        )
+
+    @property
+    def unique_id(self) -> str:
+        """Return the unique ID for this sensor."""
+        return f"{self.coordinator.data.info.uuid}_{self._key}"
+
+    @property
+    def unit_of_measurement(self) -> str:
+        """Return the unit this state is expressed in."""
+        return self._unit_of_measurement
+
+
+class IPPMarkerSensor(IPPSensor):
+    """Defines an IPP marker sensor."""
+
+    def __init__(
+        self, entry_id: str, coordinator: IPPDataUpdateCoordinator, marker_index: int
+    ) -> None:
+        """Initialize IPP marker sensor."""
+        self.marker_index = marker_index
+
+        super().__init__(
+            coordinator=coordinator,
+            entry_id=entry_id,
+            icon="mdi:water",
+            key=f"marker_{marker_index}",
+            name=f"{coordinator.data.info.name} {coordinator.data.markers[marker_index].name}",
+            unit_of_measurement=UNIT_PERCENTAGE,
+        )
+
+    @property
+    def device_state_attributes(self) -> Optional[Dict[str, Any]]:
+        """Return the state attributes of the entity."""
+        return {
+            ATTR_MARKER_HIGH_LEVEL: self.coordinator.data.markers[
+                self.marker_index
+            ].high_level,
+            ATTR_MARKER_LOW_LEVEL: self.coordinator.data.markers[
+                self.marker_index
+            ].low_level,
+            ATTR_MARKER_TYPE: self.coordinator.data.markers[
+                self.marker_index
+            ].marker_type,
+        }
+
+    @property
+    def state(self) -> Union[None, str, int, float]:
+        """Return the state of the sensor."""
+        return self.coordinator.data.markers[self.marker_index].level
+
+
+class IPPPrinterSensor(IPPSensor):
+    """Defines an IPP printer sensor."""
+
+    def __init__(self, entry_id: str, coordinator: IPPDataUpdateCoordinator) -> None:
+        """Initialize IPP printer sensor."""
+        super().__init__(
+            coordinator=coordinator,
+            entry_id=entry_id,
+            icon="mdi:printer",
+            key="printer",
+            name=coordinator.data.info.name,
+            unit_of_measurement=None,
+        )
+
+    @property
+    def device_state_attributes(self) -> Optional[Dict[str, Any]]:
+        """Return the state attributes of the entity."""
+        return {
+            ATTR_INFO: self.coordinator.data.info.printer_info,
+            ATTR_SERIAL: self.coordinator.data.info.serial,
+            ATTR_LOCATION: self.coordinator.data.info.location,
+            ATTR_STATE_MESSAGE: self.coordinator.data.state.message,
+            ATTR_STATE_REASON: self.coordinator.data.state.reasons,
+            ATTR_COMMAND_SET: self.coordinator.data.info.command_set,
+            ATTR_URI_SUPPORTED: self.coordinator.data.info.printer_uri_supported,
+        }
+
+    @property
+    def state(self) -> Union[None, str, int, float]:
+        """Return the state of the sensor."""
+        return self.coordinator.data.state.printer_state
+
+
+class IPPUptimeSensor(IPPSensor):
+    """Defines a IPP uptime sensor."""
+
+    def __init__(self, entry_id: str, coordinator: IPPDataUpdateCoordinator) -> None:
+        """Initialize IPP uptime sensor."""
+        super().__init__(
+            coordinator=coordinator,
+            enabled_default=False,
+            entry_id=entry_id,
+            icon="mdi:clock-outline",
+            key="uptime",
+            name=f"{coordinator.data.info.name} Uptime",
+        )
+
+    @property
+    def state(self) -> Union[None, str, int, float]:
+        """Return the state of the sensor."""
+        uptime = utcnow() - timedelta(seconds=self.coordinator.data.info.uptime)
+        return uptime.replace(microsecond=0).isoformat()
+
+    @property
+    def device_class(self) -> Optional[str]:
+        """Return the class of this sensor."""
+        return DEVICE_CLASS_TIMESTAMP

homeassistant/components/ipp/strings.json

@@ -0,0 +1,32 @@
+{
+  "config": {
+    "title": "Internet Printing Protocol (IPP)",
+    "flow_title": "Printer: {name}",
+    "step": {
+      "user": {
+        "title": "Link your printer",
+        "description": "Set up your printer via Internet Printing Protocol (IPP) to integrate with Home Assistant.",
+        "data": {
+          "host": "Host or IP address",
+          "port": "Port",
+          "base_path": "Relative path to the printer",
+          "ssl": "Printer supports communication over SSL/TLS",
+          "verify_ssl": "Printer uses a proper SSL certificate"
+        }
+      },
+      "zeroconf_confirm": {
+        "description": "Do you want to add the printer named `{name}` to Home Assistant?",
+        "title": "Discovered printer"
+      }
+    },
+    "error": {
+      "connection_error": "Failed to connect to printer.",
+      "connection_upgrade": "Failed to connect to printer. Please try again with SSL/TLS option checked."
+    },
+    "abort": {
+      "already_configured": "This printer is already configured.",
+      "connection_error": "Failed to connect to printer.",
+      "connection_upgrade": "Failed to connect to printer due to connection upgrade being required."
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -54,6 +54,7 @@ FLOWS = [
     "ifttt",
     "ios",
     "ipma",
+    "ipp",
     "iqvia",
     "izone",
     "konnected",

homeassistant/generated/zeroconf.py

@@ -25,6 +25,12 @@ ZEROCONF = {
     "_hap._tcp.local.": [
         "homekit_controller"
     ],
+    "_ipp._tcp.local.": [
+        "ipp"
+    ],
+    "_ipps._tcp.local.": [
+        "ipp"
+    ],
     "_printer._tcp.local.": [
         "brother"
     ],

requirements_all.txt

@@ -1335,6 +1335,9 @@ pyintesishome==1.7.1
 # homeassistant.components.ipma
 pyipma==2.0.5
 
+# homeassistant.components.ipp
+pyipp==0.8.1
+
 # homeassistant.components.iqvia
 pyiqvia==0.2.1
 

requirements_test_all.txt

@@ -518,6 +518,9 @@ pyicloud==0.9.6.1
 # homeassistant.components.ipma
 pyipma==2.0.5
 
+# homeassistant.components.ipp
+pyipp==0.8.1
+
 # homeassistant.components.iqvia
 pyiqvia==0.2.1
 

tests/components/ipp/__init__.py

@@ -0,0 +1,95 @@
+"""Tests for the IPP integration."""
+import os
+
+from homeassistant.components.ipp.const import CONF_BASE_PATH, CONF_UUID, DOMAIN
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_NAME,
+    CONF_PORT,
+    CONF_SSL,
+    CONF_TYPE,
+    CONF_VERIFY_SSL,
+)
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+ATTR_HOSTNAME = "hostname"
+ATTR_PROPERTIES = "properties"
+
+IPP_ZEROCONF_SERVICE_TYPE = "_ipp._tcp.local."
+IPPS_ZEROCONF_SERVICE_TYPE = "_ipps._tcp.local."
+
+ZEROCONF_NAME = "EPSON123456"
+ZEROCONF_HOST = "1.2.3.4"
+ZEROCONF_HOSTNAME = "EPSON123456.local."
+ZEROCONF_PORT = 631
+
+
+MOCK_USER_INPUT = {
+    CONF_HOST: "EPSON123456.local",
+    CONF_PORT: 361,
+    CONF_SSL: False,
+    CONF_VERIFY_SSL: False,
+    CONF_BASE_PATH: "/ipp/print",
+}
+
+MOCK_ZEROCONF_IPP_SERVICE_INFO = {
+    CONF_TYPE: IPP_ZEROCONF_SERVICE_TYPE,
+    CONF_NAME: ZEROCONF_NAME,
+    CONF_HOST: ZEROCONF_HOST,
+    ATTR_HOSTNAME: ZEROCONF_HOSTNAME,
+    CONF_PORT: ZEROCONF_PORT,
+    ATTR_PROPERTIES: {"rp": "ipp/print"},
+}
+
+MOCK_ZEROCONF_IPPS_SERVICE_INFO = {
+    CONF_TYPE: IPPS_ZEROCONF_SERVICE_TYPE,
+    CONF_NAME: ZEROCONF_NAME,
+    CONF_HOST: ZEROCONF_HOST,
+    ATTR_HOSTNAME: ZEROCONF_HOSTNAME,
+    CONF_PORT: ZEROCONF_PORT,
+    ATTR_PROPERTIES: {"rp": "ipp/print"},
+}
+
+
+def load_fixture_binary(filename):
+    """Load a binary fixture."""
+    path = os.path.join(os.path.dirname(__file__), "..", "..", "fixtures", filename)
+    with open(path, "rb") as fptr:
+        return fptr.read()
+
+
+async def init_integration(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker, skip_setup: bool = False,
+) -> MockConfigEntry:
+    """Set up the IPP integration in Home Assistant."""
+
+    fixture = "ipp/get-printer-attributes.bin"
+    aioclient_mock.post(
+        "http://EPSON123456.local:631/ipp/print",
+        content=load_fixture_binary(fixture),
+        headers={"Content-Type": "application/ipp"},
+    )
+
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id="cfe92100-67c4-11d4-a45f-f8d027761251",
+        data={
+            CONF_HOST: "EPSON123456.local",
+            CONF_PORT: 631,
+            CONF_SSL: False,
+            CONF_VERIFY_SSL: True,
+            CONF_BASE_PATH: "/ipp/print",
+            CONF_UUID: "cfe92100-67c4-11d4-a45f-f8d027761251",
+        },
+    )
+
+    entry.add_to_hass(hass)
+
+    if not skip_setup:
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    return entry

tests/components/ipp/test_config_flow.py

@@ -0,0 +1,306 @@
+"""Tests for the IPP config flow."""
+import aiohttp
+from pyipp import IPPConnectionUpgradeRequired
+
+from homeassistant import data_entry_flow
+from homeassistant.components.ipp import config_flow
+from homeassistant.components.ipp.const import CONF_BASE_PATH, CONF_UUID
+from homeassistant.config_entries import SOURCE_USER, SOURCE_ZEROCONF
+from homeassistant.const import CONF_HOST, CONF_NAME, CONF_SSL
+from homeassistant.core import HomeAssistant
+
+from . import (
+    MOCK_USER_INPUT,
+    MOCK_ZEROCONF_IPP_SERVICE_INFO,
+    MOCK_ZEROCONF_IPPS_SERVICE_INFO,
+    init_integration,
+    load_fixture_binary,
+)
+
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def test_show_user_form(hass: HomeAssistant) -> None:
+    """Test that the user set up form is served."""
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_USER},
+    )
+
+    assert result["step_id"] == "user"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+
+async def test_show_zeroconf_confirm_form(hass: HomeAssistant) -> None:
+    """Test that the zeroconf confirmation form is served."""
+    flow = config_flow.IPPFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+    flow.discovery_info = {CONF_NAME: "EPSON123456"}
+
+    result = await flow.async_step_zeroconf_confirm()
+
+    assert result["step_id"] == "zeroconf_confirm"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["description_placeholders"] == {CONF_NAME: "EPSON123456"}
+
+
+async def test_show_zeroconf_form(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test that the zeroconf confirmation form is served."""
+    aioclient_mock.post(
+        "http://EPSON123456.local:631/ipp/print",
+        content=load_fixture_binary("ipp/get-printer-attributes.bin"),
+        headers={"Content-Type": "application/ipp"},
+    )
+
+    flow = config_flow.IPPFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+
+    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    result = await flow.async_step_zeroconf(discovery_info)
+
+    assert flow.discovery_info[CONF_HOST] == "EPSON123456.local"
+    assert flow.discovery_info[CONF_NAME] == "EPSON123456"
+
+    assert result["step_id"] == "zeroconf_confirm"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["description_placeholders"] == {CONF_NAME: "EPSON123456"}
+
+
+async def test_connection_error(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we show user form on IPP connection error."""
+    aioclient_mock.post(
+        "http://EPSON123456.local:631/ipp/print", exc=aiohttp.ClientError
+    )
+
+    user_input = MOCK_USER_INPUT.copy()
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_USER}, data=user_input,
+    )
+
+    assert result["step_id"] == "user"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["errors"] == {"base": "connection_error"}
+
+
+async def test_zeroconf_connection_error(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort zeroconf flow on IPP connection error."""
+    aioclient_mock.post("http://EPSON123456.local/ipp/print", exc=aiohttp.ClientError)
+
+    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info,
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "connection_error"
+
+
+async def test_zeroconf_confirm_connection_error(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort zeroconf flow on IPP connection error."""
+    aioclient_mock.post("http://EPSON123456.local/ipp/print", exc=aiohttp.ClientError)
+
+    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN,
+        context={
+            "source": SOURCE_ZEROCONF,
+            CONF_HOST: "EPSON123456.local",
+            CONF_NAME: "EPSON123456",
+        },
+        data=discovery_info,
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "connection_error"
+
+
+async def test_user_connection_upgrade_required(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we show the user form if connection upgrade required by server."""
+    aioclient_mock.post(
+        "http://EPSON123456.local:631/ipp/print", exc=IPPConnectionUpgradeRequired
+    )
+
+    user_input = MOCK_USER_INPUT.copy()
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_USER}, data=user_input,
+    )
+
+    assert result["step_id"] == "user"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["errors"] == {"base": "connection_upgrade"}
+
+
+async def test_zeroconf_connection_upgrade_required(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort zeroconf flow on IPP connection error."""
+    aioclient_mock.post(
+        "http://EPSON123456.local/ipp/print", exc=IPPConnectionUpgradeRequired
+    )
+
+    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info,
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "connection_upgrade"
+
+
+async def test_user_device_exists_abort(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort user flow if printer already configured."""
+    await init_integration(hass, aioclient_mock)
+
+    user_input = MOCK_USER_INPUT.copy()
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_USER}, data=user_input,
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "already_configured"
+
+
+async def test_zeroconf_device_exists_abort(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort zeroconf flow if printer already configured."""
+    await init_integration(hass, aioclient_mock)
+
+    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info,
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "already_configured"
+
+
+async def test_zeroconf_with_uuid_device_exists_abort(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test we abort zeroconf flow if printer already configured."""
+    await init_integration(hass, aioclient_mock)
+
+    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info["properties"]["UUID"] = "cfe92100-67c4-11d4-a45f-f8d027761251"
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info,
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+    assert result["reason"] == "already_configured"
+
+
+async def test_full_user_flow_implementation(
+    hass: HomeAssistant, aioclient_mock
+) -> None:
+    """Test the full manual user flow from start to finish."""
+    aioclient_mock.post(
+        "http://EPSON123456.local:631/ipp/print",
+        content=load_fixture_binary("ipp/get-printer-attributes.bin"),
+        headers={"Content-Type": "application/ipp"},
+    )
+
+    result = await hass.config_entries.flow.async_init(
+        config_flow.DOMAIN, context={"source": SOURCE_USER},
+    )
+
+    assert result["step_id"] == "user"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={CONF_HOST: "EPSON123456.local", CONF_BASE_PATH: "/ipp/print"},
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result["title"] == "EPSON123456.local"
+
+    assert result["data"]
+    assert result["data"][CONF_HOST] == "EPSON123456.local"
+    assert result["data"][CONF_UUID] == "cfe92100-67c4-11d4-a45f-f8d027761251"
+
+
+async def test_full_zeroconf_flow_implementation(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the full manual user flow from start to finish."""
+    aioclient_mock.post(
+        "http://EPSON123456.local:631/ipp/print",
+        content=load_fixture_binary("ipp/get-printer-attributes.bin"),
+        headers={"Content-Type": "application/ipp"},
+    )
+
+    flow = config_flow.IPPFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+
+    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    result = await flow.async_step_zeroconf(discovery_info)
+
+    assert flow.discovery_info
+    assert flow.discovery_info[CONF_HOST] == "EPSON123456.local"
+    assert flow.discovery_info[CONF_NAME] == "EPSON123456"
+
+    assert result["step_id"] == "zeroconf_confirm"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+
+    result = await flow.async_step_zeroconf_confirm(
+        user_input={CONF_HOST: "EPSON123456.local"}
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result["title"] == "EPSON123456"
+
+    assert result["data"]
+    assert result["data"][CONF_HOST] == "EPSON123456.local"
+    assert result["data"][CONF_UUID] == "cfe92100-67c4-11d4-a45f-f8d027761251"
+    assert not result["data"][CONF_SSL]
+
+
+async def test_full_zeroconf_tls_flow_implementation(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the full manual user flow from start to finish."""
+    aioclient_mock.post(
+        "https://EPSON123456.local:631/ipp/print",
+        content=load_fixture_binary("ipp/get-printer-attributes.bin"),
+        headers={"Content-Type": "application/ipp"},
+    )
+
+    flow = config_flow.IPPFlowHandler()
+    flow.hass = hass
+    flow.context = {"source": SOURCE_ZEROCONF}
+
+    discovery_info = MOCK_ZEROCONF_IPPS_SERVICE_INFO.copy()
+    result = await flow.async_step_zeroconf(discovery_info)
+
+    assert result["step_id"] == "zeroconf_confirm"
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["description_placeholders"] == {CONF_NAME: "EPSON123456"}
+
+    result = await flow.async_step_zeroconf_confirm(
+        user_input={CONF_HOST: "EPSON123456.local"}
+    )
+
+    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result["title"] == "EPSON123456"
+
+    assert result["data"]
+    assert result["data"][CONF_HOST] == "EPSON123456.local"
+    assert result["data"][CONF_NAME] == "EPSON123456"
+    assert result["data"][CONF_UUID] == "cfe92100-67c4-11d4-a45f-f8d027761251"
+    assert result["data"][CONF_SSL]

tests/components/ipp/test_init.py

@@ -0,0 +1,42 @@
+"""Tests for the IPP integration."""
+import aiohttp
+
+from homeassistant.components.ipp.const import DOMAIN
+from homeassistant.config_entries import (
+    ENTRY_STATE_LOADED,
+    ENTRY_STATE_NOT_LOADED,
+    ENTRY_STATE_SETUP_RETRY,
+)
+from homeassistant.core import HomeAssistant
+
+from tests.components.ipp import init_integration
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def test_config_entry_not_ready(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the IPP configuration entry not ready."""
+    aioclient_mock.post(
+        "http://EPSON123456.local:631/ipp/print", exc=aiohttp.ClientError
+    )
+
+    entry = await init_integration(hass, aioclient_mock)
+    assert entry.state == ENTRY_STATE_SETUP_RETRY
+
+
+async def test_unload_config_entry(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the IPP configuration entry unloading."""
+    entry = await init_integration(hass, aioclient_mock)
+
+    assert hass.data[DOMAIN]
+    assert entry.entry_id in hass.data[DOMAIN]
+    assert entry.state == ENTRY_STATE_LOADED
+
+    await hass.config_entries.async_unload(entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert entry.entry_id not in hass.data[DOMAIN]
+    assert entry.state == ENTRY_STATE_NOT_LOADED

tests/components/ipp/test_sensor.py

@@ -0,0 +1,96 @@
+"""Tests for the IPP sensor platform."""
+from datetime import datetime
+
+from asynctest import patch
+
+from homeassistant.components.ipp.const import DOMAIN
+from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
+from homeassistant.const import ATTR_ICON, ATTR_UNIT_OF_MEASUREMENT, UNIT_PERCENTAGE
+from homeassistant.core import HomeAssistant
+from homeassistant.util import dt as dt_util
+
+from tests.components.ipp import init_integration
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+
+async def test_sensors(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the creation and values of the IPP sensors."""
+    entry = await init_integration(hass, aioclient_mock, skip_setup=True)
+    registry = await hass.helpers.entity_registry.async_get_registry()
+
+    # Pre-create registry entries for disabled by default sensors
+    registry.async_get_or_create(
+        SENSOR_DOMAIN,
+        DOMAIN,
+        "cfe92100-67c4-11d4-a45f-f8d027761251_uptime",
+        suggested_object_id="epson_xp_6000_series_uptime",
+        disabled_by=None,
+    )
+
+    test_time = datetime(2019, 11, 11, 9, 10, 32, tzinfo=dt_util.UTC)
+    with patch("homeassistant.components.ipp.sensor.utcnow", return_value=test_time):
+        await hass.config_entries.async_setup(entry.entry_id)
+        await hass.async_block_till_done()
+
+    state = hass.states.get("sensor.epson_xp_6000_series")
+    assert state
+    assert state.attributes.get(ATTR_ICON) == "mdi:printer"
+    assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
+
+    state = hass.states.get("sensor.epson_xp_6000_series_black_ink")
+    assert state
+    assert state.attributes.get(ATTR_ICON) == "mdi:water"
+    assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is UNIT_PERCENTAGE
+    assert state.state == "58"
+
+    state = hass.states.get("sensor.epson_xp_6000_series_photo_black_ink")
+    assert state
+    assert state.attributes.get(ATTR_ICON) == "mdi:water"
+    assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is UNIT_PERCENTAGE
+    assert state.state == "98"
+
+    state = hass.states.get("sensor.epson_xp_6000_series_cyan_ink")
+    assert state
+    assert state.attributes.get(ATTR_ICON) == "mdi:water"
+    assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is UNIT_PERCENTAGE
+    assert state.state == "91"
+
+    state = hass.states.get("sensor.epson_xp_6000_series_yellow_ink")
+    assert state
+    assert state.attributes.get(ATTR_ICON) == "mdi:water"
+    assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is UNIT_PERCENTAGE
+    assert state.state == "95"
+
+    state = hass.states.get("sensor.epson_xp_6000_series_magenta_ink")
+    assert state
+    assert state.attributes.get(ATTR_ICON) == "mdi:water"
+    assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is UNIT_PERCENTAGE
+    assert state.state == "73"
+
+    state = hass.states.get("sensor.epson_xp_6000_series_uptime")
+    assert state
+    assert state.attributes.get(ATTR_ICON) == "mdi:clock-outline"
+    assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) is None
+    assert state.state == "2019-10-26T15:37:00+00:00"
+
+    entry = registry.async_get("sensor.epson_xp_6000_series_uptime")
+    assert entry
+    assert entry.unique_id == "cfe92100-67c4-11d4-a45f-f8d027761251_uptime"
+
+
+async def test_disabled_by_default_sensors(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test the disabled by default IPP sensors."""
+    await init_integration(hass, aioclient_mock)
+    registry = await hass.helpers.entity_registry.async_get_registry()
+
+    state = hass.states.get("sensor.epson_xp_6000_series_uptime")
+    assert state is None
+
+    entry = registry.async_get("sensor.epson_xp_6000_series_uptime")
+    assert entry
+    assert entry.disabled
+    assert entry.disabled_by == "integration"