mirror of
https://github.com/home-assistant/core.git
synced 2026-09-26 09:23:17 -04:00
New: Threema Integration (#165993)
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: Norbert Rittel <norbert@rittel.de> Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Lukas
co-authored by
Claude Opus 4.6
Norbert Rittel
Joostlek
Claude Opus 4.6
Norbert Rittel
Joostlek
parent
123a469bd5
commit
125bcf42e5
@@ -603,6 +603,7 @@ homeassistant.components.teltonika.*
|
||||
homeassistant.components.teslemetry.*
|
||||
homeassistant.components.text.*
|
||||
homeassistant.components.thethingsnetwork.*
|
||||
homeassistant.components.threema.*
|
||||
homeassistant.components.threshold.*
|
||||
homeassistant.components.tibber.*
|
||||
homeassistant.components.tile.*
|
||||
|
||||
Generated
+2
@@ -1910,6 +1910,8 @@ CLAUDE.md @home-assistant/core
|
||||
/tests/components/thethingsnetwork/ @angelnu
|
||||
/homeassistant/components/thread/ @home-assistant/core
|
||||
/tests/components/thread/ @home-assistant/core
|
||||
/homeassistant/components/threema/ @LukasQ
|
||||
/tests/components/threema/ @LukasQ
|
||||
/homeassistant/components/tibber/ @danielhiversen
|
||||
/tests/components/tibber/ @danielhiversen
|
||||
/homeassistant/components/tile/ @bachya
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
"""The Threema Gateway integration."""
|
||||
|
||||
import logging
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
|
||||
|
||||
from .client import ThreemaAPIClient, ThreemaAuthError, ThreemaConnectionError
|
||||
from .const import CONF_API_SECRET, CONF_GATEWAY_ID, CONF_PRIVATE_KEY, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.NOTIFY]
|
||||
|
||||
type ThreemaConfigEntry = ConfigEntry[ThreemaAPIClient]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ThreemaConfigEntry) -> bool:
|
||||
"""Set up Threema Gateway from a config entry."""
|
||||
client = ThreemaAPIClient(
|
||||
hass,
|
||||
gateway_id=entry.data[CONF_GATEWAY_ID],
|
||||
api_secret=entry.data[CONF_API_SECRET],
|
||||
private_key=entry.data.get(CONF_PRIVATE_KEY),
|
||||
)
|
||||
|
||||
try:
|
||||
await client.validate_credentials()
|
||||
except ThreemaAuthError as err:
|
||||
raise ConfigEntryError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="invalid_auth",
|
||||
) from err
|
||||
except ThreemaConnectionError as err:
|
||||
raise ConfigEntryNotReady(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="cannot_connect",
|
||||
) from err
|
||||
|
||||
entry.runtime_data = client
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
|
||||
return True
|
||||
|
||||
|
||||
async def _async_update_listener(
|
||||
hass: HomeAssistant, entry: ThreemaConfigEntry
|
||||
) -> None:
|
||||
"""Reload entry when config is updated (e.g. subentry added/removed)."""
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ThreemaConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Threema Gateway API client, wired to Home Assistant's shared aiohttp session."""
|
||||
|
||||
from aiothreema import (
|
||||
ThreemaAuthError,
|
||||
ThreemaConnectionError,
|
||||
ThreemaGatewayClient,
|
||||
ThreemaSendError,
|
||||
derive_public_key,
|
||||
generate_key_pair,
|
||||
)
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
|
||||
__all__ = [
|
||||
"ThreemaAPIClient",
|
||||
"ThreemaAuthError",
|
||||
"ThreemaConnectionError",
|
||||
"ThreemaSendError",
|
||||
"derive_public_key",
|
||||
"generate_key_pair",
|
||||
]
|
||||
|
||||
|
||||
class ThreemaAPIClient(ThreemaGatewayClient):
|
||||
"""Threema Gateway client bound to Home Assistant's shared aiohttp session.
|
||||
|
||||
The Gateway HTTP protocol and end-to-end encryption live in the
|
||||
`aiothreema` library; this class only wires it up to Home Assistant.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
gateway_id: str,
|
||||
api_secret: str,
|
||||
private_key: str | None = None,
|
||||
) -> None:
|
||||
"""Initialize the client with Home Assistant's shared session."""
|
||||
super().__init__(
|
||||
gateway_id,
|
||||
api_secret,
|
||||
private_key,
|
||||
session=async_get_clientsession(hass),
|
||||
)
|
||||
@@ -0,0 +1,276 @@
|
||||
"""Config flow for Threema Gateway integration."""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, override
|
||||
|
||||
import probatio
|
||||
|
||||
from homeassistant.config_entries import (
|
||||
ConfigEntry,
|
||||
ConfigFlow,
|
||||
ConfigFlowResult,
|
||||
ConfigSubentryFlow,
|
||||
SubentryFlowResult,
|
||||
)
|
||||
from homeassistant.const import CONF_NAME, CONF_RECIPIENT
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.selector import (
|
||||
TextSelector,
|
||||
TextSelectorConfig,
|
||||
TextSelectorType,
|
||||
)
|
||||
|
||||
from .client import (
|
||||
ThreemaAPIClient,
|
||||
ThreemaAuthError,
|
||||
ThreemaConnectionError,
|
||||
derive_public_key,
|
||||
generate_key_pair,
|
||||
)
|
||||
from .const import (
|
||||
CONF_API_SECRET,
|
||||
CONF_GATEWAY_ID,
|
||||
CONF_PRIVATE_KEY,
|
||||
DOMAIN,
|
||||
SUBENTRY_TYPE_RECIPIENT,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
_KEY_HEX_LENGTH = 64
|
||||
_KEY_PREFIXES = ("private:", "public:")
|
||||
_CONF_PUBLIC_KEY = "public_key"
|
||||
_GATEWAY_ID_REGEX = re.compile(r"^\*[A-Z0-9]{7}$")
|
||||
|
||||
|
||||
def _strip_key_prefix(value: str, expected_prefix: str) -> str | None:
|
||||
"""Strip the expected Threema key-export prefix, if present.
|
||||
|
||||
Returns None if the value carries a *different* key-type prefix (e.g.
|
||||
a 'public:' key pasted into the private-key field), so the mismatch
|
||||
can be rejected instead of silently accepted as the wrong key type.
|
||||
"""
|
||||
lowered = value.lower()
|
||||
for prefix in _KEY_PREFIXES:
|
||||
if lowered.startswith(prefix):
|
||||
if prefix != expected_prefix:
|
||||
return None
|
||||
return value[len(prefix) :].strip()
|
||||
return value
|
||||
|
||||
|
||||
def _is_valid_key_hex(value: str) -> bool:
|
||||
"""Return True if value is a 64-character hex string (32-byte NaCl key)."""
|
||||
if len(value) != _KEY_HEX_LENGTH:
|
||||
return False
|
||||
try:
|
||||
bytes.fromhex(value)
|
||||
except ValueError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
class ThreemaConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for Threema Gateway."""
|
||||
|
||||
VERSION = 1
|
||||
MINOR_VERSION = 1
|
||||
|
||||
@classmethod
|
||||
@callback
|
||||
@override
|
||||
def async_get_supported_subentry_types(
|
||||
cls, config_entry: ConfigEntry
|
||||
) -> dict[str, type[ConfigSubentryFlow]]:
|
||||
"""Return subentry types supported by this integration."""
|
||||
return {SUBENTRY_TYPE_RECIPIENT: RecipientSubentryFlowHandler}
|
||||
|
||||
_gateway_id: str | None = None
|
||||
_api_secret: str | None = None
|
||||
_private_key: str | None = None
|
||||
_public_key: str | None = None
|
||||
|
||||
@override
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the initial step - choose setup type."""
|
||||
return self.async_show_menu(
|
||||
step_id="user",
|
||||
menu_options=["credentials", "setup_new"],
|
||||
)
|
||||
|
||||
async def async_step_setup_new(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Generate keys for a new Gateway ID."""
|
||||
if user_input is not None:
|
||||
return await self.async_step_credentials()
|
||||
|
||||
try:
|
||||
private_key, public_key = await self.hass.async_add_executor_job(
|
||||
generate_key_pair
|
||||
)
|
||||
except Exception:
|
||||
_LOGGER.exception("Failed to generate key pair")
|
||||
return self.async_abort(reason="key_generation_failed")
|
||||
|
||||
self._private_key = private_key
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="setup_new",
|
||||
description_placeholders={
|
||||
"public_key": public_key,
|
||||
"private_key": private_key,
|
||||
},
|
||||
)
|
||||
|
||||
async def async_step_credentials(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Collect Gateway credentials."""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
if user_input is not None:
|
||||
gateway_id = user_input[CONF_GATEWAY_ID].strip().upper()
|
||||
|
||||
if not _GATEWAY_ID_REGEX.match(gateway_id):
|
||||
errors["base"] = "invalid_gateway_id"
|
||||
else:
|
||||
await self.async_set_unique_id(gateway_id)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
self._gateway_id = gateway_id
|
||||
self._api_secret = user_input[CONF_API_SECRET].strip()
|
||||
|
||||
raw_private_key = user_input.get(CONF_PRIVATE_KEY, "").strip()
|
||||
self._private_key = raw_private_key or None
|
||||
raw_public_key = user_input.get(_CONF_PUBLIC_KEY, "").strip()
|
||||
self._public_key = raw_public_key or None
|
||||
|
||||
private_key = (
|
||||
_strip_key_prefix(raw_private_key, "private:")
|
||||
if raw_private_key
|
||||
else None
|
||||
)
|
||||
public_key = (
|
||||
_strip_key_prefix(raw_public_key, "public:")
|
||||
if raw_public_key
|
||||
else None
|
||||
)
|
||||
|
||||
if raw_private_key and private_key is None:
|
||||
errors[CONF_PRIVATE_KEY] = "invalid_key"
|
||||
elif raw_public_key and public_key is None:
|
||||
errors[_CONF_PUBLIC_KEY] = "invalid_key"
|
||||
elif private_key and not _is_valid_key_hex(private_key):
|
||||
errors[CONF_PRIVATE_KEY] = "invalid_key"
|
||||
elif public_key and not _is_valid_key_hex(public_key):
|
||||
errors[_CONF_PUBLIC_KEY] = "invalid_key"
|
||||
elif public_key and not private_key:
|
||||
errors[_CONF_PUBLIC_KEY] = "public_key_requires_private_key"
|
||||
elif (
|
||||
private_key
|
||||
and public_key
|
||||
and derive_public_key(private_key).lower() != public_key.lower()
|
||||
):
|
||||
errors[_CONF_PUBLIC_KEY] = "key_mismatch"
|
||||
else:
|
||||
client = ThreemaAPIClient(
|
||||
self.hass,
|
||||
gateway_id=gateway_id,
|
||||
api_secret=self._api_secret,
|
||||
private_key=private_key,
|
||||
)
|
||||
|
||||
try:
|
||||
await client.validate_credentials()
|
||||
except ThreemaAuthError:
|
||||
errors["base"] = "invalid_auth"
|
||||
except ThreemaConnectionError:
|
||||
errors["base"] = "cannot_connect"
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error validating credentials")
|
||||
errors["base"] = "unknown"
|
||||
else:
|
||||
data: dict[str, str] = {
|
||||
CONF_GATEWAY_ID: self._gateway_id,
|
||||
CONF_API_SECRET: self._api_secret,
|
||||
}
|
||||
if private_key:
|
||||
data[CONF_PRIVATE_KEY] = private_key
|
||||
|
||||
return self.async_create_entry(
|
||||
title=f"Threema {self._gateway_id}",
|
||||
data=data,
|
||||
)
|
||||
|
||||
schema = probatio.Schema(
|
||||
{
|
||||
probatio.Required(CONF_GATEWAY_ID, default=self._gateway_id or ""): str,
|
||||
probatio.Required(
|
||||
CONF_API_SECRET, default=self._api_secret or ""
|
||||
): TextSelector(TextSelectorConfig(type=TextSelectorType.PASSWORD)),
|
||||
probatio.Optional(
|
||||
CONF_PRIVATE_KEY, default=self._private_key or ""
|
||||
): TextSelector(TextSelectorConfig(type=TextSelectorType.PASSWORD)),
|
||||
probatio.Optional(
|
||||
_CONF_PUBLIC_KEY, default=self._public_key or ""
|
||||
): TextSelector(TextSelectorConfig(type=TextSelectorType.TEXT)),
|
||||
}
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="credentials",
|
||||
data_schema=schema,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
|
||||
_RECIPIENT_ID_REGEX = re.compile(r"^[0-9A-Za-z]{8}$")
|
||||
|
||||
|
||||
class RecipientSubentryFlowHandler(ConfigSubentryFlow):
|
||||
"""Handle adding a Threema recipient as a subentry."""
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> SubentryFlowResult:
|
||||
"""Handle the recipient subentry step."""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
if user_input is not None:
|
||||
recipient_id = user_input[CONF_RECIPIENT].strip().upper()
|
||||
|
||||
if not _RECIPIENT_ID_REGEX.match(recipient_id):
|
||||
errors[CONF_RECIPIENT] = "invalid_recipient_id"
|
||||
else:
|
||||
# Check for duplicate recipients
|
||||
for subentry in self._get_entry().subentries.values():
|
||||
if subentry.data.get(CONF_RECIPIENT) == recipient_id:
|
||||
return self.async_abort(reason="already_configured")
|
||||
|
||||
raw_name = user_input.get(CONF_NAME, "").strip()
|
||||
title = f"{raw_name} ({recipient_id})" if raw_name else recipient_id
|
||||
|
||||
data: dict[str, str] = {CONF_RECIPIENT: recipient_id}
|
||||
if raw_name:
|
||||
data[CONF_NAME] = raw_name
|
||||
|
||||
return self.async_create_entry(
|
||||
title=title,
|
||||
data=data,
|
||||
unique_id=recipient_id,
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=probatio.Schema(
|
||||
{
|
||||
probatio.Required(CONF_RECIPIENT): str,
|
||||
probatio.Optional(CONF_NAME): str,
|
||||
}
|
||||
),
|
||||
errors=errors,
|
||||
)
|
||||
@@ -0,0 +1,8 @@
|
||||
"""Constants for the Threema Gateway integration."""
|
||||
|
||||
DOMAIN = "threema"
|
||||
|
||||
CONF_GATEWAY_ID = "gateway_id"
|
||||
CONF_API_SECRET = "api_secret"
|
||||
CONF_PRIVATE_KEY = "private_key"
|
||||
SUBENTRY_TYPE_RECIPIENT = "recipient"
|
||||
@@ -0,0 +1,11 @@
|
||||
{
|
||||
"domain": "threema",
|
||||
"name": "Threema",
|
||||
"codeowners": ["@LukasQ"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/threema",
|
||||
"integration_type": "service",
|
||||
"iot_class": "cloud_push",
|
||||
"quality_scale": "bronze",
|
||||
"requirements": ["aiothreema==0.1.0"]
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Notify platform for Threema Gateway integration."""
|
||||
|
||||
from typing import override
|
||||
|
||||
from homeassistant.components.notify import NotifyEntity, NotifyEntityFeature
|
||||
from homeassistant.config_entries import ConfigSubentry
|
||||
from homeassistant.const import CONF_RECIPIENT
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from . import ThreemaConfigEntry
|
||||
from .client import ThreemaAuthError, ThreemaConnectionError, ThreemaSendError
|
||||
from .const import DOMAIN, SUBENTRY_TYPE_RECIPIENT
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: ThreemaConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up Threema notify entities from config entry subentries."""
|
||||
for subentry in entry.get_subentries_of_type(SUBENTRY_TYPE_RECIPIENT):
|
||||
async_add_entities(
|
||||
[ThreemaNotifyEntity(entry, subentry)],
|
||||
config_subentry_id=subentry.subentry_id,
|
||||
)
|
||||
|
||||
|
||||
class ThreemaNotifyEntity(NotifyEntity):
|
||||
"""Notify entity for sending messages to a Threema recipient."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_name = None
|
||||
_attr_supported_features = NotifyEntityFeature.TITLE
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
entry: ThreemaConfigEntry,
|
||||
subentry: ConfigSubentry,
|
||||
) -> None:
|
||||
"""Initialize the notify entity."""
|
||||
self._client = entry.runtime_data
|
||||
self._recipient_id: str = subentry.data[CONF_RECIPIENT]
|
||||
|
||||
self._attr_unique_id = f"{self._client.gateway_id}_{self._recipient_id}"
|
||||
self._attr_device_info = DeviceInfo(
|
||||
entry_type=DeviceEntryType.SERVICE,
|
||||
manufacturer="Threema",
|
||||
identifiers={(DOMAIN, self._attr_unique_id)},
|
||||
name=subentry.title,
|
||||
)
|
||||
|
||||
@override
|
||||
async def async_send_message(self, message: str, title: str | None = None) -> None:
|
||||
"""Send a message to the configured Threema recipient."""
|
||||
text = f"*{title}*\n{message}" if title else message
|
||||
try:
|
||||
await self._client.send_text_message(self._recipient_id, text)
|
||||
except ThreemaAuthError as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="invalid_auth",
|
||||
) from err
|
||||
except (ThreemaSendError, ThreemaConnectionError) as err:
|
||||
raise HomeAssistantError(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="send_error",
|
||||
translation_placeholders={"error": str(err)},
|
||||
) from err
|
||||
@@ -0,0 +1,100 @@
|
||||
rules:
|
||||
# Bronze
|
||||
action-setup:
|
||||
status: exempt
|
||||
comment: only entity actions
|
||||
appropriate-polling:
|
||||
status: exempt
|
||||
comment: Integration only sends messages, no polling required.
|
||||
brands: done
|
||||
common-modules: done
|
||||
config-flow: done
|
||||
config-flow-test-coverage: done
|
||||
dependency-transparency: done
|
||||
docs-actions: done
|
||||
docs-conditions:
|
||||
status: exempt
|
||||
comment: Integration does not provide any automation conditions.
|
||||
docs-high-level-description: done
|
||||
docs-installation-instructions: done
|
||||
docs-removal-instructions: done
|
||||
docs-triggers:
|
||||
status: exempt
|
||||
comment: Integration does not provide any automation triggers.
|
||||
entity-event-setup:
|
||||
status: exempt
|
||||
comment: Integration does not use event subscriptions.
|
||||
entity-unique-id: done
|
||||
has-entity-name: done
|
||||
runtime-data: done
|
||||
test-before-configure: done
|
||||
test-before-setup: done
|
||||
unique-config-entry: done
|
||||
|
||||
# Silver
|
||||
action-exceptions: done
|
||||
config-entry-unloading: done
|
||||
docs-configuration-parameters: done
|
||||
docs-installation-parameters: done
|
||||
entity-unavailable:
|
||||
status: exempt
|
||||
comment: Notify entities are stateless send-only, availability is not applicable.
|
||||
integration-owner: done
|
||||
log-when-unavailable:
|
||||
status: exempt
|
||||
comment: Notify entities are stateless send-only.
|
||||
parallel-updates:
|
||||
status: exempt
|
||||
comment: No polling or data updates.
|
||||
reauthentication-flow: todo
|
||||
test-coverage: done
|
||||
|
||||
# Gold
|
||||
devices: done
|
||||
diagnostics: todo
|
||||
discovery:
|
||||
status: exempt
|
||||
comment: Cloud service, not discoverable on the network.
|
||||
discovery-update-info:
|
||||
status: exempt
|
||||
comment: Cloud service, not discoverable on the network.
|
||||
docs-data-update:
|
||||
status: exempt
|
||||
comment: Integration does not poll or update data.
|
||||
docs-examples: done
|
||||
docs-known-limitations: done
|
||||
docs-supported-devices:
|
||||
status: exempt
|
||||
comment: Service integration, no physical devices.
|
||||
docs-supported-functions: done
|
||||
docs-troubleshooting: done
|
||||
docs-use-cases: done
|
||||
dynamic-devices:
|
||||
status: exempt
|
||||
comment: No devices in current scope.
|
||||
entity-category:
|
||||
status: exempt
|
||||
comment: Notify entities have no applicable category.
|
||||
entity-device-class:
|
||||
status: exempt
|
||||
comment: No applicable device class for notify entities.
|
||||
entity-disabled-by-default:
|
||||
status: exempt
|
||||
comment: Notify entities should be enabled by default for immediate use.
|
||||
entity-translations: done
|
||||
exception-translations: done
|
||||
icon-translations:
|
||||
status: exempt
|
||||
comment: Notify entities use default platform icon.
|
||||
reconfiguration-flow: todo
|
||||
repair-issues:
|
||||
status: exempt
|
||||
comment: No scenarios requiring repair issues.
|
||||
stale-devices:
|
||||
status: exempt
|
||||
comment: No devices in current scope.
|
||||
|
||||
# Platinum
|
||||
async-dependency: todo
|
||||
inject-websession: todo
|
||||
strict-typing: done
|
||||
@@ -0,0 +1,85 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
|
||||
"key_generation_failed": "Failed to generate encryption keys."
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
|
||||
"invalid_gateway_id": "Gateway ID must start with * followed by 7 alphanumeric characters.",
|
||||
"invalid_key": "Key must be a 64-character hex string (32 bytes).",
|
||||
"key_mismatch": "The public key does not match the private key.",
|
||||
"public_key_requires_private_key": "Enter the private key too to verify it against the public key.",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
},
|
||||
"step": {
|
||||
"credentials": {
|
||||
"data": {
|
||||
"api_secret": "API secret",
|
||||
"gateway_id": "Gateway ID",
|
||||
"private_key": "Private key (optional, enables E2E encryption)",
|
||||
"public_key": "Public key to verify against (optional)"
|
||||
},
|
||||
"data_description": {
|
||||
"api_secret": "Secret you received from Threema",
|
||||
"gateway_id": "ID of your Gateway, including '*', e.g.: *ABCD123",
|
||||
"private_key": "Private key as hex string (64 characters). A 'private:' prefix, as used by Threema's key export tools, is accepted and stripped automatically. Leave empty for simple (non-E2E) mode.",
|
||||
"public_key": "Never stored — only checked once against the private key above to catch copy-paste mistakes. Paste the public key you registered at gateway.threema.ch (a 'public:' prefix is accepted)."
|
||||
},
|
||||
"description": "Enter your Threema Gateway ID and secret obtained from gateway.threema.ch. Threema enables end-to-end mode for a Gateway ID manually after you register a public key, which can take a few days — so you may well be completing this step long after the keys were generated.",
|
||||
"title": "Gateway credentials"
|
||||
},
|
||||
"setup_new": {
|
||||
"description": "Your keys have been generated. Save them now before continuing!\n\nPublic key: `{public_key}`\nPrivate key: `{private_key}`\n\nWhen registering at gateway.threema.ch, paste the public key as shown above. Store the private key securely — it cannot be recovered.",
|
||||
"title": "Keys generated"
|
||||
},
|
||||
"user": {
|
||||
"menu_options": {
|
||||
"credentials": "Add existing Gateway ID",
|
||||
"setup_new": "Generate new encryption keys"
|
||||
},
|
||||
"title": "Threema setup"
|
||||
}
|
||||
}
|
||||
},
|
||||
"config_subentries": {
|
||||
"recipient": {
|
||||
"abort": {
|
||||
"already_configured": "This recipient is already configured."
|
||||
},
|
||||
"entry_type": "Recipient",
|
||||
"error": {
|
||||
"invalid_recipient_id": "Threema ID must be exactly 8 alphanumeric characters."
|
||||
},
|
||||
"initiate_flow": {
|
||||
"user": "Add recipient"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"name": "Display name",
|
||||
"recipient": "Threema ID"
|
||||
},
|
||||
"data_description": {
|
||||
"name": "Optional friendly name for this recipient (e.g. 'Dad').",
|
||||
"recipient": "The 8-character Threema ID of the recipient."
|
||||
},
|
||||
"description": "Enter the Threema ID of the person you want to send messages to.",
|
||||
"title": "Add recipient"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
"cannot_connect": {
|
||||
"message": "[%key:common::config_flow::error::cannot_connect%]"
|
||||
},
|
||||
"invalid_auth": {
|
||||
"message": "[%key:common::config_flow::error::invalid_auth%]"
|
||||
},
|
||||
"send_error": {
|
||||
"message": "Error sending message: {error}"
|
||||
}
|
||||
}
|
||||
}
|
||||
Generated
+1
@@ -810,6 +810,7 @@ FLOWS = {
|
||||
"thermopro",
|
||||
"thethingsnetwork",
|
||||
"thread",
|
||||
"threema",
|
||||
"tibber",
|
||||
"tile",
|
||||
"tilt_ble",
|
||||
|
||||
@@ -7504,6 +7504,12 @@
|
||||
"iot_class": "local_polling",
|
||||
"single_config_entry": true
|
||||
},
|
||||
"threema": {
|
||||
"name": "Threema",
|
||||
"integration_type": "service",
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_push"
|
||||
},
|
||||
"tibber": {
|
||||
"name": "Tibber",
|
||||
"integration_type": "hub",
|
||||
|
||||
@@ -5790,6 +5790,16 @@ disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.threema.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.threshold.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
||||
Generated
+3
@@ -470,6 +470,9 @@ aiotankerkoenig==0.5.3
|
||||
# homeassistant.components.tedee
|
||||
aiotedee==0.3.0
|
||||
|
||||
# homeassistant.components.threema
|
||||
aiothreema==0.1.0
|
||||
|
||||
# homeassistant.components.tractive
|
||||
aiotractive==1.0.3
|
||||
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for the Threema Gateway integration."""
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Fixtures for Threema Gateway integration tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.threema.const import (
|
||||
CONF_API_SECRET,
|
||||
CONF_GATEWAY_ID,
|
||||
CONF_PRIVATE_KEY,
|
||||
DOMAIN,
|
||||
SUBENTRY_TYPE_RECIPIENT,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigSubentryDataWithId
|
||||
from homeassistant.const import CONF_RECIPIENT
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_GATEWAY_ID = "*TESTGWY"
|
||||
MOCK_API_SECRET = "test_secret_key_12345"
|
||||
MOCK_PRIVATE_KEY = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
|
||||
MOCK_PUBLIC_KEY = "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210"
|
||||
MOCK_RECIPIENT_ID = "ABCD1234"
|
||||
MOCK_SUBENTRY_ID = "mock_subentry_id"
|
||||
|
||||
RECIPIENT_SUBENTRY: ConfigSubentryDataWithId = {
|
||||
"data": {CONF_RECIPIENT: MOCK_RECIPIENT_ID},
|
||||
"subentry_id": MOCK_SUBENTRY_ID,
|
||||
"subentry_type": SUBENTRY_TYPE_RECIPIENT,
|
||||
"title": MOCK_RECIPIENT_ID,
|
||||
"unique_id": MOCK_RECIPIENT_ID,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_subentries() -> list[ConfigSubentryDataWithId]:
|
||||
"""Fixture providing one recipient subentry by default; override to [] when not needed."""
|
||||
return [RECIPIENT_SUBENTRY]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry(
|
||||
mock_subentries: list[ConfigSubentryDataWithId],
|
||||
) -> MockConfigEntry:
|
||||
"""Return a mocked config entry for basic mode (no encryption)."""
|
||||
return MockConfigEntry(
|
||||
title=f"Threema {MOCK_GATEWAY_ID}",
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
unique_id=MOCK_GATEWAY_ID,
|
||||
subentries_data=[*mock_subentries],
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry_with_keys(
|
||||
mock_subentries: list[ConfigSubentryDataWithId],
|
||||
) -> MockConfigEntry:
|
||||
"""Return a mocked config entry for E2E encrypted mode."""
|
||||
return MockConfigEntry(
|
||||
title=f"Threema {MOCK_GATEWAY_ID}",
|
||||
domain=DOMAIN,
|
||||
data={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: MOCK_PRIVATE_KEY,
|
||||
},
|
||||
unique_id=MOCK_GATEWAY_ID,
|
||||
subentries_data=[*mock_subentries],
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_credentials() -> Generator[AsyncMock]:
|
||||
"""Mock ThreemaAPIClient.validate_credentials to succeed by default."""
|
||||
with patch(
|
||||
"homeassistant.components.threema.client.ThreemaAPIClient.validate_credentials",
|
||||
new_callable=AsyncMock,
|
||||
) as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_send_message() -> Generator[AsyncMock]:
|
||||
"""Mock ThreemaAPIClient.send_text_message to return a message ID."""
|
||||
with patch(
|
||||
"homeassistant.components.threema.client.ThreemaAPIClient.send_text_message",
|
||||
new_callable=AsyncMock,
|
||||
return_value="mock_message_id",
|
||||
) as mock:
|
||||
yield mock
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Unit tests for the Threema Gateway API client wrapper.
|
||||
|
||||
`ThreemaAPIClient` only wires Home Assistant's shared aiohttp session into
|
||||
`aiothreema.ThreemaGatewayClient`. The Gateway HTTP protocol, encryption,
|
||||
and key generation are implemented and tested in that library itself, so
|
||||
this file only tests the wiring this integration owns.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from aiothreema import ThreemaGatewayClient
|
||||
|
||||
from homeassistant.components.threema.client import ThreemaAPIClient
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .conftest import MOCK_API_SECRET, MOCK_GATEWAY_ID, MOCK_PRIVATE_KEY
|
||||
|
||||
|
||||
def _patch_session(session: MagicMock):
|
||||
"""Return a context manager that patches async_get_clientsession."""
|
||||
return patch(
|
||||
"homeassistant.components.threema.client.async_get_clientsession",
|
||||
return_value=session,
|
||||
)
|
||||
|
||||
|
||||
def test_is_a_threema_gateway_client(hass: HomeAssistant) -> None:
|
||||
"""Test the wrapper is usable wherever a ThreemaGatewayClient is expected."""
|
||||
with _patch_session(MagicMock()):
|
||||
client = ThreemaAPIClient(hass, MOCK_GATEWAY_ID, MOCK_API_SECRET)
|
||||
assert isinstance(client, ThreemaGatewayClient)
|
||||
|
||||
|
||||
def test_uses_home_assistants_shared_session(hass: HomeAssistant) -> None:
|
||||
"""Test the client is wired to Home Assistant's shared aiohttp session."""
|
||||
mock_session = MagicMock()
|
||||
with patch(
|
||||
"homeassistant.components.threema.client.async_get_clientsession",
|
||||
return_value=mock_session,
|
||||
) as mock_get_session:
|
||||
client = ThreemaAPIClient(hass, MOCK_GATEWAY_ID, MOCK_API_SECRET)
|
||||
|
||||
mock_get_session.assert_called_once_with(hass)
|
||||
assert client._session is mock_session
|
||||
|
||||
|
||||
def test_passes_through_credentials(hass: HomeAssistant) -> None:
|
||||
"""Test gateway id, API secret, and private key are passed through as-is."""
|
||||
with _patch_session(MagicMock()):
|
||||
client = ThreemaAPIClient(
|
||||
hass, MOCK_GATEWAY_ID, MOCK_API_SECRET, private_key=MOCK_PRIVATE_KEY
|
||||
)
|
||||
assert client.gateway_id == MOCK_GATEWAY_ID
|
||||
assert client.api_secret == MOCK_API_SECRET
|
||||
assert client.private_key == MOCK_PRIVATE_KEY
|
||||
|
||||
|
||||
def test_private_key_defaults_to_none(hass: HomeAssistant) -> None:
|
||||
"""Test private_key defaults to None for simple (non-E2E) mode."""
|
||||
with _patch_session(MagicMock()):
|
||||
client = ThreemaAPIClient(hass, MOCK_GATEWAY_ID, MOCK_API_SECRET)
|
||||
assert client.private_key is None
|
||||
@@ -0,0 +1,756 @@
|
||||
"""Test the Threema Gateway config flow."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import probatio
|
||||
import pytest
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.threema.client import (
|
||||
ThreemaAuthError,
|
||||
ThreemaConnectionError,
|
||||
derive_public_key,
|
||||
)
|
||||
from homeassistant.components.threema.config_flow import _CONF_PUBLIC_KEY
|
||||
from homeassistant.components.threema.const import (
|
||||
CONF_API_SECRET,
|
||||
CONF_GATEWAY_ID,
|
||||
CONF_PRIVATE_KEY,
|
||||
DOMAIN,
|
||||
SUBENTRY_TYPE_RECIPIENT,
|
||||
)
|
||||
from homeassistant.const import CONF_NAME, CONF_RECIPIENT
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from .conftest import MOCK_API_SECRET, MOCK_GATEWAY_ID, MOCK_RECIPIENT_ID
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_setup_entry() -> Generator[None]:
|
||||
"""Patch async_setup_entry to avoid full setup during flow tests."""
|
||||
with patch("homeassistant.components.threema.async_setup_entry", return_value=True):
|
||||
yield
|
||||
|
||||
|
||||
async def test_user_flow_existing_gateway(
|
||||
hass: HomeAssistant, mock_credentials: AsyncMock
|
||||
) -> None:
|
||||
"""Test user flow with existing gateway credentials."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "credentials"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == f"Threema {MOCK_GATEWAY_ID}"
|
||||
assert result["data"] == {
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
}
|
||||
assert result["result"].unique_id == MOCK_GATEWAY_ID
|
||||
|
||||
|
||||
async def test_user_flow_existing_with_keys(
|
||||
hass: HomeAssistant, mock_credentials: AsyncMock
|
||||
) -> None:
|
||||
"""Test user flow with existing gateway including optional private key."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert (
|
||||
result["data"][CONF_PRIVATE_KEY]
|
||||
== "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
|
||||
)
|
||||
assert result["result"].unique_id == MOCK_GATEWAY_ID
|
||||
|
||||
|
||||
async def test_user_flow_new_gateway(
|
||||
hass: HomeAssistant, mock_credentials: AsyncMock
|
||||
) -> None:
|
||||
"""Test user flow with new gateway (key generation)."""
|
||||
generated_private_key = "0" * 64
|
||||
generated_public_key = "f" * 64
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.threema.config_flow.generate_key_pair",
|
||||
return_value=(generated_private_key, generated_public_key),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "setup_new"},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "setup_new"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "credentials"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: generated_private_key,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_PRIVATE_KEY] == generated_private_key
|
||||
assert result["result"].unique_id == MOCK_GATEWAY_ID
|
||||
|
||||
|
||||
async def test_user_flow_key_generation_failure(hass: HomeAssistant) -> None:
|
||||
"""Test user flow aborts when key generation fails."""
|
||||
with patch(
|
||||
"homeassistant.components.threema.config_flow.generate_key_pair",
|
||||
side_effect=RuntimeError("Key generation failed"),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "setup_new"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "key_generation_failed"
|
||||
|
||||
|
||||
async def test_credentials_invalid_gateway_id(
|
||||
hass: HomeAssistant, mock_credentials: AsyncMock
|
||||
) -> None:
|
||||
"""Test credentials step with invalid Gateway ID."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
# Gateway ID not starting with *
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: "TESTGWY1",
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": "invalid_gateway_id"}
|
||||
|
||||
# Gateway ID wrong length
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: "*TEST",
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": "invalid_gateway_id"}
|
||||
|
||||
# Right length and prefix, but invalid characters
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: "*!!!!!!!",
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": "invalid_gateway_id"}
|
||||
|
||||
# Valid Gateway ID — recover and create entry
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["result"].unique_id == MOCK_GATEWAY_ID
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"invalid_key",
|
||||
["0123456789abcdef", "g" * 64],
|
||||
ids=["wrong_length", "non_hex"],
|
||||
)
|
||||
async def test_credentials_invalid_private_key(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
invalid_key: str,
|
||||
) -> None:
|
||||
"""Test credentials step rejects a malformed private key."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: invalid_key,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_PRIVATE_KEY: "invalid_key"}
|
||||
|
||||
# Recover by clearing the invalid key (the field defaults to the
|
||||
# previous, invalid value, so it must be explicitly cleared)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"prefixed_key",
|
||||
[
|
||||
"private:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
|
||||
"PRIVATE:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
|
||||
],
|
||||
ids=["lower", "upper"],
|
||||
)
|
||||
async def test_credentials_private_key_prefix_stripped(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
prefixed_key: str,
|
||||
) -> None:
|
||||
"""Test a 'private:'-prefixed key (as exported by Threema's tools) is accepted."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: prefixed_key,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert (
|
||||
result["data"][CONF_PRIVATE_KEY]
|
||||
== "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
|
||||
)
|
||||
|
||||
|
||||
async def test_credentials_public_key_in_private_key_field_rejected(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a 'public:'-prefixed key pasted into the private-key field is rejected."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: f"public:{'a' * 64}",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_PRIVATE_KEY: "invalid_key"}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "a" * 64,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_credentials_private_key_in_public_key_field_rejected(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a 'private:'-prefixed key pasted into the public-key field is rejected."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "a" * 64,
|
||||
_CONF_PUBLIC_KEY: f"private:{'b' * 64}",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {_CONF_PUBLIC_KEY: "invalid_key"}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "a" * 64,
|
||||
_CONF_PUBLIC_KEY: "",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_credentials_public_key_without_private_key_rejected(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a public key given alone (no private key) is rejected, not discarded."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
_CONF_PUBLIC_KEY: "a" * 64,
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {_CONF_PUBLIC_KEY: "public_key_requires_private_key"}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
_CONF_PUBLIC_KEY: "",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_credentials_public_key_matches(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a public key matching the private key is accepted and not stored."""
|
||||
private_key = "1" * 64
|
||||
matching_public_key = derive_public_key(private_key)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: private_key,
|
||||
_CONF_PUBLIC_KEY: f"public:{matching_public_key}",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["data"][CONF_PRIVATE_KEY] == private_key
|
||||
assert _CONF_PUBLIC_KEY not in result["data"]
|
||||
|
||||
|
||||
async def test_credentials_public_key_mismatch(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a public key that does not match the private key is rejected."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "1" * 64,
|
||||
_CONF_PUBLIC_KEY: "f" * 64,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {_CONF_PUBLIC_KEY: "key_mismatch"}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "1" * 64,
|
||||
_CONF_PUBLIC_KEY: "",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_credentials_public_key_invalid_hex(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a malformed public key is rejected."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "1" * 64,
|
||||
_CONF_PUBLIC_KEY: "not-a-valid-key",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {_CONF_PUBLIC_KEY: "invalid_key"}
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "1" * 64,
|
||||
_CONF_PUBLIC_KEY: "",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_credentials_invalid_private_key_preserves_other_fields(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test gateway ID and API secret stay filled in after an invalid key error."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
CONF_PRIVATE_KEY: "not-a-valid-key",
|
||||
},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_PRIVATE_KEY: "invalid_key"}
|
||||
|
||||
defaults = {
|
||||
schema_key: schema_key.default()
|
||||
for schema_key in result["data_schema"].schema
|
||||
if schema_key.default is not probatio.UNDEFINED
|
||||
}
|
||||
assert defaults[CONF_GATEWAY_ID] == MOCK_GATEWAY_ID
|
||||
assert defaults[CONF_API_SECRET] == MOCK_API_SECRET
|
||||
|
||||
|
||||
async def test_credentials_already_configured(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test credentials step when gateway is already configured."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("side_effect", "expected_error"),
|
||||
[
|
||||
(ThreemaConnectionError("Connection refused"), "cannot_connect"),
|
||||
(ThreemaConnectionError("Server error"), "cannot_connect"),
|
||||
(ThreemaAuthError("Invalid credentials"), "invalid_auth"),
|
||||
(RuntimeError("Unexpected"), "unknown"),
|
||||
],
|
||||
ids=["cannot_connect", "server_error_non_auth", "invalid_auth", "unknown_error"],
|
||||
)
|
||||
async def test_credentials_error(
|
||||
hass: HomeAssistant,
|
||||
mock_credentials: AsyncMock,
|
||||
side_effect: Exception,
|
||||
expected_error: str,
|
||||
) -> None:
|
||||
"""Test credentials step with various errors."""
|
||||
mock_credentials.side_effect = side_effect
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.MENU
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{"next_step_id": "credentials"},
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {"base": expected_error}
|
||||
|
||||
mock_credentials.side_effect = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_GATEWAY_ID: MOCK_GATEWAY_ID,
|
||||
CONF_API_SECRET: MOCK_API_SECRET,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_subentry_add_recipient(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test adding a recipient via subentry flow."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.subentries.async_init(
|
||||
(mock_config_entry.entry_id, SUBENTRY_TYPE_RECIPIENT),
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_RECIPIENT: "EFGH5678"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "EFGH5678"
|
||||
assert result["data"] == {CONF_RECIPIENT: "EFGH5678"}
|
||||
assert result["unique_id"] == "EFGH5678"
|
||||
|
||||
|
||||
async def test_subentry_add_recipient_with_name(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test adding a recipient with a display name."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.subentries.async_init(
|
||||
(mock_config_entry.entry_id, SUBENTRY_TYPE_RECIPIENT),
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_RECIPIENT: "EFGH5678", "name": "Dad"},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == "Dad (EFGH5678)"
|
||||
assert result["data"] == {CONF_RECIPIENT: "EFGH5678", CONF_NAME: "Dad"}
|
||||
assert result["unique_id"] == "EFGH5678"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"invalid_id",
|
||||
["ABC", "ABCD!@#$", ""],
|
||||
ids=["too_short", "special_chars", "empty"],
|
||||
)
|
||||
async def test_subentry_invalid_recipient_id(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
invalid_id: str,
|
||||
) -> None:
|
||||
"""Test subentry flow rejects invalid Threema ID with an inline form error."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.subentries.async_init(
|
||||
(mock_config_entry.entry_id, SUBENTRY_TYPE_RECIPIENT),
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_RECIPIENT: invalid_id},
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_RECIPIENT: "invalid_recipient_id"}
|
||||
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_RECIPIENT: "EFGH5678"},
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
async def test_subentry_duplicate_recipient(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
) -> None:
|
||||
"""Test subentry flow rejects duplicate recipient."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.subentries.async_init(
|
||||
(mock_config_entry.entry_id, SUBENTRY_TYPE_RECIPIENT),
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
)
|
||||
|
||||
result = await hass.config_entries.subentries.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={CONF_RECIPIENT: MOCK_RECIPIENT_ID},
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Test the Threema Gateway integration setup."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.threema.client import (
|
||||
ThreemaAuthError,
|
||||
ThreemaConnectionError,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def test_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
) -> None:
|
||||
"""Test successful setup of a config entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("side_effect", "expected_state"),
|
||||
[
|
||||
(ThreemaConnectionError("Connection refused"), ConfigEntryState.SETUP_RETRY),
|
||||
(ThreemaAuthError("Invalid credentials"), ConfigEntryState.SETUP_ERROR),
|
||||
(ThreemaConnectionError("Server error"), ConfigEntryState.SETUP_RETRY),
|
||||
],
|
||||
ids=["connection_error", "auth_error", "server_error_non_auth"],
|
||||
)
|
||||
async def test_setup_entry_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
side_effect: Exception,
|
||||
expected_state: ConfigEntryState,
|
||||
) -> None:
|
||||
"""Test setup handles various errors correctly."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
mock_credentials.side_effect = side_effect
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert mock_config_entry.state is expected_state
|
||||
|
||||
|
||||
async def test_unload_entry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
) -> None:
|
||||
"""Test unloading a config entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_update_listener_reloads(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
) -> None:
|
||||
"""Test that update listener reloads the entry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
with patch(
|
||||
"homeassistant.config_entries.ConfigEntries.async_reload"
|
||||
) as mock_reload:
|
||||
mock_reload.return_value = None
|
||||
hass.config_entries.async_update_entry(mock_config_entry, title="Updated")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
mock_reload.assert_called_once_with(mock_config_entry.entry_id)
|
||||
@@ -0,0 +1,245 @@
|
||||
"""Test the Threema Gateway notify platform."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
|
||||
from homeassistant.components.threema.client import (
|
||||
ThreemaAuthError,
|
||||
ThreemaConnectionError,
|
||||
ThreemaSendError,
|
||||
)
|
||||
from homeassistant.components.threema.const import SUBENTRY_TYPE_RECIPIENT
|
||||
from homeassistant.config_entries import ConfigSubentryDataWithId
|
||||
from homeassistant.const import CONF_RECIPIENT
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
|
||||
from .conftest import MOCK_GATEWAY_ID, MOCK_RECIPIENT_ID, RECIPIENT_SUBENTRY
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
_SECOND_RECIPIENT_ID = "WXYZ9999"
|
||||
_SECOND_RECIPIENT_SUBENTRY: ConfigSubentryDataWithId = {
|
||||
"data": {CONF_RECIPIENT: _SECOND_RECIPIENT_ID},
|
||||
"subentry_id": "second_recipient_subentry_id",
|
||||
"subentry_type": SUBENTRY_TYPE_RECIPIENT,
|
||||
"title": "Second recipient",
|
||||
"unique_id": _SECOND_RECIPIENT_ID,
|
||||
}
|
||||
|
||||
|
||||
async def test_notify_entity_created(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test notify entity is created from subentry."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entities = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
notify_entities = [e for e in entities if e.domain == NOTIFY_DOMAIN]
|
||||
assert len(notify_entities) == 1
|
||||
assert notify_entities[0].unique_id == f"{MOCK_GATEWAY_ID}_{MOCK_RECIPIENT_ID}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mock_subentries", [[RECIPIENT_SUBENTRY, _SECOND_RECIPIENT_SUBENTRY]]
|
||||
)
|
||||
async def test_notify_entities_get_separate_devices(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
) -> None:
|
||||
"""Test each recipient gets its own device, not a shared one.
|
||||
|
||||
A device can only belong to a single config subentry; sharing one
|
||||
across recipient subentries makes Home Assistant silently reassign
|
||||
it whenever a new recipient is added, orphaning the previous one.
|
||||
"""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entities = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
notify_entities = [e for e in entities if e.domain == NOTIFY_DOMAIN]
|
||||
assert len(notify_entities) == 2
|
||||
|
||||
device_ids = {e.device_id for e in notify_entities}
|
||||
assert len(device_ids) == 2
|
||||
for device_id in device_ids:
|
||||
assert device_id is not None
|
||||
device = device_registry.async_get(device_id)
|
||||
assert device is not None
|
||||
assert device.config_entry_id == mock_config_entry.entry_id
|
||||
assert device.config_subentry_id is not None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mock_subentries", [[]])
|
||||
async def test_notify_entity_not_created_without_subentry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test no notify entity without subentries."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entities = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
notify_entities = [e for e in entities if e.domain == NOTIFY_DOMAIN]
|
||||
assert len(notify_entities) == 0
|
||||
|
||||
|
||||
async def test_send_message_simple(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test sending a message via notify entity (simple mode)."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entities = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
notify_entities = [e for e in entities if e.domain == NOTIFY_DOMAIN]
|
||||
assert len(notify_entities) == 1
|
||||
|
||||
await hass.services.async_call(
|
||||
NOTIFY_DOMAIN,
|
||||
"send_message",
|
||||
{"entity_id": notify_entities[0].entity_id, "message": "Hello from tests!"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
mock_send_message.assert_called_once_with(MOCK_RECIPIENT_ID, "Hello from tests!")
|
||||
|
||||
|
||||
async def test_send_message_with_title(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test a title is formatted as a leading bold line in the sent text."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entities = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
notify_entities = [e for e in entities if e.domain == NOTIFY_DOMAIN]
|
||||
assert len(notify_entities) == 1
|
||||
|
||||
await hass.services.async_call(
|
||||
NOTIFY_DOMAIN,
|
||||
"send_message",
|
||||
{
|
||||
"entity_id": notify_entities[0].entity_id,
|
||||
"message": "Hello from tests!",
|
||||
"title": "My Title",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
mock_send_message.assert_called_once_with(
|
||||
MOCK_RECIPIENT_ID, "*My Title*\nHello from tests!"
|
||||
)
|
||||
|
||||
|
||||
async def test_send_message_e2e(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry_with_keys: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test sending a message via notify entity (E2E mode)."""
|
||||
mock_config_entry_with_keys.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry_with_keys.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entities = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry_with_keys.entry_id
|
||||
)
|
||||
notify_entities = [e for e in entities if e.domain == NOTIFY_DOMAIN]
|
||||
assert len(notify_entities) == 1
|
||||
|
||||
await hass.services.async_call(
|
||||
NOTIFY_DOMAIN,
|
||||
"send_message",
|
||||
{"entity_id": notify_entities[0].entity_id, "message": "Hello E2E!"},
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
mock_send_message.assert_called_once_with(MOCK_RECIPIENT_ID, "Hello E2E!")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("side_effect", "expected_error"),
|
||||
[
|
||||
(ThreemaSendError("Send failed"), "Error sending message: Send failed"),
|
||||
(
|
||||
ThreemaConnectionError("Connection error"),
|
||||
"Error sending message: Connection error",
|
||||
),
|
||||
(ThreemaAuthError("Invalid credentials"), "Invalid authentication"),
|
||||
],
|
||||
ids=["send_error", "connection_error", "auth_error"],
|
||||
)
|
||||
async def test_send_message_error(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_credentials: AsyncMock,
|
||||
mock_send_message: AsyncMock,
|
||||
entity_registry: er.EntityRegistry,
|
||||
side_effect: Exception,
|
||||
expected_error: str,
|
||||
) -> None:
|
||||
"""Test notify entity raises HomeAssistantError on send/connection errors."""
|
||||
mock_send_message.side_effect = side_effect
|
||||
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
entities = er.async_entries_for_config_entry(
|
||||
entity_registry, mock_config_entry.entry_id
|
||||
)
|
||||
notify_entities = [e for e in entities if e.domain == NOTIFY_DOMAIN]
|
||||
|
||||
with pytest.raises(HomeAssistantError, match=expected_error):
|
||||
await hass.services.async_call(
|
||||
NOTIFY_DOMAIN,
|
||||
"send_message",
|
||||
{
|
||||
"entity_id": notify_entities[0].entity_id,
|
||||
"message": "Hello!",
|
||||
},
|
||||
blocking=True,
|
||||
)
|
||||
Reference in New Issue
Block a user