Change Prowl to use the prowlpy library and add tests for the Prowl component (#149034)

Co-authored-by: Joostlek <joostlek@outlook.com>
Co-authored-by: Erik Montnemery <erik@montnemery.com>

homeassistant/components/prowl/const.py

@@ -0,0 +1,3 @@
+"""Constants for the Prowl Notification service."""
+
+DOMAIN = "prowl"

homeassistant/components/prowl/manifest.json

@@ -3,6 +3,9 @@
   "name": "Prowl",
   "codeowners": [],
   "documentation": "https://www.home-assistant.io/integrations/prowl",
+  "integration_type": "service",
   "iot_class": "cloud_push",
-  "quality_scale": "legacy"
+  "loggers": ["prowl"],
+  "quality_scale": "legacy",
+  "requirements": ["prowlpy==1.0.2"]
 }

homeassistant/components/prowl/notify.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import asyncio
-from http import HTTPStatus
+from functools import partial
 import logging
+from typing import Any
 
+import prowlpy
 import voluptuous as vol
 
 from homeassistant.components.notify import (
@@ -17,12 +19,11 @@ from homeassistant.components.notify import (
 )
 from homeassistant.const import CONF_API_KEY
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import config_validation as cv
-from homeassistant.helpers.aiohttp_client import async_get_clientsession
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
 _LOGGER = logging.getLogger(__name__)
-_RESOURCE = "https://api.prowlapp.com/publicapi/"
 
 PLATFORM_SCHEMA = NOTIFY_PLATFORM_SCHEMA.extend({vol.Required(CONF_API_KEY): cv.string})
 
@@ -33,46 +34,49 @@ async def async_get_service(
     discovery_info: DiscoveryInfoType | None = None,
 ) -> ProwlNotificationService:
     """Get the Prowl notification service."""
-    return ProwlNotificationService(hass, config[CONF_API_KEY])
+    prowl = await hass.async_add_executor_job(
+        partial(prowlpy.Prowl, apikey=config[CONF_API_KEY])
+    )
+    return ProwlNotificationService(hass, prowl)
 
 
 class ProwlNotificationService(BaseNotificationService):
     """Implement the notification service for Prowl."""
 
-    def __init__(self, hass, api_key):
+    def __init__(self, hass: HomeAssistant, prowl: prowlpy.Prowl) -> None:
         """Initialize the service."""
         self._hass = hass
-        self._api_key = api_key
+        self._prowl = prowl
 
-    async def async_send_message(self, message, **kwargs):
+    async def async_send_message(self, message: str, **kwargs: Any) -> None:
         """Send the message to the user."""
-        response = None
-        session = None
-        url = f"{_RESOURCE}add"
-        data = kwargs.get(ATTR_DATA)
-        payload = {
-            "apikey": self._api_key,
-            "application": "Home-Assistant",
-            "event": kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT),
-            "description": message,
-            "priority": data["priority"] if data and "priority" in data else 0,
-        }
-        if data and data.get("url"):
-            payload["url"] = data["url"]
-
-        _LOGGER.debug("Attempting call Prowl service at %s", url)
-        session = async_get_clientsession(self._hass)
+        data = kwargs.get(ATTR_DATA, {})
+        if data is None:
+            data = {}
 
         try:
             async with asyncio.timeout(10):
-                response = await session.post(url, data=payload)
-                result = await response.text()
-
-            if response.status != HTTPStatus.OK or "error" in result:
-                _LOGGER.error(
-                    "Prowl service returned http status %d, response %s",
-                    response.status,
-                    result,
+                await self._hass.async_add_executor_job(
+                    partial(
+                        self._prowl.send,
+                        application="Home-Assistant",
+                        event=kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT),
+                        description=message,
+                        priority=data.get("priority", 0),
+                        url=data.get("url"),
+                    )
                 )
-        except TimeoutError:
-            _LOGGER.error("Timeout accessing Prowl at %s", url)
+        except TimeoutError as ex:
+            _LOGGER.error("Timeout accessing Prowl API")
+            raise HomeAssistantError("Timeout accessing Prowl API") from ex
+        except prowlpy.APIError as ex:
+            if str(ex).startswith("Invalid API key"):
+                _LOGGER.error("Invalid API key for Prowl service")
+                raise HomeAssistantError("Invalid API key for Prowl service") from ex
+            if str(ex).startswith("Not accepted"):
+                _LOGGER.error("Prowl returned: exceeded rate limit")
+                raise HomeAssistantError(
+                    "Prowl service reported: exceeded rate limit"
+                ) from ex
+            _LOGGER.error("Unexpected error when calling Prowl API: %s", str(ex))
+            raise HomeAssistantError("Unexpected error when calling Prowl API") from ex

homeassistant/generated/integrations.json

@@ -5131,7 +5131,7 @@
     },
     "prowl": {
       "name": "Prowl",
-      "integration_type": "hub",
+      "integration_type": "service",
       "config_flow": false,
       "iot_class": "cloud_push"
     },

requirements_all.txt

@@ -1728,6 +1728,9 @@ proliphix==0.4.1
 # homeassistant.components.prometheus
 prometheus-client==0.21.0
 
+# homeassistant.components.prowl
+prowlpy==1.0.2
+
 # homeassistant.components.proxmoxve
 proxmoxer==2.0.1
 

requirements_test_all.txt

@@ -1463,6 +1463,9 @@ prayer-times-calculator-offline==1.0.3
 # homeassistant.components.prometheus
 prometheus-client==0.21.0
 
+# homeassistant.components.prowl
+prowlpy==1.0.2
+
 # homeassistant.components.hardware
 # homeassistant.components.recorder
 # homeassistant.components.systemmonitor

