Compare commits

..

20 Commits

Author SHA1 Message Date
Daniel Hjelseth Høyer
205bc0456f Merge branch 'dev' into homevolt 2026-01-20 16:24:32 +01:00
Daniel Hjelseth Høyer
5aa32491c8 Merge branch 'dev' into homevolt 2026-01-15 16:25:46 +01:00
Daniel Hjelseth Høyer
dc2cd2246b Merge branch 'dev' into homevolt 2026-01-15 07:06:51 +01:00
Daniel Hjelseth Høyer
181037820b Merge branch 'dev' into homevolt 2026-01-14 21:05:45 +01:00
Daniel Hjelseth Høyer
6cf15bf70c homevolt
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-14 19:09:37 +01:00
Daniel Hjelseth Høyer
5a34c31e42 Merge branch 'dev' into homevolt 2026-01-14 18:30:20 +01:00
Daniel Hjelseth Høyer
9dcc86f12e homevolt
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-14 18:03:21 +01:00
Daniel Hjelseth Høyer
04429a6eef homevolt
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-14 17:40:51 +01:00
Daniel Hjelseth Høyer
51e2506afb homevolt
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-14 16:41:08 +01:00
Daniel Hjelseth Høyer
e49e5c7c40 Merge branch 'dev' into homevolt 2026-01-14 14:41:26 +01:00
Daniel Hjelseth Høyer
b8dfc523da homevolt
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-14 14:36:43 +01:00
Daniel Hjelseth Høyer
a25fbf57ef Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 17:20:27 +01:00
Daniel Hjelseth Høyer
dac22002b0 Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 14:53:07 +01:00
Daniel Hjelseth Høyer
e61f00a3ae Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 14:15:56 +01:00
Daniel Hjelseth Høyer
14a67c6b5d Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 10:46:49 +01:00
Daniel Hjelseth Høyer
90ae81f02b Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 10:39:46 +01:00
Daniel Hjelseth Høyer
a741f214da Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 10:35:53 +01:00
Daniel Hjelseth Høyer
21d0bd3ce2 Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 10:22:32 +01:00
Daniel Hjelseth Høyer
d9c1f4850a Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 10:09:50 +01:00
Daniel Hjelseth Høyer
335994af7e Add Homevolt integration
Signed-off-by: Daniel Hjelseth Høyer <github@dahoiv.net>
2026-01-07 09:44:06 +01:00
65 changed files with 1340 additions and 921 deletions

2
CODEOWNERS generated
View File

@@ -711,6 +711,8 @@ build.json @home-assistant/supervisor
/tests/components/homematic/ @pvizeli
/homeassistant/components/homematicip_cloud/ @hahn-th
/tests/components/homematicip_cloud/ @hahn-th
/homeassistant/components/homevolt/ @danielhiversen
/tests/components/homevolt/ @danielhiversen
/homeassistant/components/homewizard/ @DCSBL
/tests/components/homewizard/ @DCSBL
/homeassistant/components/honeywell/ @rdfurman @mkmer

View File

@@ -602,10 +602,6 @@ class AutomationEntity(BaseAutomationEntity, RestoreEntity):
"""Return a set of referenced labels."""
referenced = self.action_script.referenced_labels
if self._cond_func is not None:
for conf in self._cond_func.config:
referenced |= condition.async_extract_labels(conf)
for conf in self._trigger_config:
referenced |= set(_get_targets_from_trigger_config(conf, ATTR_LABEL_ID))
return referenced
@@ -615,10 +611,6 @@ class AutomationEntity(BaseAutomationEntity, RestoreEntity):
"""Return a set of referenced floors."""
referenced = self.action_script.referenced_floors
if self._cond_func is not None:
for conf in self._cond_func.config:
referenced |= condition.async_extract_floors(conf)
for conf in self._trigger_config:
referenced |= set(_get_targets_from_trigger_config(conf, ATTR_FLOOR_ID))
return referenced
@@ -628,10 +620,6 @@ class AutomationEntity(BaseAutomationEntity, RestoreEntity):
"""Return a set of referenced areas."""
referenced = self.action_script.referenced_areas
if self._cond_func is not None:
for conf in self._cond_func.config:
referenced |= condition.async_extract_areas(conf)
for conf in self._trigger_config:
referenced |= set(_get_targets_from_trigger_config(conf, ATTR_AREA_ID))
return referenced

View File

@@ -0,0 +1,36 @@
"""The Homevolt integration."""
from __future__ import annotations
from homevolt import Homevolt
from homeassistant.const import CONF_HOST, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .coordinator import HomevoltConfigEntry, HomevoltDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: HomevoltConfigEntry) -> bool:
"""Set up Homevolt from a config entry."""
host: str = entry.data[CONF_HOST]
password: str | None = entry.data.get(CONF_PASSWORD)
websession = async_get_clientsession(hass)
client = Homevolt(host, password, websession=websession)
coordinator = HomevoltDataUpdateCoordinator(hass, entry, client)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: HomevoltConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -0,0 +1,70 @@
"""Config flow for the Homevolt integration."""
from __future__ import annotations
import logging
from typing import Any
from homevolt import Homevolt, HomevoltAuthenticationError, HomevoltConnectionError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Optional(CONF_PASSWORD): str,
}
)
class HomevoltConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Homevolt."""
VERSION = 1
MINOR_VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
host = user_input[CONF_HOST]
password = user_input.get(CONF_PASSWORD)
websession = async_get_clientsession(self.hass)
client = Homevolt(host, password, websession=websession)
try:
await client.update_info()
device = client.get_device()
device_id = device.device_id
except HomevoltAuthenticationError:
errors["base"] = "invalid_auth"
except HomevoltConnectionError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception(
"Error occurred while connecting to the Homevolt battery"
)
errors["base"] = "unknown"
else:
await self.async_set_unique_id(device_id)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title="Homevolt Local",
data={
CONF_HOST: host,
CONF_PASSWORD: password,
},
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

View File

@@ -0,0 +1,9 @@
"""Constants for the Homevolt integration."""
from __future__ import annotations
from datetime import timedelta
DOMAIN = "homevolt"
MANUFACTURER = "Homevolt"
SCAN_INTERVAL = timedelta(seconds=15)

View File

@@ -0,0 +1,56 @@
"""Data update coordinator for Homevolt integration."""
from __future__ import annotations
import logging
from homevolt import (
Device,
Homevolt,
HomevoltAuthenticationError,
HomevoltConnectionError,
HomevoltError,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, SCAN_INTERVAL
type HomevoltConfigEntry = ConfigEntry[HomevoltDataUpdateCoordinator]
_LOGGER = logging.getLogger(__name__)
class HomevoltDataUpdateCoordinator(DataUpdateCoordinator[Device]):
"""Class to manage fetching Homevolt data."""
config_entry: HomevoltConfigEntry
def __init__(
self,
hass: HomeAssistant,
entry: HomevoltConfigEntry,
client: Homevolt,
) -> None:
"""Initialize the Homevolt coordinator."""
self.client = client
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
config_entry=entry,
)
async def _async_update_data(self) -> Device:
"""Fetch data from the Homevolt API."""
try:
await self.client.update_info()
return self.client.get_device()
except HomevoltAuthenticationError as err:
raise ConfigEntryAuthFailed from err
except (HomevoltConnectionError, HomevoltError) as err:
raise UpdateFailed(f"Error communicating with device: {err}") from err

View File

@@ -0,0 +1,12 @@
{
"domain": "homevolt",
"name": "Homevolt",
"codeowners": ["@danielhiversen"],
"config_flow": true,
"dependencies": [],
"documentation": "https://www.home-assistant.io/integrations/homevolt",
"integration_type": "device",
"iot_class": "local_polling",
"quality_scale": "bronze",
"requirements": ["homevolt==0.2.4"]
}

View File

@@ -0,0 +1,70 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration does not register custom actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration does not register custom actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: Local_polling without events
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:
status: exempt
comment: Integration does not register custom actions.
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: Integration does not have an options flow.
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class: done
entity-disabled-by-default: todo
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -0,0 +1,162 @@
"""Support for Homevolt sensors."""
from __future__ import annotations
from homevolt.models import SensorType
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS,
UnitOfElectricCurrent,
UnitOfElectricPotential,
UnitOfEnergy,
UnitOfFrequency,
UnitOfPower,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MANUFACTURER
from .coordinator import HomevoltConfigEntry, HomevoltDataUpdateCoordinator
PARALLEL_UPDATES = 0 # Coordinator-based updates
SENSORS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key=SensorType.COUNT,
state_class=SensorStateClass.TOTAL_INCREASING,
),
SensorEntityDescription(
key=SensorType.ENERGY_TOTAL,
device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
),
SensorEntityDescription(
key=SensorType.ENERGY_INCREASING,
device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
),
SensorEntityDescription(
key=SensorType.FREQUENCY,
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
),
SensorEntityDescription(
key=SensorType.PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
),
SensorEntityDescription(
key=SensorType.POWER,
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
),
SensorEntityDescription(
key=SensorType.SCHEDULE_TYPE,
),
SensorEntityDescription(
key=SensorType.SIGNAL_STRENGTH,
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS,
),
SensorEntityDescription(
key=SensorType.TEMPERATURE,
device_class=SensorDeviceClass.TEMPERATURE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
),
SensorEntityDescription(
key=SensorType.TEXT,
),
SensorEntityDescription(
key=SensorType.VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
),
SensorEntityDescription(
key=SensorType.CURRENT,
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: HomevoltConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the Homevolt sensor."""
coordinator = entry.runtime_data
entities = []
sensors_by_key = {sensor.key: sensor for sensor in SENSORS}
for sensor_key, sensor in coordinator.data.sensors.items():
if (description := sensors_by_key.get(sensor.type)) is None:
continue
entities.append(
HomevoltSensor(
description,
coordinator,
sensor_key,
)
)
async_add_entities(entities)
class HomevoltSensor(CoordinatorEntity[HomevoltDataUpdateCoordinator], SensorEntity):
"""Representation of a Homevolt sensor."""
_attr_has_entity_name = True
def __init__(
self,
description: SensorEntityDescription,
coordinator: HomevoltDataUpdateCoordinator,
sensor_key: str,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self.entity_description = description
device_id = coordinator.data.device_id
self._attr_unique_id = f"{device_id}_{sensor_key}"
sensor_data = coordinator.data.sensors[sensor_key]
self._attr_translation_key = sensor_data.slug
self._sensor_key = sensor_key
device_metadata = coordinator.data.device_metadata.get(
sensor_data.device_identifier
)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{device_id}_{sensor_data.device_identifier}")},
configuration_url=coordinator.client.base_url,
manufacturer=MANUFACTURER,
model=device_metadata.model if device_metadata else None,
name=device_metadata.name if device_metadata else None,
)
@property
def available(self) -> bool:
"""Return if entity is available."""
return super().available and self._sensor_key in self.coordinator.data.sensors
@property
def native_value(self) -> StateType:
"""Return the native value of the sensor."""
return self.coordinator.data.sensors[self._sensor_key].value

