diff --git a/homeassistant/components/group/__init__.py b/homeassistant/components/group/__init__.py
index f64979c6a66..756e75ca22b 100644
--- a/homeassistant/components/group/__init__.py
+++ b/homeassistant/components/group/__init__.py
@@ -72,6 +72,7 @@ PLATFORMS = [
     Platform.NOTIFY,
     Platform.SENSOR,
     Platform.SWITCH,
+    Platform.VALVE,
 ]
 
 _LOGGER = logging.getLogger(__name__)
diff --git a/homeassistant/components/group/config_flow.py b/homeassistant/components/group/config_flow.py
index 0433deab8ae..3624ebb7ef2 100644
--- a/homeassistant/components/group/config_flow.py
+++ b/homeassistant/components/group/config_flow.py
@@ -35,6 +35,7 @@ from .media_player import MediaPlayerGroup, async_create_preview_media_player
 from .notify import async_create_preview_notify
 from .sensor import async_create_preview_sensor
 from .switch import async_create_preview_switch
+from .valve import async_create_preview_valve
 
 _STATISTIC_MEASURES = [
     "last",
@@ -172,6 +173,7 @@ GROUP_TYPES = [
     "notify",
     "sensor",
     "switch",
+    "valve",
 ]
 
 
@@ -253,6 +255,11 @@ CONFIG_FLOW = {
         preview="group",
         validate_user_input=set_group_type("switch"),
     ),
+    "valve": SchemaFlowFormStep(
+        basic_group_config_schema("valve"),
+        preview="group",
+        validate_user_input=set_group_type("valve"),
+    ),
 }
 
 
@@ -302,6 +309,10 @@ OPTIONS_FLOW = {
         partial(light_switch_options_schema, "switch"),
         preview="group",
     ),
+    "valve": SchemaFlowFormStep(
+        partial(basic_group_options_schema, "valve"),
+        preview="group",
+    ),
 }
 
 PREVIEW_OPTIONS_SCHEMA: dict[str, vol.Schema] = {}
@@ -321,6 +332,7 @@ CREATE_PREVIEW_ENTITY: dict[
     "notify": async_create_preview_notify,
     "sensor": async_create_preview_sensor,
     "switch": async_create_preview_switch,
+    "valve": async_create_preview_valve,
 }
 
 
diff --git a/homeassistant/components/group/strings.json b/homeassistant/components/group/strings.json
index 8a9f4377a62..5f4fdb563a1 100644
--- a/homeassistant/components/group/strings.json
+++ b/homeassistant/components/group/strings.json
@@ -16,7 +16,8 @@
           "media_player": "Media player group",
           "notify": "Notify group",
           "sensor": "Sensor group",
-          "switch": "Switch group"
+          "switch": "Switch group",
+          "valve": "Valve group"
         }
       },
       "binary_sensor": {
@@ -127,6 +128,18 @@
         "data_description": {
           "all": "[%key:component::group::config::step::binary_sensor::data_description::all%]"
         }
+      },
+      "valve": {
+        "title": "[%key:component::group::config::step::user::title%]",
+        "data": {
+          "all": "[%key:component::group::config::step::binary_sensor::data::all%]",
+          "entities": "[%key:component::group::config::step::binary_sensor::data::entities%]",
+          "hide_members": "[%key:component::group::config::step::binary_sensor::data::hide_members%]",
+          "name": "[%key:common::config_flow::data::name%]"
+        },
+        "data_description": {
+          "all": "[%key:component::group::config::step::binary_sensor::data_description::all%]"
+        }
       }
     }
   },
@@ -212,6 +225,16 @@
         "data_description": {
           "all": "[%key:component::group::config::step::binary_sensor::data_description::all%]"
         }
+      },
+      "valve": {
+        "data": {
+          "all": "[%key:component::group::config::step::binary_sensor::data::all%]",
+          "entities": "[%key:component::group::config::step::binary_sensor::data::entities%]",
+          "hide_members": "[%key:component::group::config::step::binary_sensor::data::hide_members%]"
+        },
+        "data_description": {
+          "all": "[%key:component::group::config::step::binary_sensor::data_description::all%]"
+        }
       }
     }
   },
