Add Watergate Sonic Local Integration (#129686)

Co-authored-by: Mark Breen <markvader@users.noreply.github.com>

CODEOWNERS

@@ -1644,6 +1644,8 @@ build.json @home-assistant/supervisor
 /tests/components/waqi/ @joostlek
 /homeassistant/components/water_heater/ @home-assistant/core
 /tests/components/water_heater/ @home-assistant/core
+/homeassistant/components/watergate/ @adam-the-hero
+/tests/components/watergate/ @adam-the-hero
 /homeassistant/components/watson_tts/ @rutkai
 /homeassistant/components/watttime/ @bachya
 /tests/components/watttime/ @bachya

homeassistant/components/watergate/__init__.py

@@ -0,0 +1,107 @@
+"""The Watergate integration."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from http import HTTPStatus
+import logging
+
+from watergate_local_api import WatergateLocalApiClient
+from watergate_local_api.models import WebhookEvent
+
+from homeassistant.components.http import HomeAssistantView
+from homeassistant.components.webhook import (
+    Request,
+    Response,
+    async_generate_url,
+    async_register,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_IP_ADDRESS, CONF_WEBHOOK_ID, Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import WatergateDataCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+
+PLATFORMS: list[Platform] = [
+    Platform.VALVE,
+]
+
+type WatergateConfigEntry = ConfigEntry[WatergateDataCoordinator]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: WatergateConfigEntry) -> bool:
+    """Set up Watergate from a config entry."""
+    sonic_address = entry.data[CONF_IP_ADDRESS]
+    webhook_id = entry.data[CONF_WEBHOOK_ID]
+
+    _LOGGER.debug(
+        "Setting up watergate local api integration for device: IP: %s)",
+        sonic_address,
+    )
+
+    watergate_client = WatergateLocalApiClient(
+        sonic_address if sonic_address.startswith("http") else f"http://{sonic_address}"
+    )
+
+    coordinator = WatergateDataCoordinator(hass, watergate_client)
+    entry.runtime_data = coordinator
+
+    async_register(
+        hass, DOMAIN, "Watergate", webhook_id, get_webhook_handler(coordinator)
+    )
+
+    _LOGGER.debug("Registered webhook: %s", webhook_id)
+
+    await coordinator.async_config_entry_first_refresh()
+
+    await watergate_client.async_set_webhook_url(
+        async_generate_url(hass, webhook_id, allow_ip=True, prefer_external=False)
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: WatergateConfigEntry) -> bool:
+    """Unload a config entry."""
+    webhook_id = entry.data[CONF_WEBHOOK_ID]
+    hass.components.webhook.async_unregister(webhook_id)
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+
+def get_webhook_handler(
+    coordinator: WatergateDataCoordinator,
+) -> Callable[[HomeAssistant, str, Request], Awaitable[Response | None]]:
+    """Return webhook handler."""
+
+    async def async_webhook_handler(
+        hass: HomeAssistant, webhook_id: str, request: Request
+    ) -> Response | None:
+        # Handle http post calls to the path.
+        if not request.body_exists:
+            return HomeAssistantView.json(
+                result="No Body", status_code=HTTPStatus.BAD_REQUEST
+            )
+
+        body = await request.json()
+
+        _LOGGER.debug("Received webhook: %s", body)
+
+        data = WebhookEvent.parse_webhook_event(body)
+
+        body_type = body.get("type")
+
+        coordinator_data = coordinator.data
+        if body_type == Platform.VALVE and coordinator_data:
+            coordinator_data.valve_state = data.state
+
+        coordinator.async_set_updated_data(coordinator_data)
+
+        return HomeAssistantView.json(result="OK", status_code=HTTPStatus.OK)
+
+    return async_webhook_handler

homeassistant/components/watergate/config_flow.py

@@ -0,0 +1,62 @@
+"""Config flow for Watergate."""
+
+import logging
+
+import voluptuous as vol
+from watergate_local_api.watergate_api import (
+    WatergateApiException,
+    WatergateLocalApiClient,
+)
+
+from homeassistant.components.webhook import async_generate_id as webhook_generate_id
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_IP_ADDRESS, CONF_WEBHOOK_ID
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+SONIC = "Sonic"
+WATERGATE_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_IP_ADDRESS): str,
+    }
+)
+
+
+class WatergateConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Watergate config flow."""
+
+    async def async_step_user(
+        self, user_input: dict[str, str] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initiated by the user."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            watergate_client = WatergateLocalApiClient(
+                self.prepare_ip_address(user_input[CONF_IP_ADDRESS])
+            )
+            try:
+                state = await watergate_client.async_get_device_state()
+            except WatergateApiException as exception:
+                _LOGGER.error("Error connecting to Watergate device: %s", exception)
+                errors[CONF_IP_ADDRESS] = "cannot_connect"
+            else:
+                if state is None:
+                    _LOGGER.error("Device state returned as None")
+                    errors[CONF_IP_ADDRESS] = "cannot_connect"
+                else:
+                    await self.async_set_unique_id(state.serial_number)
+                    self._abort_if_unique_id_configured()
+                    return self.async_create_entry(
+                        data={**user_input, CONF_WEBHOOK_ID: webhook_generate_id()},
+                        title=SONIC,
+                    )
+
+        return self.async_show_form(
+            step_id="user", data_schema=WATERGATE_SCHEMA, errors=errors
+        )
+
+    def prepare_ip_address(self, ip_address: str) -> str:
+        """Prepare the IP address for the Watergate device."""
+        return ip_address if ip_address.startswith("http") else f"http://{ip_address}"

homeassistant/components/watergate/const.py

@@ -0,0 +1,5 @@
+"""Constants for the Watergate integration."""
+
+DOMAIN = "watergate"
+
+MANUFACTURER = "Watergate"

homeassistant/components/watergate/coordinator.py

@@ -0,0 +1,35 @@
+"""Coordinator for Watergate API."""
+
+from datetime import timedelta
+import logging
+
+from watergate_local_api import WatergateApiException, WatergateLocalApiClient
+from watergate_local_api.models import DeviceState
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class WatergateDataCoordinator(DataUpdateCoordinator[DeviceState]):
+    """Class to manage fetching watergate data."""
+
+    def __init__(self, hass: HomeAssistant, api: WatergateLocalApiClient) -> None:
+        """Initialize."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(minutes=2),
+        )
+        self.api = api
+
+    async def _async_update_data(self) -> DeviceState:
+        try:
+            state = await self.api.async_get_device_state()
+        except WatergateApiException as exc:
+            raise UpdateFailed from exc
+        return state

homeassistant/components/watergate/entity.py

@@ -0,0 +1,30 @@
+"""Watergate Base Entity Definition."""
+
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN, MANUFACTURER
+from .coordinator import WatergateDataCoordinator
+
+
+class WatergateEntity(CoordinatorEntity[WatergateDataCoordinator]):
+    """Define a base Watergate entity."""
+
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: WatergateDataCoordinator,
+        entity_name: str,
+    ) -> None:
+        """Initialize the entity."""
+        super().__init__(coordinator)
+        self._api_client = coordinator.api
+        self._attr_unique_id = f"{coordinator.data.serial_number}.{entity_name}"
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, coordinator.data.serial_number)},
+            name="Sonic",
+            serial_number=coordinator.data.serial_number,
+            manufacturer=MANUFACTURER,
+            sw_version=coordinator.data.firmware_version if coordinator.data else None,
+        )

homeassistant/components/watergate/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "watergate",
+  "name": "Watergate",
+  "codeowners": ["@adam-the-hero"],
+  "config_flow": true,
+  "dependencies": ["http", "webhook"],
+  "documentation": "https://www.home-assistant.io/integrations/watergate",
+  "iot_class": "local_push",
+  "quality_scale": "bronze",
+  "requirements": ["watergate-local-api==2024.4.1"]
+}

homeassistant/components/watergate/quality_scale.yaml

@@ -0,0 +1,43 @@
+rules:
+  # Bronze
+  action-setup:
+    status: exempt
+    comment: |
+      This integration does not provide additional 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: |
+      This integration does not provide additional actions.
+  docs-high-level-description: done
+  docs-installation-instructions: done
+  docs-removal-instructions: done
+  entity-event-setup:
+    status: exempt
+    comment: |
+      Entities of this integration does not explicitly subscribe to 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
+  config-entry-unloading: done
+  log-when-unavailable: todo
+  entity-unavailable: done
+  action-exceptions: done
+  reauthentication-flow:
+    status: exempt
+    comment: |
+      This integration does not require authentication.
+  parallel-updates: done
+  test-coverage: done
+  integration-owner: done
+  docs-installation-parameters: todo
+  docs-configuration-parameters: todo

homeassistant/components/watergate/strings.json

@@ -0,0 +1,21 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "ip_address": "[%key:common::config_flow::data::ip%]"
+        },
+        "title": "Configure Watergate device",
+        "data_description": {
+          "ip_address": "Provide an IP address of your Watergate device."
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}

homeassistant/components/watergate/valve.py

@@ -0,0 +1,82 @@
+"""Support for Watergate Valve."""
+
+from homeassistant.components.sensor import Any, HomeAssistant
+from homeassistant.components.valve import (
+    ValveDeviceClass,
+    ValveEntity,
+    ValveEntityFeature,
+    ValveState,
+)
+from homeassistant.core import callback
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import WatergateConfigEntry
+from .coordinator import WatergateDataCoordinator
+from .entity import WatergateEntity
+
+ENTITY_NAME = "valve"
+PARALLEL_UPDATES = 0
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: WatergateConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up all entries for Watergate Platform."""
+
+    async_add_entities([SonicValve(config_entry.runtime_data)])
+
+
+class SonicValve(WatergateEntity, ValveEntity):
+    """Define a Sonic Valve entity."""
+
+    _attr_supported_features = ValveEntityFeature.OPEN | ValveEntityFeature.CLOSE
+    _attr_reports_position = False
+    _valve_state: str | None = None
+    _attr_device_class = ValveDeviceClass.WATER
+    _attr_name = None
+
+    def __init__(
+        self,
+        coordinator: WatergateDataCoordinator,
+    ) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator, ENTITY_NAME)
+        self._valve_state = coordinator.data.valve_state if coordinator.data else None
+
+    @property
+    def is_closed(self) -> bool:
+        """Return if the valve is closed or not."""
+        return self._valve_state == ValveState.CLOSED
+
+    @property
+    def is_opening(self) -> bool | None:
+        """Return if the valve is opening or not."""
+        return self._valve_state == ValveState.OPENING
+
+    @property
+    def is_closing(self) -> bool | None:
+        """Return if the valve is closing or not."""
+        return self._valve_state == ValveState.CLOSING
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle data update."""
+        self._attr_available = self.coordinator.data is not None
+        self._valve_state = (
+            self.coordinator.data.valve_state if self.coordinator.data else None
+        )
+        self.async_write_ha_state()
+
+    async def async_open_valve(self, **kwargs: Any) -> None:
+        """Open the valve."""
+        await self._api_client.async_set_valve_state(ValveState.OPEN)
+        self._valve_state = ValveState.OPENING
+        self.async_write_ha_state()
+
+    async def async_close_valve(self, **kwargs: Any) -> None:
+        """Close the valve."""
+        await self._api_client.async_set_valve_state(ValveState.CLOSED)
+        self._valve_state = ValveState.CLOSING
+        self.async_write_ha_state()

homeassistant/generated/config_flows.py

@@ -665,6 +665,7 @@ FLOWS = {
         "wake_on_lan",
         "wallbox",
         "waqi",
+        "watergate",
         "watttime",
         "waze_travel_time",
         "weatherflow",

homeassistant/generated/integrations.json

@@ -6892,6 +6892,12 @@
       "config_flow": false,
       "iot_class": "cloud_polling"
     },
+    "watergate": {
+      "name": "Watergate",
+      "integration_type": "hub",
+      "config_flow": true,
+      "iot_class": "local_push"
+    },
     "watttime": {
       "name": "WattTime",
       "integration_type": "service",

requirements_all.txt

@@ -2977,6 +2977,9 @@ watchdog==2.3.1
 # homeassistant.components.waterfurnace
 waterfurnace==1.1.0
 
+# homeassistant.components.watergate
+watergate-local-api==2024.4.1
+
 # homeassistant.components.weatherflow_cloud
 weatherflow4py==1.0.6
 

requirements_test_all.txt

@@ -2378,6 +2378,9 @@ wallbox==0.7.0
 # homeassistant.components.folder_watcher
 watchdog==2.3.1
 
+# homeassistant.components.watergate
+watergate-local-api==2024.4.1
+
 # homeassistant.components.weatherflow_cloud
 weatherflow4py==1.0.6
 

tests/components/watergate/__init__.py

@@ -0,0 +1,11 @@
+"""Tests for the Watergate integration."""
+
+from homeassistant.core import HomeAssistant
+
+
+async def init_integration(hass: HomeAssistant, mock_entry) -> None:
+    """Set up the Watergate integration in Home Assistant."""
+    mock_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()

tests/components/watergate/conftest.py

@@ -0,0 +1,77 @@
+"""Fixtures for watergate platform tests."""
+
+from collections.abc import Generator
+
+import pytest
+
+from homeassistant.components.watergate.const import DOMAIN
+from homeassistant.const import CONF_IP_ADDRESS
+
+from .const import (
+    DEFAULT_DEVICE_STATE,
+    DEFAULT_SERIAL_NUMBER,
+    MOCK_CONFIG,
+    MOCK_WEBHOOK_ID,
+)
+
+from tests.common import AsyncMock, MockConfigEntry, patch
+
+
+@pytest.fixture
+def mock_watergate_client() -> Generator[AsyncMock]:
+    """Fixture to mock WatergateLocalApiClient."""
+    with (
+        patch(
+            "homeassistant.components.watergate.WatergateLocalApiClient",
+            autospec=True,
+        ) as mock_client_main,
+        patch(
+            "homeassistant.components.watergate.config_flow.WatergateLocalApiClient",
+            new=mock_client_main,
+        ),
+    ):
+        mock_client_instance = mock_client_main.return_value
+
+        mock_client_instance.async_get_device_state = AsyncMock(
+            return_value=DEFAULT_DEVICE_STATE
+        )
+        yield mock_client_instance
+
+
+@pytest.fixture
+def mock_setup_entry() -> Generator[AsyncMock]:
+    """Override async_setup_entry."""
+    with patch(
+        "homeassistant.components.watergate.async_setup_entry", return_value=True
+    ) as mock_setup_entry:
+        yield mock_setup_entry
+
+
+@pytest.fixture
+def mock_webhook_id_generation() -> Generator[None]:
+    """Fixture to mock webhook_id generation."""
+    with patch(
+        "homeassistant.components.watergate.config_flow.webhook_generate_id",
+        return_value=MOCK_WEBHOOK_ID,
+    ):
+        yield
+
+
+@pytest.fixture
+def mock_entry() -> MockConfigEntry:
+    """Create full mocked entry to be used in config_flow tests."""
+    return MockConfigEntry(
+        domain=DOMAIN,
+        title="Sonic",
+        data=MOCK_CONFIG,
+        entry_id="12345",
+        unique_id=DEFAULT_SERIAL_NUMBER,
+    )
+
+
+@pytest.fixture
+def user_input() -> dict[str, str]:
+    """Create user input for config_flow tests."""
+    return {
+        CONF_IP_ADDRESS: "192.168.1.100",
+    }

tests/components/watergate/const.py

@@ -0,0 +1,27 @@
+"""Constants for the Watergate tests."""
+
+from watergate_local_api.models import DeviceState
+
+from homeassistant.const import CONF_IP_ADDRESS, CONF_NAME, CONF_WEBHOOK_ID
+
+MOCK_WEBHOOK_ID = "webhook_id"
+
+MOCK_CONFIG = {
+    CONF_NAME: "Sonic",
+    CONF_IP_ADDRESS: "http://localhost",
+    CONF_WEBHOOK_ID: MOCK_WEBHOOK_ID,
+}
+
+DEFAULT_SERIAL_NUMBER = "a63182948ce2896a"
+
+DEFAULT_DEVICE_STATE = DeviceState(
+    "open",
+    "on",
+    True,
+    True,
+    "battery",
+    "1.0.0",
+    100,
+    {"volume": 1.2, "duration": 100},
+    DEFAULT_SERIAL_NUMBER,
+)

tests/components/watergate/snapshots/test_valve.ambr

@@ -0,0 +1,16 @@
+# serializer version: 1
+# name: test_change_valve_state_snapshot
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'water',
+      'friendly_name': 'Sonic',
+      'supported_features': <ValveEntityFeature: 3>,
+    }),
+    'context': <ANY>,
+    'entity_id': 'valve.sonic',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'open',
+  })
+# ---

tests/components/watergate/test_config_flow.py

@@ -0,0 +1,107 @@
+"""Tests for the Watergate config flow."""
+
+from collections.abc import Generator
+
+import pytest
+from watergate_local_api import WatergateApiException
+
+from homeassistant.components.watergate.const import DOMAIN
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.const import CONF_IP_ADDRESS, CONF_WEBHOOK_ID
+from homeassistant.data_entry_flow import FlowResultType
+
+from .const import DEFAULT_DEVICE_STATE, DEFAULT_SERIAL_NUMBER, MOCK_WEBHOOK_ID
+
+from tests.common import AsyncMock, HomeAssistant, MockConfigEntry
+
+
+async def test_step_user_form(
+    hass: HomeAssistant,
+    mock_watergate_client: Generator[AsyncMock],
+    mock_webhook_id_generation: Generator[None],
+    user_input: dict[str, str],
+) -> None:
+    """Test checking if registration form works end to end."""
+
+    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 CONF_IP_ADDRESS in result["data_schema"].schema
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == "Sonic"
+    assert result["data"] == {**user_input, CONF_WEBHOOK_ID: MOCK_WEBHOOK_ID}
+    assert result["result"].unique_id == DEFAULT_SERIAL_NUMBER
+
+
+@pytest.mark.parametrize(
+    "client_result",
+    [AsyncMock(return_value=None), AsyncMock(side_effect=WatergateApiException)],
+)
+async def test_step_user_form_with_exception(
+    hass: HomeAssistant,
+    mock_watergate_client: Generator[AsyncMock],
+    user_input: dict[str, str],
+    client_result: AsyncMock,
+    mock_webhook_id_generation: Generator[None],
+) -> None:
+    """Test checking if errors will be displayed when Exception is thrown while checking device state."""
+    mock_watergate_client.async_get_device_state = client_result
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["errors"][CONF_IP_ADDRESS] == "cannot_connect"
+
+    mock_watergate_client.async_get_device_state = AsyncMock(
+        return_value=DEFAULT_DEVICE_STATE
+    )
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == "Sonic"
+    assert result["data"] == {**user_input, CONF_WEBHOOK_ID: MOCK_WEBHOOK_ID}
+
+
+async def test_abort_if_id_is_not_unique(
+    hass: HomeAssistant,
+    mock_watergate_client: Generator[AsyncMock],
+    mock_entry: MockConfigEntry,
+    user_input: dict[str, str],
+) -> None:
+    """Test checking if we will inform user that this entity is already registered."""
+    mock_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 CONF_IP_ADDRESS in result["data_schema"].schema
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], user_input
+    )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"

tests/components/watergate/test_init.py

@@ -0,0 +1,81 @@
+"""Tests for the Watergate integration init module."""
+
+from collections.abc import Generator
+from unittest.mock import patch
+
+from homeassistant.components.valve import ValveState
+from homeassistant.components.watergate.const import DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from . import init_integration
+from .const import MOCK_WEBHOOK_ID
+
+from tests.common import ANY, AsyncMock, MockConfigEntry
+from tests.typing import ClientSessionGenerator
+
+
+async def test_async_setup_entry(
+    hass: HomeAssistant,
+    mock_entry: MockConfigEntry,
+    mock_watergate_client: Generator[AsyncMock],
+) -> None:
+    """Test setting up the Watergate integration."""
+    hass.config.internal_url = "http://hassio.local"
+
+    with (
+        patch("homeassistant.components.watergate.async_register") as mock_webhook,
+    ):
+        await init_integration(hass, mock_entry)
+
+        assert mock_entry.state is ConfigEntryState.LOADED
+
+        mock_webhook.assert_called_once_with(
+            hass,
+            DOMAIN,
+            "Watergate",
+            MOCK_WEBHOOK_ID,
+            ANY,
+        )
+        mock_watergate_client.async_set_webhook_url.assert_called_once_with(
+            f"http://hassio.local/api/webhook/{MOCK_WEBHOOK_ID}"
+        )
+        mock_watergate_client.async_get_device_state.assert_called_once()
+
+
+async def test_handle_webhook(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    mock_entry: MockConfigEntry,
+    mock_watergate_client: Generator[AsyncMock],
+) -> None:
+    """Test handling webhook events."""
+    await init_integration(hass, mock_entry)
+
+    entity_id = "valve.sonic"
+
+    registered_entity = hass.states.get(entity_id)
+    assert registered_entity
+    assert registered_entity.state == ValveState.OPEN
+
+    valve_change_data = {
+        "type": "valve",
+        "data": {"state": "closed"},
+    }
+    client = await hass_client_no_auth()
+    await client.post(f"/api/webhook/{MOCK_WEBHOOK_ID}", json=valve_change_data)
+
+    await hass.async_block_till_done()  # Ensure the webhook is processed
+
+    assert hass.states.get(entity_id).state == ValveState.CLOSED
+
+    valve_change_data = {
+        "type": "valve",
+        "data": {"state": "open"},
+    }
+
+    await client.post(f"/api/webhook/{MOCK_WEBHOOK_ID}", json=valve_change_data)
+
+    await hass.async_block_till_done()  # Ensure the webhook is processed
+
+    assert hass.states.get(entity_id).state == ValveState.OPEN

tests/components/watergate/test_valve.py

@@ -0,0 +1,72 @@
+"""Tests for the Watergate valve platform."""
+
+from collections.abc import Generator
+
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.valve import DOMAIN as VALVE_DOMAIN, ValveState
+from homeassistant.const import ATTR_ENTITY_ID, SERVICE_CLOSE_VALVE, SERVICE_OPEN_VALVE
+from homeassistant.core import HomeAssistant
+
+from . import init_integration
+
+from tests.common import AsyncMock, MockConfigEntry
+
+
+async def test_change_valve_state_snapshot(
+    hass: HomeAssistant,
+    snapshot: SnapshotAssertion,
+    mock_watergate_client: Generator[AsyncMock],
+    mock_entry: MockConfigEntry,
+) -> None:
+    """Test entities become unavailable after failed update."""
+    await init_integration(hass, mock_entry)
+
+    entity_id = "valve.sonic"
+
+    registered_entity = hass.states.get(entity_id)
+    assert registered_entity
+    assert registered_entity.state == ValveState.OPEN
+    assert registered_entity == snapshot
+
+
+async def test_change_valve_state(
+    hass: HomeAssistant,
+    mock_watergate_client: Generator[AsyncMock],
+    mock_entry: MockConfigEntry,
+) -> None:
+    """Test entities become unavailable after failed update."""
+    await init_integration(hass, mock_entry)
+
+    entity_id = "valve.sonic"
+
+    registered_entity = hass.states.get(entity_id)
+    assert registered_entity
+    assert registered_entity.state == ValveState.OPEN
+
+    await hass.services.async_call(
+        VALVE_DOMAIN,
+        SERVICE_CLOSE_VALVE,
+        {ATTR_ENTITY_ID: entity_id},
+        blocking=True,
+    )
+
+    registered_entity = hass.states.get(entity_id)
+    assert registered_entity
+    assert registered_entity.state == ValveState.CLOSING
+
+    mock_watergate_client.async_set_valve_state.assert_called_once_with("closed")
+    mock_watergate_client.async_set_valve_state.reset_mock()
+
+    await hass.services.async_call(
+        VALVE_DOMAIN,
+        SERVICE_OPEN_VALVE,
+        {ATTR_ENTITY_ID: entity_id},
+        blocking=True,
+    )
+
+    registered_entity = hass.states.get(entity_id)
+    assert registered_entity
+    assert registered_entity.state == ValveState.OPENING
+
+    mock_watergate_client.async_set_valve_state.assert_called_once_with("open")