Add initial test for nibe buttons (home-assistant#84950)

.coveragerc

@@ -845,7 +845,6 @@ omit =
     homeassistant/components/nibe_heatpump/__init__.py
     homeassistant/components/nibe_heatpump/climate.py
     homeassistant/components/nibe_heatpump/binary_sensor.py
-    homeassistant/components/nibe_heatpump/button.py
     homeassistant/components/nibe_heatpump/number.py
     homeassistant/components/nibe_heatpump/select.py
     homeassistant/components/nibe_heatpump/sensor.py

homeassistant/components/nibe_heatpump/__init__.py

@@ -114,7 +114,7 @@ def _on_product_info(product_info: ProductInfo):
             sw_version=str(product_info.firmware_version),
         )
 
-    if isinstance(connection, NibeGW):
+    if hasattr(connection, "PRODUCT_INFO_EVENT") and hasattr(connection, "subscribe"):
         connection.subscribe(connection.PRODUCT_INFO_EVENT, _on_product_info)
     else:
         reg.async_update_device(device_id=device_entry.id, model=heatpump.model.name)

tests/components/nibe_heatpump/__init__.py

@@ -1 +1,19 @@
 """Tests for the Nibe Heat Pump integration."""
+
+from typing import Any
+
+from homeassistant.components.nibe_heatpump import DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def async_add_entry(hass: HomeAssistant, data: dict[str, Any]) -> None:
+    """Add entry and get the coordinator."""
+    entry = MockConfigEntry(domain=DOMAIN, title="Dummy", data=data)
+
+    entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+    assert entry.state == ConfigEntryState.LOADED

tests/components/nibe_heatpump/conftest.py

@@ -0,0 +1,57 @@
+"""Test configuration for Nibe Heat Pump."""
+from collections.abc import AsyncIterator, Iterable
+from contextlib import ExitStack
+from typing import Any
+from unittest.mock import AsyncMock, Mock, patch
+
+from nibe.coil import Coil
+from nibe.connection import Connection
+from nibe.exceptions import CoilReadException
+import pytest
+
+
+@pytest.fixture(autouse=True, name="mock_connection_constructor")
+async def fixture_mock_connection_constructor():
+    """Make sure we have a dummy connection."""
+    mock_constructor = Mock()
+    with ExitStack() as stack:
+        places = [
+            "homeassistant.components.nibe_heatpump.config_flow.NibeGW",
+            "homeassistant.components.nibe_heatpump.config_flow.Modbus",
+            "homeassistant.components.nibe_heatpump.NibeGW",
+            "homeassistant.components.nibe_heatpump.Modbus",
+        ]
+        for place in places:
+            stack.enter_context(patch(place, new=mock_constructor))
+        yield mock_constructor
+
+
+@pytest.fixture(name="mock_connection")
+def fixture_mock_connection(mock_connection_constructor: Mock):
+    """Make sure we have a dummy connection."""
+    mock_connection = AsyncMock(spec=Connection)
+    mock_connection_constructor.return_value = mock_connection
+    return mock_connection
+
+
+@pytest.fixture(name="coils")
+async def fixture_coils(mock_connection):
+    """Return a dict with coil data."""
+    coils: dict[int, Any] = {}
+
+    async def read_coil(coil: Coil, timeout: float = 0) -> Coil:
+        nonlocal coils
+        if (data := coils.get(coil.address, None)) is None:
+            raise CoilReadException()
+        coil.value = data
+        return coil
+
+    async def read_coils(
+        coils: Iterable[Coil], timeout: float = 0
+    ) -> AsyncIterator[Coil]:
+        for coil in coils:
+            yield await read_coil(coil, timeout)
+
+    mock_connection.read_coil = read_coil
+    mock_connection.read_coils = read_coils
+    yield coils

tests/components/nibe_heatpump/test_button.py

@@ -0,0 +1,96 @@
+"""Test the Nibe Heat Pump config flow."""
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+from freezegun.api import FrozenDateTimeFactory
+from nibe.coil import Coil
+from nibe.coil_groups import UNIT_COILGROUPS
+from nibe.heatpump import Model
+import pytest
+
+from homeassistant.components.button import DOMAIN as PLATFORM_DOMAIN, SERVICE_PRESS
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+
+from . import async_add_entry
+
+from tests.common import async_fire_time_changed
+
+MOCK_ENTRY_DATA = {
+    "model": None,
+    "ip_address": "127.0.0.1",
+    "listening_port": 9999,
+    "remote_read_port": 10000,
+    "remote_write_port": 10001,
+    "word_swap": True,
+    "connection_type": "nibegw",
+}
+
+
+@pytest.fixture(autouse=True)
+async def fixture_single_platform():
+    """Only allow this platform to load."""
+    with patch("homeassistant.components.nibe_heatpump.PLATFORMS", [Platform.BUTTON]):
+        yield
+
+
+@pytest.mark.parametrize(
+    ("model", "entity_id"),
+    [
+        (Model.F1155, "button.f1155_alarm_reset"),
+        (Model.S320, "button.s320_reset_alarm"),
+    ],
+)
+async def test_reset_button(
+    hass: HomeAssistant,
+    mock_connection: AsyncMock,
+    model: Model,
+    entity_id: str,
+    coils: dict[int, Any],
+    freezer: FrozenDateTimeFactory,
+):
+    """Test reset button."""
+
+    unit = UNIT_COILGROUPS[model.series]["main"]
+
+    # Setup a non alarm state
+    coils[unit.alarm_reset] = 0
+    coils[unit.alarm] = 0
+
+    await async_add_entry(hass, {**MOCK_ENTRY_DATA, "model": model.name})
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.state == STATE_UNAVAILABLE
+
+    # Signal alarm
+    coils[unit.alarm] = 100
+
+    freezer.tick(60)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(entity_id)
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+    # Press button
+    await hass.services.async_call(
+        PLATFORM_DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: entity_id},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    # Verify reset was written
+    args = mock_connection.write_coil.call_args
+    assert args
+    coil: Coil = args.args[0]
+    assert coil.address == unit.alarm_reset
+    assert coil.value == 1

tests/components/nibe_heatpump/test_config_flow.py

@@ -1,8 +1,7 @@
 """Test the Nibe Heat Pump config flow."""
-from unittest.mock import AsyncMock, Mock, patch
+from unittest.mock import Mock, patch
 
 from nibe.coil import Coil
-from nibe.connection import Connection
 from nibe.exceptions import (
     AddressInUseException,
     CoilNotFoundException,
@@ -34,28 +33,6 @@
 }
 
 
-@fixture(autouse=True, name="mock_connection_constructor")
-async def fixture_mock_connection_constructor():
-    """Make sure we have a dummy connection."""
-    mock_constructor = Mock()
-    with patch(
-        "homeassistant.components.nibe_heatpump.config_flow.NibeGW",
-        new=mock_constructor,
-    ), patch(
-        "homeassistant.components.nibe_heatpump.config_flow.Modbus",
-        new=mock_constructor,
-    ):
-        yield mock_constructor
-
-
-@fixture(name="mock_connection")
-def fixture_mock_connection(mock_connection_constructor: Mock):
-    """Make sure we have a dummy connection."""
-    mock_connection = AsyncMock(spec=Connection)
-    mock_connection_constructor.return_value = mock_connection
-    return mock_connection
-
-
 @fixture(autouse=True, name="mock_setup_entry")
 async def fixture_mock_setup():
     """Make sure we never actually run setup."""