tests/components/prowl/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Prowl Notification Component."""

tests/components/prowl/conftest.py

@@ -0,0 +1,43 @@
+"""Test fixtures for Prowl."""
+
+from collections.abc import Generator
+from unittest.mock import Mock, patch
+
+import pytest
+
+from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
+from homeassistant.components.prowl.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+TEST_API_KEY = "f00f" * 10
+
+
+@pytest.fixture
+async def configure_prowl_through_yaml(
+    hass: HomeAssistant, mock_prowlpy: Generator[Mock]
+) -> Generator[None]:
+    """Configure the notify domain with YAML for the Prowl platform."""
+    await async_setup_component(
+        hass,
+        NOTIFY_DOMAIN,
+        {
+            NOTIFY_DOMAIN: [
+                {
+                    "name": DOMAIN,
+                    "platform": DOMAIN,
+                    "api_key": TEST_API_KEY,
+                },
+            ]
+        },
+    )
+    await hass.async_block_till_done()
+
+
+@pytest.fixture
+def mock_prowlpy() -> Generator[Mock]:
+    """Mock the prowlpy library."""
+
+    with patch("homeassistant.components.prowl.notify.prowlpy.Prowl") as MockProwl:
+        mock_instance = MockProwl.return_value
+        yield mock_instance

tests/components/prowl/test_notify.py

@@ -0,0 +1,133 @@
+"""Test the Prowl notifications."""
+
+from typing import Any
+from unittest.mock import Mock
+
+import prowlpy
+import pytest
+
+from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
+from homeassistant.components.prowl.const import DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+
+from .conftest import TEST_API_KEY
+
+SERVICE_DATA = {"message": "Test Notification", "title": "Test Title"}
+
+EXPECTED_SEND_PARAMETERS = {
+    "application": "Home-Assistant",
+    "event": "Test Title",
+    "description": "Test Notification",
+    "priority": 0,
+    "url": None,
+}
+
+
+@pytest.mark.usefixtures("configure_prowl_through_yaml")
+async def test_send_notification_service(
+    hass: HomeAssistant,
+    mock_prowlpy: Mock,
+) -> None:
+    """Set up Prowl, call notify service, and check API call."""
+    assert hass.services.has_service(NOTIFY_DOMAIN, DOMAIN)
+    await hass.services.async_call(
+        NOTIFY_DOMAIN,
+        DOMAIN,
+        SERVICE_DATA,
+        blocking=True,
+    )
+
+    mock_prowlpy.send.assert_called_once_with(**EXPECTED_SEND_PARAMETERS)
+
+
+@pytest.mark.parametrize(
+    ("prowlpy_side_effect", "raised_exception", "exception_message"),
+    [
+        (
+            prowlpy.APIError("Internal server error"),
+            HomeAssistantError,
+            "Unexpected error when calling Prowl API",
+        ),
+        (
+            TimeoutError,
+            HomeAssistantError,
+            "Timeout accessing Prowl API",
+        ),
+        (
+            prowlpy.APIError(f"Invalid API key: {TEST_API_KEY}"),
+            HomeAssistantError,
+            "Invalid API key for Prowl service",
+        ),
+        (
+            prowlpy.APIError(
+                "Not accepted: Your IP address has exceeded the API limit"
+            ),
+            HomeAssistantError,
+            "Prowl service reported: exceeded rate limit",
+        ),
+        (
+            SyntaxError(),
+            SyntaxError,
+            "",
+        ),
+    ],
+)
+@pytest.mark.usefixtures("configure_prowl_through_yaml")
+async def test_fail_send_notification(
+    hass: HomeAssistant,
+    mock_prowlpy: Mock,
+    prowlpy_side_effect: Exception,
+    raised_exception: type[Exception],
+    exception_message: str,
+) -> None:
+    """Sending a message via Prowl with a failure."""
+    mock_prowlpy.send.side_effect = prowlpy_side_effect
+
+    assert hass.services.has_service(NOTIFY_DOMAIN, DOMAIN)
+    with pytest.raises(raised_exception, match=exception_message):
+        await hass.services.async_call(
+            NOTIFY_DOMAIN,
+            DOMAIN,
+            SERVICE_DATA,
+            blocking=True,
+        )
+
+    mock_prowlpy.send.assert_called_once_with(**EXPECTED_SEND_PARAMETERS)
+
+
+@pytest.mark.parametrize(
+    ("service_data", "expected_send_parameters"),
+    [
+        (
+            {"message": "Test Notification", "title": "Test Title"},
+            {
+                "application": "Home-Assistant",
+                "event": "Test Title",
+                "description": "Test Notification",
+                "priority": 0,
+                "url": None,
+            },
+        )
+    ],
+)
+@pytest.mark.usefixtures("configure_prowl_through_yaml")
+async def test_other_exception_send_notification(
+    hass: HomeAssistant,
+    mock_prowlpy: Mock,
+    service_data: dict[str, Any],
+    expected_send_parameters: dict[str, Any],
+) -> None:
+    """Sending a message via Prowl with a general unhandled exception."""
+    mock_prowlpy.send.side_effect = SyntaxError
+
+    assert hass.services.has_service(NOTIFY_DOMAIN, DOMAIN)
+    with pytest.raises(SyntaxError):
+        await hass.services.async_call(
+            NOTIFY_DOMAIN,
+            DOMAIN,
+            SERVICE_DATA,
+            blocking=True,
+        )
+
+    mock_prowlpy.send.assert_called_once_with(**EXPECTED_SEND_PARAMETERS)