Add labs helper

homeassistant/components/kitchen_sink/__init__.py

@@ -11,7 +11,6 @@ from random import random
 
 import voluptuous as vol
 
-from homeassistant.components.labs import async_is_preview_feature_enabled, async_listen
 from homeassistant.components.recorder import DOMAIN as RECORDER_DOMAIN, get_instance
 from homeassistant.components.recorder.models import (
     StatisticData,
@@ -39,6 +38,7 @@ from homeassistant.helpers.issue_registry import (
     async_create_issue,
     async_delete_issue,
 )
+from homeassistant.helpers.labs import async_is_preview_feature_enabled, async_listen
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.util import dt as dt_util
 from homeassistant.util.unit_conversion import (

homeassistant/components/labs/__init__.py

@@ -7,36 +7,24 @@ in the Home Assistant Labs UI for users to enable or disable.
 
 from __future__ import annotations
 
-from collections.abc import Callable
 import logging
 
-from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.core import HomeAssistant
 from homeassistant.generated.labs import LABS_PREVIEW_FEATURES
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.storage import Store
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.loader import async_get_custom_components
 
-from .const import DOMAIN, EVENT_LABS_UPDATED, LABS_DATA, STORAGE_KEY, STORAGE_VERSION
+from .const import DOMAIN, LABS_DATA, STORAGE_KEY, STORAGE_VERSION
-from .models import (
+from .models import LabPreviewFeature, LabsData, LabsStoreData, NativeLabsStoreData
-    EventLabsUpdatedData,
-    LabPreviewFeature,
-    LabsData,
-    LabsStoreData,
-    NativeLabsStoreData,
-)
 from .websocket_api import async_setup as async_setup_ws_api
 
 _LOGGER = logging.getLogger(__name__)
 
 CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
 
-__all__ = [
+__all__ = []
-    "EVENT_LABS_UPDATED",
-    "EventLabsUpdatedData",
-    "async_is_preview_feature_enabled",
-    "async_listen",
-]
 
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
@@ -134,55 +122,3 @@ async def _async_scan_all_preview_features(
 
     _LOGGER.debug("Loaded %d total lab preview features", len(preview_features))
     return preview_features
-
-
-@callback
-def async_is_preview_feature_enabled(
-    hass: HomeAssistant, domain: str, preview_feature: str
-) -> bool:
-    """Check if a lab preview feature is enabled.
-
-    Args:
-        hass: HomeAssistant instance
-        domain: Integration domain
-        preview_feature: Preview feature name
-
-    Returns:
-        True if the preview feature is enabled, False otherwise
-    """
-    if LABS_DATA not in hass.data:
-        return False
-
-    labs_data = hass.data[LABS_DATA]
-    return (domain, preview_feature) in labs_data.data.preview_feature_status
-
-
-@callback
-def async_listen(
-    hass: HomeAssistant,
-    domain: str,
-    preview_feature: str,
-    listener: Callable[[], None],
-) -> Callable[[], None]:
-    """Listen for changes to a specific preview feature.
-
-    Args:
-        hass: HomeAssistant instance
-        domain: Integration domain
-        preview_feature: Preview feature name
-        listener: Callback to invoke when the preview feature is toggled
-
-    Returns:
-        Callable to unsubscribe from the listener
-    """
-
-    @callback
-    def _async_feature_updated(event: Event[EventLabsUpdatedData]) -> None:
-        """Handle labs feature update event."""
-        if (
-            event.data["domain"] == domain
-            and event.data["preview_feature"] == preview_feature
-        ):
-            listener()
-
-    return hass.bus.async_listen(EVENT_LABS_UPDATED, _async_feature_updated)

homeassistant/components/labs/const.py

@@ -11,6 +11,4 @@ DOMAIN = "labs"
 STORAGE_KEY = "core.labs"
 STORAGE_VERSION = 1
 
-EVENT_LABS_UPDATED = "labs_updated"
-
 LABS_DATA: HassKey[LabsData] = HassKey(DOMAIN)

homeassistant/components/labs/models.py

@@ -9,14 +9,6 @@ if TYPE_CHECKING:
     from homeassistant.helpers.storage import Store
 
 
-class EventLabsUpdatedData(TypedDict):
-    """Event data for labs_updated event."""
-
-    domain: str
-    preview_feature: str
-    enabled: bool
-
-
 @dataclass(frozen=True, kw_only=True, slots=True)
 class LabPreviewFeature:
     """Lab preview feature definition."""

homeassistant/components/labs/websocket_api.py

@@ -9,9 +9,9 @@ import voluptuous as vol
 from homeassistant.components import websocket_api
 from homeassistant.components.backup import async_get_manager
 from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.labs import EVENT_LABS_UPDATED, EventLabsUpdatedData
 
-from .const import EVENT_LABS_UPDATED, LABS_DATA
+from .const import LABS_DATA
-from .models import EventLabsUpdatedData
 
 
 @callback

homeassistant/helpers/labs.py

@@ -0,0 +1,72 @@
+"""Helpers to check labs features."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TypedDict
+
+from homeassistant.core import Event, HomeAssistant, callback
+
+EVENT_LABS_UPDATED = "labs_updated"
+
+
+class EventLabsUpdatedData(TypedDict):
+    """Event data for labs_updated event."""
+
+    domain: str
+    preview_feature: str
+    enabled: bool
+
+
+@callback
+def async_is_preview_feature_enabled(
+    hass: HomeAssistant, domain: str, preview_feature: str
+) -> bool:
+    """Check if a lab preview feature is enabled.
+
+    Args:
+        hass: HomeAssistant instance
+        domain: Integration domain
+        preview_feature: Preview feature name
+
+    Returns:
+        True if the preview feature is enabled, False otherwise
+    """
+    from homeassistant.components.labs import LABS_DATA  # noqa: PLC0415
+
+    if LABS_DATA not in hass.data:
+        return False
+
+    labs_data = hass.data[LABS_DATA]
+    return (domain, preview_feature) in labs_data.data.preview_feature_status
+
+
+@callback
+def async_listen(
+    hass: HomeAssistant,
+    domain: str,
+    preview_feature: str,
+    listener: Callable[[], None],
+) -> Callable[[], None]:
+    """Listen for changes to a specific preview feature.
+
+    Args:
+        hass: HomeAssistant instance
+        domain: Integration domain
+        preview_feature: Preview feature name
+        listener: Callback to invoke when the preview feature is toggled
+
+    Returns:
+        Callable to unsubscribe from the listener
+    """
+
+    @callback
+    def _async_feature_updated(event: Event[EventLabsUpdatedData]) -> None:
+        """Handle labs feature update event."""
+        if (
+            event.data["domain"] == domain
+            and event.data["preview_feature"] == preview_feature
+        ):
+            listener()
+
+    return hass.bus.async_listen(EVENT_LABS_UPDATED, _async_feature_updated)

tests/components/labs/test_init.py

@@ -7,14 +7,10 @@ from unittest.mock import Mock, patch
 
 import pytest
 
-from homeassistant.components.labs import (
-    EVENT_LABS_UPDATED,
-    async_is_preview_feature_enabled,
-    async_listen,
-)
 from homeassistant.components.labs.const import DOMAIN, LABS_DATA
 from homeassistant.components.labs.models import LabPreviewFeature
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers.labs import async_is_preview_feature_enabled
 from homeassistant.loader import Integration
 from homeassistant.setup import async_setup_component
 
@@ -31,66 +27,6 @@ async def test_async_setup(hass: HomeAssistant) -> None:
     assert "labs/update" in hass.data["websocket_api"]
 
 
-async def test_async_is_preview_feature_enabled_not_setup(hass: HomeAssistant) -> None:
-    """Test checking if preview feature is enabled before setup returns False."""
-    # Don't set up labs integration
-    result = async_is_preview_feature_enabled(hass, "kitchen_sink", "special_repair")
-    assert result is False
-
-
-async def test_async_is_preview_feature_enabled_nonexistent(
-    hass: HomeAssistant,
-) -> None:
-    """Test checking if non-existent preview feature is enabled."""
-    assert await async_setup_component(hass, DOMAIN, {})
-    await hass.async_block_till_done()
-
-    result = async_is_preview_feature_enabled(
-        hass, "kitchen_sink", "nonexistent_feature"
-    )
-    assert result is False
-
-
-async def test_async_is_preview_feature_enabled_when_enabled(
-    hass: HomeAssistant, hass_storage: dict[str, Any]
-) -> None:
-    """Test checking if preview feature is enabled."""
-    # Load kitchen_sink integration so preview feature exists
-    hass.config.components.add("kitchen_sink")
-
-    # Enable a preview feature via storage
-    hass_storage["core.labs"] = {
-        "version": 1,
-        "minor_version": 1,
-        "key": "core.labs",
-        "data": {
-            "preview_feature_status": [
-                {"domain": "kitchen_sink", "preview_feature": "special_repair"}
-            ]
-        },
-    }
-
-    assert await async_setup_component(hass, DOMAIN, {})
-    await hass.async_block_till_done()
-
-    result = async_is_preview_feature_enabled(hass, "kitchen_sink", "special_repair")
-    assert result is True
-
-
-async def test_async_is_preview_feature_enabled_when_disabled(
-    hass: HomeAssistant,
-) -> None:
-    """Test checking if preview feature is disabled (not in storage)."""
-    # Load kitchen_sink integration so preview feature exists
-    hass.config.components.add("kitchen_sink")
-
-    assert await async_setup_component(hass, DOMAIN, {})
-    await hass.async_block_till_done()
-
-    result = async_is_preview_feature_enabled(hass, "kitchen_sink", "special_repair")
-    assert result is False
-
-
 @pytest.mark.parametrize(
     (
         "features_to_store",
@@ -168,41 +104,6 @@ async def test_storage_cleanup_stale_features(
     assert_stored_labs_data(hass_storage, expected_cleaned_store)
 
 
-@pytest.mark.parametrize(
-    ("domain", "preview_feature", "expected"),
-    [
-        ("kitchen_sink", "special_repair", True),
-        ("other", "nonexistent", False),
-        ("kitchen_sink", "nonexistent", False),
-    ],
-)
-async def test_async_is_preview_feature_enabled(
-    hass: HomeAssistant,
-    hass_storage: dict[str, Any],
-    domain: str,
-    preview_feature: str,
-    expected: bool,
-) -> None:
-    """Test async_is_preview_feature_enabled."""
-    # Enable the kitchen_sink.special_repair preview feature via storage
-    hass_storage["core.labs"] = {
-        "version": 1,
-        "minor_version": 1,
-        "key": "core.labs",
-        "data": {
-            "preview_feature_status": [
-                {"domain": "kitchen_sink", "preview_feature": "special_repair"}
-            ]
-        },
-    }
-
-    await async_setup_component(hass, DOMAIN, {})
-    await hass.async_block_till_done()
-
-    result = async_is_preview_feature_enabled(hass, domain, preview_feature)
-    assert result is expected
-
-
 async def test_preview_feature_full_key(hass: HomeAssistant) -> None:
     """Test that preview feature full_key property returns correct format."""
     feature = LabPreviewFeature(
@@ -353,86 +254,3 @@ async def test_preview_feature_to_dict_is_built_in(
     assert feature.is_built_in is expected_default
     result = feature.to_dict(enabled=True)
     assert result["is_built_in"] is expected_default
-
-
-async def test_async_listen_helper(hass: HomeAssistant) -> None:
-    """Test the async_listen helper function for preview feature events."""
-    # Load kitchen_sink integration
-    hass.config.components.add("kitchen_sink")
-
-    assert await async_setup_component(hass, DOMAIN, {})
-    await hass.async_block_till_done()
-
-    # Track listener calls
-    listener_calls = []
-
-    def test_listener() -> None:
-        """Test listener callback."""
-        listener_calls.append("called")
-
-    # Subscribe to a specific preview feature
-    unsub = async_listen(
-        hass,
-        domain="kitchen_sink",
-        preview_feature="special_repair",
-        listener=test_listener,
-    )
-
-    # Fire event for the subscribed feature
-    hass.bus.async_fire(
-        EVENT_LABS_UPDATED,
-        {
-            "domain": "kitchen_sink",
-            "preview_feature": "special_repair",
-            "enabled": True,
-        },
-    )
-    await hass.async_block_till_done()
-
-    # Verify listener was called
-    assert len(listener_calls) == 1
-
-    # Fire event for a different feature - should not trigger listener
-    hass.bus.async_fire(
-        EVENT_LABS_UPDATED,
-        {
-            "domain": "kitchen_sink",
-            "preview_feature": "other_feature",
-            "enabled": True,
-        },
-    )
-    await hass.async_block_till_done()
-
-    # Verify listener was not called again
-    assert len(listener_calls) == 1
-
-    # Fire event for a different domain - should not trigger listener
-    hass.bus.async_fire(
-        EVENT_LABS_UPDATED,
-        {
-            "domain": "other_domain",
-            "preview_feature": "special_repair",
-            "enabled": True,
-        },
-    )
-    await hass.async_block_till_done()
-
-    # Verify listener was not called again
-    assert len(listener_calls) == 1
-
-    # Test unsubscribe
-    unsub()
-
-    # Fire event again - should not trigger listener after unsubscribe
-    hass.bus.async_fire(
-        EVENT_LABS_UPDATED,
-        {
-            "domain": "kitchen_sink",
-            "preview_feature": "special_repair",
-            "enabled": True,
-        },
-    )
-    await hass.async_block_till_done()
-
-    # Verify listener was not called after unsubscribe
-    assert len(listener_calls) == 1

tests/components/labs/test_websocket_api.py

@@ -7,12 +7,12 @@ from unittest.mock import ANY, AsyncMock, patch
 
 import pytest
 
-from homeassistant.components.labs import (
+from homeassistant.components.labs import async_setup
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.labs import (
     EVENT_LABS_UPDATED,
     async_is_preview_feature_enabled,
-    async_setup,
 )
-from homeassistant.core import HomeAssistant
 
 from . import assert_stored_labs_data
 

tests/helpers/test_labs.py

@@ -0,0 +1,194 @@
+"""Tests for the Home Assistant Labs helper."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from homeassistant.components.labs import DOMAIN as LABS_DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.labs import (
+    EVENT_LABS_UPDATED,
+    async_is_preview_feature_enabled,
+    async_listen,
+)
+from homeassistant.setup import async_setup_component
+
+
+async def test_async_is_preview_feature_enabled_not_setup(hass: HomeAssistant) -> None:
+    """Test checking if preview feature is enabled before setup returns False."""
+    # Don't set up labs integration
+    result = async_is_preview_feature_enabled(hass, "kitchen_sink", "special_repair")
+    assert result is False
+
+
+async def test_async_is_preview_feature_enabled_nonexistent(
+    hass: HomeAssistant,
+) -> None:
+    """Test checking if non-existent preview feature is enabled."""
+    assert await async_setup_component(hass, LABS_DOMAIN, {})
+    await hass.async_block_till_done()
+
+    result = async_is_preview_feature_enabled(
+        hass, "kitchen_sink", "nonexistent_feature"
+    )
+    assert result is False
+
+
+async def test_async_is_preview_feature_enabled_when_enabled(
+    hass: HomeAssistant, hass_storage: dict[str, Any]
+) -> None:
+    """Test checking if preview feature is enabled."""
+    # Load kitchen_sink integration so preview feature exists
+    hass.config.components.add("kitchen_sink")
+
+    # Enable a preview feature via storage
+    hass_storage["core.labs"] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": "core.labs",
+        "data": {
+            "preview_feature_status": [
+                {"domain": "kitchen_sink", "preview_feature": "special_repair"}
+            ]
+        },
+    }
+
+    assert await async_setup_component(hass, LABS_DOMAIN, {})
+    await hass.async_block_till_done()
+
+    result = async_is_preview_feature_enabled(hass, "kitchen_sink", "special_repair")
+    assert result is True
+
+
+async def test_async_is_preview_feature_enabled_when_disabled(
+    hass: HomeAssistant,
+) -> None:
+    """Test checking if preview feature is disabled (not in storage)."""
+    # Load kitchen_sink integration so preview feature exists
+    hass.config.components.add("kitchen_sink")
+
+    assert await async_setup_component(hass, LABS_DOMAIN, {})
+    await hass.async_block_till_done()
+
+    result = async_is_preview_feature_enabled(hass, "kitchen_sink", "special_repair")
+    assert result is False
+
+
+@pytest.mark.parametrize(
+    ("domain", "preview_feature", "expected"),
+    [
+        ("kitchen_sink", "special_repair", True),
+        ("other", "nonexistent", False),
+        ("kitchen_sink", "nonexistent", False),
+    ],
+)
+async def test_async_is_preview_feature_enabled(
+    hass: HomeAssistant,
+    hass_storage: dict[str, Any],
+    domain: str,
+    preview_feature: str,
+    expected: bool,
+) -> None:
+    """Test async_is_preview_feature_enabled."""
+    # Enable the kitchen_sink.special_repair preview feature via storage
+    hass_storage["core.labs"] = {
+        "version": 1,
+        "minor_version": 1,
+        "key": "core.labs",
+        "data": {
+            "preview_feature_status": [
+                {"domain": "kitchen_sink", "preview_feature": "special_repair"}
+            ]
+        },
+    }
+
+    await async_setup_component(hass, LABS_DOMAIN, {})
+    await hass.async_block_till_done()
+
+    result = async_is_preview_feature_enabled(hass, domain, preview_feature)
+    assert result is expected
+
+
+async def test_async_listen_helper(hass: HomeAssistant) -> None:
+    """Test the async_listen helper function for preview feature events."""
+    # Load kitchen_sink integration
+    hass.config.components.add("kitchen_sink")
+
+    assert await async_setup_component(hass, LABS_DOMAIN, {})
+    await hass.async_block_till_done()
+
+    # Track listener calls
+    listener_calls = []
+
+    def test_listener() -> None:
+        """Test listener callback."""
+        listener_calls.append("called")
+
+    # Subscribe to a specific preview feature
+    unsub = async_listen(
+        hass,
+        domain="kitchen_sink",
+        preview_feature="special_repair",
+        listener=test_listener,
+    )
+
+    # Fire event for the subscribed feature
+    hass.bus.async_fire(
+        EVENT_LABS_UPDATED,
+        {
+            "domain": "kitchen_sink",
+            "preview_feature": "special_repair",
+            "enabled": True,
+        },
+    )
+    await hass.async_block_till_done()
+
+    # Verify listener was called
+    assert len(listener_calls) == 1
+
+    # Fire event for a different feature - should not trigger listener
+    hass.bus.async_fire(
+        EVENT_LABS_UPDATED,
+        {
+            "domain": "kitchen_sink",
+            "preview_feature": "other_feature",
+            "enabled": True,
+        },
+    )
+    await hass.async_block_till_done()
+
+    # Verify listener was not called again
+    assert len(listener_calls) == 1
+
+    # Fire event for a different domain - should not trigger listener
+    hass.bus.async_fire(
+        EVENT_LABS_UPDATED,
+        {
+            "domain": "other_domain",
+            "preview_feature": "special_repair",
+            "enabled": True,
+        },
+    )
+    await hass.async_block_till_done()
+
+    # Verify listener was not called again
+    assert len(listener_calls) == 1
+
+    # Test unsubscribe
+    unsub()
+
+    # Fire event again - should not trigger listener after unsubscribe
+    hass.bus.async_fire(
+        EVENT_LABS_UPDATED,
+        {
+            "domain": "kitchen_sink",
+            "preview_feature": "special_repair",
+            "enabled": True,
+        },
+    )
+    await hass.async_block_till_done()
+
+    # Verify listener was not called after unsubscribe
+    assert len(listener_calls) == 1