View File

@@ -0,0 +1,198 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"host": "The IP address or hostname of your Homevolt battery on your local network.",
"password": "The local password configured for your Homevolt battery, if required."
},
"description": "Connect Home Assistant to your Homevolt battery over the local network.",
"title": "Homevolt Local"
}
}
},
"entity": {
"sensor": {
"available_charging_energy": {
"name": "Available charging energy"
},
"available_charging_power": {
"name": "Available charging power"
},
"available_discharge_energy": {
"name": "Available discharge energy"
},
"available_discharge_power": {
"name": "Available discharge power"
},
"average_rssi_grid": {
"name": "Grid average RSSI"
},
"average_rssi_load": {
"name": "Load average RSSI"
},
"battery_state_of_charge": {
"name": "Battery state of charge"
},
"charge_cycles": {
"name": "Charge cycles"
},
"energy_exported_grid": {
"name": "Grid exported energy"
},
"energy_exported_load": {
"name": "Load exported energy"
},
"energy_imported_grid": {
"name": "Grid imported energy"
},
"energy_imported_load": {
"name": "Load imported energy"
},
"exported_energy": {
"name": "Exported energy"
},
"frequency": {
"name": "Frequency"
},
"imported_energy": {
"name": "Imported energy"
},
"l1_current": {
"name": "L1 current"
},
"l1_current_grid": {
"name": "Grid L1 current"
},
"l1_current_load": {
"name": "Load L1 current"
},
"l1_l2_voltage": {
"name": "L1-L2 voltage"
},
"l1_power_grid": {
"name": "Grid L1 power"
},
"l1_power_load": {
"name": "Load L1 power"
},
"l1_voltage": {
"name": "L1 voltage"
},
"l1_voltage_grid": {
"name": "Grid L1 voltage"
},
"l1_voltage_load": {
"name": "Load L1 voltage"
},
"l2_current": {
"name": "L2 current"
},
"l2_current_grid": {
"name": "Grid L2 current"
},
"l2_current_load": {
"name": "Load L2 current"
},
"l2_l3_voltage": {
"name": "L2-L3 voltage"
},
"l2_power_grid": {
"name": "Grid L2 power"
},
"l2_power_load": {
"name": "Load L2 power"
},
"l2_voltage": {
"name": "L2 voltage"
},
"l2_voltage_grid": {
"name": "Grid L2 voltage"
},
"l2_voltage_load": {
"name": "Load L2 voltage"
},
"l3_current": {
"name": "L3 current"
},
"l3_current_grid": {
"name": "Grid L3 current"
},
"l3_current_load": {
"name": "Load L3 current"
},
"l3_l1_voltage": {
"name": "L3-L1 voltage"
},
"l3_power_grid": {
"name": "Grid L3 power"
},
"l3_power_load": {
"name": "Load L3 power"
},
"l3_voltage": {
"name": "L3 voltage"
},
"l3_voltage_grid": {
"name": "Grid L3 voltage"
},
"l3_voltage_load": {
"name": "Load L3 voltage"
},
"power": {
"name": "Power"
},
"power_grid": {
"name": "Grid power"
},
"power_load": {
"name": "Load power"
},
"rssi_grid": {
"name": "Grid RSSI"
},
"rssi_load": {
"name": "Load RSSI"
},
"schedule_id": {
"name": "Schedule ID"
},
"schedule_max_discharge": {
"name": "Schedule max discharge"
},
"schedule_max_power": {
"name": "Schedule max power"
},
"schedule_power_setpoint": {
"name": "Schedule power setpoint"
},
"schedule_type": {
"name": "Schedule type"
},
"state_of_charge": {
"name": "State of charge"
},
"system_temperature": {
"name": "System temperature"
},
"tmax": {
"name": "Maximum temperature"
},
"tmin": {
"name": "Minimum temperature"
}
}
}
}

View File

@@ -28,7 +28,6 @@ from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_track_time_interval
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.util.ssl import SSL_ALPN_HTTP11_HTTP2
from .const import DOMAIN, UPDATE_INTERVAL
from .entity import AqualinkEntity
@@ -67,11 +66,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: AqualinkConfigEntry) ->
username = entry.data[CONF_USERNAME]
password = entry.data[CONF_PASSWORD]
aqualink = AqualinkClient(
username,
password,
httpx_client=get_async_client(hass, alpn_protocols=SSL_ALPN_HTTP11_HTTP2),
)
aqualink = AqualinkClient(username, password, httpx_client=get_async_client(hass))
try:
await aqualink.login()
except AqualinkServiceException as login_exception:

View File

@@ -15,7 +15,6 @@ import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.util.ssl import SSL_ALPN_HTTP11_HTTP2
from .const import DOMAIN
@@ -37,11 +36,7 @@ class AqualinkFlowHandler(ConfigFlow, domain=DOMAIN):
try:
async with AqualinkClient(
username,
password,
httpx_client=get_async_client(
self.hass, alpn_protocols=SSL_ALPN_HTTP11_HTTP2
),
username, password, httpx_client=get_async_client(self.hass)
):
pass
except AqualinkServiceUnauthorizedException:

View File

@@ -7,5 +7,5 @@
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["kostal"],
"requirements": ["pykoplenti==1.5.0"]
"requirements": ["pykoplenti==1.3.0"]
}

View File

@@ -1,47 +1,24 @@
"""Provides triggers for lights."""
from typing import Any
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.trigger import (
EntityNumericalStateAttributeChangedTriggerBase,
EntityNumericalStateAttributeCrossedThresholdTriggerBase,
Trigger,
make_entity_numerical_state_attribute_changed_trigger,
make_entity_numerical_state_attribute_crossed_threshold_trigger,
make_entity_target_state_trigger,
)
from . import ATTR_BRIGHTNESS
from .const import DOMAIN
def _convert_uint8_to_percentage(value: Any) -> float:
"""Convert a uint8 value (0-255) to a percentage (0-100)."""
return (float(value) / 255.0) * 100.0
class BrightnessChangedTrigger(EntityNumericalStateAttributeChangedTriggerBase):
"""Trigger for brightness changed."""
_domain = DOMAIN
_attribute = ATTR_BRIGHTNESS
_converter = staticmethod(_convert_uint8_to_percentage)
class BrightnessCrossedThresholdTrigger(
EntityNumericalStateAttributeCrossedThresholdTriggerBase
):
"""Trigger for brightness crossed threshold."""
_domain = DOMAIN
_attribute = ATTR_BRIGHTNESS
_converter = staticmethod(_convert_uint8_to_percentage)
TRIGGERS: dict[str, type[Trigger]] = {
"brightness_changed": BrightnessChangedTrigger,
"brightness_crossed_threshold": BrightnessCrossedThresholdTrigger,
"brightness_changed": make_entity_numerical_state_attribute_changed_trigger(
DOMAIN, ATTR_BRIGHTNESS
),
"brightness_crossed_threshold": make_entity_numerical_state_attribute_crossed_threshold_trigger(
DOMAIN, ATTR_BRIGHTNESS
),
"turned_off": make_entity_target_state_trigger(DOMAIN, STATE_OFF),
"turned_on": make_entity_target_state_trigger(DOMAIN, STATE_ON),
}

View File

@@ -22,10 +22,7 @@
number:
selector:
number:
max: 100
min: 0
mode: box
unit_of_measurement: "%"
entity:
selector:
entity:

View File

@@ -8,9 +8,6 @@
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"user": {
"data": {

View File

@@ -178,7 +178,6 @@ class OneDriveBackupAgent(BackupAgent):
file,
upload_chunk_size=upload_chunk_size,
session=async_get_clientsession(self._hass),
smart_chunk_size=True,
)
except HashMismatchError as err:
raise BackupAgentError(

View File

@@ -10,5 +10,5 @@
"iot_class": "cloud_polling",
"loggers": ["onedrive_personal_sdk"],
"quality_scale": "platinum",
"requirements": ["onedrive-personal-sdk==0.1.1"]
"requirements": ["onedrive-personal-sdk==0.1.0"]
}

View File

@@ -13,9 +13,6 @@
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"reauth_confirm": {
"data": {

View File

@@ -9,7 +9,6 @@
}
],
"documentation": "https://www.home-assistant.io/integrations/qnap_qsw",
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["aioqsw"],
"requirements": ["aioqsw==0.4.2"]

View File

@@ -5,7 +5,6 @@
"codeowners": ["@rabbit-air"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/rabbitair",
"integration_type": "device",
"iot_class": "local_polling",
"requirements": ["python-rabbitair==0.0.8"],
"zeroconf": ["_rabbitair._udp.local."]

View File

@@ -13,7 +13,6 @@
}
],
"documentation": "https://www.home-assistant.io/integrations/radiotherm",
"integration_type": "device",
"iot_class": "local_polling",
"loggers": ["radiotherm"],
"requirements": ["radiotherm==2.1.0"]

View File

