Skip to content

Commit

Permalink
Update opentherm_gw tests to avoid patching internals (#125152)
Browse files Browse the repository at this point in the history 
* Update tests to avoid patching internals

* * Use fixtures for tests
* Update variable names in tests for clarity

* Use hass.config_entries.async_setup instead of setup.async_setup_component
  • Loading branch information
@mvn23
mvn23 authored Sep 3, 2024
1 parent 8f26cff commit 8e03f3a
Show file tree
Hide file tree
Showing 3 changed files with 157 additions and 195 deletions.
41 changes: 41 additions & 0 deletions tests/components/opentherm_gw/conftest.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
"""Test configuration for opentherm_gw."""

from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch

from pyotgw.vars import OTGW, OTGW_ABOUT
import pytest

VERSION_TEST = "4.2.5"
MINIMAL_STATUS = {OTGW: {OTGW_ABOUT: f"OpenTherm Gateway {VERSION_TEST}"}}


@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.opentherm_gw.async_setup_entry",
return_value=True,
) as mock_setup_entry:
yield mock_setup_entry


@pytest.fixture
def mock_pyotgw() -> Generator[MagicMock]:
"""Mock a pyotgw.OpenThermGateway object."""
with (
patch(
"homeassistant.components.opentherm_gw.OpenThermGateway",
return_value=MagicMock(
connect=AsyncMock(return_value=MINIMAL_STATUS),
set_control_setpoint=AsyncMock(),
set_max_relative_mod=AsyncMock(),
disconnect=AsyncMock(),
),
) as mock_gateway,
patch(
"homeassistant.components.opentherm_gw.config_flow.pyotgw.OpenThermGateway",
new=mock_gateway,
),
):
yield mock_gateway
223 changes: 88 additions & 135 deletions tests/components/opentherm_gw/test_config_flow.py
View file
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
"""Test the Opentherm Gateway config flow."""

from unittest.mock import patch
from unittest.mock import AsyncMock, MagicMock

from pyotgw.vars import OTGW, OTGW_ABOUT
from serial import SerialException

from homeassistant import config_entries
Expand All @@ -25,10 +24,12 @@

from tests.common import MockConfigEntry

MINIMAL_STATUS = {OTGW: {OTGW_ABOUT: "OpenTherm Gateway 4.2.5"}}


async def test_form_user(hass: HomeAssistant) -> None:
async def test_form_user(
hass: HomeAssistant,
mock_pyotgw: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test we get the form."""

result = await hass.config_entries.flow.async_init(
Expand All @@ -37,27 +38,10 @@ async def test_form_user(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}

with (
patch(
"homeassistant.components.opentherm_gw.async_setup",
return_value=True,
) as mock_setup,
patch(
"homeassistant.components.opentherm_gw.async_setup_entry",
return_value=True,
) as mock_setup_entry,
patch(
"pyotgw.OpenThermGateway.connect", return_value=MINIMAL_STATUS
) as mock_pyotgw_connect,
patch(
"pyotgw.OpenThermGateway.disconnect", return_value=None
) as mock_pyotgw_disconnect,
patch("pyotgw.status.StatusManager._process_updates", return_value=None),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB0"}
)
await hass.async_block_till_done()
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB0"}
)
await hass.async_block_till_done()

assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "Test Entry 1"
Expand All @@ -66,37 +50,21 @@ async def test_form_user(hass: HomeAssistant) -> None:
CONF_DEVICE: "/dev/ttyUSB0",
CONF_ID: "test_entry_1",
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_pyotgw_connect.mock_calls) == 1
assert len(mock_pyotgw_disconnect.mock_calls) == 1
assert mock_pyotgw.return_value.connect.await_count == 1
assert mock_pyotgw.return_value.disconnect.await_count == 1


async def test_form_import(hass: HomeAssistant) -> None:
async def test_form_import(
hass: HomeAssistant,
mock_pyotgw: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test import from existing config."""

with (
patch(
"homeassistant.components.opentherm_gw.async_setup",
return_value=True,
) as mock_setup,
patch(
"homeassistant.components.opentherm_gw.async_setup_entry",
return_value=True,
) as mock_setup_entry,
patch(
"pyotgw.OpenThermGateway.connect", return_value=MINIMAL_STATUS
) as mock_pyotgw_connect,
patch(
"pyotgw.OpenThermGateway.disconnect", return_value=None
) as mock_pyotgw_disconnect,
patch("pyotgw.status.StatusManager._process_updates", return_value=None),
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_ID: "legacy_gateway", CONF_DEVICE: "/dev/ttyUSB1"},
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={CONF_ID: "legacy_gateway", CONF_DEVICE: "/dev/ttyUSB1"},
)

assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "legacy_gateway"
Expand All @@ -105,13 +73,15 @@ async def test_form_import(hass: HomeAssistant) -> None:
CONF_DEVICE: "/dev/ttyUSB1",
CONF_ID: "legacy_gateway",
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_pyotgw_connect.mock_calls) == 1
assert len(mock_pyotgw_disconnect.mock_calls) == 1
assert mock_pyotgw.return_value.connect.await_count == 1
assert mock_pyotgw.return_value.disconnect.await_count == 1


