jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-24 05:37:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Check for duplicate host/port and integration name in modbus (#54664)

Browse Source 
* Check for duplicate host/port and integration name.

* Change to use set().

* Please CI.

* Add basic tests.
This commit is contained in:
jan iversen 2021-08-25 12:29:00 +02:00 committed by GitHub
parent a23f4dac62
commit 7df8d0c973
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 120 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/modbus/__init__.py
View File

@ -121,6 +121,7 @@ from .const import (
from .modbus import ModbusHub, async_modbus_setup
from .validators import (
duplicate_entity_validator,
duplicate_modbus_validator,
number_validator,
scan_interval_validator,
struct_validator,
@ -338,6 +339,7 @@ CONFIG_SCHEMA = vol.Schema(
cv.ensure_list,
scan_interval_validator,
duplicate_entity_validator,
duplicate_modbus_validator,
[
vol.Any(SERIAL_SCHEMA, ETHERNET_SCHEMA),
],

31
homeassistant/components/modbus/validators.py
View File

@ -11,11 +11,14 @@ import voluptuous as vol
from homeassistant.const import (
CONF_ADDRESS,
CONF_COUNT,
CONF_HOST,
CONF_NAME,
CONF_PORT,
CONF_SCAN_INTERVAL,
CONF_SLAVE,
CONF_STRUCTURE,
CONF_TIMEOUT,
CONF_TYPE,
)
from .const import (
@ -37,8 +40,10 @@ from .const import (
DATA_TYPE_UINT16,
DATA_TYPE_UINT32,
DATA_TYPE_UINT64,
DEFAULT_HUB,
DEFAULT_SCAN_INTERVAL,
PLATFORMS,
SERIAL,
)
_LOGGER = logging.getLogger(__name__)
@ -221,5 +226,29 @@ def duplicate_entity_validator(config: dict) -> dict:
for i in reversed(errors):
del config[hub_index][conf_key][i]
return config
def duplicate_modbus_validator(config: list) -> list:
"""Control modbus connection for duplicates."""
hosts: set[str] = set()
names: set[str] = set()
errors = []
for index, hub in enumerate(config):
name = hub.get(CONF_NAME, DEFAULT_HUB)
host = hub[CONF_PORT] if hub[CONF_TYPE] == SERIAL else hub[CONF_HOST]
if host in hosts:
err = f"Modbus {name}  contains duplicate host/port {host}, not loaded!"
_LOGGER.warning(err)
errors.append(index)
elif name in names:
err = f"Modbus {name}  is duplicate, second entry not loaded!"
_LOGGER.warning(err)
errors.append(index)
else:
hosts.add(host)
names.add(name)
for i in reversed(errors):
del config[i]
return config

88
tests/components/modbus/test_init.py
View File

@ -59,6 +59,8 @@ from homeassistant.components.modbus.const import (
UDP,
)
from homeassistant.components.modbus.validators import (
duplicate_entity_validator,
duplicate_modbus_validator,
number_validator,
struct_validator,
)
@ -202,6 +204,92 @@ async def test_exception_struct_validator(do_config):
pytest.fail("struct_validator missing exception")
@pytest.mark.parametrize(
"do_config",
[
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
},
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST + "2",
CONF_PORT: TEST_PORT_TCP,
},
],
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
},
{
CONF_NAME: TEST_MODBUS_NAME + "2",
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
},
],
],
)
async def test_duplicate_modbus_validator(do_config):
"""Test duplicate modbus validator."""
duplicate_modbus_validator(do_config)
assert len(do_config) == 1
@pytest.mark.parametrize(
"do_config",
[
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
},
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 119,
},
],
}
],
[
{
CONF_NAME: TEST_MODBUS_NAME,
CONF_TYPE: TCP,
CONF_HOST: TEST_MODBUS_HOST,
CONF_PORT: TEST_PORT_TCP,
CONF_SENSORS: [
{
CONF_NAME: TEST_ENTITY_NAME,
CONF_ADDRESS: 117,
},
{
CONF_NAME: TEST_ENTITY_NAME + "2",
CONF_ADDRESS: 117,
},
],
}
],
],
)
async def test_duplicate_entity_validator(do_config):
"""Test duplicate entity validator."""
duplicate_entity_validator(do_config)
assert len(do_config[0][CONF_SENSORS]) == 1
@pytest.mark.parametrize(
"do_config",
[