@@ -5,7 +5,6 @@
"config_flow": true,
"dependencies": ["usb"],
"documentation": "https://www.home-assistant.io/integrations/rainforest_raven",
"integration_type": "hub",
"iot_class": "local_polling",
"requirements": ["aioraven==0.7.1"],
"usb": [

View File

@@ -15,7 +15,6 @@
"config_flow": true,
"dependencies": ["bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/rapt_ble",
"integration_type": "device",
"iot_class": "local_push",
"requirements": ["rapt-ble==0.1.2"]
}

View File

@@ -4,7 +4,6 @@
"codeowners": ["@ashionky"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/refoss",
"integration_type": "hub",
"iot_class": "local_polling",
"requirements": ["refoss-ha==1.2.5"],
"single_config_entry": true

View File

@@ -10,7 +10,6 @@
}
],
"documentation": "https://www.home-assistant.io/integrations/rehlko",
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["aiokem"],
"quality_scale": "silver",

View File

@@ -4,7 +4,6 @@
"codeowners": ["@jimmyd-be"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/renson",
"integration_type": "device",
"iot_class": "local_polling",
"requirements": ["renson-endura-delta==1.7.2"]
}

View File

@@ -4,7 +4,6 @@
"codeowners": ["@danielhiversen", "@elupus", "@RobBie1221"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/rfxtrx",
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["RFXtrx"],
"requirements": ["pyRFXtrx==0.31.1"]

View File

@@ -4,7 +4,6 @@
"codeowners": ["@milanmeu", "@frenck", "@quebulm"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/rituals_perfume_genie",
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["pyrituals"],
"requirements": ["pyrituals==0.0.7"]

View File

@@ -4,7 +4,6 @@
"codeowners": ["@xeniter"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/romy",
"integration_type": "device",
"iot_class": "local_polling",
"requirements": ["romy==0.0.10"],
"zeroconf": ["_aicu-http._tcp.local."]

View File

@@ -22,7 +22,6 @@
}
],
"documentation": "https://www.home-assistant.io/integrations/roomba",
"integration_type": "device",
"iot_class": "local_push",
"loggers": ["paho_mqtt", "roombapy"],
"requirements": ["roombapy==1.9.0"],

View File

@@ -4,7 +4,6 @@
"codeowners": ["@pavoni"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/roon",
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["roonapi"],
"requirements": ["roonapi==0.1.6"]

View File

@@ -4,7 +4,6 @@
"codeowners": [],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/rova",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["rova"],
"requirements": ["rova==0.4.1"]

View File

@@ -4,7 +4,6 @@
"codeowners": ["@noahhusby"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/russound_rio",
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["aiorussound"],
"quality_scale": "silver",

View File

@@ -10,7 +10,6 @@
}
],
"documentation": "https://www.home-assistant.io/integrations/ruuvi_gateway",
"integration_type": "device",
"iot_class": "local_polling",
"requirements": ["aioruuvigateway==0.1.0"]
}

View File

@@ -15,7 +15,6 @@
"config_flow": true,
"dependencies": ["bluetooth_adapters"],
"documentation": "https://www.home-assistant.io/integrations/ruuvitag_ble",
"integration_type": "device",
"iot_class": "local_push",
"requirements": ["ruuvitag-ble==0.4.0"]
}

View File

@@ -4,7 +4,6 @@
"codeowners": ["@OnFreund", "@elad-bar", "@maorcc"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/rympro",
"integration_type": "hub",
"iot_class": "cloud_polling",
"requirements": ["pyrympro==0.0.9"]
}

View File

@@ -4,7 +4,6 @@
"codeowners": ["@shaiu", "@jpbede"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/sabnzbd",
"integration_type": "service",
"iot_class": "local_polling",
"loggers": ["pysabnzbd"],
"quality_scale": "bronze",

View File

@@ -4,7 +4,6 @@
"codeowners": ["@dknowles2"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/schlage",
"integration_type": "hub",
"iot_class": "cloud_polling",
"requirements": ["pyschlage==2025.9.0"]
}

View File

@@ -18,7 +18,6 @@
}
],
"documentation": "https://www.home-assistant.io/integrations/sense",
"integration_type": "hub",
"iot_class": "cloud_polling",
"loggers": ["sense_energy"],
"requirements": ["sense-energy==0.13.8"]

View File