async def test_form_duplicate_entries(hass: HomeAssistant) -> None:
async def test_form_duplicate_entries(
hass: HomeAssistant,
mock_pyotgw: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test duplicate device or id errors."""
flow1 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
Expand All @@ -123,87 +93,76 @@ async def test_form_duplicate_entries(hass: HomeAssistant) -> None:
DOMAIN, context={"source": config_entries.SOURCE_USER}
)

with (
patch(
"homeassistant.components.opentherm_gw.async_setup",
return_value=True,
) as mock_setup,
patch(
"homeassistant.components.opentherm_gw.async_setup_entry",
return_value=True,
) as mock_setup_entry,
patch(
"pyotgw.OpenThermGateway.connect", return_value=MINIMAL_STATUS
) as mock_pyotgw_connect,
patch(
"pyotgw.OpenThermGateway.disconnect", return_value=None
) as mock_pyotgw_disconnect,
patch("pyotgw.status.StatusManager._process_updates", return_value=None),
):
result1 = await hass.config_entries.flow.async_configure(
flow1["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB0"}
)
result2 = await hass.config_entries.flow.async_configure(
flow2["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB1"}
)
result3 = await hass.config_entries.flow.async_configure(
flow3["flow_id"], {CONF_NAME: "Test Entry 2", CONF_DEVICE: "/dev/ttyUSB0"}
)
result1 = await hass.config_entries.flow.async_configure(
flow1["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB0"}
)
assert result1["type"] is FlowResultType.CREATE_ENTRY

result2 = await hass.config_entries.flow.async_configure(
flow2["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB1"}
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "id_exists"}

result3 = await hass.config_entries.flow.async_configure(
flow3["flow_id"], {CONF_NAME: "Test Entry 2", CONF_DEVICE: "/dev/ttyUSB0"}
)
assert result3["type"] is FlowResultType.FORM
assert result3["errors"] == {"base": "already_configured"}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_pyotgw_connect.mock_calls) == 1
assert len(mock_pyotgw_disconnect.mock_calls) == 1

assert mock_pyotgw.return_value.connect.await_count == 1
assert mock_pyotgw.return_value.disconnect.await_count == 1


async def test_form_connection_timeout(hass: HomeAssistant) -> None:
async def test_form_connection_timeout(
hass: HomeAssistant,
mock_pyotgw: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test we handle connection timeout."""
result = await hass.config_entries.flow.async_init(
flow = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)

with (
patch(
"pyotgw.OpenThermGateway.connect", side_effect=(TimeoutError)
) as mock_connect,
patch("pyotgw.status.StatusManager._process_updates", return_value=None),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_NAME: "Test Entry 1", CONF_DEVICE: "socket://192.0.2.254:1234"},
)
mock_pyotgw.return_value.connect.side_effect = TimeoutError

assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "timeout_connect"}
assert len(mock_connect.mock_calls) == 1
result = await hass.config_entries.flow.async_configure(
flow["flow_id"],
{CONF_NAME: "Test Entry 1", CONF_DEVICE: "socket://192.0.2.254:1234"},
)

assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "timeout_connect"}

assert mock_pyotgw.return_value.connect.await_count == 1


async def test_form_connection_error(hass: HomeAssistant) -> None:
async def test_form_connection_error(
hass: HomeAssistant,
mock_pyotgw: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test we handle serial connection error."""
result = await hass.config_entries.flow.async_init(
flow = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)

with (
patch(
"pyotgw.OpenThermGateway.connect", side_effect=(SerialException)
) as mock_connect,
patch("pyotgw.status.StatusManager._process_updates", return_value=None),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB0"}
)
mock_pyotgw.return_value.connect.side_effect = SerialException

assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
assert len(mock_connect.mock_calls) == 1
result = await hass.config_entries.flow.async_configure(
flow["flow_id"], {CONF_NAME: "Test Entry 1", CONF_DEVICE: "/dev/ttyUSB0"}
)

assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
assert mock_pyotgw.return_value.connect.await_count == 1

async def test_options_form(hass: HomeAssistant) -> None:

async def test_options_form(
hass: HomeAssistant,
mock_pyotgw: MagicMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test the options form."""
entry = MockConfigEntry(
domain=DOMAIN,
Expand All @@ -217,23 +176,17 @@ async def test_options_form(hass: HomeAssistant) -> None:
)
entry.add_to_hass(hass)

with (
patch("homeassistant.components.opentherm_gw.async_setup", return_value=True),
patch(
"homeassistant.components.opentherm_gw.async_setup_entry", return_value=True
),
):
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()

result = await hass.config_entries.options.async_init(
flow = await hass.config_entries.options.async_init(
entry.entry_id, context={"source": "test"}, data=None
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert flow["type"] is FlowResultType.FORM
assert flow["step_id"] == "init"

result = await hass.config_entries.options.async_configure(
result["flow_id"],
flow["flow_id"],
user_input={
CONF_FLOOR_TEMP: True,
CONF_READ_PRECISION: PRECISION_HALVES,
Expand All @@ -248,12 +201,12 @@ async def test_options_form(hass: HomeAssistant) -> None:
assert result["data"][CONF_TEMPORARY_OVRD_MODE] is True
assert result["data"][CONF_FLOOR_TEMP] is True

result = await hass.config_entries.options.async_init(
flow = await hass.config_entries.options.async_init(
entry.entry_id, context={"source": "test"}, data=None
)

result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={CONF_READ_PRECISION: 0}
flow["flow_id"], user_input={CONF_READ_PRECISION: 0}
)

assert result["type"] is FlowResultType.CREATE_ENTRY
Expand All @@ -262,12 +215,12 @@ async def test_options_form(hass: HomeAssistant) -> None:
assert result["data"][CONF_TEMPORARY_OVRD_MODE] is True
assert result["data"][CONF_FLOOR_TEMP] is True

result = await hass.config_entries.options.async_init(
flow = await hass.config_entries.options.async_init(
entry.entry_id, context={"source": "test"}, data=None
)

result = await hass.config_entries.options.async_configure(
result["flow_id"],
flow["flow_id"],
user_input={
CONF_FLOOR_TEMP: False,
CONF_READ_PRECISION: PRECISION_TENTHS,
Expand Down
Loading

0 comments on commit 8e03f3a

Please sign in to comment.