diff --git a/homeassistant/components/group/valve.py b/homeassistant/components/group/valve.py
new file mode 100644
index 00000000000..29fe72cb576
--- /dev/null
+++ b/homeassistant/components/group/valve.py
@@ -0,0 +1,262 @@
+"""Platform allowing several valves to be grouped into one valve."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant.components.valve import (
+    ATTR_CURRENT_POSITION,
+    ATTR_POSITION,
+    DOMAIN as VALVE_DOMAIN,
+    PLATFORM_SCHEMA as VALVE_PLATFORM_SCHEMA,
+    ValveEntity,
+    ValveEntityFeature,
+    ValveState,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    ATTR_SUPPORTED_FEATURES,
+    CONF_ENTITIES,
+    CONF_NAME,
+    CONF_UNIQUE_ID,
+    SERVICE_CLOSE_VALVE,
+    SERVICE_OPEN_VALVE,
+    SERVICE_SET_VALVE_POSITION,
+    SERVICE_STOP_VALVE,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import HomeAssistant, State, callback
+from homeassistant.helpers import config_validation as cv, entity_registry as er
+from homeassistant.helpers.entity_platform import (
+    AddConfigEntryEntitiesCallback,
+    AddEntitiesCallback,
+)
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+
+from .entity import GroupEntity
+from .util import reduce_attribute
+
+KEY_OPEN_CLOSE = "open_close"
+KEY_STOP = "stop"
+KEY_SET_POSITION = "set_position"
+
+DEFAULT_NAME = "Valve Group"
+
+# No limit on parallel updates to enable a group calling another group
+PARALLEL_UPDATES = 0
+
+PLATFORM_SCHEMA = VALVE_PLATFORM_SCHEMA.extend(
+    {
+        vol.Required(CONF_ENTITIES): cv.entities_domain(VALVE_DOMAIN),
+        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
+        vol.Optional(CONF_UNIQUE_ID): cv.string,
+    }
+)
+
+
+async def async_setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    async_add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> None:
+    """Set up the Valve Group platform."""
+    async_add_entities(
+        [
+            ValveGroup(
+                config.get(CONF_UNIQUE_ID), config[CONF_NAME], config[CONF_ENTITIES]
+            )
+        ]
+    )
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Initialize Valve Group config entry."""
+    registry = er.async_get(hass)
+    entities = er.async_validate_entity_ids(
+        registry, config_entry.options[CONF_ENTITIES]
+    )
+
+    async_add_entities(
+        [ValveGroup(config_entry.entry_id, config_entry.title, entities)]
+    )
+
+
+@callback
+def async_create_preview_valve(
+    hass: HomeAssistant, name: str, validated_config: dict[str, Any]
+) -> ValveGroup:
+    """Create a preview valve."""
+    return ValveGroup(
+        None,
+        name,
+        validated_config[CONF_ENTITIES],
+    )
+
+
+class ValveGroup(GroupEntity, ValveEntity):
+    """Representation of a ValveGroup."""
+
+    _attr_available: bool = False
+    _attr_current_valve_position: int | None = None
+    _attr_is_closed: bool | None = None
+    _attr_is_closing: bool | None = False
+    _attr_is_opening: bool | None = False
+    _attr_reports_position: bool = False
+
+    def __init__(self, unique_id: str | None, name: str, entities: list[str]) -> None:
+        """Initialize a ValveGroup entity."""
+        self._entity_ids = entities
+        self._valves: dict[str, set[str]] = {
+            KEY_OPEN_CLOSE: set(),
+            KEY_STOP: set(),
+            KEY_SET_POSITION: set(),
+        }
+
+        self._attr_name = name
+        self._attr_extra_state_attributes = {ATTR_ENTITY_ID: entities}
+        self._attr_unique_id = unique_id
+
+    @callback
+    def async_update_supported_features(
+        self,
+        entity_id: str,
+        new_state: State | None,
+    ) -> None:
+        """Update dictionaries with supported features."""
+        if not new_state:
+            for values in self._valves.values():
+                values.discard(entity_id)
+            return
+
+        features = new_state.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
+
+        if features & (ValveEntityFeature.OPEN | ValveEntityFeature.CLOSE):
+            self._valves[KEY_OPEN_CLOSE].add(entity_id)
+        else:
+            self._valves[KEY_OPEN_CLOSE].discard(entity_id)
+        if features & (ValveEntityFeature.STOP):
+            self._valves[KEY_STOP].add(entity_id)
+        else:
+            self._valves[KEY_STOP].discard(entity_id)
+        if features & (ValveEntityFeature.SET_POSITION):
+            self._valves[KEY_SET_POSITION].add(entity_id)
+        else:
+            self._valves[KEY_SET_POSITION].discard(entity_id)
+
+    async def async_open_valve(self) -> None:
+        """Open the valves."""
+        data = {ATTR_ENTITY_ID: self._valves[KEY_OPEN_CLOSE]}
+        await self.hass.services.async_call(
+            VALVE_DOMAIN, SERVICE_OPEN_VALVE, data, blocking=True, context=self._context
+        )
+
+    async def async_handle_open_valve(self) -> None:  # type: ignore[misc]
+        """Open the valves.
+
+        Override the base class to avoid calling the set position service
+        for all valves. Transfer the service call to the base class and let
+        it decide if the valve uses set position or open service.
+        """
+        await self.async_open_valve()
+
+    async def async_close_valve(self) -> None:
+        """Close valves."""
+        data = {ATTR_ENTITY_ID: self._valves[KEY_OPEN_CLOSE]}
+        await self.hass.services.async_call(
+            VALVE_DOMAIN,
+            SERVICE_CLOSE_VALVE,
+            data,
+            blocking=True,
+            context=self._context,
+        )
+
+    async def async_handle_close_valve(self) -> None:  # type: ignore[misc]
+        """Close the valves.
+
+        Override the base class to avoid calling the set position service
+        for all valves. Transfer the service call to the base class and let
+        it decide if the valve uses set position or close service.
+        """
+        await self.async_close_valve()
+
+    async def async_set_valve_position(self, position: int) -> None:
+        """Move the valves to a specific position."""
+        data = {
+            ATTR_ENTITY_ID: self._valves[KEY_SET_POSITION],
+            ATTR_POSITION: position,
+        }
+        await self.hass.services.async_call(
+            VALVE_DOMAIN,
+            SERVICE_SET_VALVE_POSITION,
+            data,
+            blocking=True,
+            context=self._context,
+        )
+
+    async def async_stop_valve(self) -> None:
+        """Stop the valves."""
+        data = {ATTR_ENTITY_ID: self._valves[KEY_STOP]}
+        await self.hass.services.async_call(
+            VALVE_DOMAIN, SERVICE_STOP_VALVE, data, blocking=True, context=self._context
+        )
+
+    @callback
+    def async_update_group_state(self) -> None:
+        """Update state and attributes."""
+        states = [
+            state
+            for entity_id in self._entity_ids
+            if (state := self.hass.states.get(entity_id)) is not None
+        ]
+
+        # Set group as unavailable if all members are unavailable or missing
+        self._attr_available = any(state.state != STATE_UNAVAILABLE for state in states)
+
+        self._attr_is_closed = True
+        self._attr_is_closing = False
+        self._attr_is_opening = False
+        self._attr_reports_position = False
+        self._update_assumed_state_from_members()
+        for state in states:
+            if state.attributes.get(ATTR_CURRENT_POSITION) is not None:
+                self._attr_reports_position = True
+            if state.state == ValveState.OPEN:
+                self._attr_is_closed = False
+                continue
+            if state.state == ValveState.CLOSED:
+                continue
+            if state.state == ValveState.CLOSING:
+                self._attr_is_closing = True
+                continue
+            if state.state == ValveState.OPENING:
+                self._attr_is_opening = True
+                continue
+
+        valid_state = any(
+            state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE) for state in states
+        )
+        if not valid_state:
+            # Set as unknown if all members are unknown or unavailable
+            self._attr_is_closed = None
+
+        self._attr_current_valve_position = reduce_attribute(
+            states, ATTR_CURRENT_POSITION
+        )
+
+        supported_features = ValveEntityFeature(0)
+        if self._valves[KEY_OPEN_CLOSE]:
+            supported_features |= ValveEntityFeature.OPEN | ValveEntityFeature.CLOSE
+        if self._valves[KEY_STOP]:
+            supported_features |= ValveEntityFeature.STOP
+        if self._valves[KEY_SET_POSITION]:
+            supported_features |= ValveEntityFeature.SET_POSITION
+        self._attr_supported_features = supported_features
diff --git a/tests/components/group/test_config_flow.py b/tests/components/group/test_config_flow.py
index b1bb6e5d7bb..86ad65c69a9 100644
--- a/tests/components/group/test_config_flow.py
+++ b/tests/components/group/test_config_flow.py
@@ -1,4 +1,4 @@
-"""Test the Switch config flow."""
+"""Test the Group config flow."""
 
 from typing import Any
 from unittest.mock import patch