@@ -18,9 +18,6 @@
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"pick_implementation": {
"data": {

View File

@@ -10,9 +10,6 @@
"unknown": "[%key:common::config_flow::error::unknown%]",
"wrong_account": "Wrong account: Please authenticate with {username}."
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"reauth_confirm": {
"description": "The Twitch integration needs to re-authenticate your account",

View File

@@ -21,9 +21,6 @@
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"oauth_discovery": {
"description": "Home Assistant has found a Withings device on your network. Be aware that the setup of Withings is more complicated than many other integrations. Press **Submit** to continue setting up Withings."

View File

@@ -15,9 +15,6 @@
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"oauth_discovery": {
"description": "Home Assistant has found an Xbox device on your network. Press **Submit** to continue setting up the Xbox integration.",

View File

@@ -17,9 +17,6 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
},
"initiate_flow": {
"user": "[%key:common::config_flow::initiate_flow::account%]"
},
"step": {
"channels": {
"data": { "channels": "YouTube channels" },

View File

@@ -294,6 +294,7 @@ FLOWS = {
"homekit",
"homekit_controller",
"homematicip_cloud",
"homevolt",
"homewizard",
"homeworks",
"honeywell",

View File

@@ -2836,6 +2836,12 @@
"zwave"
]
},
"homevolt": {
"name": "Homevolt",
"integration_type": "device",
"config_flow": true,
"iot_class": "local_polling"
},
"homewizard": {
"name": "HomeWizard",
"integration_type": "device",
@@ -5375,7 +5381,7 @@
"name": "QNAP"
},
"qnap_qsw": {
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling",
"name": "QNAP QSW"
@@ -5413,7 +5419,7 @@
},
"rabbitair": {
"name": "Rabbit Air",
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},
@@ -5438,7 +5444,7 @@
},
"radiotherm": {
"name": "Radio Thermostat",
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},
@@ -5473,7 +5479,7 @@
},
"rapt_ble": {
"name": "RAPT Bluetooth",
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push"
},
@@ -5571,7 +5577,7 @@
},
"renson": {
"name": "Renson",
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},
@@ -5679,13 +5685,13 @@
},
"romy": {
"name": "ROMY Vacuum Cleaner",
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},
"roomba": {
"name": "iRobot Roomba and Braava",
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push"
},
@@ -5720,7 +5726,7 @@
},
"rova": {
"name": "ROVA",
"integration_type": "service",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
@@ -5763,13 +5769,13 @@
"name": "Ruuvi",
"integrations": {
"ruuvi_gateway": {
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling",
"name": "Ruuvi Gateway"
},
"ruuvitag_ble": {
"integration_type": "device",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push",
"name": "Ruuvi BLE"
@@ -5784,7 +5790,7 @@
},
"sabnzbd": {
"name": "SABnzbd",
"integration_type": "service",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_polling"
},

View File

@@ -370,13 +370,9 @@ def _async_get_connector(
return connectors[connector_key]
if verify_ssl:
ssl_context: SSLContext = ssl_util.client_context(
ssl_cipher, ssl_util.SSL_ALPN_HTTP11
)
ssl_context: SSLContext = ssl_util.client_context(ssl_cipher)
else:
ssl_context = ssl_util.client_context_no_verify(
ssl_cipher, ssl_util.SSL_ALPN_HTTP11
)
ssl_context = ssl_util.client_context_no_verify(ssl_cipher)
connector = HomeAssistantTCPConnector(
family=family,

View File

@@ -17,7 +17,6 @@ from typing import (
TYPE_CHECKING,
Any,
Final,
Literal,
Protocol,
TypedDict,
Unpack,
@@ -29,10 +28,7 @@ from typing import (
import voluptuous as vol
from homeassistant.const import (
ATTR_AREA_ID,
ATTR_DEVICE_CLASS,
ATTR_FLOOR_ID,
ATTR_LABEL_ID,
CONF_ABOVE,
CONF_AFTER,
CONF_ATTRIBUTE,
@@ -1350,18 +1346,13 @@ def async_extract_entities(config: ConfigType | Template) -> set[str]:
if entity_ids is not None:
referenced.update(entity_ids)
if target_entities := _get_targets_from_condition_config(
config, CONF_ENTITY_ID
):
referenced.update(target_entities)
return referenced
@callback
def async_extract_devices(config: ConfigType | Template) -> set[str]:
"""Extract devices from a condition."""
referenced: set[str] = set()
referenced = set()
to_process = deque([config])
while to_process:
@@ -1375,75 +1366,15 @@ def async_extract_devices(config: ConfigType | Template) -> set[str]:
to_process.extend(config["conditions"])
continue
if condition == "device":
if (device_id := config.get(CONF_DEVICE_ID)) is not None:
referenced.add(device_id)
if condition != "device":
continue
if target_devices := _get_targets_from_condition_config(config, CONF_DEVICE_ID):
referenced.update(target_devices)
if (device_id := config.get(CONF_DEVICE_ID)) is not None:
referenced.add(device_id)
return referenced
@callback
def async_extract_areas(config: ConfigType | Template) -> set[str]:
"""Extract areas from a condition."""
return _async_extract_targets(config, ATTR_AREA_ID)
@callback
def async_extract_floors(config: ConfigType | Template) -> set[str]:
"""Extract floors from a condition."""
return _async_extract_targets(config, ATTR_FLOOR_ID)
@callback
def async_extract_labels(config: ConfigType | Template) -> set[str]:
"""Extract labels from a condition."""
return _async_extract_targets(config, ATTR_LABEL_ID)
@callback
def _async_extract_targets(
config: ConfigType | Template,
target_type: Literal["entity_id", "device_id", "area_id", "floor_id", "label_id"],
) -> set[str]:
"""Extract targets from a condition."""
referenced: set[str] = set()
to_process = deque([config])
while to_process:
config = to_process.popleft()
if isinstance(config, Template):
continue
condition = config[CONF_CONDITION]
if condition in ("and", "not", "or"):
to_process.extend(config["conditions"])
continue
if targets := _get_targets_from_condition_config(config, target_type):
referenced.update(targets)
return referenced
@callback
def _get_targets_from_condition_config(
config: ConfigType,
target: Literal["entity_id", "device_id", "area_id", "floor_id", "label_id"],
) -> list[str]:
"""Extract targets from a condition target config."""
if not (target_conf := config.get(CONF_TARGET)):
return []
if not (targets := target_conf.get(target)):
return []
return [targets] if isinstance(targets, str) else targets
def _load_conditions_file(integration: Integration) -> dict[str, Any]:
"""Load conditions file for an integration."""
try:

View File

@@ -17,9 +17,6 @@ from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.loader import bind_hass
from homeassistant.util.hass_dict import HassKey
from homeassistant.util.ssl import (
SSL_ALPN_HTTP11,
SSL_ALPN_HTTP11_HTTP2,
SSLALPNProtocols,
SSLCipherList,
client_context,
create_no_verify_ssl_context,
@@ -31,9 +28,9 @@ from .frame import warn_use
# and we want to keep the connection open for a while so we
# don't have to reconnect every time so we use 15s to match aiohttp.
KEEP_ALIVE_TIMEOUT = 15
# Shared httpx clients keyed by (verify_ssl, alpn_protocols)
DATA_ASYNC_CLIENT: HassKey[dict[tuple[bool, SSLALPNProtocols], httpx.AsyncClient]] = (
HassKey("httpx_async_client")
DATA_ASYNC_CLIENT: HassKey[httpx.AsyncClient] = HassKey("httpx_async_client")
DATA_ASYNC_CLIENT_NOVERIFY: HassKey[httpx.AsyncClient] = HassKey(
"httpx_async_client_noverify"
)
DEFAULT_LIMITS = limits = httpx.Limits(keepalive_expiry=KEEP_ALIVE_TIMEOUT)
SERVER_SOFTWARE = (
@@ -45,26 +42,15 @@ USER_AGENT = "User-Agent"
@callback
@bind_hass
def get_async_client(
hass: HomeAssistant,
verify_ssl: bool = True,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_HTTP11,
) -> httpx.AsyncClient:
def get_async_client(hass: HomeAssistant, verify_ssl: bool = True) -> httpx.AsyncClient:
"""Return default httpx AsyncClient.
This method must be run in the event loop.
Pass alpn_protocols=SSL_ALPN_HTTP11_HTTP2 to get a client configured for HTTP/2.
Clients are cached separately by ALPN protocol to ensure proper SSL context
configuration (ALPN protocols differ between HTTP versions).
"""
client_key = (verify_ssl, alpn_protocols)
clients = hass.data.setdefault(DATA_ASYNC_CLIENT, {})
key = DATA_ASYNC_CLIENT if verify_ssl else DATA_ASYNC_CLIENT_NOVERIFY
if (client := clients.get(client_key)) is None:
client = clients[client_key] = create_async_httpx_client(
hass, verify_ssl, alpn_protocols=alpn_protocols
)
if (client := hass.data.get(key)) is None:
client = hass.data[key] = create_async_httpx_client(hass, verify_ssl)
return client
@@ -91,7 +77,6 @@ def create_async_httpx_client(
verify_ssl: bool = True,
auto_cleanup: bool = True,
ssl_cipher_list: SSLCipherList = SSLCipherList.PYTHON_DEFAULT,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_HTTP11,
**kwargs: Any,
) -> httpx.AsyncClient:
"""Create a new httpx.AsyncClient with kwargs, i.e. for cookies.
@@ -99,22 +84,13 @@ def create_async_httpx_client(
If auto_cleanup is False, the client will be
automatically closed on homeassistant_stop.
Pass alpn_protocols=SSL_ALPN_HTTP11_HTTP2 for HTTP/2 support (automatically
enables httpx http2 mode).
This method must be run in the event loop.
"""
# Use the requested ALPN protocols directly to ensure proper SSL context
# bucketing. httpx/httpcore mutates SSL contexts by calling set_alpn_protocols(),
# so we pre-set the correct protocols to prevent shared context corruption.
ssl_context = (
client_context(ssl_cipher_list, alpn_protocols)
client_context(ssl_cipher_list)
if verify_ssl
else create_no_verify_ssl_context(ssl_cipher_list, alpn_protocols)
else create_no_verify_ssl_context(ssl_cipher_list)
)
# Enable httpx HTTP/2 mode when HTTP/2 protocol is requested
if alpn_protocols == SSL_ALPN_HTTP11_HTTP2:
kwargs.setdefault("http2", True)
client = HassHttpXAsyncClient(
verify=ssl_context,
headers={USER_AGENT: SERVER_SOFTWARE},

View File

@@ -594,8 +594,6 @@ class EntityNumericalStateAttributeChangedTriggerBase(EntityTriggerBase):
_above: None | float | str
_below: None | float | str
_converter: Callable[[Any], float] = float
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the state trigger."""
super().__init__(hass, config)
@@ -618,7 +616,7 @@ class EntityNumericalStateAttributeChangedTriggerBase(EntityTriggerBase):
return False
try:
current_value = self._converter(_attribute_value)
current_value = float(_attribute_value)
except (TypeError, ValueError):
# Attribute is not a valid number, don't trigger
return False
@@ -708,8 +706,6 @@ class EntityNumericalStateAttributeCrossedThresholdTriggerBase(EntityTriggerBase
_upper_limit: float | str | None = None
_threshold_type: ThresholdType
_converter: Callable[[Any], float] = float
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
"""Initialize the state trigger."""
super().__init__(hass, config)
@@ -745,7 +741,7 @@ class EntityNumericalStateAttributeCrossedThresholdTriggerBase(EntityTriggerBase
return False
try:
current_value = self._converter(_attribute_value)
current_value = float(_attribute_value)
except (TypeError, ValueError):
# Attribute is not a valid number, don't trigger
return False

View File

@@ -8,17 +8,6 @@ import ssl
import certifi
# Type alias for ALPN protocols tuple (None means no ALPN protocols set)
type SSLALPNProtocols = tuple[str, ...] | None
# ALPN protocol configurations
# No ALPN protocols - used for libraries that don't support/need ALPN (e.g., aioimap)
SSL_ALPN_NONE: SSLALPNProtocols = None
# HTTP/1.1 only - used by default and for aiohttp (which doesn't support HTTP/2)
SSL_ALPN_HTTP11: SSLALPNProtocols = ("http/1.1",)
# HTTP/1.1 with HTTP/2 support - used when httpx http2=True
SSL_ALPN_HTTP11_HTTP2: SSLALPNProtocols = ("http/1.1", "h2")
class SSLCipherList(StrEnum):
"""SSL cipher lists."""
@@ -75,10 +64,7 @@ SSL_CIPHER_LISTS = {
@cache
def _client_context_no_verify(
ssl_cipher_list: SSLCipherList,
alpn_protocols: SSLALPNProtocols,
) -> ssl.SSLContext:
def _client_context_no_verify(ssl_cipher_list: SSLCipherList) -> ssl.SSLContext:
# This is a copy of aiohttp's create_default_context() function, with the
# ssl verify turned off.
# https://github.com/aio-libs/aiohttp/blob/33953f110e97eecc707e1402daa8d543f38a189b/aiohttp/connector.py#L911
@@ -92,18 +78,12 @@ def _client_context_no_verify(
sslcontext.set_default_verify_paths()
if ssl_cipher_list != SSLCipherList.PYTHON_DEFAULT:
sslcontext.set_ciphers(SSL_CIPHER_LISTS[ssl_cipher_list])
# Set ALPN protocols to prevent downstream libraries (e.g., httpx/httpcore)
# from mutating the shared SSL context with different protocol settings.
# If alpn_protocols is None, don't set ALPN (for libraries like aioimap).
if alpn_protocols is not None:
sslcontext.set_alpn_protocols(list(alpn_protocols))
return sslcontext
def _create_client_context(
ssl_cipher_list: SSLCipherList = SSLCipherList.PYTHON_DEFAULT,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_NONE,
) -> ssl.SSLContext:
"""Return an independent SSL context for making requests."""
# Reuse environment variable definition from requests, since it's already a
@@ -116,11 +96,6 @@ def _create_client_context(
)
if ssl_cipher_list != SSLCipherList.PYTHON_DEFAULT:
sslcontext.set_ciphers(SSL_CIPHER_LISTS[ssl_cipher_list])
# Set ALPN protocols to prevent downstream libraries (e.g., httpx/httpcore)
# from mutating the shared SSL context with different protocol settings.
# If alpn_protocols is None, don't set ALPN (for libraries like aioimap).
if alpn_protocols is not None:
sslcontext.set_alpn_protocols(list(alpn_protocols))
return sslcontext
@@ -128,63 +103,63 @@ def _create_client_context(
@cache
def _client_context(
ssl_cipher_list: SSLCipherList = SSLCipherList.PYTHON_DEFAULT,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_NONE,
) -> ssl.SSLContext:
# Cached version of _create_client_context
return _create_client_context(ssl_cipher_list, alpn_protocols)
return _create_client_context(ssl_cipher_list)
# Pre-warm the cache for ALL SSL context configurations at module load time.
# This is critical because creating SSL contexts loads certificates from disk,
# which is blocking I/O that must not happen in the event loop.
_SSL_ALPN_PROTOCOLS = (SSL_ALPN_NONE, SSL_ALPN_HTTP11, SSL_ALPN_HTTP11_HTTP2)
for _cipher in SSLCipherList:
for _alpn in _SSL_ALPN_PROTOCOLS:
_client_context(_cipher, _alpn)
_client_context_no_verify(_cipher, _alpn)
# Create this only once and reuse it
_DEFAULT_SSL_CONTEXT = _client_context(SSLCipherList.PYTHON_DEFAULT)
_DEFAULT_NO_VERIFY_SSL_CONTEXT = _client_context_no_verify(SSLCipherList.PYTHON_DEFAULT)
_NO_VERIFY_SSL_CONTEXTS = {
SSLCipherList.INTERMEDIATE: _client_context_no_verify(SSLCipherList.INTERMEDIATE),
SSLCipherList.MODERN: _client_context_no_verify(SSLCipherList.MODERN),
SSLCipherList.INSECURE: _client_context_no_verify(SSLCipherList.INSECURE),
}
_SSL_CONTEXTS = {
SSLCipherList.INTERMEDIATE: _client_context(SSLCipherList.INTERMEDIATE),
SSLCipherList.MODERN: _client_context(SSLCipherList.MODERN),
SSLCipherList.INSECURE: _client_context(SSLCipherList.INSECURE),
}
def get_default_context() -> ssl.SSLContext:
"""Return the default SSL context."""
return _client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11)
return _DEFAULT_SSL_CONTEXT
def get_default_no_verify_context() -> ssl.SSLContext:
"""Return the default SSL context that does not verify the server certificate."""
return _client_context_no_verify(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11)
return _DEFAULT_NO_VERIFY_SSL_CONTEXT
def client_context_no_verify(
ssl_cipher_list: SSLCipherList = SSLCipherList.PYTHON_DEFAULT,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_NONE,
) -> ssl.SSLContext:
"""Return a SSL context with no verification with a specific ssl cipher."""
return _client_context_no_verify(ssl_cipher_list, alpn_protocols)
return _NO_VERIFY_SSL_CONTEXTS.get(ssl_cipher_list, _DEFAULT_NO_VERIFY_SSL_CONTEXT)
def client_context(
ssl_cipher_list: SSLCipherList = SSLCipherList.PYTHON_DEFAULT,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_NONE,
) -> ssl.SSLContext:
"""Return an SSL context for making requests."""
return _client_context(ssl_cipher_list, alpn_protocols)
return _SSL_CONTEXTS.get(ssl_cipher_list, _DEFAULT_SSL_CONTEXT)
def create_client_context(
ssl_cipher_list: SSLCipherList = SSLCipherList.PYTHON_DEFAULT,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_NONE,
) -> ssl.SSLContext:
"""Return an independent SSL context for making requests."""
# This explicitly uses the non-cached version to create a client context
return _create_client_context(ssl_cipher_list, alpn_protocols)
return _create_client_context(ssl_cipher_list)
def create_no_verify_ssl_context(
ssl_cipher_list: SSLCipherList = SSLCipherList.PYTHON_DEFAULT,
alpn_protocols: SSLALPNProtocols = SSL_ALPN_NONE,
) -> ssl.SSLContext:
"""Return an SSL context that does not verify the server certificate."""
return _client_context_no_verify(ssl_cipher_list, alpn_protocols)
return _client_context_no_verify(ssl_cipher_list)
def server_context_modern() -> ssl.SSLContext:

7
requirements_all.txt generated
View File

@@ -1226,6 +1226,9 @@ homelink-integration-api==0.0.1
# homeassistant.components.homematicip_cloud
homematicip==2.4.0
# homeassistant.components.homevolt
homevolt==0.2.4
# homeassistant.components.horizon
horimote==0.4.1
@@ -1646,7 +1649,7 @@ omnilogic==0.4.5
ondilo==0.5.0
# homeassistant.components.onedrive
onedrive-personal-sdk==0.1.1
onedrive-personal-sdk==0.1.0
# homeassistant.components.onvif
onvif-zeep-async==4.0.4
@@ -2147,7 +2150,7 @@ pykmtronic==0.3.0
pykodi==0.2.7
# homeassistant.components.kostal_plenticore
pykoplenti==1.5.0
pykoplenti==1.3.0
# homeassistant.components.kraken
pykrakenapi==0.1.8

View File

@@ -1084,6 +1084,9 @@ homelink-integration-api==0.0.1
# homeassistant.components.homematicip_cloud
homematicip==2.4.0
# homeassistant.components.homevolt
homevolt==0.2.4
# homeassistant.components.remember_the_milk
httplib2==0.20.4
@@ -1429,7 +1432,7 @@ omnilogic==0.4.5
ondilo==0.5.0
# homeassistant.components.onedrive
onedrive-personal-sdk==0.1.1
onedrive-personal-sdk==0.1.0
# homeassistant.components.onvif
onvif-zeep-async==4.0.4
@@ -1821,7 +1824,7 @@ pykmtronic==0.3.0
pykodi==0.2.7
# homeassistant.components.kostal_plenticore
pykoplenti==1.5.0
pykoplenti==1.3.0
# homeassistant.components.kraken
pykrakenapi==0.1.8

View File

@@ -2232,7 +2232,7 @@ async def test_extraction_functions(
assert automation.blueprint_in_automation(hass, "automation.test3") is None
async def test_extraction_functions_with_trigger_targets(
async def test_extraction_functions_with_targets(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
hass_ws_client: WebSocketGenerator,
@@ -2428,211 +2428,6 @@ async def test_extraction_functions_with_trigger_targets(
}
async def test_extraction_functions_with_condition_targets(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test extraction functions with targets in conditions."""
config_entry = MockConfigEntry(domain="fake_integration", data={})
config_entry.mock_state(hass, ConfigEntryState.LOADED)
config_entry.add_to_hass(hass)
condition_device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "00:00:00:00:00:02")},
)
await async_setup_component(hass, "homeassistant", {})
await async_setup_component(hass, "light", {"light": {"platform": "demo"}})
await hass.async_block_till_done()
# Enable the new_triggers_conditions feature flag to allow new-style conditions
assert await async_setup_component(hass, "labs", {})
ws_client = await hass_ws_client(hass)
await ws_client.send_json_auto_id(
{
"type": "labs/update",
"domain": "automation",
"preview_feature": "new_triggers_conditions",
"enabled": True,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
await hass.async_block_till_done()
assert await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: [
{
"alias": "test1",
"triggers": [
{"trigger": "state", "entity_id": "sensor.trigger_state"},
],
"conditions": [
# Single entity_id in target
{
"condition": "light.is_on",
"target": {"entity_id": "light.condition_entity"},
"options": {"behavior": "any"},
},
# Multiple entity_ids in target
{
"condition": "light.is_on",
"target": {
"entity_id": [
"light.condition_entity_list1",
"light.condition_entity_list2",
]
},
"options": {"behavior": "any"},
},
# Single device_id in target
{
"condition": "light.is_on",
"target": {"device_id": condition_device.id},
"options": {"behavior": "any"},
},
# Multiple device_ids in target
{
"condition": "light.is_on",
"target": {
"device_id": [
"target-device-1",
"target-device-2",
]
},
"options": {"behavior": "any"},
},
# Single area_id in target
{
"condition": "light.is_on",
"target": {"area_id": "area-condition-single"},
"options": {"behavior": "any"},
},
# Multiple area_ids in target
{
"condition": "light.is_on",
"target": {
"area_id": ["area-condition-1", "area-condition-2"]
},
"options": {"behavior": "any"},
},
# Single floor_id in target
{
"condition": "light.is_on",
"target": {"floor_id": "floor-condition-single"},
"options": {"behavior": "any"},
},
# Multiple floor_ids in target
{
"condition": "light.is_on",
"target": {
"floor_id": ["floor-condition-1", "floor-condition-2"]
},
"options": {"behavior": "any"},
},
# Single label_id in target
{
"condition": "light.is_on",
"target": {"label_id": "label-condition-single"},
"options": {"behavior": "any"},
},
# Multiple label_ids in target
{
"condition": "light.is_on",
"target": {
"label_id": ["label-condition-1", "label-condition-2"]
},
"options": {"behavior": "any"},
},
# Combined targets
{
"condition": "light.is_on",
"target": {
"entity_id": "light.combined_entity",
"device_id": "combined-device",
"area_id": "combined-area",
"floor_id": "combined-floor",
"label_id": "combined-label",
},
"options": {"behavior": "any"},
},
],
"actions": [
{
"action": "test.script",
"data": {"entity_id": "light.action_entity"},
},
],
},
]
},
)
# Test entity extraction from condition targets
assert set(automation.entities_in_automation(hass, "automation.test1")) == {
"sensor.trigger_state",
"light.condition_entity",
"light.condition_entity_list1",
"light.condition_entity_list2",
"light.combined_entity",
"light.action_entity",
}
# Test device extraction from condition targets
assert set(automation.devices_in_automation(hass, "automation.test1")) == {
condition_device.id,
"target-device-1",
"target-device-2",
"combined-device",
}
# Test area extraction from condition targets
assert set(automation.areas_in_automation(hass, "automation.test1")) == {
"area-condition-single",
"area-condition-1",
"area-condition-2",
"combined-area",
}
# Test floor extraction from condition targets
assert set(automation.floors_in_automation(hass, "automation.test1")) == {
"floor-condition-single",
"floor-condition-1",
"floor-condition-2",
"combined-floor",
}
# Test label extraction from condition targets
assert set(automation.labels_in_automation(hass, "automation.test1")) == {
"label-condition-single",
"label-condition-1",
"label-condition-2",
"combined-label",
}
# Test automations_with_* functions
assert set(automation.automations_with_entity(hass, "light.condition_entity")) == {
"automation.test1"
}
assert set(automation.automations_with_device(hass, condition_device.id)) == {
"automation.test1"
}
assert set(automation.automations_with_area(hass, "area-condition-single")) == {
"automation.test1"
}
assert set(automation.automations_with_floor(hass, "floor-condition-single")) == {
"automation.test1"
}
assert set(automation.automations_with_label(hass, "label-condition-single")) == {
"automation.test1"
}
async def test_logbook_humanify_automation_triggered_event(hass: HomeAssistant) -> None:
"""Test humanifying Automation Trigger event."""
hass.config.components.add("recorder")

View File

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

View File

@@ -0,0 +1,110 @@
"""Common fixtures for the Homevolt tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from homevolt import Device, DeviceMetadata, Sensor, SensorType
import pytest
from homeassistant.components.homevolt.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.homevolt.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
title="Homevolt",
domain=DOMAIN,
data={
CONF_HOST: "127.0.0.1",
CONF_PASSWORD: "test-password",
},
unique_id="40580137858664",
)
@pytest.fixture
def mock_homevolt_client() -> Generator[MagicMock]:
"""Return a mocked Homevolt client."""
with (
patch(
"homeassistant.components.homevolt.Homevolt",
autospec=True,
) as homevolt_mock,
patch(
"homeassistant.components.homevolt.config_flow.Homevolt",
new=homevolt_mock,
),
):
client = homevolt_mock.return_value
client.base_url = "http://127.0.0.1"
client.update_info = AsyncMock()
# Create a mock Device with sensors
device = MagicMock(spec=Device)
device.device_id = "40580137858664"
device.sensors = {
"L1 Voltage": Sensor(
value=234.5,
type=SensorType.VOLTAGE,
device_identifier="ems_40580137858664",
slug="l1_voltage",
),
"Battery State of Charge": Sensor(
value=80.6,
type=SensorType.PERCENTAGE,
device_identifier="ems_40580137858664",
slug="battery_state_of_charge",
),
"Power": Sensor(
value=-12,
type=SensorType.POWER,
device_identifier="ems_40580137858664",
slug="power",
),
}
device.device_metadata = {
"ems_40580137858664": DeviceMetadata(
name="Homevolt EMS",
model="EMS-1000",
),
}
client.get_device.return_value = device
yield client
@pytest.fixture
def platforms() -> list[Platform]:
"""Return the platforms to test."""
return [Platform.SENSOR]
@pytest.fixture
async def init_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_homevolt_client: MagicMock,
platforms: list[Platform],
) -> MockConfigEntry:
"""Set up the Homevolt integration for testing."""
mock_config_entry.add_to_hass(hass)
with patch("homeassistant.components.homevolt.PLATFORMS", platforms):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
return mock_config_entry

View File

@@ -0,0 +1,169 @@
# serializer version: 1
# name: test_entities[sensor.homevolt_ems_battery_state_of_charge-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.homevolt_ems_battery_state_of_charge',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Battery state of charge',
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery state of charge',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_state_of_charge',
'unique_id': '40580137858664_Battery State of Charge',
'unit_of_measurement': '%',
})
# ---
# name: test_entities[sensor.homevolt_ems_battery_state_of_charge-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Homevolt EMS Battery state of charge',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.homevolt_ems_battery_state_of_charge',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '80.6',
})
# ---
# name: test_entities[sensor.homevolt_ems_l1_voltage-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.homevolt_ems_l1_voltage',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'L1 voltage',
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.VOLTAGE: 'voltage'>,
'original_icon': None,
'original_name': 'L1 voltage',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'l1_voltage',
'unique_id': '40580137858664_L1 Voltage',
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
})
# ---
# name: test_entities[sensor.homevolt_ems_l1_voltage-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'voltage',
'friendly_name': 'Homevolt EMS L1 voltage',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfElectricPotential.VOLT: 'V'>,
}),
'context': <ANY>,
'entity_id': 'sensor.homevolt_ems_l1_voltage',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '234.5',
})
# ---
# name: test_entities[sensor.homevolt_ems_power-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.homevolt_ems_power',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Power',
'options': dict({
'sensor': dict({
'suggested_display_precision': 0,
}),
}),
'original_device_class': <SensorDeviceClass.POWER: 'power'>,
'original_icon': None,
'original_name': 'Power',
'platform': 'homevolt',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'power',
'unique_id': '40580137858664_Power',
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
})
# ---
# name: test_entities[sensor.homevolt_ems_power-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'power',
'friendly_name': 'Homevolt EMS Power',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfPower.WATT: 'W'>,
}),
'context': <ANY>,
'entity_id': 'sensor.homevolt_ems_power',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '-12',
})
# ---

View File

@@ -0,0 +1,170 @@
"""Tests for the Homevolt config flow."""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock, patch
from homevolt import HomevoltAuthenticationError, HomevoltConnectionError
import pytest
from homeassistant.components.homevolt.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
async def test_full_flow_success(
hass: HomeAssistant, mock_setup_entry: AsyncMock
) -> None:
"""Test a complete successful user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
user_input = {
CONF_HOST: "192.168.1.100",
CONF_PASSWORD: "test-password",
}
with (
patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
),
patch(
"homeassistant.components.homevolt.config_flow.Homevolt.get_device",
) as mock_get_device,
):
mock_device = MagicMock()
mock_device.device_id = "40580137858664"
mock_get_device.return_value = mock_device
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Homevolt Local"
assert result["data"] == user_input
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "expected_error"),
[
(HomevoltAuthenticationError, "invalid_auth"),
(HomevoltConnectionError, "cannot_connect"),
(Exception, "unknown"),
],
)
async def test_step_user_errors(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
exception: Exception,
expected_error: str,
) -> None:
"""Test error cases for the user step with recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
user_input = {
CONF_HOST: "192.168.1.100",
CONF_PASSWORD: "test-password",
}
with patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
) as mock_update_info:
mock_update_info.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": expected_error}
with (
patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
),
patch(
"homeassistant.components.homevolt.config_flow.Homevolt.get_device",
) as mock_get_device,
):
mock_device = MagicMock()
mock_device.device_id = "40580137858664"
mock_get_device.return_value = mock_device
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Homevolt Local"
assert result["data"] == user_input
assert len(mock_setup_entry.mock_calls) == 1
async def test_duplicate_entry(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
) -> None:
"""Test that a duplicate device_id aborts the flow."""
existing_entry = MockConfigEntry(
domain=DOMAIN,
data={CONF_HOST: "192.168.1.100", CONF_PASSWORD: "test-password"},
unique_id="40580137858664",
)
existing_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
user_input = {
CONF_HOST: "192.168.1.200",
CONF_PASSWORD: "test-password",
}
with (
patch(
"homeassistant.components.homevolt.config_flow.Homevolt.update_info",
new_callable=AsyncMock,
),
patch(
"homeassistant.components.homevolt.config_flow.Homevolt.get_device",
) as mock_get_device,
):
mock_device = MagicMock()
mock_device.device_id = "40580137858664"
mock_get_device.return_value = mock_device
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"

View File

@@ -0,0 +1,65 @@
"""Test the Homevolt init module."""
from __future__ import annotations
from unittest.mock import MagicMock
from homevolt import HomevoltAuthenticationError, HomevoltConnectionError
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_load_unload_entry(
hass: HomeAssistant,
mock_homevolt_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test load and unload 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_remove(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
async def test_config_entry_not_ready(
hass: HomeAssistant,
mock_homevolt_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the Homevolt configuration entry not ready."""
mock_homevolt_client.update_info.side_effect = HomevoltConnectionError(
"Connection failed"
)
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.SETUP_RETRY
async def test_config_entry_auth_failed(
hass: HomeAssistant,
mock_homevolt_client: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the Homevolt configuration entry authentication failed."""
mock_homevolt_client.update_info.side_effect = HomevoltAuthenticationError(
"Authentication failed"
)
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.SETUP_ERROR

View File

@@ -0,0 +1,60 @@
"""Tests for the Homevolt sensor platform."""
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.homevolt.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
pytestmark = pytest.mark.usefixtures(
"entity_registry_enabled_by_default", "init_integration"
)
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the sensor entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "40580137858664_ems_40580137858664")}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id
async def test_sensor_exposes_values_from_coordinator(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_homevolt_client,
) -> None:
"""Ensure sensor entities are created and expose values from the coordinator."""
unique_id = "40580137858664_L1 Voltage"
entity_id = entity_registry.async_get_entity_id("sensor", DOMAIN, unique_id)
assert entity_id is not None
state = hass.states.get(entity_id)
assert state is not None
assert float(state.state) == 234.5
mock_homevolt_client.get_device.return_value.sensors["L1 Voltage"].value = 240.1
coordinator = mock_config_entry.runtime_data
await coordinator.async_refresh()
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state is not None
assert float(state.state) == 240.1

View File

@@ -5,25 +5,14 @@ from typing import Any
import pytest
from homeassistant.components.light import ATTR_BRIGHTNESS
from homeassistant.const import (
ATTR_LABEL_ID,
CONF_ABOVE,
CONF_BELOW,
CONF_ENTITY_ID,
STATE_OFF,
STATE_ON,
)
from homeassistant.const import ATTR_LABEL_ID, CONF_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers.trigger import (
CONF_LOWER_LIMIT,
CONF_THRESHOLD_TYPE,
CONF_UPPER_LIMIT,
ThresholdType,
)
from tests.components import (
TriggerStateDescription,
arm_trigger,
parametrize_numerical_attribute_changed_trigger_states,
parametrize_numerical_attribute_crossed_threshold_trigger_states,
parametrize_target_entities,
parametrize_trigger_states,
set_or_remove_state,
@@ -37,131 +26,6 @@ async def target_lights(hass: HomeAssistant) -> list[str]:
return (await target_entities(hass, "light"))["included"]
def parametrize_brightness_changed_trigger_states(
trigger: str, state: str, attribute: str
) -> list[tuple[str, dict[str, Any], list[TriggerStateDescription]]]:
"""Parametrize states and expected service call counts for brightness changed triggers.
Note: The brightness in the trigger configuration is in percentage (0-100) scale,
the underlying attribute in the state is in uint8 (0-255) scale.
"""
return [
*parametrize_trigger_states(
trigger=trigger,
trigger_options={},
target_states=[
(state, {attribute: 0}),
(state, {attribute: 128}),
(state, {attribute: 255}),
],
other_states=[(state, {attribute: None})],
retrigger_on_target_state=True,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={CONF_ABOVE: 10},
target_states=[
(state, {attribute: 128}),
(state, {attribute: 255}),
],
other_states=[
(state, {attribute: None}),
(state, {attribute: 0}),
],
retrigger_on_target_state=True,
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={CONF_BELOW: 90},
target_states=[
(state, {attribute: 0}),
(state, {attribute: 128}),
],
other_states=[
(state, {attribute: None}),
(state, {attribute: 255}),
],
retrigger_on_target_state=True,
),
]
def parametrize_brightness_crossed_threshold_trigger_states(
trigger: str, state: str, attribute: str
) -> list[tuple[str, dict[str, Any], list[TriggerStateDescription]]]:
"""Parametrize states and expected service call counts for brightness crossed threshold triggers.
Note: The brightness in the trigger configuration is in percentage (0-100) scale,
the underlying attribute in the state is in uint8 (0-255) scale.
"""
return [
*parametrize_trigger_states(
trigger=trigger,
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
},
target_states=[
(state, {attribute: 128}),
(state, {attribute: 153}),
],
other_states=[
(state, {attribute: None}),
(state, {attribute: 0}),
(state, {attribute: 255}),
],
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE,
CONF_LOWER_LIMIT: 10,
CONF_UPPER_LIMIT: 90,
},
target_states=[
(state, {attribute: 0}),
(state, {attribute: 255}),
],
other_states=[
(state, {attribute: None}),
(state, {attribute: 128}),
(state, {attribute: 153}),
],
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.ABOVE,
CONF_LOWER_LIMIT: 10,
},
target_states=[
(state, {attribute: 128}),
(state, {attribute: 255}),
],
other_states=[
(state, {attribute: None}),
(state, {attribute: 0}),
],
),
*parametrize_trigger_states(
trigger=trigger,
trigger_options={
CONF_THRESHOLD_TYPE: ThresholdType.BELOW,
CONF_UPPER_LIMIT: 90,
},
target_states=[
(state, {attribute: 0}),
(state, {attribute: 128}),
],
other_states=[
(state, {attribute: None}),
(state, {attribute: 255}),
],
),
]
@pytest.mark.parametrize(
"trigger_key",
[
@@ -250,10 +114,10 @@ async def test_light_state_trigger_behavior_any(
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_brightness_changed_trigger_states(
*parametrize_numerical_attribute_changed_trigger_states(
"light.brightness_changed", STATE_ON, ATTR_BRIGHTNESS
),
*parametrize_brightness_crossed_threshold_trigger_states(
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"light.brightness_crossed_threshold", STATE_ON, ATTR_BRIGHTNESS
),
],
@@ -361,7 +225,7 @@ async def test_light_state_trigger_behavior_first(
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_brightness_crossed_threshold_trigger_states(
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"light.brightness_crossed_threshold", STATE_ON, ATTR_BRIGHTNESS
),
],
@@ -469,7 +333,7 @@ async def test_light_state_trigger_behavior_last(
@pytest.mark.parametrize(
("trigger", "trigger_options", "states"),
[
*parametrize_brightness_crossed_threshold_trigger_states(
*parametrize_numerical_attribute_crossed_threshold_trigger_states(
"light.brightness_crossed_threshold", STATE_ON, ATTR_BRIGHTNESS
),
],

View File

@@ -22,7 +22,6 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client as client
from homeassistant.util import ssl as ssl_util
from homeassistant.util.color import RGBColor
from homeassistant.util.ssl import SSLCipherList
@@ -414,29 +413,3 @@ async def test_resolver_is_singleton(hass: HomeAssistant) -> None:
assert isinstance(session3._connector, aiohttp.TCPConnector)
assert session._connector._resolver is session2._connector._resolver
assert session._connector._resolver is session3._connector._resolver
async def test_connector_uses_http11_alpn(hass: HomeAssistant) -> None:
"""Test that connector uses HTTP/1.1 ALPN protocols."""
with patch.object(
ssl_util, "client_context", wraps=ssl_util.client_context
) as mock_client_context:
client.async_get_clientsession(hass)
# Verify client_context was called with HTTP/1.1 ALPN
mock_client_context.assert_called_once_with(
SSLCipherList.PYTHON_DEFAULT, ssl_util.SSL_ALPN_HTTP11
)
async def test_connector_no_verify_uses_http11_alpn(hass: HomeAssistant) -> None:
"""Test that connector without SSL verification uses HTTP/1.1 ALPN protocols."""
with patch.object(
ssl_util, "client_context_no_verify", wraps=ssl_util.client_context_no_verify
) as mock_client_context_no_verify:
client.async_get_clientsession(hass, verify_ssl=False)
# Verify client_context_no_verify was called with HTTP/1.1 ALPN
mock_client_context_no_verify.assert_called_once_with(
SSLCipherList.PYTHON_DEFAULT, ssl_util.SSL_ALPN_HTTP11
)

View File

@@ -8,7 +8,6 @@ import pytest
from homeassistant.const import EVENT_HOMEASSISTANT_CLOSE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import httpx_client as client
from homeassistant.util.ssl import SSL_ALPN_HTTP11, SSL_ALPN_HTTP11_HTTP2
from tests.common import MockModule, extract_stack_to_frame, mock_integration
@@ -17,20 +16,14 @@ async def test_get_async_client_with_ssl(hass: HomeAssistant) -> None:
"""Test init async client with ssl."""
client.get_async_client(hass)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11)],
httpx.AsyncClient,
)
assert isinstance(hass.data[client.DATA_ASYNC_CLIENT], httpx.AsyncClient)
async def test_get_async_client_without_ssl(hass: HomeAssistant) -> None:
"""Test init async client without ssl."""
client.get_async_client(hass, verify_ssl=False)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(False, SSL_ALPN_HTTP11)],
httpx.AsyncClient,
)
assert isinstance(hass.data[client.DATA_ASYNC_CLIENT_NOVERIFY], httpx.AsyncClient)
async def test_create_async_httpx_client_with_ssl_and_cookies(
@@ -41,7 +34,7 @@ async def test_create_async_httpx_client_with_ssl_and_cookies(
httpx_client = client.create_async_httpx_client(hass, cookies={"bla": True})
assert isinstance(httpx_client, httpx.AsyncClient)
assert hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11)] != httpx_client
assert hass.data[client.DATA_ASYNC_CLIENT] != httpx_client
async def test_create_async_httpx_client_without_ssl_and_cookies(
@@ -54,37 +47,31 @@ async def test_create_async_httpx_client_without_ssl_and_cookies(
hass, verify_ssl=False, cookies={"bla": True}
)
assert isinstance(httpx_client, httpx.AsyncClient)
assert hass.data[client.DATA_ASYNC_CLIENT][(False, SSL_ALPN_HTTP11)] != httpx_client
assert hass.data[client.DATA_ASYNC_CLIENT_NOVERIFY] != httpx_client
async def test_get_async_client_cleanup(hass: HomeAssistant) -> None:
"""Test init async client with ssl."""
client.get_async_client(hass)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11)],
httpx.AsyncClient,
)
assert isinstance(hass.data[client.DATA_ASYNC_CLIENT], httpx.AsyncClient)
hass.bus.async_fire(EVENT_HOMEASSISTANT_CLOSE)
await hass.async_block_till_done()
assert hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11)].is_closed
assert hass.data[client.DATA_ASYNC_CLIENT].is_closed
async def test_get_async_client_cleanup_without_ssl(hass: HomeAssistant) -> None:
"""Test init async client without ssl."""
client.get_async_client(hass, verify_ssl=False)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(False, SSL_ALPN_HTTP11)],
httpx.AsyncClient,
)
assert isinstance(hass.data[client.DATA_ASYNC_CLIENT_NOVERIFY], httpx.AsyncClient)
hass.bus.async_fire(EVENT_HOMEASSISTANT_CLOSE)
await hass.async_block_till_done()
assert hass.data[client.DATA_ASYNC_CLIENT][(False, SSL_ALPN_HTTP11)].is_closed
assert hass.data[client.DATA_ASYNC_CLIENT_NOVERIFY].is_closed
async def test_get_async_client_patched_close(hass: HomeAssistant) -> None:
@@ -92,10 +79,7 @@ async def test_get_async_client_patched_close(hass: HomeAssistant) -> None:
with patch("httpx.AsyncClient.aclose") as mock_aclose:
httpx_session = client.get_async_client(hass)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11)],
httpx.AsyncClient,
)
assert isinstance(hass.data[client.DATA_ASYNC_CLIENT], httpx.AsyncClient)
with pytest.raises(RuntimeError):
await httpx_session.aclose()
@@ -108,10 +92,7 @@ async def test_get_async_client_context_manager(hass: HomeAssistant) -> None:
with patch("httpx.AsyncClient.aclose") as mock_aclose:
httpx_session = client.get_async_client(hass)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11)],
httpx.AsyncClient,
)
assert isinstance(hass.data[client.DATA_ASYNC_CLIENT], httpx.AsyncClient)
async with httpx_session:
pass
@@ -119,80 +100,6 @@ async def test_get_async_client_context_manager(hass: HomeAssistant) -> None:
assert mock_aclose.call_count == 0
async def test_get_async_client_http2(hass: HomeAssistant) -> None:
"""Test init async client with HTTP/2 support."""
http1_client = client.get_async_client(hass)
http2_client = client.get_async_client(hass, alpn_protocols=SSL_ALPN_HTTP11_HTTP2)
# HTTP/1.1 and HTTP/2 clients should be different (different SSL contexts)
assert http1_client is not http2_client
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11)],
httpx.AsyncClient,
)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11_HTTP2)],
httpx.AsyncClient,
)
# Same parameters should return cached client
assert client.get_async_client(hass) is http1_client
assert (
client.get_async_client(hass, alpn_protocols=SSL_ALPN_HTTP11_HTTP2)
is http2_client
)
async def test_get_async_client_http2_cleanup(hass: HomeAssistant) -> None:
"""Test cleanup of HTTP/2 async client."""
client.get_async_client(hass, alpn_protocols=SSL_ALPN_HTTP11_HTTP2)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11_HTTP2)],
httpx.AsyncClient,
)
hass.bus.async_fire(EVENT_HOMEASSISTANT_CLOSE)
await hass.async_block_till_done()
assert hass.data[client.DATA_ASYNC_CLIENT][(True, SSL_ALPN_HTTP11_HTTP2)].is_closed
async def test_get_async_client_http2_without_ssl(hass: HomeAssistant) -> None:
"""Test init async client with HTTP/2 and without SSL."""
http2_client = client.get_async_client(
hass, verify_ssl=False, alpn_protocols=SSL_ALPN_HTTP11_HTTP2
)
assert isinstance(
hass.data[client.DATA_ASYNC_CLIENT][(False, SSL_ALPN_HTTP11_HTTP2)],
httpx.AsyncClient,
)
# Same parameters should return cached client
assert (
client.get_async_client(
hass, verify_ssl=False, alpn_protocols=SSL_ALPN_HTTP11_HTTP2
)
is http2_client
)
async def test_create_async_httpx_client_http2(hass: HomeAssistant) -> None:
"""Test create async client with HTTP/2 uses correct ALPN protocols."""
http1_client = client.create_async_httpx_client(hass)
http2_client = client.create_async_httpx_client(
hass, alpn_protocols=SSL_ALPN_HTTP11_HTTP2
)
# Different clients (not cached)
assert http1_client is not http2_client
# Both should be valid clients
assert isinstance(http1_client, httpx.AsyncClient)
assert isinstance(http2_client, httpx.AsyncClient)
async def test_warning_close_session_integration(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:

View File

@@ -1,58 +1,78 @@
"""Test Home Assistant ssl utility functions."""
from unittest.mock import MagicMock, Mock, patch
import pytest
from homeassistant.util.ssl import (
SSL_ALPN_HTTP11,
SSL_ALPN_HTTP11_HTTP2,
SSL_ALPN_NONE,
SSLCipherList,
client_context,
client_context_no_verify,
create_client_context,
create_no_verify_ssl_context,
get_default_context,
get_default_no_verify_context,
)
@pytest.fixture
def mock_sslcontext():
"""Mock the ssl lib."""
return MagicMock(set_ciphers=Mock(return_value=True))
def test_client_context(mock_sslcontext) -> None:
"""Test client context."""
with patch("homeassistant.util.ssl.ssl.SSLContext", return_value=mock_sslcontext):
client_context()
mock_sslcontext.set_ciphers.assert_not_called()
client_context(SSLCipherList.MODERN)
mock_sslcontext.set_ciphers.assert_not_called()
client_context(SSLCipherList.INTERMEDIATE)
mock_sslcontext.set_ciphers.assert_not_called()
client_context(SSLCipherList.INSECURE)
mock_sslcontext.set_ciphers.assert_not_called()
def test_no_verify_ssl_context(mock_sslcontext) -> None:
"""Test no verify ssl context."""
with patch("homeassistant.util.ssl.ssl.SSLContext", return_value=mock_sslcontext):
create_no_verify_ssl_context()
mock_sslcontext.set_ciphers.assert_not_called()
create_no_verify_ssl_context(SSLCipherList.MODERN)
mock_sslcontext.set_ciphers.assert_not_called()
create_no_verify_ssl_context(SSLCipherList.INTERMEDIATE)
mock_sslcontext.set_ciphers.assert_not_called()
create_no_verify_ssl_context(SSLCipherList.INSECURE)
mock_sslcontext.set_ciphers.assert_not_called()
def test_ssl_context_caching() -> None:
"""Test that SSLContext instances are cached correctly."""
assert client_context() is client_context(SSLCipherList.PYTHON_DEFAULT)
assert create_no_verify_ssl_context() is create_no_verify_ssl_context(
SSLCipherList.PYTHON_DEFAULT
)
def test_ssl_context_cipher_bucketing() -> None:
"""Test that SSL contexts are bucketed by cipher list."""
default_ctx = client_context(SSLCipherList.PYTHON_DEFAULT)
modern_ctx = client_context(SSLCipherList.MODERN)
intermediate_ctx = client_context(SSLCipherList.INTERMEDIATE)
insecure_ctx = client_context(SSLCipherList.INSECURE)
def test_create_client_context(mock_sslcontext) -> None:
"""Test create client context."""
with patch("homeassistant.util.ssl.ssl.SSLContext", return_value=mock_sslcontext):
client_context()
mock_sslcontext.set_ciphers.assert_not_called()
# Different cipher lists should return different contexts
assert default_ctx is not modern_ctx
assert default_ctx is not intermediate_ctx
assert default_ctx is not insecure_ctx
assert modern_ctx is not intermediate_ctx
assert modern_ctx is not insecure_ctx
assert intermediate_ctx is not insecure_ctx
client_context(SSLCipherList.MODERN)
mock_sslcontext.set_ciphers.assert_not_called()
# Same parameters should return cached context
assert client_context(SSLCipherList.PYTHON_DEFAULT) is default_ctx
assert client_context(SSLCipherList.MODERN) is modern_ctx
client_context(SSLCipherList.INTERMEDIATE)
mock_sslcontext.set_ciphers.assert_not_called()
def test_no_verify_ssl_context_cipher_bucketing() -> None:
"""Test that no-verify SSL contexts are bucketed by cipher list."""
default_ctx = create_no_verify_ssl_context(SSLCipherList.PYTHON_DEFAULT)
modern_ctx = create_no_verify_ssl_context(SSLCipherList.MODERN)
# Different cipher lists should return different contexts
assert default_ctx is not modern_ctx
# Same parameters should return cached context
assert create_no_verify_ssl_context(SSLCipherList.PYTHON_DEFAULT) is default_ctx
assert create_no_verify_ssl_context(SSLCipherList.MODERN) is modern_ctx
client_context(SSLCipherList.INSECURE)
mock_sslcontext.set_ciphers.assert_not_called()
def test_create_client_context_independent() -> None:
@@ -62,129 +82,3 @@ def test_create_client_context_independent() -> None:
independent_context_2 = create_client_context()
assert shared_context is not independent_context_1
assert independent_context_1 is not independent_context_2
def test_ssl_context_alpn_bucketing() -> None:
"""Test that SSL contexts are bucketed by ALPN protocols.
Different ALPN protocol configurations should return different cached contexts
to prevent downstream libraries (e.g., httpx/httpcore) from mutating shared
contexts with incompatible settings.
"""
# HTTP/1.1, HTTP/2, and no-ALPN contexts should all be different
http1_context = client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11)
http2_context = client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11_HTTP2)
no_alpn_context = client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_NONE)
assert http1_context is not http2_context
assert http1_context is not no_alpn_context
assert http2_context is not no_alpn_context
# Same parameters should return cached context
assert (
client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11) is http1_context
)
assert (
client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11_HTTP2)
is http2_context
)
assert (
client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_NONE) is no_alpn_context
)
# No-verify contexts should also be bucketed by ALPN
http1_no_verify = client_context_no_verify(
SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11
)
http2_no_verify = client_context_no_verify(
SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11_HTTP2
)
no_alpn_no_verify = client_context_no_verify(
SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_NONE
)
assert http1_no_verify is not http2_no_verify
assert http1_no_verify is not no_alpn_no_verify
assert http2_no_verify is not no_alpn_no_verify
# create_no_verify_ssl_context should also work with ALPN
assert (
create_no_verify_ssl_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11)
is http1_no_verify
)
assert (
create_no_verify_ssl_context(
SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11_HTTP2
)
is http2_no_verify
)
assert (
create_no_verify_ssl_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_NONE)
is no_alpn_no_verify
)
def test_ssl_context_insecure_alpn_bucketing() -> None:
"""Test that INSECURE cipher list SSL contexts are bucketed by ALPN protocols.
INSECURE cipher list is used by some integrations that need to connect to
devices with outdated TLS implementations.
"""
# HTTP/1.1, HTTP/2, and no-ALPN contexts should all be different
http1_context = client_context(SSLCipherList.INSECURE, SSL_ALPN_HTTP11)
http2_context = client_context(SSLCipherList.INSECURE, SSL_ALPN_HTTP11_HTTP2)
no_alpn_context = client_context(SSLCipherList.INSECURE, SSL_ALPN_NONE)
assert http1_context is not http2_context
assert http1_context is not no_alpn_context
assert http2_context is not no_alpn_context
# Same parameters should return cached context
assert client_context(SSLCipherList.INSECURE, SSL_ALPN_HTTP11) is http1_context
assert (
client_context(SSLCipherList.INSECURE, SSL_ALPN_HTTP11_HTTP2) is http2_context
)
assert client_context(SSLCipherList.INSECURE, SSL_ALPN_NONE) is no_alpn_context
# No-verify contexts should also be bucketed by ALPN
http1_no_verify = client_context_no_verify(SSLCipherList.INSECURE, SSL_ALPN_HTTP11)
http2_no_verify = client_context_no_verify(
SSLCipherList.INSECURE, SSL_ALPN_HTTP11_HTTP2
)
no_alpn_no_verify = client_context_no_verify(SSLCipherList.INSECURE, SSL_ALPN_NONE)
assert http1_no_verify is not http2_no_verify
assert http1_no_verify is not no_alpn_no_verify
assert http2_no_verify is not no_alpn_no_verify
# create_no_verify_ssl_context should also work with ALPN
assert (
create_no_verify_ssl_context(SSLCipherList.INSECURE, SSL_ALPN_HTTP11)
is http1_no_verify
)
assert (
create_no_verify_ssl_context(SSLCipherList.INSECURE, SSL_ALPN_HTTP11_HTTP2)
is http2_no_verify
)
assert (
create_no_verify_ssl_context(SSLCipherList.INSECURE, SSL_ALPN_NONE)
is no_alpn_no_verify
)
def test_get_default_context_uses_http1_alpn() -> None:
"""Test that get_default_context returns context with HTTP1 ALPN."""
default_ctx = get_default_context()
default_no_verify_ctx = get_default_no_verify_context()
# Default contexts should be the same as explicitly requesting HTTP1 ALPN
assert default_ctx is client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11)
assert default_no_verify_ctx is client_context_no_verify(
SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11
)
def test_client_context_default_no_alpn() -> None:
"""Test that client_context defaults to no ALPN for backward compatibility."""
# Default (no ALPN) should be different from HTTP1 ALPN
default_ctx = client_context()
http1_ctx = client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_HTTP11)
assert default_ctx is not http1_ctx
assert default_ctx is client_context(SSLCipherList.PYTHON_DEFAULT, SSL_ALPN_NONE)