jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 14:17:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Parametrize DSMR serial config flow tests (#103524)

Browse Source
This commit is contained in:
Joost Lekkerkerker 2023-11-07 14:13:57 +01:00 committed by GitHub
parent c13744f4cf
commit b636a4d5cf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 68 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

255
tests/components/dsmr/test_config_flow.py
View File

@ -2,14 +2,17 @@
import asyncio import asyncio
from itertools import chain, repeat from itertools import chain, repeat
import os import os
from typing import Any
from unittest.mock import DEFAULT, AsyncMock, MagicMock, patch, sentinel from unittest.mock import DEFAULT, AsyncMock, MagicMock, patch, sentinel
import pytest
import serial import serial
import serial.tools.list_ports import serial.tools.list_ports
from homeassistant import config_entries, data_entry_flow from homeassistant import config_entries, data_entry_flow
from homeassistant.components.dsmr import DOMAIN, config_flow from homeassistant.components.dsmr import DOMAIN, config_flow
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@ -123,9 +126,68 @@ async def test_setup_network_rfxtrx(
assert result["data"] == {**entry_data, **SERIAL_DATA} assert result["data"] == {**entry_data, **SERIAL_DATA}
@pytest.mark.parametrize(
("version", "entry_data"),
[
(
"2.2",
{
"port": "/dev/ttyUSB1234",
"dsmr_version": "2.2",
"protocol": "dsmr_protocol",
"serial_id": "12345678",
"serial_id_gas": "123456789",
},
),
(
"5B",
{
"port": "/dev/ttyUSB1234",
"dsmr_version": "5B",
"protocol": "dsmr_protocol",
"serial_id": "12345678",
"serial_id_gas": "123456789",
},
),
(
"5L",
{
"port": "/dev/ttyUSB1234",
"dsmr_version": "5L",
"protocol": "dsmr_protocol",
"serial_id": "12345678",
"serial_id_gas": "123456789",
},
),
(
"5S",
{
"port": "/dev/ttyUSB1234",
"dsmr_version": "5S",
"protocol": "dsmr_protocol",
"serial_id": None,
"serial_id_gas": None,
},
),
(
"Q3D",
{
"port": "/dev/ttyUSB1234",
"dsmr_version": "Q3D",
"protocol": "dsmr_protocol",
"serial_id": "12345678",
"serial_id_gas": None,
},
),
],
)
@patch("serial.tools.list_ports.comports", return_value=[com_port()]) @patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_serial( async def test_setup_serial(
com_mock, hass: HomeAssistant, dsmr_connection_send_validate_fixture com_mock,
hass: HomeAssistant,
dsmr_connection_send_validate_fixture,
version: str,
entry_data: dict[str, Any],
) -> None: ) -> None:
"""Test we can setup serial.""" """Test we can setup serial."""
port = com_port() port = com_port()
@ -134,7 +196,7 @@ async def test_setup_serial(
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
) )
assert result["type"] == "form" assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "user" assert result["step_id"] == "user"
assert result["errors"] is None assert result["errors"] is None
@ -143,26 +205,20 @@ async def test_setup_serial(
{"type": "Serial"}, {"type": "Serial"},
) )
assert result["type"] == "form" assert result["type"] == FlowResultType.FORM
assert result["step_id"] == "setup_serial" assert result["step_id"] == "setup_serial"
assert result["errors"] == {} assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True): with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure( result = await hass.config_entries.flow.async_configure(
result["flow_id"], result["flow_id"],
{"port": port.device, "dsmr_version": "2.2"}, {"port": port.device, "dsmr_version": version},
) )
await hass.async_block_till_done() await hass.async_block_till_done()
entry_data = { assert result["type"] == FlowResultType.CREATE_ENTRY
"port": port.device,
"dsmr_version": "2.2",
"protocol": "dsmr_protocol",
}
assert result["type"] == "create_entry"
assert result["title"] == port.device assert result["title"] == port.device
assert result["data"] == {**entry_data, **SERIAL_DATA} assert result["data"] == entry_data
@patch("serial.tools.list_ports.comports", return_value=[com_port()]) @patch("serial.tools.list_ports.comports", return_value=[com_port()])
@ -215,181 +271,6 @@ async def test_setup_serial_rfxtrx(
assert result["data"] == {**entry_data, **SERIAL_DATA} assert result["data"] == {**entry_data, **SERIAL_DATA}
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_5B(
com_mock, hass: HomeAssistant, dsmr_connection_send_validate_fixture
) -> None:
"""Test we can setup serial."""
port = com_port()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"port": port.device, "dsmr_version": "5B"},
)
await hass.async_block_till_done()
entry_data = {
"port": port.device,
"dsmr_version": "5B",
"protocol": "dsmr_protocol",
"serial_id": "12345678",
"serial_id_gas": "123456789",
}
assert result["type"] == "create_entry"
assert result["title"] == port.device
assert result["data"] == entry_data
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_5L(
com_mock, hass: HomeAssistant, dsmr_connection_send_validate_fixture
) -> None:
"""Test we can setup serial."""
port = com_port()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"port": port.device, "dsmr_version": "5L"},
)
await hass.async_block_till_done()
entry_data = {
"port": port.device,
"dsmr_version": "5L",
"protocol": "dsmr_protocol",
"serial_id": "12345678",
"serial_id_gas": "123456789",
}
assert result["type"] == "create_entry"
assert result["title"] == port.device
assert result["data"] == entry_data
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_5S(
com_mock, hass: HomeAssistant, dsmr_connection_send_validate_fixture
) -> None:
"""Test we can setup serial."""
port = com_port()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {"port": port.device, "dsmr_version": "5S"}
)
await hass.async_block_till_done()
entry_data = {
"port": port.device,
"dsmr_version": "5S",
"protocol": "dsmr_protocol",
"serial_id": None,
"serial_id_gas": None,
}
assert result["type"] == "create_entry"
assert result["title"] == port.device
assert result["data"] == entry_data
@patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_Q3D(
com_mock, hass: HomeAssistant, dsmr_connection_send_validate_fixture
) -> None:
"""Test we can setup serial."""
port = com_port()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == "form"
assert result["step_id"] == "user"
assert result["errors"] is None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"type": "Serial"},
)
assert result["type"] == "form"
assert result["step_id"] == "setup_serial"
assert result["errors"] == {}
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"port": port.device, "dsmr_version": "Q3D"},
)
await hass.async_block_till_done()
entry_data = {
"port": port.device,
"dsmr_version": "Q3D",
"protocol": "dsmr_protocol",
"serial_id": "12345678",
"serial_id_gas": None,
}
assert result["type"] == "create_entry"
assert result["title"] == port.device
assert result["data"] == entry_data
@patch("serial.tools.list_ports.comports", return_value=[com_port()]) @patch("serial.tools.list_ports.comports", return_value=[com_port()])
async def test_setup_serial_manual( async def test_setup_serial_manual(
com_mock, hass: HomeAssistant, dsmr_connection_send_validate_fixture com_mock, hass: HomeAssistant, dsmr_connection_send_validate_fixture