home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-06-25 01:21:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add config flow to pushbullet (#74240)

Browse Source 
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Rami Mosleh
2022-11-02 17:11:44 +02:00
committed by GitHub
parent 71920cd687
commit fc3843f5e2
21 changed files with 755 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -999,6 +999,7 @@ omit =
homeassistant/components/proxmoxve/*
homeassistant/components/proxy/camera.py
homeassistant/components/pulseaudio_loopback/switch.py
homeassistant/components/pushbullet/api.py
homeassistant/components/pushbullet/notify.py
homeassistant/components/pushbullet/sensor.py
homeassistant/components/pushover/notify.py

2
CODEOWNERS
View File

@ -880,6 +880,8 @@ build.json @home-assistant/supervisor
/tests/components/pure_energie/ @klaasnicolaas
/homeassistant/components/push/ @dgomes
/tests/components/push/ @dgomes
/homeassistant/components/pushbullet/ @engrbm87
/tests/components/pushbullet/ @engrbm87
/homeassistant/components/pushover/ @engrbm87
/tests/components/pushover/ @engrbm87
/homeassistant/components/pvoutput/ @frenck

78
homeassistant/components/pushbullet/__init__.py
View File

@ -1 +1,79 @@
"""The pushbullet component."""
from __future__ import annotations
import logging
from pushbullet import InvalidKeyError, PushBullet, PushbulletError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_API_KEY,
CONF_NAME,
EVENT_HOMEASSISTANT_START,
Platform,
)
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import discovery
from homeassistant.helpers.typing import ConfigType
from .api import PushBulletNotificationProvider
from .const import DATA_HASS_CONFIG, DOMAIN
PLATFORMS = [Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the pushbullet component."""
hass.data[DATA_HASS_CONFIG] = config
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up pushbullet from a config entry."""
try:
pushbullet = await hass.async_add_executor_job(
PushBullet, entry.data[CONF_API_KEY]
)
except InvalidKeyError:
_LOGGER.error("Invalid API key for Pushbullet")
return False
except PushbulletError as err:
raise ConfigEntryNotReady from err
pb_provider = PushBulletNotificationProvider(hass, pushbullet)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = pb_provider
def start_listener(event: Event) -> None:
"""Start the listener thread."""
_LOGGER.debug("Starting listener for pushbullet")
pb_provider.start()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, start_listener)
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
{CONF_NAME: entry.data[CONF_NAME], "entry_id": entry.entry_id},
hass.data[DATA_HASS_CONFIG],
)
)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
pb_provider: PushBulletNotificationProvider = hass.data[DOMAIN].pop(
entry.entry_id
)
await hass.async_add_executor_job(pb_provider.close)
return unload_ok

32
homeassistant/components/pushbullet/api.py Normal file
View File

@ -0,0 +1,32 @@
"""Pushbullet Notification provider."""
from typing import Any
from pushbullet import Listener, PushBullet
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import dispatcher_send
from .const import DATA_UPDATED
class PushBulletNotificationProvider(Listener):
"""Provider for an account, leading to one or more sensors."""
def __init__(self, hass: HomeAssistant, pushbullet: PushBullet) -> None:
"""Start to retrieve pushes from the given Pushbullet instance."""
self.hass = hass
self.pushbullet = pushbullet
self.data: dict[str, Any] = {}
super().__init__(account=pushbullet, on_push=self.update_data)
self.daemon = True
def update_data(self, data: dict[str, Any]) -> None:
"""Update the current data.
Currently only monitors pushes but might be extended to monitor
different kinds of Pushbullet events.
"""
if data["type"] == "push":
self.data = data["push"]
dispatcher_send(self.hass, DATA_UPDATED)

63
homeassistant/components/pushbullet/config_flow.py Normal file
View File

@ -0,0 +1,63 @@
"""Config flow for pushbullet integration."""
from __future__ import annotations
from typing import Any
from pushbullet import InvalidKeyError, PushBullet, PushbulletError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_API_KEY, CONF_NAME
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import selector
from .const import DEFAULT_NAME, DOMAIN
CONFIG_SCHEMA = vol.Schema(
{
vol.Optional(CONF_NAME, default=DEFAULT_NAME): selector.TextSelector(),
vol.Required(CONF_API_KEY): selector.TextSelector(),
}
)
class PushBulletConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for pushbullet integration."""
async def async_step_import(self, import_config: dict[str, Any]) -> FlowResult:
"""Handle import from config."""
import_config[CONF_NAME] = import_config.get(CONF_NAME, DEFAULT_NAME)
return await self.async_step_user(import_config)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors = {}
if user_input is not None:
self._async_abort_entries_match({CONF_NAME: user_input[CONF_NAME]})
try:
pushbullet = await self.hass.async_add_executor_job(
PushBullet, user_input[CONF_API_KEY]
)
except InvalidKeyError:
errors[CONF_API_KEY] = "invalid_api_key"
except PushbulletError:
errors["base"] = "cannot_connect"
if not errors:
await self.async_set_unique_id(pushbullet.user_info["iden"])
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_NAME],
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=CONFIG_SCHEMA,
errors=errors,
)

12
homeassistant/components/pushbullet/const.py Normal file
View File

@ -0,0 +1,12 @@
"""Constants for the pushbullet integration."""
from typing import Final
DOMAIN: Final = "pushbullet"
DEFAULT_NAME: Final = "Pushbullet"
DATA_HASS_CONFIG: Final = "pushbullet_hass_config"
DATA_UPDATED: Final = "pushbullet_data_updated"
ATTR_URL: Final = "url"
ATTR_FILE: Final = "file"
ATTR_FILE_URL: Final = "file_url"

3
homeassistant/components/pushbullet/manifest.json
View File

@ -3,7 +3,8 @@
"name": "Pushbullet",
"documentation": "https://www.home-assistant.io/integrations/pushbullet",
"requirements": ["pushbullet.py==0.11.0"],
"codeowners": [],
"codeowners": ["@engrbm87"],
"config_flow": true,
"iot_class": "cloud_polling",
"loggers": ["pushbullet"]
}

172
homeassistant/components/pushbullet/notify.py
View File

@ -1,8 +1,13 @@
"""Pushbullet platform for notify component."""
from __future__ import annotations
import logging
import mimetypes
from typing import Any
from pushbullet import InvalidKeyError, PushBullet, PushError
from pushbullet import PushBullet, PushError
from pushbullet.channel import Channel
from pushbullet.device import Device
import voluptuous as vol
from homeassistant.components.notify import (
@ -13,59 +18,69 @@ from homeassistant.components.notify import (
PLATFORM_SCHEMA,
BaseNotificationService,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import ATTR_FILE, ATTR_FILE_URL, ATTR_URL, DOMAIN
_LOGGER = logging.getLogger(__name__)
ATTR_URL = "url"
ATTR_FILE = "file"
ATTR_FILE_URL = "file_url"
ATTR_LIST = "list"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({vol.Required(CONF_API_KEY): cv.string})
def get_service(hass, config, discovery_info=None):
async def async_get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> PushBulletNotificationService | None:
"""Get the Pushbullet notification service."""
try:
pushbullet = PushBullet(config[CONF_API_KEY])
except InvalidKeyError:
_LOGGER.error("Wrong API key supplied")
if discovery_info is None:
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml",
breaks_in_ha_version="2023.2.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
return None
return PushBulletNotificationService(pushbullet)
pushbullet: PushBullet = hass.data[DOMAIN][discovery_info["entry_id"]].pushbullet
return PushBulletNotificationService(hass, pushbullet)
class PushBulletNotificationService(BaseNotificationService):
"""Implement the notification service for Pushbullet."""
def __init__(self, pb): # pylint: disable=invalid-name
def __init__(self, hass: HomeAssistant, pushbullet: PushBullet) -> None:
"""Initialize the service."""
self.pushbullet = pb
self.pbtargets = {}
self.refresh()
self.hass = hass
self.pushbullet = pushbullet
def refresh(self):
"""Refresh devices, contacts, etc.
pbtargets stores all targets available from this Pushbullet instance
into a dict. These are Pushbullet objects!. It sacrifices a bit of
memory for faster processing at send_message.
As of sept 2015, contacts were replaced by chats. This is not
implemented in the module yet.
"""
self.pushbullet.refresh()
self.pbtargets = {
@property
def pbtargets(self) -> dict[str, dict[str, Device | Channel]]:
"""Return device and channel detected targets."""
return {
"device": {tgt.nickname.lower(): tgt for tgt in self.pushbullet.devices},
"channel": {
tgt.channel_tag.lower(): tgt for tgt in self.pushbullet.channels
},
}
def send_message(self, message=None, **kwargs):
def send_message(self, message: str, **kwargs: Any) -> None:
"""Send a message to a specified target.
If no target specified, a 'normal' push will be sent to all devices
@ -73,24 +88,25 @@ class PushBulletNotificationService(BaseNotificationService):
Email is special, these are assumed to always exist. We use a special
call which doesn't require a push object.
"""
targets = kwargs.get(ATTR_TARGET)
title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
data = kwargs.get(ATTR_DATA)
refreshed = False
targets: list[str] = kwargs.get(ATTR_TARGET, [])
title: str = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
data: dict[str, Any] = kwargs[ATTR_DATA] or {}
if not targets:
# Backward compatibility, notify all devices in own account.
self._push_data(message, title, data, self.pushbullet)
_LOGGER.info("Sent notification to self")
_LOGGER.debug("Sent notification to self")
return
# refresh device and channel targets
self.pushbullet.refresh()
# Main loop, process all targets specified.
for target in targets:
try:
ttype, tname = target.split("/", 1)
except ValueError:
_LOGGER.error("Invalid target syntax: %s", target)
continue
except ValueError as err:
raise ValueError(f"Invalid target syntax: '{target}'") from err
# Target is email, send directly, don't use a target object.
# This also seems to work to send to all devices in own account.
@ -107,71 +123,57 @@ class PushBulletNotificationService(BaseNotificationService):
_LOGGER.info("Sent sms notification to %s", tname)
continue
# Refresh if name not found. While awaiting periodic refresh
# solution in component, poor mans refresh.
if ttype not in self.pbtargets:
_LOGGER.error("Invalid target syntax: %s", target)
continue
raise ValueError(f"Invalid target syntax: {target}")
tname = tname.lower()
if tname not in self.pbtargets[ttype] and not refreshed:
self.refresh()
refreshed = True
if tname not in self.pbtargets[ttype]:
raise ValueError(f"Target: {target} doesn't exist")
# Attempt push_note on a dict value. Keys are types & target
# name. Dict pbtargets has all *actual* targets.
try:
self._push_data(message, title, data, self.pbtargets[ttype][tname])
_LOGGER.info("Sent notification to %s/%s", ttype, tname)
except KeyError:
_LOGGER.error("No such target: %s/%s", ttype, tname)
continue
self._push_data(message, title, data, self.pbtargets[ttype][tname])
_LOGGER.debug("Sent notification to %s/%s", ttype, tname)
def _push_data(self, message, title, data, pusher, email=None, phonenumber=None):
def _push_data(
self,
message: str,
title: str,
data: dict[str, Any],
pusher: PushBullet,
email: str | None = None,
phonenumber: str | None = None,
):
"""Create the message content."""
kwargs = {"body": message, "title": title}
if email:
kwargs["email"] = email
if data is None:
data = {}
data_list = data.get(ATTR_LIST)
url = data.get(ATTR_URL)
filepath = data.get(ATTR_FILE)
file_url = data.get(ATTR_FILE_URL)
try:
email_kwargs = {}
if email:
email_kwargs["email"] = email
if phonenumber:
device = pusher.devices[0]
pusher.push_sms(device, phonenumber, message)
elif url:
pusher.push_link(title, url, body=message, **email_kwargs)
elif filepath:
if phonenumber and pusher.devices:
pusher.push_sms(pusher.devices[0], phonenumber, message)
return
if url := data.get(ATTR_URL):
pusher.push_link(url=url, **kwargs)
return
if filepath := data.get(ATTR_FILE):
if not self.hass.config.is_allowed_path(filepath):
_LOGGER.error("Filepath is not valid or allowed")
return
raise ValueError("Filepath is not valid or allowed")
with open(filepath, "rb") as fileh:
filedata = self.pushbullet.upload_file(fileh, filepath)
if filedata.get("file_type") == "application/x-empty":
_LOGGER.error("Can not send an empty file")
return
filedata.update(email_kwargs)
pusher.push_file(title=title, body=message, **filedata)
elif file_url:
if not file_url.startswith("http"):
_LOGGER.error("URL should start with http or https")
return
if filedata.get("file_type") == "application/x-empty":
raise ValueError("Cannot send an empty file")
kwargs.update(filedata)
pusher.push_file(**kwargs)
elif (file_url := data.get(ATTR_FILE_URL)) and vol.Url(file_url):
pusher.push_file(
title=title,
body=message,
file_name=file_url,
file_url=file_url,
file_type=(mimetypes.guess_type(file_url)[0]),
**email_kwargs,
**kwargs,
)
elif data_list:
pusher.push_list(title, data_list, **email_kwargs)
else:
pusher.push_note(title, message, **email_kwargs)
pusher.push_note(**kwargs)
except PushError as err:
_LOGGER.error("Notify failed: %s", err)
raise HomeAssistantError(f"Notify failed: {err}") from err

138
homeassistant/components/pushbullet/sensor.py
View File

@ -1,10 +1,6 @@
"""Pushbullet platform for sensor component."""
from __future__ import annotations
import logging
import threading
from pushbullet import InvalidKeyError, Listener, PushBullet
import voluptuous as vol
from homeassistant.components.sensor import (
@ -12,18 +8,25 @@ from homeassistant.components.sensor import (
SensorEntity,
SensorEntityDescription,
)
from homeassistant.const import CONF_API_KEY, CONF_MONITORED_CONDITIONS
from homeassistant.core import HomeAssistant
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_API_KEY, CONF_MONITORED_CONDITIONS, CONF_NAME
from homeassistant.core import HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_LOGGER = logging.getLogger(__name__)
from .api import PushBulletNotificationProvider
from .const import DATA_UPDATED, DOMAIN
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="application_name",
name="Application name",
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="body",
@ -32,26 +35,32 @@ SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="notification_id",
name="Notification ID",
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="notification_tag",
name="Notification tag",
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="package_name",
name="Package name",
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="receiver_email",
name="Receiver email",
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="sender_email",
name="Sender email",
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="source_device_iden",
name="Sender device ID",
entity_registry_enabled_default=False,
),
SensorEntityDescription(
key="title",
@ -60,6 +69,7 @@ SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="type",
name="Type",
entity_registry_enabled_default=False,
),
)
@ -75,94 +85,88 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Pushbullet Sensor platform."""
async_create_issue(
hass,
DOMAIN,
"deprecated_yaml",
breaks_in_ha_version="2023.2.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
)
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
try:
pushbullet = PushBullet(config.get(CONF_API_KEY))
except InvalidKeyError:
_LOGGER.error("Wrong API key for Pushbullet supplied")
return
pbprovider = PushBulletNotificationProvider(pushbullet)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the Pushbullet sensors from config entry."""
pb_provider: PushBulletNotificationProvider = hass.data[DOMAIN][entry.entry_id]
monitored_conditions = config[CONF_MONITORED_CONDITIONS]
entities = [
PushBulletNotificationSensor(pbprovider, description)
PushBulletNotificationSensor(entry.data[CONF_NAME], pb_provider, description)
for description in SENSOR_TYPES
if description.key in monitored_conditions
]
add_entities(entities)
async_add_entities(entities)
class PushBulletNotificationSensor(SensorEntity):
"""Representation of a Pushbullet Sensor."""
_attr_should_poll = False
_attr_has_entity_name = True
def __init__(
self,
pb, # pylint: disable=invalid-name
name: str,
pb_provider: PushBulletNotificationProvider,
description: SensorEntityDescription,
):
) -> None:
"""Initialize the Pushbullet sensor."""
self.entity_description = description
self.pushbullet = pb
self.pb_provider = pb_provider
self._attr_unique_id = (
f"{pb_provider.pushbullet.user_info['iden']}-{description.key}"
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, pb_provider.pushbullet.user_info["iden"])},
name=name,
entry_type=DeviceEntryType.SERVICE,
)
self._attr_name = f"Pushbullet {description.key}"
def update(self) -> None:
@callback
def async_update_callback(self) -> None:
"""Fetch the latest data from the sensor.
This will fetch the 'sensor reading' into self._state but also all
attributes into self._state_attributes.
"""
try:
self._attr_native_value = self.pushbullet.data[self.entity_description.key]
self._attr_extra_state_attributes = self.pushbullet.data
self._attr_native_value = self.pb_provider.data[self.entity_description.key]
self._attr_extra_state_attributes = self.pb_provider.data
except (KeyError, TypeError):
pass
self.async_write_ha_state()
class PushBulletNotificationProvider:
"""Provider for an account, leading to one or more sensors."""
def __init__(self, pushbullet):
"""Start to retrieve pushes from the given Pushbullet instance."""
self.pushbullet = pushbullet
self._data = None
self.listener = None
self.thread = threading.Thread(target=self.retrieve_pushes)
self.thread.daemon = True
self.thread.start()
def on_push(self, data):
"""Update the current data.
Currently only monitors pushes but might be extended to monitor
different kinds of Pushbullet events.
"""
if data["type"] == "push":
self._data = data["push"]
@property
def data(self):
"""Return the current data stored in the provider."""
return self._data
def retrieve_pushes(self):
"""Retrieve_pushes.
Spawn a new Listener and links it to self.on_push.
"""
self.listener = Listener(account=self.pushbullet, on_push=self.on_push)
_LOGGER.debug("Getting pushes")
try:
self.listener.run_forever()
finally:
self.listener.close()
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
self.async_on_remove(
async_dispatcher_connect(
self.hass, DATA_UPDATED, self.async_update_callback
)
)

25
homeassistant/components/pushbullet/strings.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_api_key": "[%key:common::config_flow::error::invalid_api_key%]"
},
"step": {
"user": {
"data": {
"name": "[%key:common::config_flow::data::name%]",
"api_key": "[%key:common::config_flow::data::api_key%]"
}
}
}
},
"issues": {
"deprecated_yaml": {
"title": "The Pushbullet YAML configuration is being removed",
"description": "Configuring Pushbullet using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the Pushbullet YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue."
}
}
}

25
homeassistant/components/pushbullet/translations/en.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"abort": {
"already_configured": "Service is already configured"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_api_key": "Invalid API key"
},
"step": {
"user": {
"data": {
"api_key": "API Key",
"name": "Name"
}
}
}
},
"issues": {
"deprecated_yaml": {
"description": "Configuring Pushbullet using YAML is being removed.\n\nYour existing YAML configuration has been imported into the UI automatically.\n\nRemove the Pushbullet YAML configuration from your configuration.yaml file and restart Home Assistant to fix this issue.",
"title": "The Pushbullet YAML configuration is being removed"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -304,6 +304,7 @@ FLOWS = {
"prusalink",
"ps4",
"pure_energie",
"pushbullet",
"pushover",
"pvoutput",
"pvpc_hourly_pricing",

2
homeassistant/generated/integrations.json
View File

@ -4085,7 +4085,7 @@
"pushbullet": {
"name": "Pushbullet",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "cloud_polling"
},
"pushover": {

4
tests/components/pushbullet/__init__.py
View File

@ -1 +1,5 @@
"""Tests for the pushbullet component."""
from homeassistant.const import CONF_API_KEY, CONF_NAME
MOCK_CONFIG = {CONF_NAME: "pushbullet", CONF_API_KEY: "MYAPIKEY"}

28
tests/components/pushbullet/conftest.py Normal file
View File

@ -0,0 +1,28 @@
"""Conftest for pushbullet integration."""
from pushbullet import PushBullet
import pytest
from requests_mock import Mocker
from tests.common import load_fixture
@pytest.fixture(autouse=True)
def requests_mock_fixture(requests_mock: Mocker) -> None:
"""Fixture to provide a aioclient mocker."""
requests_mock.get(
PushBullet.DEVICES_URL,
text=load_fixture("devices.json", "pushbullet"),
)
requests_mock.get(
PushBullet.ME_URL,
text=load_fixture("user_info.json", "pushbullet"),
)
requests_mock.get(
PushBullet.CHATS_URL,
text=load_fixture("chats.json", "pushbullet"),
)
requests_mock.get(
PushBullet.CHANNELS_URL,
text=load_fixture("channels.json", "pushbullet"),
)

14
tests/components/pushbullet/fixtures/channels.json Normal file
View File

@ -0,0 +1,14 @@
{
"channels": [
{
"active": true,
"created": 1412047948.579029,
"description": "Sample channel.",
"iden": "ujxPklLhvyKsjAvkMyTVh6",
"image_url": "https://dl.pushbulletusercontent.com/abc123/image.jpg",
"modified": 1412047948.579031,
"name": "Sample channel",
"tag": "sample-channel"
}
]
}

18
tests/components/pushbullet/fixtures/chats.json Normal file
View File

@ -0,0 +1,18 @@
{
"chats": [
{
"active": true,
"created": 1412047948.579029,
"iden": "ujpah72o0sjAoRtnM0jc",
"modified": 1412047948.579031,
"with": {
"email": "someone@example.com",
"email_normalized": "someone@example.com",
"iden": "ujlMns72k",
"image_url": "https://dl.pushbulletusercontent.com/acb123/example.jpg",
"name": "Someone",
"type": "user"
}
}
]
}

10
tests/components/pushbullet/fixtures/user_info.json Normal file
View File

@ -0,0 +1,10 @@
{
"created": 1381092887.398433,
"email": "example@email.com",
"email_normalized": "example@email.com",
"iden": "ujpah72o0",
"image_url": "https://static.pushbullet.com/missing-image/55a7dc-45",
"max_upload_size": 26214400,
"modified": 1441054560.741007,
"name": "Some name"
}

134
tests/components/pushbullet/test_config_flow.py Normal file
View File

@ -0,0 +1,134 @@
"""Test pushbullet config flow."""
from unittest.mock import patch
from pushbullet import InvalidKeyError, PushbulletError
import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.pushbullet.const import DOMAIN
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from . import MOCK_CONFIG
from tests.common import MockConfigEntry
@pytest.fixture(autouse=True)
def pushbullet_setup_fixture():
"""Patch pushbullet setup entry."""
with patch(
"homeassistant.components.pushbullet.async_setup_entry", return_value=True
):
yield
async def test_flow_user(hass: HomeAssistant, requests_mock_fixture) -> None:
"""Test user initialized flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=MOCK_CONFIG,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "pushbullet"
assert result["data"] == MOCK_CONFIG
async def test_flow_user_already_configured(
hass: HomeAssistant, requests_mock_fixture
) -> None:
"""Test user initialized flow with duplicate server."""
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
unique_id="ujpah72o0",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=MOCK_CONFIG,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_flow_name_already_configured(hass: HomeAssistant) -> None:
"""Test user initialized flow with duplicate server."""
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
unique_id="MYAPIKEY",
)
entry.add_to_hass(hass)
new_config = MOCK_CONFIG.copy()
new_config[CONF_API_KEY] = "NEWKEY"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=new_config,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_flow_invalid_key(hass: HomeAssistant) -> None:
"""Test user initialized flow with invalid api key."""
with patch(
"homeassistant.components.pushbullet.config_flow.PushBullet",
side_effect=InvalidKeyError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=MOCK_CONFIG,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {CONF_API_KEY: "invalid_api_key"}
async def test_flow_conn_error(hass: HomeAssistant) -> None:
"""Test user initialized flow with conn error."""
with patch(
"homeassistant.components.pushbullet.config_flow.PushBullet",
side_effect=PushbulletError,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data=MOCK_CONFIG,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
async def test_import(hass: HomeAssistant, requests_mock_fixture) -> None:
"""Test user initialized flow with unreachable server."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data=MOCK_CONFIG,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "pushbullet"
assert result["data"] == MOCK_CONFIG

84
tests/components/pushbullet/test_init.py Normal file
View File

@ -0,0 +1,84 @@
"""Test pushbullet integration."""
from unittest.mock import patch
from pushbullet import InvalidKeyError, PushbulletError
from homeassistant.components.pushbullet.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import EVENT_HOMEASSISTANT_START
from homeassistant.core import HomeAssistant
from . import MOCK_CONFIG
from tests.common import MockConfigEntry
async def test_async_setup_entry_success(
hass: HomeAssistant, requests_mock_fixture
) -> None:
"""Test pushbullet successful setup."""
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
with patch(
"homeassistant.components.pushbullet.api.PushBulletNotificationProvider.start"
) as mock_start:
hass.bus.async_fire(EVENT_HOMEASSISTANT_START)
await hass.async_block_till_done()
mock_start.assert_called_once()
async def test_setup_entry_failed_invalid_key(hass: HomeAssistant) -> None:
"""Test pushbullet failed setup due to invalid key."""
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.pushbullet.PushBullet",
side_effect=InvalidKeyError,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_ERROR
async def test_setup_entry_failed_conn_error(hass: HomeAssistant) -> None:
"""Test pushbullet failed setup due to conn error."""
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.pushbullet.PushBullet",
side_effect=PushbulletError,
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.SETUP_RETRY
async def test_async_unload_entry(hass: HomeAssistant, requests_mock_fixture) -> None:
"""Test pushbullet unload entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.LOADED
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.state == ConfigEntryState.NOT_LOADED

219
tests/components/pushbullet/test_notify.py
View File

@ -1,109 +1,65 @@
"""The tests for the pushbullet notification platform."""
"""Test pushbullet notification platform."""
from http import HTTPStatus
import json
from unittest.mock import patch
from pushbullet import PushBullet
import pytest
from requests_mock import Mocker
import homeassistant.components.notify as notify
from homeassistant.setup import async_setup_component
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
from homeassistant.components.pushbullet.const import DOMAIN
from tests.common import assert_setup_component, load_fixture
from . import MOCK_CONFIG
from tests.common import MockConfigEntry
@pytest.fixture
def mock_pushbullet():
"""Mock pushbullet."""
with patch.object(
PushBullet,
"_get_data",
return_value=json.loads(load_fixture("devices.json", "pushbullet")),
):
yield
async def test_pushbullet_config(hass, mock_pushbullet):
"""Test setup."""
config = {
notify.DOMAIN: {
"name": "test",
"platform": "pushbullet",
"api_key": "MYFAKEKEY",
}
}
with assert_setup_component(1) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert handle_config[notify.DOMAIN]
async def test_pushbullet_config_bad(hass):
"""Test set up the platform with bad/missing configuration."""
config = {notify.DOMAIN: {"platform": "pushbullet"}}
with assert_setup_component(0) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert not handle_config[notify.DOMAIN]
async def test_pushbullet_push_default(hass, requests_mock, mock_pushbullet):
async def test_pushbullet_push_default(hass, requests_mock: Mocker):
"""Test pushbullet push to default target."""
config = {
notify.DOMAIN: {
"name": "test",
"platform": "pushbullet",
"api_key": "MYFAKEKEY",
}
}
with assert_setup_component(1) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert handle_config[notify.DOMAIN]
requests_mock.register_uri(
"POST",
"https://api.pushbullet.com/v2/pushes",
status_code=HTTPStatus.OK,
json={"mock_response": "Ok"},
)
data = {"title": "Test Title", "message": "Test Message"}
await hass.services.async_call(notify.DOMAIN, "test", data)
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
data = {"title": "Test Title", "message": "Test Message"}
await hass.services.async_call(NOTIFY_DOMAIN, "pushbullet", data)
await hass.async_block_till_done()
assert requests_mock.called
assert requests_mock.call_count == 1
expected_body = {"body": "Test Message", "title": "Test Title", "type": "note"}
assert requests_mock.last_request
assert requests_mock.last_request.json() == expected_body
async def test_pushbullet_push_device(hass, requests_mock, mock_pushbullet):
async def test_pushbullet_push_device(hass, requests_mock):
"""Test pushbullet push to default target."""
config = {
notify.DOMAIN: {
"name": "test",
"platform": "pushbullet",
"api_key": "MYFAKEKEY",
}
}
with assert_setup_component(1) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert handle_config[notify.DOMAIN]
requests_mock.register_uri(
"POST",
"https://api.pushbullet.com/v2/pushes",
status_code=HTTPStatus.OK,
json={"mock_response": "Ok"},
)
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
data = {
"title": "Test Title",
"message": "Test Message",
"target": ["device/DESKTOP"],
}
await hass.services.async_call(notify.DOMAIN, "test", data)
await hass.services.async_call(NOTIFY_DOMAIN, "pushbullet", data)
await hass.async_block_till_done()
assert requests_mock.called
assert requests_mock.call_count == 1
expected_body = {
"body": "Test Message",
@ -114,35 +70,29 @@ async def test_pushbullet_push_device(hass, requests_mock, mock_pushbullet):
assert requests_mock.last_request.json() == expected_body
async def test_pushbullet_push_devices(hass, requests_mock, mock_pushbullet):
async def test_pushbullet_push_devices(hass, requests_mock):
"""Test pushbullet push to default target."""
config = {
notify.DOMAIN: {
"name": "test",
"platform": "pushbullet",
"api_key": "MYFAKEKEY",
}
}
with assert_setup_component(1) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert handle_config[notify.DOMAIN]
requests_mock.register_uri(
"POST",
"https://api.pushbullet.com/v2/pushes",
status_code=HTTPStatus.OK,
json={"mock_response": "Ok"},
)
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
data = {
"title": "Test Title",
"message": "Test Message",
"target": ["device/DESKTOP", "device/My iPhone"],
}
await hass.services.async_call(notify.DOMAIN, "test", data)
await hass.services.async_call(NOTIFY_DOMAIN, "pushbullet", data)
await hass.async_block_till_done()
assert requests_mock.called
assert requests_mock.call_count == 2
assert len(requests_mock.request_history) == 2
expected_body = {
"body": "Test Message",
@ -150,45 +100,39 @@ async def test_pushbullet_push_devices(hass, requests_mock, mock_pushbullet):
"title": "Test Title",
"type": "note",
}
assert requests_mock.request_history[0].json() == expected_body
assert requests_mock.request_history[-2].json() == expected_body
expected_body = {
"body": "Test Message",
"device_iden": "identity2",
"title": "Test Title",
"type": "note",
}
assert requests_mock.request_history[1].json() == expected_body
assert requests_mock.request_history[-1].json() == expected_body
async def test_pushbullet_push_email(hass, requests_mock, mock_pushbullet):
async def test_pushbullet_push_email(hass, requests_mock):
"""Test pushbullet push to default target."""
config = {
notify.DOMAIN: {
"name": "test",
"platform": "pushbullet",
"api_key": "MYFAKEKEY",
}
}
with assert_setup_component(1) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert handle_config[notify.DOMAIN]
requests_mock.register_uri(
"POST",
"https://api.pushbullet.com/v2/pushes",
status_code=HTTPStatus.OK,
json={"mock_response": "Ok"},
)
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
data = {
"title": "Test Title",
"message": "Test Message",
"target": ["email/user@host.net"],
}
await hass.services.async_call(notify.DOMAIN, "test", data)
await hass.services.async_call(NOTIFY_DOMAIN, "pushbullet", data)
await hass.async_block_till_done()
assert requests_mock.called
assert requests_mock.call_count == 1
assert len(requests_mock.request_history) == 1
expected_body = {
"body": "Test Message",
@ -196,38 +140,30 @@ async def test_pushbullet_push_email(hass, requests_mock, mock_pushbullet):
"title": "Test Title",
"type": "note",
}
assert requests_mock.request_history[0].json() == expected_body
assert requests_mock.last_request.json() == expected_body
async def test_pushbullet_push_mixed(hass, requests_mock, mock_pushbullet):
async def test_pushbullet_push_mixed(hass, requests_mock):
"""Test pushbullet push to default target."""
config = {
notify.DOMAIN: {
"name": "test",
"platform": "pushbullet",
"api_key": "MYFAKEKEY",
}
}
with assert_setup_component(1) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert handle_config[notify.DOMAIN]
requests_mock.register_uri(
"POST",
"https://api.pushbullet.com/v2/pushes",
status_code=HTTPStatus.OK,
json={"mock_response": "Ok"},
)
entry = MockConfigEntry(
domain=DOMAIN,
data=MOCK_CONFIG,
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
data = {
"title": "Test Title",
"message": "Test Message",
"target": ["device/DESKTOP", "email/user@host.net"],
}
await hass.services.async_call(notify.DOMAIN, "test", data)
await hass.services.async_call(NOTIFY_DOMAIN, "pushbullet", data)
await hass.async_block_till_done()
assert requests_mock.called
assert requests_mock.call_count == 2
assert len(requests_mock.request_history) == 2
expected_body = {
"body": "Test Message",
@ -235,40 +171,11 @@ async def test_pushbullet_push_mixed(hass, requests_mock, mock_pushbullet):
"title": "Test Title",
"type": "note",
}
assert requests_mock.request_history[0].json() == expected_body
assert requests_mock.request_history[-2].json() == expected_body
expected_body = {
"body": "Test Message",
"email": "user@host.net",
"title": "Test Title",
"type": "note",
}
assert requests_mock.request_history[1].json() == expected_body
async def test_pushbullet_push_no_file(hass, requests_mock, mock_pushbullet):
"""Test pushbullet push to default target."""
config = {
notify.DOMAIN: {
"name": "test",
"platform": "pushbullet",
"api_key": "MYFAKEKEY",
}
}
with assert_setup_component(1) as handle_config:
assert await async_setup_component(hass, notify.DOMAIN, config)
await hass.async_block_till_done()
assert handle_config[notify.DOMAIN]
requests_mock.register_uri(
"POST",
"https://api.pushbullet.com/v2/pushes",
status_code=HTTPStatus.OK,
json={"mock_response": "Ok"},
)
data = {
"title": "Test Title",
"message": "Test Message",
"target": ["device/DESKTOP", "device/My iPhone"],
"data": {"file": "not_a_file"},
}
assert not await hass.services.async_call(notify.DOMAIN, "test", data)
await hass.async_block_till_done()
assert requests_mock.request_history[-1].json() == expected_body