@@ -60,6 +60,7 @@ from tests.typing import WebSocketGenerator
         ),
         ("switch", "on", "on", {}, {}, {"all": False}, {}),
         ("switch", "on", "on", {}, {"all": True}, {"all": True}, {}),
+        ("valve", "open", "open", {}, {}, {}, {}),
     ],
 )
 async def test_config_flow(
@@ -148,6 +149,7 @@ async def test_config_flow(
         ("notify", {}),
         ("media_player", {}),
         ("switch", {}),
+        ("valve", {}),
     ],
 )
 async def test_config_flow_hides_members(
@@ -222,6 +224,7 @@ async def test_config_flow_hides_members(
             {"ignore_non_numeric": False, "type": "sum"},
         ),
         ("switch", "on", {"all": False}, {}),
+        ("valve", "open", {}, {}),
     ],
 )
 async def test_options(
@@ -404,6 +407,7 @@ async def test_all_options(
         ("notify", {}),
         ("media_player", {}),
         ("switch", {}),
+        ("valve", {}),
     ],
 )
 async def test_options_flow_hides_members(
@@ -487,6 +491,7 @@ LOCK_ATTRS = [{"supported_features": 1}, {}]
 NOTIFY_ATTRS = [{"supported_features": 0}, {}]
 MEDIA_PLAYER_ATTRS = [{"supported_features": 0}, {}]
 SENSOR_ATTRS = [{"icon": "mdi:calculator"}, {"max_entity_id": "sensor.input_two"}]
+VALVE_ATTRS = [{"supported_features": 0}, {}]
 
 
 @pytest.mark.parametrize(
@@ -503,6 +508,7 @@ SENSOR_ATTRS = [{"icon": "mdi:calculator"}, {"max_entity_id": "sensor.input_two"
         ("media_player", {}, ["on", "off"], "on", MEDIA_PLAYER_ATTRS),
         ("sensor", {"type": "max"}, ["10", "20"], "20.0", SENSOR_ATTRS),
         ("switch", {}, ["on", "off"], "on", [{}, {}]),
+        ("valve", {}, ["open", "closed"], "open", VALVE_ATTRS),
     ],
 )
 async def test_config_flow_preview(
@@ -621,6 +627,7 @@ async def test_config_flow_preview(
             SENSOR_ATTRS,
         ),
         ("switch", {}, {}, ["on", "off"], "on", [{}, {}]),
+        ("valve", {}, {}, ["open", "closed"], "open", VALVE_ATTRS),
     ],
 )
 async def test_option_flow_preview(
diff --git a/tests/components/group/test_valve.py b/tests/components/group/test_valve.py
new file mode 100644
index 00000000000..9d8fd2670f2
--- /dev/null
+++ b/tests/components/group/test_valve.py
@@ -0,0 +1,688 @@
+"""The tests for the group valve platform."""
+
+import asyncio
+from datetime import timedelta
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant.components.group.valve import DEFAULT_NAME
+from homeassistant.components.valve import (
+    ATTR_CURRENT_POSITION,
+    ATTR_POSITION,
+    DOMAIN as VALVE_DOMAIN,
+    ValveState,
+)
+from homeassistant.const import (
+    ATTR_ASSUMED_STATE,
+    ATTR_ENTITY_ID,
+    ATTR_FRIENDLY_NAME,
+    ATTR_SUPPORTED_FEATURES,
+    CONF_ENTITIES,
+    CONF_UNIQUE_ID,
+    SERVICE_CLOSE_VALVE,
+    SERVICE_OPEN_VALVE,
+    SERVICE_SET_VALVE_POSITION,
+    SERVICE_STOP_VALVE,
+    SERVICE_TOGGLE,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+from homeassistant.setup import async_setup_component
+from homeassistant.util import dt as dt_util
+
+from tests.common import assert_setup_component, async_fire_time_changed
+
+VALVE_GROUP = "valve.valve_group"
+DEMO_VALVE1 = "valve.front_garden"
+DEMO_VALVE2 = "valve.orchard"
+DEMO_VALVE_POS1 = "valve.back_garden"
+DEMO_VALVE_POS2 = "valve.trees"
+
+CONFIG_ALL = {
+    VALVE_DOMAIN: [
+        {"platform": "demo"},
+        {
+            "platform": "group",
+            CONF_ENTITIES: [DEMO_VALVE1, DEMO_VALVE2, DEMO_VALVE_POS1, DEMO_VALVE_POS2],
+        },
+    ]
+}
+
+CONFIG_POS = {
+    VALVE_DOMAIN: [
+        {"platform": "demo"},
+        {
+            "platform": "group",
+            CONF_ENTITIES: [DEMO_VALVE_POS1, DEMO_VALVE_POS2],
+        },
+    ]
+}
+
+
+CONFIG_ATTRIBUTES = {
+    VALVE_DOMAIN: {
+        "platform": "group",
+        CONF_ENTITIES: [DEMO_VALVE1, DEMO_VALVE2, DEMO_VALVE_POS1, DEMO_VALVE_POS2],
+        CONF_UNIQUE_ID: "unique_identifier",
+    }
+}
+
+
+@pytest.fixture(scope="module", autouse=True)
+def patch_demo_open_close_delay():
+    """Patch demo valve open/close delay."""
+    with patch("homeassistant.components.demo.valve.OPEN_CLOSE_DELAY", 0):
+        yield
+
+
+@pytest.fixture
+async def setup_comp(
+    hass: HomeAssistant, config_count: tuple[dict[str, Any], int]
+) -> None:
+    """Set up group valve component."""
+    config, count = config_count
+    with assert_setup_component(count, VALVE_DOMAIN):
+        await async_setup_component(hass, VALVE_DOMAIN, config)
+    await hass.async_block_till_done()
+    await hass.async_start()
+    await hass.async_block_till_done()
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ATTRIBUTES, 1)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_state(hass: HomeAssistant) -> None:
+    """Test handling of state.
+
+    The group state is unknown if all group members are unknown or unavailable.
+    Otherwise, the group state is opening if at least one group member is opening.
+    Otherwise, the group state is closing if at least one group member is closing.
+    Otherwise, the group state is open if at least one group member is open.
+    Otherwise, the group state is closed.
+    """
+    state = hass.states.get(VALVE_GROUP)
+    # No entity has a valid state -> group state unavailable
+    assert state.state == STATE_UNAVAILABLE
+    assert state.attributes[ATTR_FRIENDLY_NAME] == DEFAULT_NAME
+    assert ATTR_ENTITY_ID not in state.attributes
+    assert ATTR_ASSUMED_STATE not in state.attributes
+    assert state.attributes[ATTR_SUPPORTED_FEATURES] == 0
+    assert ATTR_CURRENT_POSITION not in state.attributes
+
+    # Test group members exposed as attribute
+    hass.states.async_set(DEMO_VALVE1, STATE_UNKNOWN, {})
+    await hass.async_block_till_done()
+    state = hass.states.get(VALVE_GROUP)
+    assert state.attributes[ATTR_ENTITY_ID] == [
+        DEMO_VALVE1,
+        DEMO_VALVE2,
+        DEMO_VALVE_POS1,
+        DEMO_VALVE_POS2,
+    ]
+
+    # The group state is unavailable if all group members are unavailable.
+    hass.states.async_set(DEMO_VALVE1, STATE_UNAVAILABLE, {})
+    hass.states.async_set(DEMO_VALVE_POS1, STATE_UNAVAILABLE, {})
+    hass.states.async_set(DEMO_VALVE_POS2, STATE_UNAVAILABLE, {})
+    hass.states.async_set(DEMO_VALVE2, STATE_UNAVAILABLE, {})
+    await hass.async_block_till_done()
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == STATE_UNAVAILABLE
+
+    # The group state is unknown if all group members are unknown or unavailable.
+    for state_1 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
+        for state_2 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
+            for state_3 in (STATE_UNAVAILABLE, STATE_UNKNOWN):
+                hass.states.async_set(DEMO_VALVE1, state_1, {})
+                hass.states.async_set(DEMO_VALVE_POS1, state_2, {})
+                hass.states.async_set(DEMO_VALVE_POS2, state_3, {})
+                hass.states.async_set(DEMO_VALVE2, STATE_UNKNOWN, {})
+                await hass.async_block_till_done()
+                state = hass.states.get(VALVE_GROUP)
+                assert state.state == STATE_UNKNOWN
+
+    # At least one member opening -> group opening
+    for state_1 in (
+        ValveState.CLOSED,
+        ValveState.CLOSING,
+        ValveState.OPEN,
+        ValveState.OPENING,
+        STATE_UNAVAILABLE,
+        STATE_UNKNOWN,
+    ):
+        for state_2 in (
+            ValveState.CLOSED,
+            ValveState.CLOSING,
+            ValveState.OPEN,
+            ValveState.OPENING,
+            STATE_UNAVAILABLE,
+            STATE_UNKNOWN,
+        ):
+            for state_3 in (
+                ValveState.CLOSED,
+                ValveState.CLOSING,
+                ValveState.OPEN,
+                ValveState.OPENING,
+                STATE_UNAVAILABLE,
+                STATE_UNKNOWN,
+            ):
+                hass.states.async_set(DEMO_VALVE1, state_1, {})
+                hass.states.async_set(DEMO_VALVE_POS1, state_2, {})
+                hass.states.async_set(DEMO_VALVE_POS2, state_3, {})
+                hass.states.async_set(DEMO_VALVE2, ValveState.OPENING, {})
+                await hass.async_block_till_done()
+                state = hass.states.get(VALVE_GROUP)
+                assert state.state == ValveState.OPENING
+
+    # At least one member closing -> group closing
+    for state_1 in (
+        ValveState.CLOSED,
+        ValveState.CLOSING,
+        ValveState.OPEN,
+        STATE_UNAVAILABLE,
+        STATE_UNKNOWN,
+    ):
+        for state_2 in (
+            ValveState.CLOSED,
+            ValveState.CLOSING,
+            ValveState.OPEN,
+            STATE_UNAVAILABLE,
+            STATE_UNKNOWN,
+        ):
+            for state_3 in (
+                ValveState.CLOSED,
+                ValveState.CLOSING,
+                ValveState.OPEN,
+                STATE_UNAVAILABLE,
+                STATE_UNKNOWN,
+            ):
+                hass.states.async_set(DEMO_VALVE1, state_1, {})
+                hass.states.async_set(DEMO_VALVE_POS1, state_2, {})
+                hass.states.async_set(DEMO_VALVE_POS2, state_3, {})
+                hass.states.async_set(DEMO_VALVE2, ValveState.CLOSING, {})
+                await hass.async_block_till_done()
+                state = hass.states.get(VALVE_GROUP)
+                assert state.state == ValveState.CLOSING
+
+    # At least one member open -> group open
+    for state_1 in (
+        ValveState.CLOSED,
+        ValveState.OPEN,
+        STATE_UNAVAILABLE,
+        STATE_UNKNOWN,
+    ):
+        for state_2 in (
+            ValveState.CLOSED,
+            ValveState.OPEN,
+            STATE_UNAVAILABLE,
+            STATE_UNKNOWN,
+        ):
+            for state_3 in (
+                ValveState.CLOSED,
+                ValveState.OPEN,
+                STATE_UNAVAILABLE,
+                STATE_UNKNOWN,
+            ):
+                hass.states.async_set(DEMO_VALVE1, state_1, {})
+                hass.states.async_set(DEMO_VALVE_POS1, state_2, {})
+                hass.states.async_set(DEMO_VALVE_POS2, state_3, {})
+                hass.states.async_set(DEMO_VALVE2, ValveState.OPEN, {})
+                await hass.async_block_till_done()
+                state = hass.states.get(VALVE_GROUP)
+                assert state.state == ValveState.OPEN
+
+    # At least one member closed -> group closed
+    for state_1 in (ValveState.CLOSED, STATE_UNAVAILABLE, STATE_UNKNOWN):
+        for state_2 in (ValveState.CLOSED, STATE_UNAVAILABLE, STATE_UNKNOWN):
+            for state_3 in (ValveState.CLOSED, STATE_UNAVAILABLE, STATE_UNKNOWN):
+                hass.states.async_set(DEMO_VALVE1, state_1, {})
+                hass.states.async_set(DEMO_VALVE_POS1, state_2, {})
+                hass.states.async_set(DEMO_VALVE_POS2, state_3, {})
+                hass.states.async_set(DEMO_VALVE2, ValveState.CLOSED, {})
+                await hass.async_block_till_done()
+                state = hass.states.get(VALVE_GROUP)
+                assert state.state == ValveState.CLOSED
+
+    # All group members removed from the state machine -> unavailable
+    hass.states.async_remove(DEMO_VALVE1)
+    hass.states.async_remove(DEMO_VALVE_POS1)
+    hass.states.async_remove(DEMO_VALVE_POS2)
+    hass.states.async_remove(DEMO_VALVE2)
+    await hass.async_block_till_done()
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == STATE_UNAVAILABLE
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ATTRIBUTES, 1)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_attributes(
+    hass: HomeAssistant, entity_registry: er.EntityRegistry
+) -> None:
+    """Test handling of state attributes."""
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == STATE_UNAVAILABLE
+    assert state.attributes[ATTR_FRIENDLY_NAME] == DEFAULT_NAME
+    assert ATTR_ENTITY_ID not in state.attributes
+    assert ATTR_ASSUMED_STATE not in state.attributes
+    assert state.attributes[ATTR_SUPPORTED_FEATURES] == 0
+    assert ATTR_CURRENT_POSITION not in state.attributes
+
+    # Set entity as closed
+    hass.states.async_set(DEMO_VALVE1, ValveState.CLOSED, {})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.CLOSED
+    assert state.attributes[ATTR_ENTITY_ID] == [
+        DEMO_VALVE1,
+        DEMO_VALVE2,
+        DEMO_VALVE_POS1,
+        DEMO_VALVE_POS2,
+    ]
+
+    # Set entity as opening
+    hass.states.async_set(DEMO_VALVE1, ValveState.OPENING, {})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPENING
+
+    # Set entity as closing
+    hass.states.async_set(DEMO_VALVE1, ValveState.CLOSING, {})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.CLOSING
+
+    # Set entity as unknown again
+    hass.states.async_set(DEMO_VALVE1, STATE_UNKNOWN, {})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == STATE_UNKNOWN
+
+    # Add Entity that supports open / close / stop
+    hass.states.async_set(DEMO_VALVE1, ValveState.OPEN, {ATTR_SUPPORTED_FEATURES: 11})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+    assert ATTR_ASSUMED_STATE not in state.attributes
+    assert state.attributes[ATTR_SUPPORTED_FEATURES] == 11
+    assert ATTR_CURRENT_POSITION not in state.attributes
+
+    # Add Entity that supports set_valve_position
+    hass.states.async_set(
+        DEMO_VALVE_POS1,
+        ValveState.OPEN,
+        {ATTR_SUPPORTED_FEATURES: 4, ATTR_CURRENT_POSITION: 70},
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+    assert ATTR_ASSUMED_STATE not in state.attributes
+    assert state.attributes[ATTR_SUPPORTED_FEATURES] == 15
+    assert state.attributes[ATTR_CURRENT_POSITION] == 70
+
+    ### Test state when group members have different states ###
+
+    # Valves
+    hass.states.async_remove(DEMO_VALVE_POS1)
+    hass.states.async_remove(DEMO_VALVE_POS2)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+    assert ATTR_ASSUMED_STATE not in state.attributes
+    assert state.attributes[ATTR_SUPPORTED_FEATURES] == 11
+    assert ATTR_CURRENT_POSITION not in state.attributes
+
+    # Test entity registry integration
+    entry = entity_registry.async_get(VALVE_GROUP)
+    assert entry
+    assert entry.unique_id == "unique_identifier"
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ALL, 2)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_open_valves(hass: HomeAssistant) -> None:
+    """Test open valve function."""
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_OPEN_VALVE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+
+    for _ in range(10):
+        future = dt_util.utcnow() + timedelta(seconds=1)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+    assert state.attributes[ATTR_CURRENT_POSITION] == 100
+
+    assert hass.states.get(DEMO_VALVE1).state == ValveState.OPEN
+    assert hass.states.get(DEMO_VALVE_POS1).attributes[ATTR_CURRENT_POSITION] == 100
+    assert hass.states.get(DEMO_VALVE_POS2).attributes[ATTR_CURRENT_POSITION] == 100
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ALL, 2)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_close_valves(hass: HomeAssistant) -> None:
+    """Test close valve function."""
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_CLOSE_VALVE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+
+    for _ in range(10):
+        future = dt_util.utcnow() + timedelta(seconds=1)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.CLOSED
+    assert state.attributes[ATTR_CURRENT_POSITION] == 0
+
+    assert hass.states.get(DEMO_VALVE1).state == ValveState.CLOSED
+    assert hass.states.get(DEMO_VALVE_POS1).attributes[ATTR_CURRENT_POSITION] == 0
+    assert hass.states.get(DEMO_VALVE_POS2).attributes[ATTR_CURRENT_POSITION] == 0
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ALL, 2)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_toggle_valves(hass: HomeAssistant) -> None:
+    """Test toggle valve function."""
+    # Start valves in open state
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_OPEN_VALVE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+    for _ in range(10):
+        future = dt_util.utcnow() + timedelta(seconds=1)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+
+    # Toggle will close valves
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_TOGGLE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+    for _ in range(10):
+        future = dt_util.utcnow() + timedelta(seconds=1)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.CLOSED
+    assert state.attributes[ATTR_CURRENT_POSITION] == 0
+
+    assert hass.states.get(DEMO_VALVE1).state == ValveState.CLOSED
+    assert hass.states.get(DEMO_VALVE_POS1).attributes[ATTR_CURRENT_POSITION] == 0
+    assert hass.states.get(DEMO_VALVE_POS2).attributes[ATTR_CURRENT_POSITION] == 0
+
+    # Toggle again will open valves
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_TOGGLE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+    for _ in range(10):
+        future = dt_util.utcnow() + timedelta(seconds=1)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+    assert state.attributes[ATTR_CURRENT_POSITION] == 100
+
+    assert hass.states.get(DEMO_VALVE1).state == ValveState.OPEN
+    assert hass.states.get(DEMO_VALVE_POS1).attributes[ATTR_CURRENT_POSITION] == 100
+    assert hass.states.get(DEMO_VALVE_POS2).attributes[ATTR_CURRENT_POSITION] == 100
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ALL, 2)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_stop_valves(hass: HomeAssistant) -> None:
+    """Test stop valve function."""
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_OPEN_VALVE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+    future = dt_util.utcnow() + timedelta(seconds=1)
+    async_fire_time_changed(hass, future)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPENING
+
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_STOP_VALVE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+    future = dt_util.utcnow() + timedelta(seconds=1)
+    async_fire_time_changed(hass, future)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+    assert state.attributes[ATTR_CURRENT_POSITION] == 60  # (40 + 80) / 2
+
+    assert hass.states.get(DEMO_VALVE1).state == ValveState.OPEN
+    assert hass.states.get(DEMO_VALVE_POS1).attributes[ATTR_CURRENT_POSITION] == 80
+    assert hass.states.get(DEMO_VALVE_POS2).attributes[ATTR_CURRENT_POSITION] == 40
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ALL, 2)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_set_valve_position(hass: HomeAssistant) -> None:
+    """Test set valve position function."""
+    await hass.services.async_call(
+        VALVE_DOMAIN,
+        SERVICE_SET_VALVE_POSITION,
+        {ATTR_ENTITY_ID: VALVE_GROUP, ATTR_POSITION: 50},
+        blocking=True,
+    )
+    for _ in range(4):
+        future = dt_util.utcnow() + timedelta(seconds=1)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.state == ValveState.OPEN
+    assert state.attributes[ATTR_CURRENT_POSITION] == 50
+
+    assert hass.states.get(DEMO_VALVE1).state == ValveState.OPEN
+    assert hass.states.get(DEMO_VALVE_POS1).attributes[ATTR_CURRENT_POSITION] == 50
+    assert hass.states.get(DEMO_VALVE_POS2).attributes[ATTR_CURRENT_POSITION] == 50
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_POS, 2)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_is_opening_closing(hass: HomeAssistant) -> None:
+    """Test is_opening property."""
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_OPEN_VALVE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+    await hass.async_block_till_done()
+
+    # Both valves opening -> opening
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.OPENING
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.OPENING
+    assert hass.states.get(VALVE_GROUP).state == ValveState.OPENING
+
+    for _ in range(10):
+        future = dt_util.utcnow() + timedelta(seconds=1)
+        async_fire_time_changed(hass, future)
+        await hass.async_block_till_done()
+
+    await hass.services.async_call(
+        VALVE_DOMAIN, SERVICE_CLOSE_VALVE, {ATTR_ENTITY_ID: VALVE_GROUP}, blocking=True
+    )
+
+    # Both valves closing -> closing
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.CLOSING
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.CLOSING
+    assert hass.states.get(VALVE_GROUP).state == ValveState.CLOSING
+
+    hass.states.async_set(
+        DEMO_VALVE_POS1, ValveState.OPENING, {ATTR_SUPPORTED_FEATURES: 11}
+    )
+    await hass.async_block_till_done()
+
+    # Closing + Opening -> Opening
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.CLOSING
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.OPENING
+    assert hass.states.get(VALVE_GROUP).state == ValveState.OPENING
+
+    hass.states.async_set(
+        DEMO_VALVE_POS1, ValveState.CLOSING, {ATTR_SUPPORTED_FEATURES: 11}
+    )
+    await hass.async_block_till_done()
+
+    # Both valves closing -> closing
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.CLOSING
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.CLOSING
+    assert hass.states.get(VALVE_GROUP).state == ValveState.CLOSING
+
+    # Closed + Closing -> Closing
+    hass.states.async_set(
+        DEMO_VALVE_POS1, ValveState.CLOSED, {ATTR_SUPPORTED_FEATURES: 11}
+    )
+    await hass.async_block_till_done()
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.CLOSING
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.CLOSED
+    assert hass.states.get(VALVE_GROUP).state == ValveState.CLOSING
+
+    # Open + Closing -> Closing
+    hass.states.async_set(
+        DEMO_VALVE_POS1, ValveState.OPEN, {ATTR_SUPPORTED_FEATURES: 11}
+    )
+    await hass.async_block_till_done()
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.CLOSING
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.OPEN
+    assert hass.states.get(VALVE_GROUP).state == ValveState.CLOSING
+
+    # Closed + Opening -> Closing
+    hass.states.async_set(
+        DEMO_VALVE_POS2, ValveState.OPENING, {ATTR_SUPPORTED_FEATURES: 11}
+    )
+    hass.states.async_set(
+        DEMO_VALVE_POS1, ValveState.CLOSED, {ATTR_SUPPORTED_FEATURES: 11}
+    )
+    await hass.async_block_till_done()
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.OPENING
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.CLOSED
+    assert hass.states.get(VALVE_GROUP).state == ValveState.OPENING
+
+    # Open + Opening -> Closing
+    hass.states.async_set(
+        DEMO_VALVE_POS1, ValveState.OPEN, {ATTR_SUPPORTED_FEATURES: 11}
+    )
+    await hass.async_block_till_done()
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.OPENING
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.OPEN
+    assert hass.states.get(VALVE_GROUP).state == ValveState.OPENING
+
+
+@pytest.mark.parametrize("config_count", [(CONFIG_ATTRIBUTES, 1)])
+@pytest.mark.usefixtures("setup_comp")
+async def test_assumed_state(hass: HomeAssistant) -> None:
+    """Test assumed_state attribute behavior."""
+    # No members with assumed_state -> group doesn't have assumed_state in attributes
+    hass.states.async_set(DEMO_VALVE1, ValveState.OPEN, {})
+    hass.states.async_set(DEMO_VALVE_POS1, ValveState.OPEN, {})
+    hass.states.async_set(DEMO_VALVE_POS2, ValveState.CLOSED, {})
+    hass.states.async_set(DEMO_VALVE2, ValveState.CLOSED, {})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert ATTR_ASSUMED_STATE not in state.attributes
+
+    # One member with assumed_state=True -> group has assumed_state=True
+    hass.states.async_set(DEMO_VALVE1, ValveState.OPEN, {ATTR_ASSUMED_STATE: True})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.attributes.get(ATTR_ASSUMED_STATE) is True
+
+    # Multiple members with assumed_state=True -> group has assumed_state=True
+    hass.states.async_set(
+        DEMO_VALVE_POS2, ValveState.CLOSED, {ATTR_ASSUMED_STATE: True}
+    )
+    hass.states.async_set(DEMO_VALVE2, ValveState.CLOSED, {ATTR_ASSUMED_STATE: True})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.attributes.get(ATTR_ASSUMED_STATE) is True
+
+    # Unavailable member with assumed_state=True -> group has assumed_state=True
+    hass.states.async_set(DEMO_VALVE1, ValveState.OPEN, {})
+    hass.states.async_set(DEMO_VALVE_POS2, ValveState.CLOSED, {})
+    hass.states.async_set(DEMO_VALVE2, STATE_UNAVAILABLE, {ATTR_ASSUMED_STATE: True})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.attributes.get(ATTR_ASSUMED_STATE) is True
+
+    # Unknown member with assumed_state=True -> group has assumed_state=True
+    hass.states.async_set(DEMO_VALVE2, STATE_UNKNOWN, {ATTR_ASSUMED_STATE: True})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert state.attributes.get(ATTR_ASSUMED_STATE) is True
+
+    # All members without assumed_state -> group doesn't have assumed_state in attributes
+    hass.states.async_set(DEMO_VALVE2, ValveState.CLOSED, {})
+    await hass.async_block_till_done()
+
+    state = hass.states.get(VALVE_GROUP)
+    assert ATTR_ASSUMED_STATE not in state.attributes
+
+
+async def test_nested_group(hass: HomeAssistant) -> None:
+    """Test nested valve group."""
+    await async_setup_component(
+        hass,
+        VALVE_DOMAIN,
+        {
+            VALVE_DOMAIN: [
+                {"platform": "demo"},
+                {
+                    "platform": "group",
+                    "entities": ["valve.bedroom_group"],
+                    "name": "Nested Group",
+                },
+                {
+                    "platform": "group",
+                    CONF_ENTITIES: [DEMO_VALVE_POS1, DEMO_VALVE_POS2],
+                    "name": "Bedroom Group",
+                },
+            ]
+        },
+    )
+    await hass.async_block_till_done()
+    await hass.async_start()
+    await hass.async_block_till_done()
+
+    state = hass.states.get("valve.bedroom_group")
+    assert state is not None
+    assert state.state == ValveState.OPEN
+    assert state.attributes.get(ATTR_ENTITY_ID) == [DEMO_VALVE_POS1, DEMO_VALVE_POS2]
+
+    state = hass.states.get("valve.nested_group")
+    assert state is not None
+    assert state.state == ValveState.OPEN
+    assert state.attributes.get(ATTR_ENTITY_ID) == ["valve.bedroom_group"]
+
+    # Test controlling the nested group
+    async with asyncio.timeout(0.5):
+        await hass.services.async_call(
+            VALVE_DOMAIN,
+            SERVICE_CLOSE_VALVE,
+            {ATTR_ENTITY_ID: "valve.nested_group"},
+            blocking=True,
+        )
+    assert hass.states.get(DEMO_VALVE_POS1).state == ValveState.CLOSING
+    assert hass.states.get(DEMO_VALVE_POS2).state == ValveState.CLOSING
+    assert hass.states.get("valve.bedroom_group").state == ValveState.CLOSING
+    assert hass.states.get("valve.nested_group").state == ValveState.CLOSING