Compare commits

...

23 Commits

Author SHA1 Message Date
Petro31 d294b04b79 Add EntityComponent to device_tracker (#171507) 2026-05-21 16:10:20 +02:00
Markus Tuominen 8b0e9060b3 Set _attr_has_entity_name on tplink_omada OmadaClientScannerEntity (#171680) 2026-05-21 16:41:37 +03:00
MoonDevLT 39066b6e3a Fix missing exceptions translation key missing_device_info in lunatone (#171569) 2026-05-21 14:59:48 +02:00
Max Michels a23a9b350b Replace duplicate constants with homeassistant.const imports (#171701) 2026-05-21 14:57:58 +02:00
chiro79 fdaa807ca8 Switch to aiopvpc-ng (#171025) 2026-05-21 14:54:23 +02:00
A. Gideonse f290dcc03f Update Indevolt integration quality scale to platinum (#170320) 2026-05-21 14:53:06 +02:00
Markus Tuominen 654408cc76 Set _attr_has_entity_name on sonos SonosFavoritesEntity (#171678) 2026-05-21 13:42:21 +02:00
Max Michels 1f814faad8 Replace duplicate constants with homeassistant.const imports (#171702) 2026-05-21 13:36:14 +02:00
Markus Tuominen 6e00eecfcd Set _attr_has_entity_name on lunatone LunatoneLineBroadcastLight (#171682) 2026-05-21 13:19:42 +02:00
Robert Resch 8c8620c511 Add check requirements yanked and CVE check (#171641)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
2026-05-21 12:54:15 +02:00
Wendelin cca8825ca5 Add comment optional attribute to automation items (#171091) 2026-05-21 12:52:54 +02:00
Max Michels 92fbcc29a5 Replace duplicate constants with homeassistant.const imports (#171700) 2026-05-21 12:51:19 +02:00
Shay Levy 1c28833f39 Fix LG WebOS TV translation placeholders mismatches (#171696) 2026-05-21 13:33:36 +03:00
Christian Lackas cfdef77222 homematicip_cloud: migrate entity names to has_entity_name (#169273) 2026-05-21 12:29:43 +02:00
epenet 49720475da Bump renault-api to 0.5.10 (#171692) 2026-05-21 12:16:29 +02:00
Markus Tuominen 7967b84cc6 Set _attr_has_entity_name on omie OMIEPriceSensor (#171671) 2026-05-21 12:14:03 +02:00
Markus Tuominen c715557813 Set _attr_has_entity_name on smartthings SmartThingsScene (#171672) 2026-05-21 12:10:06 +02:00
Markus Tuominen 79e5330782 Set _attr_has_entity_name on ekeybionyx EkeyEvent (#171668) 2026-05-21 12:03:01 +02:00
Max Michels 5210ca64b1 Replace duplicate constants with homeassistant.const imports (#171669) 2026-05-21 12:02:12 +02:00
Markus Tuominen 65283e3d77 Set _attr_has_entity_name on fitbit battery sensors (#171670) 2026-05-21 12:01:27 +02:00
mhuiskes 427cb9f8db Remove unnecessary intermediate variables in zeversolar diagnostics (#171691) 2026-05-21 11:55:34 +02:00
Erik Montnemery a09e042d42 Add test of FlowHandler show_advanced_options property (#171681) 2026-05-21 11:47:42 +02:00
Shay Levy 072e9b51a2 Fix Shelly translation placeholders mismatches (#171685) 2026-05-21 11:47:20 +02:00
65 changed files with 910 additions and 224 deletions
Generated
+2 -2
View File
@@ -1413,8 +1413,8 @@ CLAUDE.md @home-assistant/core
/tests/components/pushover/ @engrbm87
/homeassistant/components/pvoutput/ @frenck
/tests/components/pvoutput/ @frenck
/homeassistant/components/pvpc_hourly_pricing/ @azogue
/tests/components/pvpc_hourly_pricing/ @azogue
/homeassistant/components/pvpc_hourly_pricing/ @azogue @chiro79
/tests/components/pvpc_hourly_pricing/ @azogue @chiro79
/homeassistant/components/pyload/ @tr4nt0r
/tests/components/pyload/ @tr4nt0r
/homeassistant/components/qbittorrent/ @geoffreylagaisse @finder39
@@ -1,11 +1,18 @@
"""Provide functionality to keep track of devices."""
import asyncio
from typing import Any
from homeassistant.const import ATTR_GPS_ACCURACY, STATE_HOME # noqa: F401
from homeassistant.core import HomeAssistant
from homeassistant.helpers import discovery
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.typing import ConfigType
from .config_entry import ( # noqa: F401
DATA_COMPONENT,
BaseScannerEntity,
BaseTrackerEntity,
ScannerEntity,
ScannerEntityDescription,
TrackerEntity,
@@ -33,6 +40,8 @@ from .const import ( # noqa: F401
DEFAULT_TRACK_NEW,
DOMAIN,
ENTITY_ID_FORMAT,
LOGGER,
PLATFORM_TYPE_LEGACY,
SCAN_INTERVAL,
SourceType,
)
@@ -44,7 +53,9 @@ from .legacy import ( # noqa: F401
SOURCE_TYPES,
AsyncSeeCallback,
DeviceScanner,
DeviceTracker,
SeeCallback,
async_create_platform_type,
async_setup_integration as async_setup_legacy_integration,
see,
)
@@ -57,5 +68,43 @@ def is_on(hass: HomeAssistant, entity_id: str) -> bool:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the device tracker."""
async_setup_legacy_integration(hass, config)
component = hass.data[DATA_COMPONENT] = EntityComponent[BaseTrackerEntity](
LOGGER, DOMAIN, hass, SCAN_INTERVAL
)
component.config = {}
component.register_shutdown()
# The tracker is loaded in the async_setup_legacy_integration task so
# we create a future to avoid waiting on it here so that only
# async_platform_discovered will have to wait in the rare event
# a custom component still uses the legacy device tracker discovery.
tracker_future: asyncio.Future[DeviceTracker] = hass.loop.create_future()
async def async_platform_discovered(
p_type: str, info: dict[str, Any] | None
) -> None:
"""Load a platform."""
platform = await async_create_platform_type(hass, config, p_type, {})
if platform is None:
return
if platform.type != PLATFORM_TYPE_LEGACY:
await component.async_setup_platform(p_type, {}, info)
return
tracker = await tracker_future
await platform.async_setup_legacy(hass, tracker, info)
discovery.async_listen_platform(hass, DOMAIN, async_platform_discovered)
#
# Legacy and platforms load in a non-awaited tracked task
# to ensure device tracker setup can continue and config
# entry integrations are not waiting for legacy device
# tracker platforms to be set up.
#
hass.async_create_task(
async_setup_legacy_integration(hass, config, tracker_future),
eager_start=True,
)
return True
@@ -37,11 +37,7 @@ from homeassistant.const import (
)
from homeassistant.core import Event, HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
config_validation as cv,
discovery,
entity_registry as er,
)
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.event import (
async_track_time_interval,
async_track_utc_time_change,
@@ -204,40 +200,7 @@ def see(
hass.services.call(DOMAIN, SERVICE_SEE, data)
@callback
def async_setup_integration(hass: HomeAssistant, config: ConfigType) -> None:
"""Set up the legacy integration."""
# The tracker is loaded in the _async_setup_integration task so
# we create a future to avoid waiting on it here so that only
# async_platform_discovered will have to wait in the rare event
# a custom component still uses the legacy device tracker discovery.
tracker_future: asyncio.Future[DeviceTracker] = hass.loop.create_future()
async def async_platform_discovered(
p_type: str, info: dict[str, Any] | None
) -> None:
"""Load a platform."""
platform = await async_create_platform_type(hass, config, p_type, {})
if platform is None or platform.type != PLATFORM_TYPE_LEGACY:
return
tracker = await tracker_future
await platform.async_setup_legacy(hass, tracker, info)
discovery.async_listen_platform(hass, DOMAIN, async_platform_discovered)
#
# Legacy and platforms load in a non-awaited tracked task
# to ensure device tracker setup can continue and config
# entry integrations are not waiting for legacy device
# tracker platforms to be set up.
#
hass.async_create_task(
_async_setup_integration(hass, config, tracker_future), eager_start=True
)
async def _async_setup_integration(
async def async_setup_integration(
hass: HomeAssistant,
config: ConfigType,
tracker_future: asyncio.Future[DeviceTracker],
+1 -1
View File
@@ -26,12 +26,12 @@ async def async_setup_entry(
async_add_entities(EkeyEvent(data) for data in entry.data["webhooks"])
# pylint: disable-next=home-assistant-missing-has-entity-name
class EkeyEvent(EventEntity):
"""Ekey Event."""
_attr_device_class = EventDeviceClass.BUTTON
_attr_event_types = ["event happened"]
_attr_has_entity_name = True
def __init__(
self,
+2 -6
View File
@@ -509,14 +509,12 @@ FITBIT_RESOURCE_BATTERY = FitbitSensorEntityDescription(
icon="mdi:battery",
scope=FitbitScope.DEVICE,
entity_category=EntityCategory.DIAGNOSTIC,
has_entity_name=True,
)
FITBIT_RESOURCE_BATTERY_LEVEL = FitbitSensorEntityDescription(
key="devices/battery_level",
translation_key="battery_level",
scope=FitbitScope.DEVICE,
entity_category=EntityCategory.DIAGNOSTIC,
has_entity_name=True,
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
)
@@ -649,13 +647,12 @@ class FitbitSensor(SensorEntity):
self.async_schedule_update_ha_state(force_refresh=True)
# has_entity_name=True is supplied by the FITBIT_RESOURCE_BATTERY description
# pylint: disable-next=home-assistant-missing-has-entity-name
class FitbitBatterySensor(CoordinatorEntity[FitbitDeviceCoordinator], SensorEntity):
"""Implementation of a Fitbit battery sensor."""
entity_description: FitbitSensorEntityDescription
_attr_attribution = ATTRIBUTION
_attr_has_entity_name = True
def __init__(
self,
@@ -709,14 +706,13 @@ class FitbitBatterySensor(CoordinatorEntity[FitbitDeviceCoordinator], SensorEnti
self.async_write_ha_state()
# has_entity_name=True is supplied by the FITBIT_RESOURCE_BATTERY_LEVEL description
# pylint: disable-next=home-assistant-missing-has-entity-name
class FitbitBatteryLevelSensor(
CoordinatorEntity[FitbitDeviceCoordinator], SensorEntity
):
"""Implementation of a Fitbit battery level sensor."""
entity_description: FitbitSensorEntityDescription
_attr_has_entity_name = True
_attr_attribution = ATTRIBUTION
def __init__(
@@ -180,27 +180,24 @@ async def async_setup_entry(
class HomematicipCloudConnectionSensor(HomematicipGenericEntity, BinarySensorEntity):
"""Representation of the HomematicIP cloud connection sensor."""
_attr_translation_key = "cloud_connection"
def __init__(self, hap: HomematicipHAP) -> None:
"""Initialize the cloud connection sensor."""
super().__init__(hap, hap.home, feature_id="cloud_connection")
@property
def name(self) -> str:
"""Return the name cloud connection entity."""
name = "Cloud Connection"
# Add a prefix to the name if the homematic ip home has a name.
return name if not self._home.name else f"{self._home.name} {name}"
@property
def device_info(self) -> DeviceInfo:
"""Return device specific attributes."""
# Adds a sensor to the existing HAP device
# Merges into the existing HAP device registered in __init__.py.
# Name must match __init__.py logic for has_entity_name to work.
label = self._home.label or ""
return DeviceInfo(
identifiers={
# Serial numbers of Homematic IP device
(DOMAIN, self._home.id)
}
},
name=label,
)
@property
@@ -579,6 +576,7 @@ class HomematicipPluggableMainsFailureSurveillanceSensor(
class HomematicipSecurityZoneSensorGroup(HomematicipGenericEntity, BinarySensorEntity):
"""Representation of the HomematicIP security zone sensor group."""
_attr_has_entity_name = False
_attr_device_class = BinarySensorDeviceClass.SAFETY
def __init__(
@@ -74,6 +74,7 @@ class HomematicipHeatingGroup(HomematicipGenericEntity, ClimateEntity):
basically enabled in the hmip app.
"""
_attr_has_entity_name = False
_attr_supported_features = (
ClimateEntityFeature.PRESET_MODE | ClimateEntityFeature.TARGET_TEMPERATURE
)
@@ -320,6 +320,7 @@ class HomematicipGarageDoorModule(HomematicipGenericEntity, CoverEntity):
class HomematicipCoverShutterGroup(HomematicipGenericEntity, CoverEntity):
"""Representation of the HomematicIP cover shutter group."""
_attr_has_entity_name = False
_attr_device_class = CoverDeviceClass.SHUTTER
def __init__(self, hap: HomematicipHAP, device, post: str = "ShutterGroup") -> None:
@@ -74,6 +74,7 @@ GROUP_ATTRIBUTES = {
class HomematicipGenericEntity(Entity):
"""Representation of the HomematicIP generic entity."""
_attr_has_entity_name = True
_attr_should_poll = False
def __init__(
@@ -112,6 +113,14 @@ class HomematicipGenericEntity(Entity):
# Marker showing that the HmIP device hase been removed.
self.hmip_device_removed = False
# Compute entity name based on has_entity_name mode.
if not self._attr_has_entity_name:
# Legacy mode (groups, special entities): compose the full name
# including device/group label and home prefix.
self._attr_name = self._compute_legacy_name()
else:
self._setup_entity_name()
@property
def device_info(self) -> DeviceInfo | None:
"""Return device specific attributes."""
@@ -120,6 +129,14 @@ class HomematicipGenericEntity(Entity):
device_id = str(self._device.id)
home_id = str(self._device.homeId)
# Include the home name in the device name so that the
# previous "{home} {device}" naming is preserved after
# switching to has_entity_name=True.
device_name = self._device.label
home_name = getattr(self._home, "name", None)
if device_name and home_name:
device_name = f"{home_name} {device_name}"
return DeviceInfo(
identifiers={
# Serial numbers of Homematic IP device
@@ -127,7 +144,7 @@ class HomematicipGenericEntity(Entity):
},
manufacturer=self._device.oem,
model=self._device.modelType,
name=self._device.label,
name=device_name,
sw_version=self._device.firmwareVersion,
# Link to the homematic ip access point.
via_device=(DOMAIN, home_id),
@@ -200,38 +217,93 @@ class HomematicipGenericEntity(Entity):
self.async_remove(force_remove=True), eager_start=False
)
@property
def name(self) -> str:
"""Return the name of the generic entity."""
def _compute_legacy_name(self) -> str:
"""Compute the full legacy name for entities without has_entity_name.
name = ""
# Try to get a label from a channel.
functional_channels = getattr(self._device, "functionalChannels", None)
if functional_channels and self.functional_channel:
if self._is_multi_channel:
label = getattr(self.functional_channel, "label", None)
if label:
name = str(label)
elif len(functional_channels) > 1:
label = getattr(functional_channels[1], "label", None)
if label:
name = str(label)
# Use device label, if name is not defined by channel label.
if not name:
name = self._device.label or ""
if self._post:
name = f"{name} {self._post}"
elif self._is_multi_channel:
name = f"{name} Channel{self.get_channel_index()}"
# Add a prefix to the name if the homematic ip home has a name.
Used by group entities and other special cases where has_entity_name
is False. Includes device/group label, post suffix, and home prefix.
"""
name = self._device.label or ""
if self._post:
name = f"{name} {self._post}" if name else self._post
home_name = getattr(self._home, "name", None)
if name and home_name:
name = f"{home_name} {name}"
return name
def _setup_entity_name(self) -> None:
"""Set up entity naming for has_entity_name mode.
With has_entity_name=True, HA composes the full friendly name as
"{device_name} {entity_name}". This method sets the appropriate
naming attributes.
For multi-channel entities, channel labels provide _attr_name (dynamic).
For entities with _post, _attr_name is derived from the post suffix,
with the first letter capitalized for display consistency.
For primary entities, HA uses device_class as the name.
"""
# Multi-channel entities: use channel label as entity name.
if self._is_multi_channel and self.functional_channel:
label = getattr(self.functional_channel, "label", None)
if label:
label_str = str(label)
device_label = self._device.label or ""
# Strip device name prefix from channel label to avoid
# duplication when HA composes "{device_name} {entity_name}".
# E.g., device "Licht Flur" + channel "Licht Flur 5" -> "5".
if device_label and label_str.startswith(device_label):
stripped = label_str[len(device_label) :].strip()
if stripped:
self._attr_name = stripped
# Otherwise channel label equals device label (modulo
# whitespace); leave _attr_name unset so HA composes just
# the device name without duplicating it.
return
self._attr_name = label_str
return
# Fallback: use post suffix or generic channel name.
if self._post:
self._attr_name = self._post[0].upper() + self._post[1:]
else:
self._attr_name = f"Channel{self.get_channel_index()}"
return
# Entities with a post suffix: use it as the entity name,
# capitalizing the first letter for display consistency.
if self._post:
self._attr_name = self._post[0].upper() + self._post[1:]
return
# Non-multi-channel entities on devices with multiple channels:
# use the first functional channel's label as name context.
# This preserves names like "Treppe CH" for single-function entities
# on multi-channel devices (e.g., HmIP-BSL switch channel).
functional_channels = getattr(self._device, "functionalChannels", None)
if functional_channels and len(functional_channels) > 1:
ch1 = (
functional_channels.get(1)
if isinstance(functional_channels, dict)
else functional_channels[1]
)
label = getattr(ch1, "label", None) if ch1 else None
if label:
label_str = str(label)
device_label = self._device.label or ""
# Strip device name prefix to avoid duplication.
if device_label and label_str.startswith(device_label):
stripped = label_str[len(device_label) :].strip()
if stripped:
self._attr_name = stripped
# Otherwise channel label equals device label (modulo
# whitespace); leave _attr_name unset.
return
self._attr_name = label_str
return
# Primary entity on device: leave unset so HA derives name from
# device_class or translation_key.
@property
def available(self) -> bool:
"""Return if entity is available."""
@@ -82,7 +82,6 @@ class HomematicipDoorBellEvent(HomematicipGenericEntity, EventEntity):
super().__init__(
hap,
device,
post=description.key,
channel=channel,
is_multi_channel=False,
feature_id="doorbell",
@@ -1070,9 +1070,7 @@ class HmipSmokeDetectorSensor(HomematicipGenericEntity, SensorEntity):
description: HmipSmokeDetectorSensorDescription,
) -> None:
"""Initialize the smoke detector sensor."""
super().__init__(
hap, device, post=description.key, feature_id="smoke_detector_sensor"
)
super().__init__(hap, device, feature_id="smoke_detector_sensor")
self.entity_description = description
self._sensor_unique_id = f"{device.id}_{description.key}"
@@ -37,6 +37,11 @@
}
},
"entity": {
"binary_sensor": {
"cloud_connection": {
"name": "Cloud connection"
}
},
"light": {
"optical_signal_light": {
"state_attributes": {
@@ -142,6 +142,8 @@ class HomematicipSwitch(HomematicipMultiSwitch, SwitchEntity):
class HomematicipGroupSwitch(HomematicipGenericEntity, SwitchEntity):
"""Representation of the HomematicIP switching group."""
_attr_has_entity_name = False
def __init__(self, hap: HomematicipHAP, device, post: str = "Group") -> None:
"""Initialize switching group."""
device.modelType = f"HmIP-{post}"
@@ -74,11 +74,6 @@ class HomematicipWeatherSensor(HomematicipGenericEntity, WeatherEntity):
"""Initialize the weather sensor."""
super().__init__(hap, device, feature_id="weather")
@property
def name(self) -> str:
"""Return the name of the sensor."""
return self._device.label
@property
def native_temperature(self) -> float:
"""Return the platform temperature."""
@@ -118,6 +113,7 @@ class HomematicipWeatherSensorPro(HomematicipWeatherSensor):
class HomematicipHomeWeather(HomematicipGenericEntity, WeatherEntity):
"""Representation of the HomematicIP home weather."""
_attr_has_entity_name = False
_attr_native_temperature_unit = UnitOfTemperature.CELSIUS
_attr_native_wind_speed_unit = UnitOfSpeed.KILOMETERS_PER_HOUR
_attr_attribution = "Powered by Homematic IP"
@@ -12,6 +12,6 @@
"documentation": "https://www.home-assistant.io/integrations/indevolt",
"integration_type": "device",
"iot_class": "local_polling",
"quality_scale": "silver",
"quality_scale": "platinum",
"requirements": ["indevolt-api==1.8.1"]
}
@@ -77,7 +77,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: LunatoneConfigEntry) ->
await coordinator_info.async_config_entry_first_refresh()
if info_api.data is None or info_api.serial_number is None:
# pylint: disable-next=home-assistant-exception-translation-key-missing
raise ConfigEntryError(
translation_domain=DOMAIN, translation_key="missing_device_info"
)
+2 -1
View File
@@ -207,7 +207,6 @@ class LunatoneLight(
await self.coordinator.async_refresh()
# pylint: disable-next=home-assistant-missing-has-entity-name
class LunatoneLineBroadcastLight(
CoordinatorEntity[LunatoneInfoDataUpdateCoordinator], LightEntity
):
@@ -217,6 +216,8 @@ class LunatoneLineBroadcastLight(
_attr_assumed_state = True
_attr_color_mode = ColorMode.BRIGHTNESS
_attr_has_entity_name = True
_attr_name = None
_attr_supported_color_modes = {ColorMode.BRIGHTNESS}
def __init__(
@@ -35,5 +35,10 @@
"description": "Enter the URL of your Lunatone device.\nHome Assistant will use this address to connect to the device API."
}
}
},
"exceptions": {
"missing_device_info": {
"message": "Unable to read device information. Please verify the device's network connection."
}
}
}
@@ -11,7 +11,7 @@ from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import CONF_NAME
from homeassistant.const import CONF_COUNTRY, CONF_LANGUAGE, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@@ -22,10 +22,6 @@ _LOGGER = logging.getLogger(__name__)
ATTRIBUTION = "Information provided by MeteoAlarm"
# pylint: disable-next=home-assistant-duplicate-const
CONF_COUNTRY = "country"
# pylint: disable-next=home-assistant-duplicate-const
CONF_LANGUAGE = "language"
CONF_PROVINCE = "province"
DEFAULT_NAME = "meteoalarm"
@@ -10,8 +10,6 @@ DEFAULT_DETECTION_TIME: Final = 300
ATTR_MANUFACTURER: Final = "Mikrotik"
ATTR_SERIAL_NUMBER: Final = "serial-number"
ATTR_FIRMWARE: Final = "current-firmware"
# pylint: disable-next=home-assistant-duplicate-const
ATTR_MODEL: Final = "model"
CONF_ARP_PING: Final = "arp_ping"
CONF_FORCE_DHCP: Final = "force_dhcp"
@@ -9,7 +9,13 @@ import librouteros
from librouteros.login import plain as login_plain, token as login_token
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, CONF_VERIFY_SSL
from homeassistant.const import (
ATTR_MODEL,
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@@ -17,7 +23,6 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from .const import (
ARP,
ATTR_FIRMWARE,
ATTR_MODEL,
ATTR_SERIAL_NUMBER,
CAPSMAN,
CONF_ARP_PING,
@@ -23,7 +23,7 @@ from music_assistant_models.errors import (
from music_assistant_models.player import Player
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import CONF_URL, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.const import CONF_TOKEN, CONF_URL, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
@@ -38,7 +38,7 @@ from homeassistant.helpers.issue_registry import (
async_delete_issue,
)
from .const import ATTR_CONF_EXPOSE_PLAYER_TO_HA, CONF_TOKEN, DOMAIN, LOGGER
from .const import ATTR_CONF_EXPOSE_PLAYER_TO_HA, DOMAIN, LOGGER
from .helpers import get_music_assistant_client
from .services import register_actions
@@ -21,7 +21,7 @@ from homeassistant.config_entries import (
ConfigFlow,
ConfigFlowResult,
)
from homeassistant.const import CONF_URL
from homeassistant.const import CONF_TOKEN, CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.config_entry_oauth2_flow import (
@@ -31,13 +31,7 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
from .const import (
AUTH_SCHEMA_VERSION,
CONF_TOKEN,
DOMAIN,
HASSIO_DISCOVERY_SCHEMA_VERSION,
LOGGER,
)
from .const import AUTH_SCHEMA_VERSION, DOMAIN, HASSIO_DISCOVERY_SCHEMA_VERSION, LOGGER
DEFAULT_TITLE = "Music Assistant"
DEFAULT_URL = "http://mass.local:8095"
@@ -12,9 +12,6 @@ AUTH_SCHEMA_VERSION = 28
# Schema version where hassio discovery support was added
HASSIO_DISCOVERY_SCHEMA_VERSION = 28
# pylint: disable-next=home-assistant-duplicate-const
CONF_TOKEN = "token"
ATTR_IS_GROUP = "is_group"
ATTR_GROUP_MEMBERS = "group_members"
ATTR_GROUP_PARENTS = "group_parents"
+1 -3
View File
@@ -23,7 +23,6 @@ _ATTRIBUTION = "Data provided by OMIE.es"
SENSOR_DESCRIPTIONS: dict[str, SensorEntityDescription] = {
key: SensorEntityDescription(
key=key,
has_entity_name=True,
translation_key=key,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=f"{CURRENCY_EURO}/{UnitOfEnergy.KILO_WATT_HOUR}",
@@ -33,11 +32,10 @@ SENSOR_DESCRIPTIONS: dict[str, SensorEntityDescription] = {
}
# has_entity_name=True is supplied by every SENSOR_DESCRIPTIONS entry
# pylint: disable-next=home-assistant-missing-has-entity-name
class OMIEPriceSensor(CoordinatorEntity[OMIECoordinator], SensorEntity):
"""OMIE price sensor."""
_attr_has_entity_name = True
_attr_should_poll = False
_attr_attribution = _ATTRIBUTION
@@ -3,7 +3,7 @@
from collections.abc import Mapping
from typing import Any
from aiopvpc import DEFAULT_POWER_KW, PVPCData
from esios_api import DEFAULT_POWER_KW, PVPCData
import voluptuous as vol
from homeassistant.config_entries import (
@@ -63,9 +63,10 @@ class TariffSelectorConfigFlow(ConfigFlow, domain=DOMAIN):
if user_input is not None:
await self.async_set_unique_id(user_input[ATTR_TARIFF])
self._abort_if_unique_id_configured()
calc_name = f"{DEFAULT_NAME} - {user_input[ATTR_TARIFF]}"
if not user_input[CONF_USE_API_TOKEN]:
return self.async_create_entry(
title=DEFAULT_NAME,
title=calc_name,
data={
ATTR_TARIFF: user_input[ATTR_TARIFF],
ATTR_POWER: user_input[ATTR_POWER],
@@ -74,7 +75,7 @@ class TariffSelectorConfigFlow(ConfigFlow, domain=DOMAIN):
},
)
self._name = DEFAULT_NAME
self._name = calc_name
self._tariff = user_input[ATTR_TARIFF]
self._power = user_input[ATTR_POWER]
self._power_p3 = user_input[ATTR_POWER_P3]
@@ -150,7 +151,7 @@ class TariffSelectorConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle re-authentication with ESIOS Token."""
self._api_token = entry_data.get(CONF_API_TOKEN)
self._use_api_token = self._api_token is not None
self._name = DEFAULT_NAME
self._name = f"{DEFAULT_NAME} - {entry_data[ATTR_TARIFF]}"
self._tariff = entry_data[ATTR_TARIFF]
self._power = entry_data[ATTR_POWER]
self._power_p3 = entry_data[ATTR_POWER_P3]
@@ -1,6 +1,6 @@
"""Constant values for pvpc_hourly_pricing."""
from aiopvpc.const import TARIFFS
from esios_api.const import TARIFFS
import voluptuous as vol
DOMAIN = "pvpc_hourly_pricing"
@@ -3,7 +3,7 @@
from datetime import timedelta
import logging
from aiopvpc import BadApiTokenAuthError, EsiosApiData, PVPCData
from esios_api import BadApiTokenAuthError, EsiosApiData, PVPCData
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_TOKEN
@@ -1,6 +1,6 @@
"""Helper functions to relate sensors keys and unique ids."""
from aiopvpc.const import (
from esios_api.const import (
ALL_SENSORS,
KEY_INJECTION,
KEY_MAG,
@@ -1,11 +1,11 @@
{
"domain": "pvpc_hourly_pricing",
"name": "Spain electricity hourly pricing (PVPC)",
"codeowners": ["@azogue"],
"codeowners": ["@azogue", "@chiro79"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/pvpc_hourly_pricing",
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["aiopvpc"],
"requirements": ["aiopvpc==4.3.1"]
"loggers": ["esios_api"],
"requirements": ["esios_api==4.4.0"]
}
@@ -5,7 +5,7 @@ from datetime import datetime
import logging
from typing import Any
from aiopvpc.const import KEY_INJECTION, KEY_MAG, KEY_OMIE, KEY_PVPC
from esios_api.const import KEY_INJECTION, KEY_MAG, KEY_OMIE, KEY_PVPC
from homeassistant.components.sensor import (
SensorEntity,
@@ -8,5 +8,5 @@
"iot_class": "cloud_polling",
"loggers": ["renault_api"],
"quality_scale": "silver",
"requirements": ["renault-api==0.5.9"]
"requirements": ["renault-api==0.5.10"]
}
@@ -401,7 +401,6 @@ class ShellyBlockCoordinator(ShellyCoordinatorBase[BlockDevice]):
"""Fetch data."""
if self.sleep_period:
# Sleeping device, no point polling it, just mark it unavailable
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_error_sleeping_device",
@@ -671,7 +670,6 @@ class ShellyRpcCoordinator(ShellyCoordinatorBase[RpcDevice]):
if self.sleep_period:
# Sleeping device, no point polling it, just mark it unavailable
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_error_sleeping_device",
+1 -1
View File
@@ -673,7 +673,7 @@
"message": "An error occurred while reconnecting to {device}"
},
"update_error_sleeping_device": {
"message": "Sleeping device did not update within {period} seconds interval"
"message": "Sleeping device {device} did not update within {period} seconds interval"
}
},
"issues": {
+1 -5
View File
@@ -344,14 +344,10 @@ class RpcUpdateEntity(ShellyRpcAttributeEntity, UpdateEntity):
translation_placeholders={"device": self.coordinator.name},
) from err
except RpcCallError as err:
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="ota_update_rpc_error",
translation_placeholders={
"entity": self.entity_id,
"device": self.coordinator.name,
},
translation_placeholders={"device": self.coordinator.name},
) from err
except InvalidAuthError:
await self.coordinator.async_shutdown_device_and_start_reauth()
@@ -22,10 +22,11 @@ async def async_setup_entry(
async_add_entities(SmartThingsScene(scene, client) for scene in scenes.values())
# pylint: disable-next=home-assistant-missing-has-entity-name
class SmartThingsScene(Scene):
"""Define a SmartThings scene."""
_attr_has_entity_name = True
def __init__(self, scene: STScene, client: SmartThings) -> None:
"""Init the scene class."""
self.client = client
-2
View File
@@ -9,8 +9,6 @@ ATTR_HTML: Final = "html"
ATTR_SENDER_NAME: Final = "sender_name"
CONF_ENCRYPTION: Final = "encryption"
# pylint: disable-next=home-assistant-duplicate-const
CONF_DEBUG: Final = "debug"
CONF_SERVER: Final = "server"
CONF_SENDER_NAME: Final = "sender_name"
+1 -1
View File
@@ -24,6 +24,7 @@ from homeassistant.components.notify import (
BaseNotificationService,
)
from homeassistant.const import (
CONF_DEBUG,
CONF_PASSWORD,
CONF_PORT,
CONF_RECIPIENT,
@@ -44,7 +45,6 @@ from homeassistant.util.ssl import create_client_context
from .const import (
ATTR_HTML,
ATTR_IMAGES,
CONF_DEBUG,
CONF_ENCRYPTION,
CONF_SENDER_NAME,
CONF_SERVER,
+1 -1
View File
@@ -191,11 +191,11 @@ class SonosAudioInputFormatSensorEntity(SonosPollingEntity, SensorEntity):
"""Provide a stub for required ABC method."""
# pylint: disable-next=home-assistant-missing-has-entity-name
class SonosFavoritesEntity(SensorEntity):
"""Representation of a Sonos favorites info entity."""
_attr_entity_registry_enabled_default = False
_attr_has_entity_name = True
_attr_name = "Sonos favorites"
_attr_translation_key = "favorites"
_attr_native_unit_of_measurement = "items"
@@ -38,12 +38,12 @@ async def async_setup_entry(
)
# pylint: disable-next=home-assistant-missing-has-entity-name
class OmadaClientScannerEntity(
CoordinatorEntity[OmadaClientsCoordinator], ScannerEntity
):
"""Entity for a client connected to the Omada network."""
_attr_has_entity_name = True
_client_details: OmadaWirelessClient | None = None
def __init__(
+1 -5
View File
@@ -48,14 +48,10 @@ class LgWebOSNotificationService(BaseNotificationService):
icon_path = data.get(ATTR_ICON) if data else None
if not client.tv_state.is_on:
# pylint: disable-next=home-assistant-exception-placeholder-mismatch
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="notify_device_off",
translation_placeholders={
"name": str(self._entry.title),
"func": __name__,
},
translation_placeholders={"name": str(self._entry.title)},
)
try:
await client.send_message(message, icon_path=icon_path)
@@ -14,10 +14,9 @@ async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: ZeversolarConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
data: ZeverSolarData = config_entry.runtime_data.data
payload: dict[str, Any] = {
return {
"wifi_enabled": data.wifi_enabled,
"serial_or_registry_id": data.serial_or_registry_id,
"registry_key": data.registry_key,
@@ -33,8 +32,6 @@ async def async_get_config_entry_diagnostics(
"meter_status": data.meter_status.value,
}
return payload
async def async_get_device_diagnostics(
hass: HomeAssistant, entry: ZeversolarConfigEntry, device: DeviceEntry
@@ -42,15 +39,13 @@ async def async_get_device_diagnostics(
"""Return diagnostics for a device entry."""
coordinator = entry.runtime_data
updateInterval = (
None
if coordinator.update_interval is None
else coordinator.update_interval.total_seconds()
)
return {
"name": coordinator.name,
"always_update": coordinator.always_update,
"last_update_success": coordinator.last_update_success,
"update_interval": updateInterval,
"update_interval": (
None
if coordinator.update_interval is None
else coordinator.update_interval.total_seconds()
),
}
+1
View File
@@ -91,6 +91,7 @@ CONF_COMMAND_ON: Final = "command_on"
CONF_COMMAND_OPEN: Final = "command_open"
CONF_COMMAND_STATE: Final = "command_state"
CONF_COMMAND_STOP: Final = "command_stop"
CONF_COMMENT: Final = "comment"
CONF_CONDITION: Final = "condition"
CONF_CONDITIONS: Final = "conditions"
CONF_CONTINUE_ON_ERROR: Final = "continue_on_error"
@@ -38,6 +38,7 @@ from homeassistant.const import (
CONF_ATTRIBUTE,
CONF_BELOW,
CONF_CHOOSE,
CONF_COMMENT,
CONF_CONDITION,
CONF_CONDITIONS,
CONF_CONTINUE_ON_ERROR,
@@ -1458,6 +1459,7 @@ SCRIPT_SCHEMA = vol.All(ensure_list, [script_action])
SCRIPT_ACTION_BASE_SCHEMA: VolDictType = {
vol.Optional(CONF_ALIAS): string,
vol.Remove(CONF_COMMENT): str, # Is only used in frontend
vol.Optional(CONF_CONTINUE_ON_ERROR): boolean,
vol.Optional(CONF_ENABLED): vol.Any(boolean, template),
}
@@ -1525,6 +1527,7 @@ NUMERIC_STATE_THRESHOLD_SCHEMA = vol.Any(
CONDITION_BASE_SCHEMA: VolDictType = {
vol.Optional(CONF_ALIAS): string,
vol.Remove(CONF_COMMENT): str, # Is only used in frontend
vol.Optional(CONF_ENABLED): vol.Any(boolean, template),
}
@@ -1859,6 +1862,7 @@ TRIGGER_BASE_SCHEMA = vol.Schema(
vol.Optional(CONF_ID): str,
vol.Optional(CONF_VARIABLES): SCRIPT_VARIABLES_SCHEMA,
vol.Optional(CONF_ENABLED): vol.Any(boolean, template),
vol.Remove(CONF_COMMENT): str, # Is only used in frontend
}
)
+4 -4
View File
@@ -374,9 +374,6 @@ aiopurpleair==2025.08.1
# homeassistant.components.hunterdouglas_powerview
aiopvapi==3.3.0
# homeassistant.components.pvpc_hourly_pricing
aiopvpc==4.3.1
# homeassistant.components.lidarr
# homeassistant.components.radarr
# homeassistant.components.sonarr
@@ -951,6 +948,9 @@ epson-projector==0.6.0
# homeassistant.components.eq3btsmart
eq3btsmart==2.3.0
# homeassistant.components.pvpc_hourly_pricing
esios_api==4.4.0
# homeassistant.components.esphome
esphome-dashboard-api==1.3.0
@@ -2865,7 +2865,7 @@ refoss-ha==1.2.5
regenmaschine==2024.03.0
# homeassistant.components.renault
renault-api==0.5.9
renault-api==0.5.10
# homeassistant.components.renson
renson-endura-delta==1.7.2
View File
+2
View File
@@ -27,6 +27,8 @@ class CheckKind(StrEnum):
RELEASE_PIPELINE = "release_pipeline"
PR_LINK = "pr_link"
ASYNC_BLOCKING = "async_blocking"
YANKED = "yanked"
VULNERABILITIES = "vulnerabilities"
@dataclass(slots=True)
+50 -2
View File
@@ -1,6 +1,6 @@
"""PyPI metadata + PEP 740 provenance attestation lookups."""
from dataclasses import dataclass
from dataclasses import dataclass, field
import logging
import re
from typing import Any
@@ -69,6 +69,17 @@ _HEADERS = {
_TIMEOUT = 30.0
@dataclass(slots=True, frozen=True)
class Vulnerability:
"""One advisory entry for a specific package version (OSV / PyPA / GHSA)."""
id: str
aliases: tuple[str, ...]
summary: str
fixed_in: tuple[str, ...]
link: str
@dataclass(slots=True)
class PypiPackageInfo:
"""The subset of PyPI metadata we care about for a specific version."""
@@ -77,6 +88,9 @@ class PypiPackageInfo:
repo_url: str | None
file_provenance_urls: list[str] # may be empty
found: bool # False if the version doesn't exist on PyPI
yanked: bool = False
yanked_reason: str | None = None
vulnerabilities: list[Vulnerability] = field(default_factory=list)
@dataclass(slots=True)
@@ -157,9 +171,43 @@ def fetch_package_info(name: str, version: str) -> PypiPackageInfo:
repo_url=_pick_repo_url(project_urls),
file_provenance_urls=provenance_urls,
found=True,
yanked=bool(info.get("yanked")),
yanked_reason=_safe(info.get("yanked_reason")),
vulnerabilities=_parse_vulnerabilities(versioned.get("vulnerabilities")),
)
def _parse_vulnerabilities(raw: Any) -> list[Vulnerability]:
"""Extract non-withdrawn advisories from the PyPI `vulnerabilities` field."""
if not isinstance(raw, list):
return []
out: list[Vulnerability] = []
for entry in raw:
if not isinstance(entry, dict):
continue
if entry.get("withdrawn"):
# Withdrawn means the advisory was removed by the maintainer
# and should not be treated as valid.
continue
vid = _safe(entry.get("id"))
if not vid:
continue
aliases_raw = entry.get("aliases") or []
aliases = tuple(a for a in (_safe(str(x)) for x in aliases_raw if x) if a)
fixed_raw = entry.get("fixed_in") or []
fixed_in = tuple(f for f in (_safe(str(x)) for x in fixed_raw if x) if f)
out.append(
Vulnerability(
id=vid,
aliases=aliases,
summary=_safe(entry.get("summary")) or "",
fixed_in=fixed_in,
link=_safe(entry.get("link")) or "",
)
)
return out
def check_provenance(pkg: PypiPackageInfo) -> ProvenanceResult:
"""Resolve the provenance attestation, if any, to a Step 2b verdict."""
if not pkg.found:
@@ -176,7 +224,7 @@ def check_provenance(pkg: PypiPackageInfo) -> ProvenanceResult:
if not bundle:
continue
any_bundle_fetched = True
for entry in bundle.get("attestation_bundles", []) or []:
for entry in bundle.get("attestation_bundles") or []:
publisher = entry.get("publisher") or {}
kind = publisher.get("kind")
if not kind:
+2
View File
@@ -16,6 +16,8 @@ HEADER = "## Check requirements"
# Column / bullet labels per check kind, in display order.
_CHECK_DISPLAY: tuple[tuple[CheckKind, str], ...] = (
(CheckKind.VULNERABILITIES, "No Advisories"),
(CheckKind.YANKED, "Not Yanked"),
(CheckKind.REPO_PUBLIC, "Repo Public"),
(CheckKind.CI_UPLOAD, "CI Upload"),
(CheckKind.RELEASE_PIPELINE, "Release Pipeline"),
+59
View File
@@ -1,6 +1,9 @@
"""Orchestrate the deterministic requirements checks for one PR.
What the runner resolves itself (deterministic):
- `yanked`: PASS if the new release is live on PyPI, FAIL if it was yanked.
- `vulnerabilities`: FAIL if PyPI reports any non-withdrawn OSV / GHSA / CVE
advisory for the new version; PASS otherwise.
- `ci_upload`: PASS / WARN / FAIL based on PEP 740 attestation on PyPI.
- `release_pipeline`: PASS only when the attestation already identifies a
recognised CI publisher; otherwise NEEDS_AGENT.
@@ -21,6 +24,60 @@ from .pypi import PypiPackageInfo, check_provenance, fetch_package_info
from .render import render_comment
def _resolve_yanked(pkg: PackageChange, pypi_info: PypiPackageInfo) -> None:
"""Mark the release as yanked / not yanked."""
if not pypi_info.found:
pkg.checks[CheckKind.YANKED] = CheckResult(
CheckStatus.FAIL,
f"Version {pkg.new_version} not found on PyPI.",
)
return
if pypi_info.yanked:
reason = pypi_info.yanked_reason or "no reason provided by uploader"
pkg.checks[CheckKind.YANKED] = CheckResult(
CheckStatus.FAIL,
f"Version {pkg.new_version} is yanked on PyPI ({reason}). "
"Home Assistant should not depend on a yanked release.",
)
return
pkg.checks[CheckKind.YANKED] = CheckResult(
CheckStatus.PASS,
f"Version {pkg.new_version} is a live (non-yanked) release.",
)
def _resolve_vulnerabilities(pkg: PackageChange, pypi_info: PypiPackageInfo) -> None:
"""Flag versions with active OSV / GHSA / CVE advisories on PyPI."""
if not pypi_info.found:
pkg.checks[CheckKind.VULNERABILITIES] = CheckResult(
CheckStatus.FAIL,
f"Version {pkg.new_version} not found on PyPI.",
)
return
vulns = pypi_info.vulnerabilities
if not vulns:
pkg.checks[CheckKind.VULNERABILITIES] = CheckResult(
CheckStatus.PASS,
f"No active advisories reported by PyPI for version {pkg.new_version}.",
)
return
entries: list[str] = []
for vuln in vulns:
# Prefer a CVE alias as the primary label when present.
cve = next((a for a in vuln.aliases if a.upper().startswith("CVE-")), None)
label = cve or vuln.id
fixed = ", ".join(vuln.fixed_in) if vuln.fixed_in else "no fix listed"
if vuln.link:
entries.append(f"[{label}]({vuln.link}) (fixed in: {fixed})")
else:
entries.append(f"{label} (fixed in: {fixed})")
pkg.checks[CheckKind.VULNERABILITIES] = CheckResult(
CheckStatus.FAIL,
f"PyPI reports {len(vulns)} active advisories for version "
f"{pkg.new_version}: " + "; ".join(entries) + ".",
)
def _resolve_ci_upload_and_release_pipeline(
pkg: PackageChange, pypi_info: PypiPackageInfo
) -> None:
@@ -69,6 +126,8 @@ def run_checks(
for pkg in packages:
pypi_info = fetch_package_info(pkg.name, pkg.new_version)
pkg.repo_url = pypi_info.repo_url
_resolve_yanked(pkg, pypi_info)
_resolve_vulnerabilities(pkg, pypi_info)
_resolve_ci_upload_and_release_pipeline(pkg, pypi_info)
if not pypi_info.found:
fail = CheckResult(
-1
View File
@@ -203,7 +203,6 @@ FORBIDDEN_PACKAGE_EXCEPTIONS: dict[str, dict[str, set[str]]] = {
"opengarage": {"open-garage": {"async-timeout"}},
"overkiz": {"pyoverkiz": {"backoff"}},
"prosegur": {"pyprosegur": {"backoff"}},
"pvpc_hourly_pricing": {"aiopvpc": {"async-timeout"}},
"radio_browser": {"radios": {"backoff"}},
"remote_rpi_gpio": {
# https://github.com/waveform80/colorzero/issues/9
+93 -1
View File
@@ -8,7 +8,12 @@ from unittest.mock import call, patch
import pytest
from homeassistant.components import device_tracker, zone
from homeassistant.components.device_tracker import SourceType, const, legacy
from homeassistant.components.device_tracker import (
SourceType,
TrackerEntity,
const,
legacy,
)
from homeassistant.const import (
ATTR_ENTITY_PICTURE,
ATTR_FRIENDLY_NAME,
@@ -19,11 +24,15 @@ from homeassistant.const import (
CONF_PLATFORM,
STATE_HOME,
STATE_NOT_HOME,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, State, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import discovery
from homeassistant.helpers.discovery import DiscoveryInfoType
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.json import JSONEncoder
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
@@ -31,9 +40,13 @@ from . import common
from .common import MockScanner, mock_legacy_device_tracker_setup
from tests.common import (
MockModule,
MockPlatform,
RegistryEntryWithDefaults,
assert_setup_component,
async_fire_time_changed,
mock_integration,
mock_platform,
mock_registry,
mock_restore_cache,
patch_yaml_files,
@@ -729,3 +742,82 @@ def test_see_schema_allowing_ios_calls() -> None:
"hostname": "beer",
}
)
async def test_modern_platform_setup(hass: HomeAssistant) -> None:
"""Test modern platform setup."""
test_domain = "test"
entity1 = TrackerEntity()
entity1.entity_id = "device_tracker.test1"
entity1._attr_source_type = SourceType.ROUTER
entity2 = TrackerEntity()
entity2.entity_id = "device_tracker.test2"
entity2._attr_location_name = "home"
entity2._attr_location_accuracy = 1
entity2._attr_latitude = 10.0
entity2._attr_longitude = 5.0
entity2._attr_source_type = SourceType.GPS
entity3 = TrackerEntity()
entity3.entity_id = "device_tracker.test3"
entity3._attr_location_name = "not_home"
entity3._attr_source_type = SourceType.ROUTER
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> bool:
async_add_entities([entity1, entity2, entity3])
return True
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
hass.async_create_task(
discovery.async_load_platform(
hass, "device_tracker", test_domain, {}, config
)
)
return True
mock_integration(
hass,
MockModule(test_domain, async_setup=async_setup),
)
mock_platform(
hass,
f"{test_domain}.device_tracker",
MockPlatform(async_setup_platform=async_setup_platform),
)
await async_setup_component(hass, "homeassistant", {})
await async_setup_component(hass, "device_tracker", {})
await async_setup_component(hass, test_domain, {})
await hass.async_block_till_done()
state = hass.states.get(entity1.entity_id)
assert state
assert state.state == STATE_UNKNOWN
assert state.attributes == {"in_zones": [], "source_type": SourceType.ROUTER}
state = hass.states.get(entity2.entity_id)
assert state
assert state.state == STATE_HOME
assert state.attributes == {
"in_zones": [],
"source_type": SourceType.GPS,
"latitude": 10.0,
"longitude": 5.0,
"gps_accuracy": 1,
}
state = hass.states.get(entity3.entity_id)
assert state
assert state.state == STATE_NOT_HOME
assert state.attributes == {
"in_zones": [],
"source_type": SourceType.ROUTER,
}
+3 -1
View File
@@ -44,7 +44,9 @@ def get_and_check_entity_basics(
assert ha_state is not None
if device_model:
assert ha_state.attributes[ATTR_MODEL_TYPE] == device_model
assert ha_state.name == entity_name
assert ha_state.name == entity_name, (
f"Expected '{entity_name}', got '{ha_state.name}'"
)
hmip_device = mock_hap.hmip_device_by_entity_id.get(entity_id)
@@ -131,8 +131,8 @@ async def test_hmip_home_cloud_connection_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipCloudConnectionSensor."""
entity_id = "binary_sensor.cloud_connection"
entity_name = "Home Cloud Connection"
entity_id = "binary_sensor.home_cloud_connection"
entity_name = "Home Cloud connection"
device_model = None
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Cloud Connection"]
@@ -154,11 +154,11 @@ async def test_hmip_acceleration_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipAccelerationSensor."""
entity_id = "binary_sensor.garagentor"
entity_name = "Garagentor"
entity_id = "binary_sensor.garagentor_moving"
entity_name = "Garagentor Moving"
device_model = "HmIP-SAM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Garagentor"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -193,11 +193,11 @@ async def test_hmip_tilt_vibration_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipTiltVibrationSensor."""
entity_id = "binary_sensor.garage_neigungs_und_erschutterungssensor"
entity_name = "Garage Neigungs- und Erschütterungssensor"
entity_id = "binary_sensor.garage_neigungs_und_erschutterungssensor_moving"
entity_name = "Garage Neigungs- und Erschütterungssensor Moving"
device_model = "HmIP-STV"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Garage Neigungs- und Erschütterungssensor"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -231,11 +231,11 @@ async def test_hmip_contact_interface(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipContactInterface."""
entity_id = "binary_sensor.kontakt_schnittstelle_unterputz_1_fach"
entity_name = "Kontakt-Schnittstelle Unterputz 1-fach"
entity_id = "binary_sensor.kontakt_schnittstelle_unterputz_1_fach_opening"
entity_name = "Kontakt-Schnittstelle Unterputz 1-fach Opening"
device_model = "HmIP-FCI1"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Kontakt-Schnittstelle Unterputz 1-fach"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -256,11 +256,11 @@ async def test_hmip_shutter_contact(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipShutterContact."""
entity_id = "binary_sensor.fenstergriffsensor"
entity_name = "Fenstergriffsensor"
entity_id = "binary_sensor.fenstergriffsensor_door"
entity_name = "Fenstergriffsensor Door"
device_model = "HmIP-SRH"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Fenstergriffsensor"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -298,11 +298,11 @@ async def test_hmip_shutter_contact_optical(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipShutterContact."""
entity_id = "binary_sensor.sitzplatzture"
entity_name = "Sitzplatzt\u00fcre"
entity_id = "binary_sensor.sitzplatzture_door"
entity_name = "Sitzplatzt\u00fcre Door"
device_model = "HmIP-SWDO-PL"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Sitzplatzt\u00fcre"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -330,11 +330,11 @@ async def test_hmip_motion_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipMotionDetector."""
entity_id = "binary_sensor.bewegungsmelder_fur_55er_rahmen_innen"
entity_name = "Bewegungsmelder für 55er Rahmen innen"
entity_id = "binary_sensor.bewegungsmelder_fur_55er_rahmen_innen_motion"
entity_name = "Bewegungsmelder für 55er Rahmen innen Motion"
device_model = "HmIP-SMI55"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Bewegungsmelder für 55er Rahmen innen"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -351,12 +351,10 @@ async def test_hmip_presence_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipPresenceDetector."""
entity_id = "binary_sensor.spi_1"
entity_name = "SPI_1"
entity_id = "binary_sensor.spi_1_presence"
entity_name = "SPI_1 Presence"
device_model = "HmIP-SPI"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
)
mock_hap = await default_mock_hap_factory.async_get_mock_hap(test_devices=["SPI_1"])
ha_state, hmip_device = get_and_check_entity_basics(
hass, mock_hap, entity_id, entity_name, device_model
@@ -377,11 +375,11 @@ async def test_hmip_pluggable_mains_failure_surveillance_sensor(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipPresenceDetector."""
entity_id = "binary_sensor.netzausfalluberwachung"
entity_name = "Netzausfallüberwachung"
entity_id = "binary_sensor.netzausfalluberwachung_power"
entity_name = "Netzausfallüberwachung Power"
device_model = "HmIP-PMFS"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Netzausfallüberwachung"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -398,11 +396,11 @@ async def test_hmip_smoke_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipSmokeDetector."""
entity_id = "binary_sensor.rauchwarnmelder"
entity_name = "Rauchwarnmelder"
entity_id = "binary_sensor.rauchwarnmelder_smoke"
entity_name = "Rauchwarnmelder Smoke"
device_model = "HmIP-SWSD"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Rauchwarnmelder"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -453,11 +451,11 @@ async def test_hmip_water_detector(
hass: HomeAssistant, default_mock_hap_factory: HomeFactory
) -> None:
"""Test HomematicipWaterDetector."""
entity_id = "binary_sensor.wassersensor"
entity_name = "Wassersensor"
entity_id = "binary_sensor.wassersensor_moisture"
entity_name = "Wassersensor Moisture"
device_model = "HmIP-SWD"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=[entity_name]
test_devices=["Wassersensor"]
)
ha_state, hmip_device = get_and_check_entity_basics(
@@ -212,8 +212,8 @@ async def test_hap_with_name(
) -> None:
"""Test hap with name."""
home_name = "TestName"
entity_id = "light.treppe_testname_treppe_ch"
entity_name = "Treppe TestName Treppe CH"
entity_id = "light.testname_treppe_ch"
entity_name = "TestName Treppe CH"
device_model = "HmIP-BSL"
hmip_config_entry.add_to_hass(hass)
@@ -0,0 +1,115 @@
"""Unit tests for HomematicipGenericEntity naming helpers."""
from types import SimpleNamespace
import pytest
from homeassistant.components.homematicip_cloud.entity import HomematicipGenericEntity
def _make_entity(
*,
device_label: str,
channels: dict[int, str],
channel: int,
is_multi_channel: bool,
post: str | None = None,
) -> HomematicipGenericEntity:
"""Build a HomematicipGenericEntity bypassing __init__ for unit testing.
Only the attributes read by _setup_entity_name() are populated.
"""
entity = HomematicipGenericEntity.__new__(HomematicipGenericEntity)
entity._device = SimpleNamespace(
label=device_label,
functionalChannels={
idx: SimpleNamespace(label=label, index=idx)
for idx, label in channels.items()
},
)
entity._post = post
entity._channel = channel
entity._channel_real_index = channel
entity._is_multi_channel = is_multi_channel
entity.functional_channel = entity._device.functionalChannels.get(channel)
entity._attr_name = None
return entity
@pytest.mark.parametrize(
("device_label", "channel_label"),
[
("Thermostat EG Wohnzimmer", "Thermostat EG Wohnzimmer"),
("Thermostat EG Wohnzimmer", "Thermostat EG Wohnzimmer "),
("Thermostat EG Wohnzimmer ", "Thermostat EG Wohnzimmer "),
],
ids=["exact-match", "trailing-space-on-channel", "trailing-space-on-both"],
)
def test_multi_channel_label_equals_device_label_leaves_name_unset(
device_label: str, channel_label: str
) -> None:
"""When channel label equals device label, _attr_name must stay None.
Otherwise HA composes "{device_name} {entity_name}" and the user sees the
label duplicated, e.g. "Thermostat EG Wohnzimmer Thermostat EG Wohnzimmer".
"""
entity = _make_entity(
device_label=device_label,
channels={3: channel_label, 1: "primary"},
channel=3,
is_multi_channel=True,
)
entity._setup_entity_name()
assert entity._attr_name is None
def test_multi_channel_label_extends_device_label_keeps_suffix() -> None:
"""When channel label starts with device label + suffix, keep the suffix."""
entity = _make_entity(
device_label="Licht Flur",
channels={5: "Licht Flur 5"},
channel=5,
is_multi_channel=True,
)
entity._setup_entity_name()
assert entity._attr_name == "5"
def test_multi_channel_label_unrelated_to_device_label_uses_full_label() -> None:
"""When channel label does not start with device label, use it verbatim."""
entity = _make_entity(
device_label="DRS8-4",
channels={2: "Flur OG Lampe"},
channel=2,
is_multi_channel=True,
)
entity._setup_entity_name()
assert entity._attr_name == "Flur OG Lampe"
def test_primary_entity_with_channel_1_label_equals_device_label_unset() -> None:
"""Non-multi-channel entity on device whose ch1 label equals device label.
Same duplication risk as the multi-channel case; same fix.
"""
entity = _make_entity(
device_label="Thermostat EG Wohnzimmer",
channels={1: "Thermostat EG Wohnzimmer ", 2: "other"},
channel=2, # non-multi-channel entity sits on some channel
is_multi_channel=False,
)
entity._setup_entity_name()
assert entity._attr_name is None
def test_post_suffix_capitalised() -> None:
"""Post suffix becomes the entity name with first letter uppercased."""
entity = _make_entity(
device_label="Bewegungsmelder Küche",
channels={1: "primary"},
channel=1,
is_multi_channel=False,
post="battery",
)
entity._setup_entity_name()
assert entity._attr_name == "Battery"
@@ -807,7 +807,7 @@ async def test_hmip_water_valve_current_water_flow(
) -> None:
"""Test HomematicipCurrentWaterFlow."""
entity_id = "sensor.bewaesserungsaktor_currentwaterflow"
entity_name = "Bewaesserungsaktor currentWaterFlow"
entity_name = "Bewaesserungsaktor CurrentWaterFlow"
device_model = "ELV-SH-WSM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Bewaesserungsaktor"]
@@ -830,7 +830,7 @@ async def test_hmip_water_valve_water_volume(
) -> None:
"""Test HomematicipWaterVolume."""
entity_id = "sensor.bewaesserungsaktor_watervolume"
entity_name = "Bewaesserungsaktor waterVolume"
entity_name = "Bewaesserungsaktor WaterVolume"
device_model = "ELV-SH-WSM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Bewaesserungsaktor"]
@@ -850,7 +850,7 @@ async def test_hmip_water_valve_water_volume_since_open(
) -> None:
"""Test HomematicipWaterVolumeSinceOpen."""
entity_id = "sensor.bewaesserungsaktor_watervolumesinceopen"
entity_name = "Bewaesserungsaktor waterVolumeSinceOpen"
entity_name = "Bewaesserungsaktor WaterVolumeSinceOpen"
device_model = "ELV-SH-WSM"
mock_hap = await default_mock_hap_factory.async_get_mock_hap(
test_devices=["Bewaesserungsaktor"]
@@ -870,7 +870,7 @@ async def test_hmip_smoke_detector_dirt_level(
) -> None:
"""Test HomematicipSmokeDetectorDirtLevel."""
entity_id = "sensor.rauchwarnmelder_dirt_level"
entity_name = "Rauchwarnmelder Dirt_level"
entity_name = "Rauchwarnmelder Dirt level"
device_model = "HmIP-SWSD"
# Pre-register the entity as enabled before platform loads
@@ -910,7 +910,7 @@ async def test_hmip_smoke_detector_alarm_counter(
) -> None:
"""Test HomematicipSmokeDetectorAlarmCounter."""
entity_id = "sensor.rauchwarnmelder_smoke_alarm_counter"
entity_name = "Rauchwarnmelder Smoke_alarm_counter"
entity_name = "Rauchwarnmelder Alarm counter"
device_model = "HmIP-SWSD"
# Pre-register the entity as enabled before platform loads
@@ -947,7 +947,7 @@ async def test_hmip_smoke_detector_test_counter(
) -> None:
"""Test HomematicipSmokeDetectorTestCounter."""
entity_id = "sensor.rauchwarnmelder_smoke_test_counter"
entity_name = "Rauchwarnmelder Smoke_test_counter"
entity_name = "Rauchwarnmelder Test counter"
device_model = "HmIP-SWSD"
# Pre-register the entity as enabled before platform loads
@@ -18,7 +18,7 @@
'domain': 'light',
'entity_category': None,
'entity_id': 'light.dali_line_0',
'has_entity_name': False,
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
@@ -79,7 +79,7 @@
'domain': 'light',
'entity_category': None,
'entity_id': 'light.dali_line_1',
'has_entity_name': False,
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
@@ -22,7 +22,6 @@ from homeassistant.components.music_assistant.config_flow import (
)
from homeassistant.components.music_assistant.const import (
AUTH_SCHEMA_VERSION,
CONF_TOKEN,
DEFAULT_NAME,
DOMAIN,
)
@@ -34,6 +33,7 @@ from homeassistant.config_entries import (
SOURCE_ZEROCONF,
ConfigEntryState,
)
from homeassistant.const import CONF_TOKEN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.hassio import HassioServiceInfo
+2 -2
View File
@@ -1027,7 +1027,7 @@ async def test_block_sleeping_device_connection_error(
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert "Sleeping device did not update" in caplog.text
assert "Sleeping device Test name did not update" in caplog.text
assert (state := hass.states.get(entity_id))
assert state.state == STATE_UNAVAILABLE
@@ -1081,7 +1081,7 @@ async def test_rpc_sleeping_device_connection_error(
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
assert "Sleeping device did not update" in caplog.text
assert "Sleeping device Test name did not update" in caplog.text
assert (state := hass.states.get(entity_id))
assert state.state == STATE_UNAVAILABLE
@@ -14,7 +14,7 @@
'domain': 'scene',
'entity_category': None,
'entity_id': 'scene.away',
'has_entity_name': False,
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
@@ -67,7 +67,7 @@
'domain': 'scene',
'entity_category': None,
'entity_id': 'scene.home',
'has_entity_name': False,
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
+53
View File
@@ -1,6 +1,7 @@
"""Test config validators."""
from collections import OrderedDict
from collections.abc import Callable
from datetime import date, datetime, timedelta
import enum
from functools import partial
@@ -2031,3 +2032,55 @@ def test_stop_action_schema_error_false_with_response() -> None:
# no error with response_variable should work
config = schema({"stop": "Done", "response_variable": "result"})
assert config["response_variable"] == "result"
_COMMENT_SCHEMA_PARAMS = [
pytest.param(
cv.TRIGGER_BASE_SCHEMA,
{"platform": "event"},
id="trigger_base",
),
pytest.param(
cv.CONDITION_SCHEMA,
{"condition": "state", "entity_id": "sun.sun", "state": "above_horizon"},
id="condition",
),
pytest.param(
cv.script_action,
{"action": "test.foo"},
id="script_action",
),
]
@pytest.mark.parametrize(("validator", "base_config"), _COMMENT_SCHEMA_PARAMS)
@pytest.mark.usefixtures("hass")
def test_base_schemas_accept_comment(
validator: Callable[[dict[str, Any]], dict[str, Any]],
base_config: dict[str, Any],
) -> None:
"""Test that the comment field is accepted and stripped from the output."""
validated = validator({**base_config, "comment": "Single line"})
assert "comment" not in validated
@pytest.mark.parametrize(("validator", "base_config"), _COMMENT_SCHEMA_PARAMS)
@pytest.mark.parametrize(
"invalid_comment",
[
pytest.param(None, id="none"),
pytest.param(42, id="int"),
pytest.param(True, id="bool"),
pytest.param([], id="list"),
pytest.param({}, id="dict"),
],
)
@pytest.mark.usefixtures("hass")
def test_base_schemas_reject_invalid_comment(
validator: Callable[[dict[str, Any]], dict[str, Any]],
base_config: dict[str, Any],
invalid_comment: Any,
) -> None:
"""Test that script, condition, trigger base schemas reject non-string comments."""
with pytest.raises(vol.Invalid):
validator({**base_config, "comment": invalid_comment})
@@ -204,6 +204,85 @@ def test_fetch_package_info_picks_repo_url_from_project_urls(
assert info.repo_url == expected_repo_url
def test_fetch_package_info_extracts_yanked_fields(
requests_mock: rm.Mocker,
) -> None:
"""The fetcher lifts `yanked` and `yanked_reason` from PyPI."""
requests_mock.get(
_versioned_url("foo", "1.0"),
json={
"info": {
"project_urls": {},
"yanked": True,
"yanked_reason": "broken on 3.14",
},
"urls": [],
},
)
info = fetch_package_info("foo", "1.0")
assert info.found is True
assert info.yanked is True
assert info.yanked_reason == "broken on 3.14"
def test_fetch_package_info_extracts_vulnerabilities(
requests_mock: rm.Mocker,
) -> None:
"""Active OSV / GHSA / CVE advisories on PyPI are surfaced and parsed."""
requests_mock.get(
_versioned_url("foo", "1.0"),
json={
"info": {"project_urls": {}},
"urls": [],
"vulnerabilities": [
{
"id": "GHSA-aaaa-bbbb-cccc",
"aliases": ["CVE-2099-12345"],
"summary": "remote code execution",
"fixed_in": ["1.1", "1.2"],
"link": "https://osv.dev/vulnerability/GHSA-aaaa-bbbb-cccc",
"withdrawn": None,
},
{
"id": "GHSA-dddd-eeee-ffff",
"aliases": [],
"summary": "withdrawn advisory",
"fixed_in": [],
"link": "https://osv.dev/vulnerability/GHSA-dddd-eeee-ffff",
"withdrawn": "2024-01-01T00:00:00Z",
},
],
},
)
info = fetch_package_info("foo", "1.0")
# The withdrawn advisory is filtered out.
assert len(info.vulnerabilities) == 1
vuln = info.vulnerabilities[0]
assert vuln.id == "GHSA-aaaa-bbbb-cccc"
assert vuln.aliases == ("CVE-2099-12345",)
assert vuln.fixed_in == ("1.1", "1.2")
assert "remote code execution" in vuln.summary
def test_fetch_package_info_defaults_when_yanked_fields_absent(
requests_mock: rm.Mocker,
) -> None:
"""Missing `yanked` keys default to False / None."""
requests_mock.get(
_versioned_url("foo", "1.0"),
json={"info": {"project_urls": {}}, "urls": []},
)
info = fetch_package_info("foo", "1.0")
assert info.yanked is False
assert info.yanked_reason is None
def test_fetch_package_info_strips_dangerous_chars_from_repo_url(
requests_mock: rm.Mocker,
) -> None:
+144 -1
View File
@@ -5,7 +5,11 @@ import json
import pytest
from script.check_requirements.models import CheckKind, CheckStatus
from script.check_requirements.pypi import ProvenanceResult, PypiPackageInfo
from script.check_requirements.pypi import (
ProvenanceResult,
PypiPackageInfo,
Vulnerability,
)
from script.check_requirements.runner import run_checks
@@ -268,6 +272,145 @@ def test_runner_async_blocking_version_bump_diff_only(
assert "diff" in detail
def test_runner_yanked_release_fails(monkeypatch: pytest.MonkeyPatch) -> None:
"""A yanked release on PyPI must FAIL the yanked check."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
yanked=True,
yanked_reason="critical bug",
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.YANKED].status == CheckStatus.FAIL
assert "yanked" in pkg.checks[CheckKind.YANKED].details
assert "critical bug" in pkg.checks[CheckKind.YANKED].details
def test_runner_non_yanked_release_passes(monkeypatch: pytest.MonkeyPatch) -> None:
"""A normal (non-yanked) release passes the yanked check."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.YANKED].status == CheckStatus.PASS
def test_runner_active_vulnerabilities_fail(monkeypatch: pytest.MonkeyPatch) -> None:
"""An active CVE/GHSA on PyPI fails the vulnerabilities check with a link."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
vulnerabilities=[
Vulnerability(
id="GHSA-xxxx-xxxx-xxxx",
aliases=("CVE-2099-12345",),
summary="rce",
fixed_in=("1.2.0",),
link="https://osv.dev/vulnerability/GHSA-xxxx-xxxx-xxxx",
),
],
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.VULNERABILITIES].status == CheckStatus.FAIL
details = pkg.checks[CheckKind.VULNERABILITIES].details
# CVE alias is preferred as the link label when present.
assert "[CVE-2099-12345](" in details
assert "[GHSA-xxxx-xxxx-xxxx](" not in details
assert "fixed in: 1.2.0" in details
assert "https://osv.dev/vulnerability/GHSA-xxxx-xxxx-xxxx" in details
def test_runner_no_vulnerabilities_passes(monkeypatch: pytest.MonkeyPatch) -> None:
"""An empty `vulnerabilities` list passes the check."""
_patch_pypi(
monkeypatch,
PypiPackageInfo(
project_urls={"Source": "https://github.com/x/y"},
repo_url="https://github.com/x/y",
file_provenance_urls=["whatever"],
found=True,
),
ProvenanceResult(
has_attestation=True,
publisher_kind="GitHub",
recognized_publisher=True,
detail="ok",
),
)
diff = (
"diff --git a/requirements_all.txt b/requirements_all.txt\n"
"--- a/requirements_all.txt\n"
"+++ b/requirements_all.txt\n"
"@@ -1 +1 @@\n"
"-pkg==1.0.0\n"
"+pkg==1.1.0\n"
)
result = run_checks(pr_number=1, diff_text=diff)
pkg = result.packages[0]
assert pkg.checks[CheckKind.VULNERABILITIES].status == CheckStatus.PASS
def test_runner_serialises_to_json(monkeypatch: pytest.MonkeyPatch) -> None:
"""The artifact contract: `to_dict()` is JSON-serialisable with expected keys."""
_patch_pypi(
+31
View File
@@ -3,6 +3,7 @@
import asyncio
import dataclasses
import logging
from typing import Any
from unittest.mock import Mock, patch
import pytest
@@ -1272,3 +1273,33 @@ def test_nested_section_in_serializer() -> None:
{"collapsed": False},
)
)
@pytest.mark.parametrize(
("context", "expected_show_advanced"),
[
({}, False),
({"show_advanced_options": False}, False),
({"show_advanced_options": True}, True),
],
)
async def test_show_advanced_options(
manager: MockFlowManager, context: dict[str, Any], expected_show_advanced: bool
) -> None:
"""Test FlowHandler show_advanced_options property."""
@manager.mock_reg_handler("test")
class TestFlow(data_entry_flow.FlowHandler):
VERSION = 5
async def async_step_init(self, info):
assert self.show_advanced_options == expected_show_advanced
return self.async_create_entry(title="hello", data={})
await manager.async_init("test", context=context, data={})
assert len(manager.async_progress()) == 0
assert len(manager.mock_created_entries) == 1
entry = manager.mock_created_entries[0]
assert entry["handler"] == "test"
assert entry["title"] == "hello"