Refactor dsmr tests (#39724)

This commit is contained in:
Rob Bierbooms 2020-10-04 14:27:22 +02:00 committed by GitHub
parent fd065e7aea
commit e97792adf0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 115 additions and 146 deletions

View File

@ -0,0 +1,71 @@
"""Common test tools."""
import asyncio
from dsmr_parser.clients.protocol import DSMRProtocol
from dsmr_parser.obis_references import EQUIPMENT_IDENTIFIER, EQUIPMENT_IDENTIFIER_GAS
from dsmr_parser.objects import CosemObject
import pytest
from tests.async_mock import MagicMock, patch
@pytest.fixture
async def dsmr_connection_fixture(hass):
"""Fixture that mocks serial connection."""
transport = MagicMock(spec=asyncio.Transport)
protocol = MagicMock(spec=DSMRProtocol)
async def connection_factory(*args, **kwargs):
"""Return mocked out Asyncio classes."""
return (transport, protocol)
connection_factory = MagicMock(wraps=connection_factory)
with patch(
"homeassistant.components.dsmr.sensor.create_dsmr_reader", connection_factory
), patch(
"homeassistant.components.dsmr.sensor.create_tcp_dsmr_reader",
connection_factory,
):
yield (connection_factory, transport, protocol)
@pytest.fixture
async def dsmr_connection_send_validate_fixture(hass):
"""Fixture that mocks serial connection."""
transport = MagicMock(spec=asyncio.Transport)
protocol = MagicMock(spec=DSMRProtocol)
async def connection_factory(*args, **kwargs):
"""Return mocked out Asyncio classes."""
return (transport, protocol)
connection_factory = MagicMock(wraps=connection_factory)
protocol.telegram = {
EQUIPMENT_IDENTIFIER: CosemObject([{"value": "12345678", "unit": ""}]),
EQUIPMENT_IDENTIFIER_GAS: CosemObject([{"value": "123456789", "unit": ""}]),
}
async def wait_closed():
if isinstance(connection_factory.call_args_list[0][0][2], str):
# TCP
telegram_callback = connection_factory.call_args_list[0][0][3]
else:
# Serial
telegram_callback = connection_factory.call_args_list[0][0][2]
telegram_callback(protocol.telegram)
protocol.wait_closed = wait_closed
with patch(
"homeassistant.components.dsmr.config_flow.create_dsmr_reader",
connection_factory,
), patch(
"homeassistant.components.dsmr.config_flow.create_tcp_dsmr_reader",
connection_factory,
):
yield (connection_factory, transport, protocol)

View File

@ -2,64 +2,18 @@
import asyncio import asyncio
from itertools import chain, repeat from itertools import chain, repeat
from dsmr_parser.clients.protocol import DSMRProtocol
from dsmr_parser.obis_references import EQUIPMENT_IDENTIFIER, EQUIPMENT_IDENTIFIER_GAS
from dsmr_parser.objects import CosemObject
import pytest
import serial import serial
from homeassistant import config_entries, setup from homeassistant import config_entries, setup
from homeassistant.components.dsmr import DOMAIN from homeassistant.components.dsmr import DOMAIN
from tests.async_mock import DEFAULT, AsyncMock, Mock, patch from tests.async_mock import DEFAULT, AsyncMock, patch
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
SERIAL_DATA = {"serial_id": "12345678", "serial_id_gas": "123456789"} SERIAL_DATA = {"serial_id": "12345678", "serial_id_gas": "123456789"}
@pytest.fixture async def test_import_usb(hass, dsmr_connection_send_validate_fixture):
def mock_connection_factory(monkeypatch):
"""Mock the create functions for serial and TCP Asyncio connections."""
transport = Mock(spec=asyncio.Transport)
protocol = Mock(spec=DSMRProtocol)
async def connection_factory(*args, **kwargs):
"""Return mocked out Asyncio classes."""
return (transport, protocol)
connection_factory = Mock(wraps=connection_factory)
# apply the mock to both connection factories
monkeypatch.setattr(
"homeassistant.components.dsmr.config_flow.create_dsmr_reader",
connection_factory,
)
monkeypatch.setattr(
"homeassistant.components.dsmr.config_flow.create_tcp_dsmr_reader",
connection_factory,
)
protocol.telegram = {
EQUIPMENT_IDENTIFIER: CosemObject([{"value": "12345678", "unit": ""}]),
EQUIPMENT_IDENTIFIER_GAS: CosemObject([{"value": "123456789", "unit": ""}]),
}
async def wait_closed():
if isinstance(connection_factory.call_args_list[0][0][2], str):
# TCP
telegram_callback = connection_factory.call_args_list[0][0][3]
else:
# Serial
telegram_callback = connection_factory.call_args_list[0][0][2]
telegram_callback(protocol.telegram)
protocol.wait_closed = wait_closed
return connection_factory, transport, protocol
async def test_import_usb(hass, mock_connection_factory):
"""Test we can import.""" """Test we can import."""
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})
@ -82,9 +36,11 @@ async def test_import_usb(hass, mock_connection_factory):
assert result["data"] == {**entry_data, **SERIAL_DATA} assert result["data"] == {**entry_data, **SERIAL_DATA}
async def test_import_usb_failed_connection(hass, monkeypatch, mock_connection_factory): async def test_import_usb_failed_connection(
hass, dsmr_connection_send_validate_fixture
):
"""Test we can import.""" """Test we can import."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})
@ -101,12 +57,12 @@ async def test_import_usb_failed_connection(hass, monkeypatch, mock_connection_f
side_effect=chain([serial.serialutil.SerialException], repeat(DEFAULT)), side_effect=chain([serial.serialutil.SerialException], repeat(DEFAULT)),
) )
monkeypatch.setattr( with patch(
"homeassistant.components.dsmr.async_setup_entry", return_value=True
), patch(
"homeassistant.components.dsmr.config_flow.create_dsmr_reader", "homeassistant.components.dsmr.config_flow.create_dsmr_reader",
first_fail_connection_factory, first_fail_connection_factory,
) ):
with patch("homeassistant.components.dsmr.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN,
context={"source": config_entries.SOURCE_IMPORT}, context={"source": config_entries.SOURCE_IMPORT},
@ -117,9 +73,9 @@ async def test_import_usb_failed_connection(hass, monkeypatch, mock_connection_f
assert result["reason"] == "cannot_connect" assert result["reason"] == "cannot_connect"
async def test_import_usb_no_data(hass, monkeypatch, mock_connection_factory): async def test_import_usb_no_data(hass, dsmr_connection_send_validate_fixture):
"""Test we can import.""" """Test we can import."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})
@ -149,9 +105,9 @@ async def test_import_usb_no_data(hass, monkeypatch, mock_connection_factory):
assert result["reason"] == "cannot_communicate" assert result["reason"] == "cannot_communicate"
async def test_import_usb_wrong_telegram(hass, mock_connection_factory): async def test_import_usb_wrong_telegram(hass, dsmr_connection_send_validate_fixture):
"""Test we can import.""" """Test we can import."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_send_validate_fixture
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})
@ -175,7 +131,7 @@ async def test_import_usb_wrong_telegram(hass, mock_connection_factory):
assert result["reason"] == "cannot_communicate" assert result["reason"] == "cannot_communicate"
async def test_import_network(hass, mock_connection_factory): async def test_import_network(hass, dsmr_connection_send_validate_fixture):
"""Test we can import from network.""" """Test we can import from network."""
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})
@ -199,7 +155,7 @@ async def test_import_network(hass, mock_connection_factory):
assert result["data"] == {**entry_data, **SERIAL_DATA} assert result["data"] == {**entry_data, **SERIAL_DATA}
async def test_import_update(hass, mock_connection_factory): async def test_import_update(hass, dsmr_connection_send_validate_fixture):
"""Test we can import.""" """Test we can import."""
await setup.async_setup_component(hass, "persistent_notification", {}) await setup.async_setup_component(hass, "persistent_notification", {})

View File

@ -10,46 +10,17 @@ import datetime
from decimal import Decimal from decimal import Decimal
from itertools import chain, repeat from itertools import chain, repeat
import pytest
from homeassistant.components.dsmr.const import DOMAIN from homeassistant.components.dsmr.const import DOMAIN
from homeassistant.components.dsmr.sensor import DerivativeDSMREntity from homeassistant.components.dsmr.sensor import DerivativeDSMREntity
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.const import ENERGY_KILO_WATT_HOUR, TIME_HOURS, VOLUME_CUBIC_METERS from homeassistant.const import ENERGY_KILO_WATT_HOUR, TIME_HOURS, VOLUME_CUBIC_METERS
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
import tests.async_mock from tests.async_mock import DEFAULT, MagicMock
from tests.async_mock import DEFAULT, MagicMock, Mock
from tests.common import MockConfigEntry, patch from tests.common import MockConfigEntry, patch
@pytest.fixture async def test_setup_platform(hass, dsmr_connection_fixture):
def mock_connection_factory(monkeypatch):
"""Mock the create functions for serial and TCP Asyncio connections."""
from dsmr_parser.clients.protocol import DSMRProtocol
transport = tests.async_mock.Mock(spec=asyncio.Transport)
protocol = tests.async_mock.Mock(spec=DSMRProtocol)
async def connection_factory(*args, **kwargs):
"""Return mocked out Asyncio classes."""
return (transport, protocol)
connection_factory = Mock(wraps=connection_factory)
# apply the mock to both connection factories
monkeypatch.setattr(
"homeassistant.components.dsmr.sensor.create_dsmr_reader", connection_factory
)
monkeypatch.setattr(
"homeassistant.components.dsmr.sensor.create_tcp_dsmr_reader",
connection_factory,
)
return connection_factory, transport, protocol
async def test_setup_platform(hass, mock_connection_factory):
"""Test setup of platform.""" """Test setup of platform."""
async_add_entities = MagicMock() async_add_entities = MagicMock()
@ -87,9 +58,9 @@ async def test_setup_platform(hass, mock_connection_factory):
assert entry.data == {**entry_data, **serial_data} assert entry.data == {**entry_data, **serial_data}
async def test_default_setup(hass, mock_connection_factory): async def test_default_setup(hass, dsmr_connection_fixture):
"""Test the default setup.""" """Test the default setup."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import ( from dsmr_parser.obis_references import (
CURRENT_ELECTRICITY_USAGE, CURRENT_ELECTRICITY_USAGE,
@ -157,10 +128,6 @@ async def test_default_setup(hass, mock_connection_factory):
assert gas_consumption.state == "745.695" assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == "not_loaded"
async def test_derivative(): async def test_derivative():
"""Test calculation of derivative value.""" """Test calculation of derivative value."""
@ -202,9 +169,9 @@ async def test_derivative():
assert entity.unit_of_measurement == f"{VOLUME_CUBIC_METERS}/{TIME_HOURS}" assert entity.unit_of_measurement == f"{VOLUME_CUBIC_METERS}/{TIME_HOURS}"
async def test_v4_meter(hass, mock_connection_factory): async def test_v4_meter(hass, dsmr_connection_fixture):
"""Test if v4 meter is correctly parsed.""" """Test if v4 meter is correctly parsed."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import ( from dsmr_parser.obis_references import (
ELECTRICITY_ACTIVE_TARIFF, ELECTRICITY_ACTIVE_TARIFF,
@ -256,14 +223,10 @@ async def test_v4_meter(hass, mock_connection_factory):
assert gas_consumption.state == "745.695" assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == "not_loaded" async def test_v5_meter(hass, dsmr_connection_fixture):
async def test_v5_meter(hass, mock_connection_factory):
"""Test if v5 meter is correctly parsed.""" """Test if v5 meter is correctly parsed."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import ( from dsmr_parser.obis_references import (
ELECTRICITY_ACTIVE_TARIFF, ELECTRICITY_ACTIVE_TARIFF,
@ -315,14 +278,10 @@ async def test_v5_meter(hass, mock_connection_factory):
assert gas_consumption.state == "745.695" assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == "not_loaded" async def test_belgian_meter(hass, dsmr_connection_fixture):
async def test_belgian_meter(hass, mock_connection_factory):
"""Test if Belgian meter is correctly parsed.""" """Test if Belgian meter is correctly parsed."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import ( from dsmr_parser.obis_references import (
BELGIUM_HOURLY_GAS_METER_READING, BELGIUM_HOURLY_GAS_METER_READING,
@ -374,14 +333,10 @@ async def test_belgian_meter(hass, mock_connection_factory):
assert gas_consumption.state == "745.695" assert gas_consumption.state == "745.695"
assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS assert gas_consumption.attributes.get("unit_of_measurement") == VOLUME_CUBIC_METERS
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == "not_loaded" async def test_belgian_meter_low(hass, dsmr_connection_fixture):
async def test_belgian_meter_low(hass, mock_connection_factory):
"""Test if Belgian meter is correctly parsed.""" """Test if Belgian meter is correctly parsed."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
from dsmr_parser.obis_references import ELECTRICITY_ACTIVE_TARIFF from dsmr_parser.obis_references import ELECTRICITY_ACTIVE_TARIFF
from dsmr_parser.objects import CosemObject from dsmr_parser.objects import CosemObject
@ -417,14 +372,10 @@ async def test_belgian_meter_low(hass, mock_connection_factory):
assert power_tariff.state == "low" assert power_tariff.state == "low"
assert power_tariff.attributes.get("unit_of_measurement") == "" assert power_tariff.attributes.get("unit_of_measurement") == ""
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == "not_loaded" async def test_tcp(hass, dsmr_connection_fixture):
async def test_tcp(hass, mock_connection_factory):
"""If proper config provided TCP connection should be made.""" """If proper config provided TCP connection should be made."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
entry_data = { entry_data = {
"host": "localhost", "host": "localhost",
@ -446,14 +397,10 @@ async def test_tcp(hass, mock_connection_factory):
assert connection_factory.call_args_list[0][0][0] == "localhost" assert connection_factory.call_args_list[0][0][0] == "localhost"
assert connection_factory.call_args_list[0][0][1] == "1234" assert connection_factory.call_args_list[0][0][1] == "1234"
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == "not_loaded" async def test_connection_errors_retry(hass, dsmr_connection_fixture):
async def test_connection_errors_retry(hass, monkeypatch, mock_connection_factory):
"""Connection should be retried on error during setup.""" """Connection should be retried on error during setup."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
entry_data = { entry_data = {
"port": "/dev/ttyUSB0", "port": "/dev/ttyUSB0",
@ -463,37 +410,32 @@ async def test_connection_errors_retry(hass, monkeypatch, mock_connection_factor
} }
# override the mock to have it fail the first time and succeed after # override the mock to have it fail the first time and succeed after
first_fail_connection_factory = tests.async_mock.AsyncMock( first_fail_connection_factory = MagicMock(
return_value=(transport, protocol), return_value=(transport, protocol),
side_effect=chain([TimeoutError], repeat(DEFAULT)), side_effect=chain([TimeoutError], repeat(DEFAULT)),
) )
monkeypatch.setattr(
"homeassistant.components.dsmr.sensor.create_dsmr_reader",
first_fail_connection_factory,
)
mock_entry = MockConfigEntry( mock_entry = MockConfigEntry(
domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data domain="dsmr", unique_id="/dev/ttyUSB0", data=entry_data
) )
mock_entry.add_to_hass(hass) mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id) with patch(
await hass.async_block_till_done() "homeassistant.components.dsmr.sensor.create_dsmr_reader",
first_fail_connection_factory,
):
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
# wait for sleep to resolve # wait for sleep to resolve
await hass.async_block_till_done() await hass.async_block_till_done()
assert first_fail_connection_factory.call_count >= 2, "connecting not retried" assert first_fail_connection_factory.call_count >= 2, "connecting not retried"
await hass.config_entries.async_unload(mock_entry.entry_id)
assert mock_entry.state == "not_loaded"
async def test_reconnect(hass, monkeypatch, mock_connection_factory): async def test_reconnect(hass, dsmr_connection_fixture):
"""If transport disconnects, the connection should be retried.""" """If transport disconnects, the connection should be retried."""
(connection_factory, transport, protocol) = mock_connection_factory (connection_factory, transport, protocol) = dsmr_connection_fixture
entry_data = { entry_data = {
"port": "/dev/ttyUSB0", "port": "/dev/ttyUSB0",