"""Test function in __init__.py."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any
from unittest.mock import patch
from aiohttp import ClientWebSocketResponse
from mysensors import BaseSyncGateway
from mysensors.sensor import Sensor
import pytest
from homeassistant.components.mysensors import (
CONF_BAUD_RATE,
CONF_DEVICE,
CONF_GATEWAYS,
CONF_PERSISTENCE,
CONF_PERSISTENCE_FILE,
CONF_RETAIN,
CONF_TCP_PORT,
CONF_VERSION,
DEFAULT_VERSION,
DOMAIN,
)
from homeassistant.components.mysensors.const import (
CONF_GATEWAY_TYPE,
CONF_GATEWAY_TYPE_MQTT,
CONF_GATEWAY_TYPE_SERIAL,
CONF_GATEWAY_TYPE_TCP,
CONF_TOPIC_IN_PREFIX,
CONF_TOPIC_OUT_PREFIX,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
"config, expected_calls, expected_to_succeed, expected_config_entry_data",
[
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "COM5",
CONF_PERSISTENCE_FILE: "bla.json",
CONF_BAUD_RATE: 57600,
CONF_TCP_PORT: 5003,
}
],
CONF_VERSION: "2.3",
CONF_PERSISTENCE: False,
CONF_RETAIN: True,
}
},
1,
True,
[
{
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_SERIAL,
CONF_DEVICE: "COM5",
CONF_PERSISTENCE_FILE: "bla.json",
CONF_BAUD_RATE: 57600,
CONF_VERSION: "2.3",
CONF_TCP_PORT: 5003,
CONF_TOPIC_IN_PREFIX: "",
CONF_TOPIC_OUT_PREFIX: "",
CONF_RETAIN: True,
}
],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "127.0.0.1",
CONF_PERSISTENCE_FILE: "blub.pickle",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 343,
}
],
CONF_VERSION: "2.4",
CONF_PERSISTENCE: False,
CONF_RETAIN: False,
}
},
1,
True,
[
{
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_TCP,
CONF_DEVICE: "127.0.0.1",
CONF_PERSISTENCE_FILE: "blub.pickle",
CONF_TCP_PORT: 343,
CONF_VERSION: "2.4",
CONF_BAUD_RATE: 115200,
CONF_TOPIC_IN_PREFIX: "",
CONF_TOPIC_OUT_PREFIX: "",
CONF_RETAIN: False,
}
],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "127.0.0.1",
}
],
CONF_PERSISTENCE: False,
CONF_RETAIN: False,
}
},
1,
True,
[
{
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_TCP,
CONF_DEVICE: "127.0.0.1",
CONF_TCP_PORT: 5003,
CONF_VERSION: DEFAULT_VERSION,
CONF_BAUD_RATE: 115200,
CONF_TOPIC_IN_PREFIX: "",
CONF_TOPIC_OUT_PREFIX: "",
CONF_RETAIN: False,
CONF_PERSISTENCE_FILE: "mysensors1.pickle",
}
],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "mqtt",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
CONF_TOPIC_IN_PREFIX: "intopic",
CONF_TOPIC_OUT_PREFIX: "outtopic",
}
],
CONF_PERSISTENCE: False,
CONF_RETAIN: False,
}
},
1,
True,
[
{
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_MQTT,
CONF_DEVICE: "mqtt",
CONF_VERSION: DEFAULT_VERSION,
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
CONF_TOPIC_OUT_PREFIX: "outtopic",
CONF_TOPIC_IN_PREFIX: "intopic",
CONF_RETAIN: False,
CONF_PERSISTENCE_FILE: "mysensors1.pickle",
}
],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "mqtt",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
}
],
CONF_PERSISTENCE: False,
CONF_RETAIN: False,
}
},
0,
True,
[{}],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "mqtt",
CONF_PERSISTENCE_FILE: "bla.json",
CONF_TOPIC_OUT_PREFIX: "out",
CONF_TOPIC_IN_PREFIX: "in",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
},
{
CONF_DEVICE: "COM6",
CONF_PERSISTENCE_FILE: "bla2.json",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
},
],
CONF_VERSION: "2.4",
CONF_PERSISTENCE: False,
CONF_RETAIN: False,
}
},
2,
True,
[
{
CONF_DEVICE: "mqtt",
CONF_PERSISTENCE_FILE: "bla.json",
CONF_TOPIC_OUT_PREFIX: "out",
CONF_TOPIC_IN_PREFIX: "in",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
CONF_VERSION: "2.4",
CONF_RETAIN: False,
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_MQTT,
},
{
CONF_DEVICE: "COM6",
CONF_PERSISTENCE_FILE: "bla2.json",
CONF_TOPIC_OUT_PREFIX: "",
CONF_TOPIC_IN_PREFIX: "",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
CONF_VERSION: "2.4",
CONF_RETAIN: False,
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_SERIAL,
},
],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "mqtt",
CONF_PERSISTENCE_FILE: "bla.json",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
},
{
CONF_DEVICE: "COM6",
CONF_PERSISTENCE_FILE: "bla.json",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
},
],
CONF_VERSION: "2.4",
CONF_PERSISTENCE: False,
CONF_RETAIN: False,
}
},
0,
False,
[{}],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "COMx",
CONF_PERSISTENCE_FILE: "bla.json",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
},
],
CONF_VERSION: "2.4",
CONF_PERSISTENCE: False,
CONF_RETAIN: False,
}
},
0,
True,
[{}],
),
(
{
DOMAIN: {
CONF_GATEWAYS: [
{
CONF_DEVICE: "COM1",
},
{
CONF_DEVICE: "COM2",
},
],
}
},
2,
True,
[
{
CONF_DEVICE: "COM1",
CONF_PERSISTENCE_FILE: "mysensors1.pickle",
CONF_TOPIC_OUT_PREFIX: "",
CONF_TOPIC_IN_PREFIX: "",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
CONF_VERSION: "1.4",
CONF_RETAIN: True,
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_SERIAL,
},
{
CONF_DEVICE: "COM2",
CONF_PERSISTENCE_FILE: "mysensors2.pickle",
CONF_TOPIC_OUT_PREFIX: "",
CONF_TOPIC_IN_PREFIX: "",
CONF_BAUD_RATE: 115200,
CONF_TCP_PORT: 5003,
CONF_VERSION: "1.4",
CONF_RETAIN: True,
CONF_GATEWAY_TYPE: CONF_GATEWAY_TYPE_SERIAL,
},
],
),
],
)
async def test_import(
hass: HomeAssistant,
mqtt: None,
config: ConfigType,
expected_calls: int,
expected_to_succeed: bool,
expected_config_entry_data: list[dict[str, Any]],
) -> None:
"""Test importing a gateway."""
with patch("sys.platform", "win32"), patch(
"homeassistant.components.mysensors.config_flow.try_connect", return_value=True
), patch(
"homeassistant.components.mysensors.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result = await async_setup_component(hass, DOMAIN, config)
assert result == expected_to_succeed
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == expected_calls
for idx in range(expected_calls):
config_entry = mock_setup_entry.mock_calls[idx][1][1]
expected_persistence_file = expected_config_entry_data[idx].pop(
CONF_PERSISTENCE_FILE
)
expected_persistence_path = hass.config.path(expected_persistence_file)
config_entry_data = dict(config_entry.data)
persistence_path = config_entry_data.pop(CONF_PERSISTENCE_FILE)
assert persistence_path == expected_persistence_path
assert config_entry_data == expected_config_entry_data[idx]
async def test_remove_config_entry_device(
hass: HomeAssistant,
gps_sensor: Sensor,
integration: MockConfigEntry,
gateway: BaseSyncGateway,
hass_ws_client: Callable[[HomeAssistant], Awaitable[ClientWebSocketResponse]],
) -> None:
"""Test that a device can be removed ok."""
entity_id = "sensor.gps_sensor_1_1"
node_id = 1
config_entry = integration
assert await async_setup_component(hass, "config", {})
await hass.async_block_till_done()
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, f"{config_entry.entry_id}-{node_id}")}
)
entity_registry = er.async_get(hass)
state = hass.states.get(entity_id)
assert gateway.sensors
assert gateway.sensors[node_id]
assert device_entry
assert state
client = await hass_ws_client(hass)
await client.send_json(
{
"id": 5,
"type": "config/device_registry/remove_config_entry",
"config_entry_id": config_entry.entry_id,
"device_id": device_entry.id,
}
)
response = await client.receive_json()
assert response["success"]
await hass.async_block_till_done()
assert node_id not in gateway.sensors
assert gateway.tasks.persistence.need_save is True
assert not device_registry.async_get_device(
identifiers={(DOMAIN, f"{config_entry.entry_id}-1")}
)
assert not entity_registry.async_get(entity_id)
assert not hass.states.get(entity_id)