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

Address late KNX flow tests review (#82975)

Browse Source 
* KNX flow tests review

* patch out async_setup_entry in options-flow tests

* remove unneeded hass.async_block_till_done()

* Update test_config_flow.py

* autouse setup mock

* patch out async_setup too

* rename fixture according to pytest docs

* test call count to async_setup_entry

* dict access instead of .get()
This commit is contained in:
Matthias Alphart 2022-11-30 18:54:06 +01:00 committed by GitHub
parent 949dede16b
commit a43d944309
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 206 additions and 253 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

459
tests/components/knx/test_config_flow.py
View File

@ -1,5 +1,5 @@
"""Test the KNX config flow.""" """Test the KNX config flow."""
from unittest.mock import Mock, patch from unittest.mock import patch
import pytest import pytest
from xknx.exceptions.exception import InvalidSecureConfiguration from xknx.exceptions.exception import InvalidSecureConfiguration
@ -45,6 +45,15 @@ from homeassistant.data_entry_flow import FlowResult, FlowResultType
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@pytest.fixture(name="knx_setup", autouse=True)
def fixture_knx_setup():
"""Mock KNX entry setup."""
with patch("homeassistant.components.knx.async_setup", return_value=True), patch(
"homeassistant.components.knx.async_setup_entry", return_value=True
) as mock_async_setup_entry:
yield mock_async_setup_entry
def _gateway_descriptor( def _gateway_descriptor(
ip: str, ip: str,
port: int, port: int,
@ -100,7 +109,9 @@ async def test_user_single_instance(hass):
"homeassistant.components.knx.config_flow.GatewayScanner", "homeassistant.components.knx.config_flow.GatewayScanner",
return_value=GatewayScannerMock(), return_value=GatewayScannerMock(),
) )
async def test_routing_setup(gateway_scanner_mock, hass: HomeAssistant) -> None: async def test_routing_setup(
gateway_scanner_mock, hass: HomeAssistant, knx_setup
) -> None:
"""Test routing setup.""" """Test routing setup."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER} DOMAIN, context={"source": config_entries.SOURCE_USER}
@ -114,36 +125,30 @@ async def test_routing_setup(gateway_scanner_mock, hass: HomeAssistant) -> None:
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "routing" assert result2["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result2["errors"] == {"base": "no_router_discovered"}
with patch( result3 = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", result2["flow_id"],
return_value=True, {
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Routing as 1.1.110"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675, CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
} },
)
assert len(mock_setup_entry.mock_calls) == 1 await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Routing as 1.1.110"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_LOCAL_IP: None,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
}
knx_setup.assert_called_once()
@patch( @patch(
@ -151,7 +156,7 @@ async def test_routing_setup(gateway_scanner_mock, hass: HomeAssistant) -> None:
return_value=GatewayScannerMock(), return_value=GatewayScannerMock(),
) )
async def test_routing_setup_advanced( async def test_routing_setup_advanced(
gateway_scanner_mock, hass: HomeAssistant gateway_scanner_mock, hass: HomeAssistant, knx_setup
) -> None: ) -> None:
"""Test routing setup with advanced options.""" """Test routing setup with advanced options."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -170,7 +175,6 @@ async def test_routing_setup_advanced(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "routing" assert result2["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result2["errors"] == {"base": "no_router_discovered"}
@ -185,7 +189,6 @@ async def test_routing_setup_advanced(
CONF_KNX_LOCAL_IP: "no_local_ip", CONF_KNX_LOCAL_IP: "no_local_ip",
}, },
) )
await hass.async_block_till_done()
assert result_invalid_input["type"] == FlowResultType.FORM assert result_invalid_input["type"] == FlowResultType.FORM
assert result_invalid_input["step_id"] == "routing" assert result_invalid_input["step_id"] == "routing"
assert result_invalid_input["errors"] == { assert result_invalid_input["errors"] == {
@ -196,32 +199,27 @@ async def test_routing_setup_advanced(
} }
# valid user input # valid user input
with patch( result3 = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", result2["flow_id"],
return_value=True, {
) as mock_setup_entry:
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
CONF_KNX_LOCAL_IP: "192.168.1.112",
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Routing as 1.1.110"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP, CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675, CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110", CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
} CONF_KNX_LOCAL_IP: "192.168.1.112",
},
assert len(mock_setup_entry.mock_calls) == 1 )
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Routing as 1.1.110"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
CONF_KNX_MCAST_GRP: DEFAULT_MCAST_GRP,
CONF_KNX_MCAST_PORT: 3675,
CONF_KNX_LOCAL_IP: "192.168.1.112",
CONF_KNX_INDIVIDUAL_ADDRESS: "1.1.110",
}
knx_setup.assert_called_once()
@patch( @patch(
@ -229,7 +227,7 @@ async def test_routing_setup_advanced(
return_value=GatewayScannerMock(), return_value=GatewayScannerMock(),
) )
async def test_routing_secure_manual_setup( async def test_routing_secure_manual_setup(
gateway_scanner_mock, hass: HomeAssistant gateway_scanner_mock, hass: HomeAssistant, knx_setup
) -> None: ) -> None:
"""Test routing secure setup with manual key config.""" """Test routing secure setup with manual key config."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -244,7 +242,6 @@ async def test_routing_secure_manual_setup(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "routing" assert result2["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result2["errors"] == {"base": "no_router_discovered"}
@ -291,28 +288,24 @@ async def test_routing_secure_manual_setup(
assert result_invalid_key2["step_id"] == "secure_routing_manual" assert result_invalid_key2["step_id"] == "secure_routing_manual"
assert result_invalid_key2["errors"] == {"backbone_key": "invalid_backbone_key"} assert result_invalid_key2["errors"] == {"backbone_key": "invalid_backbone_key"}
with patch( secure_routing_manual = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", result_invalid_key2["flow_id"],
return_value=True, {
) as mock_setup_entry:
secure_routing_manual = await hass.config_entries.flow.async_configure(
result_invalid_key2["flow_id"],
{
CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44",
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000,
},
)
await hass.async_block_till_done()
assert secure_routing_manual["type"] == FlowResultType.CREATE_ENTRY
assert secure_routing_manual["title"] == "Secure Routing as 0.0.123"
assert secure_routing_manual["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE,
CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44", CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44",
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", },
} )
assert len(mock_setup_entry.mock_calls) == 1 await hass.async_block_till_done()
assert secure_routing_manual["type"] == FlowResultType.CREATE_ENTRY
assert secure_routing_manual["title"] == "Secure Routing as 0.0.123"
assert secure_routing_manual["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE,
CONF_KNX_ROUTING_BACKBONE_KEY: "bbaacc44bbaacc44bbaacc44bbaacc44",
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: 2000,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123",
}
knx_setup.assert_called_once()
@patch( @patch(
@ -320,7 +313,7 @@ async def test_routing_secure_manual_setup(
return_value=GatewayScannerMock(), return_value=GatewayScannerMock(),
) )
async def test_routing_secure_keyfile( async def test_routing_secure_keyfile(
gateway_scanner_mock, hass: HomeAssistant gateway_scanner_mock, hass: HomeAssistant, knx_setup
) -> None: ) -> None:
"""Test routing secure setup with keyfile.""" """Test routing secure setup with keyfile."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -335,7 +328,6 @@ async def test_routing_secure_keyfile(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "routing" assert result2["step_id"] == "routing"
assert result2["errors"] == {"base": "no_router_discovered"} assert result2["errors"] == {"base": "no_router_discovered"}
@ -361,9 +353,6 @@ async def test_routing_secure_keyfile(
assert not result4["errors"] assert not result4["errors"]
with patch( with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.knx.config_flow.load_keyring", return_value=True "homeassistant.components.knx.config_flow.load_keyring", return_value=True
): ):
routing_secure_knxkeys = await hass.config_entries.flow.async_configure( routing_secure_knxkeys = await hass.config_entries.flow.async_configure(
@ -374,21 +363,21 @@ async def test_routing_secure_keyfile(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert routing_secure_knxkeys["type"] == FlowResultType.CREATE_ENTRY assert routing_secure_knxkeys["type"] == FlowResultType.CREATE_ENTRY
assert routing_secure_knxkeys["title"] == "Secure Routing as 0.0.123" assert routing_secure_knxkeys["title"] == "Secure Routing as 0.0.123"
assert routing_secure_knxkeys["data"] == { assert routing_secure_knxkeys["data"] == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE, CONF_KNX_CONNECTION_TYPE: CONF_KNX_ROUTING_SECURE,
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys", CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_KNXKEY_PASSWORD: "password",
CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_BACKBONE_KEY: None,
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None,
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_SECURE_USER_PASSWORD: None,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123", CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.123",
} }
assert len(mock_setup_entry.mock_calls) == 1 knx_setup.assert_called_once()
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -452,7 +441,7 @@ async def test_routing_secure_keyfile(
return_value=GatewayScannerMock(), return_value=GatewayScannerMock(),
) )
async def test_tunneling_setup_manual( async def test_tunneling_setup_manual(
gateway_scanner_mock, hass: HomeAssistant, user_input, config_entry_data gateway_scanner_mock, hass: HomeAssistant, knx_setup, user_input, config_entry_data
) -> None: ) -> None:
"""Test tunneling if no gateway was found found (or `manual` option was chosen).""" """Test tunneling if no gateway was found found (or `manual` option was chosen)."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -467,25 +456,19 @@ async def test_tunneling_setup_manual(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "manual_tunnel" assert result2["step_id"] == "manual_tunnel"
assert result2["errors"] == {"base": "no_tunnel_discovered"} assert result2["errors"] == {"base": "no_tunnel_discovered"}
with patch( result3 = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", result2["flow_id"],
return_value=True, user_input,
) as mock_setup_entry: )
result3 = await hass.config_entries.flow.async_configure( await hass.async_block_till_done()
result2["flow_id"], assert result3["type"] == FlowResultType.CREATE_ENTRY
user_input, assert result3["title"] == "Tunneling @ 192.168.0.1"
) assert result3["data"] == config_entry_data
await hass.async_block_till_done() knx_setup.assert_called_once()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.1"
assert result3["data"] == config_entry_data
assert len(mock_setup_entry.mock_calls) == 1
@patch( @patch(
@ -493,7 +476,7 @@ async def test_tunneling_setup_manual(
return_value=GatewayScannerMock(), return_value=GatewayScannerMock(),
) )
async def test_tunneling_setup_for_local_ip( async def test_tunneling_setup_for_local_ip(
gateway_scanner_mock, hass: HomeAssistant gateway_scanner_mock, hass: HomeAssistant, knx_setup
) -> None: ) -> None:
"""Test tunneling if only one gateway is found.""" """Test tunneling if only one gateway is found."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -512,7 +495,6 @@ async def test_tunneling_setup_for_local_ip(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "manual_tunnel" assert result2["step_id"] == "manual_tunnel"
assert result2["errors"] == {"base": "no_tunnel_discovered"} assert result2["errors"] == {"base": "no_tunnel_discovered"}
@ -527,7 +509,6 @@ async def test_tunneling_setup_for_local_ip(
CONF_KNX_LOCAL_IP: "192.168.1.112", CONF_KNX_LOCAL_IP: "192.168.1.112",
}, },
) )
await hass.async_block_till_done()
assert result_invalid_host["type"] == FlowResultType.FORM assert result_invalid_host["type"] == FlowResultType.FORM
assert result_invalid_host["step_id"] == "manual_tunnel" assert result_invalid_host["step_id"] == "manual_tunnel"
assert result_invalid_host["errors"] == { assert result_invalid_host["errors"] == {
@ -544,7 +525,6 @@ async def test_tunneling_setup_for_local_ip(
CONF_KNX_LOCAL_IP: "asdf", CONF_KNX_LOCAL_IP: "asdf",
}, },
) )
await hass.async_block_till_done()
assert result_invalid_local["type"] == FlowResultType.FORM assert result_invalid_local["type"] == FlowResultType.FORM
assert result_invalid_local["step_id"] == "manual_tunnel" assert result_invalid_local["step_id"] == "manual_tunnel"
assert result_invalid_local["errors"] == { assert result_invalid_local["errors"] == {
@ -553,36 +533,33 @@ async def test_tunneling_setup_for_local_ip(
} }
# valid user input # valid user input
with patch( result3 = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", result2["flow_id"],
return_value=True, {
) as mock_setup_entry: CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING,
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{
CONF_KNX_TUNNELING_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
CONF_KNX_LOCAL_IP: "192.168.1.112",
},
)
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.2"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2", CONF_HOST: "192.168.0.2",
CONF_PORT: 3675, CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: "192.168.1.112", CONF_KNX_LOCAL_IP: "192.168.1.112",
} },
)
assert len(mock_setup_entry.mock_calls) == 1 await hass.async_block_till_done()
assert result3["type"] == FlowResultType.CREATE_ENTRY
assert result3["title"] == "Tunneling @ 192.168.0.2"
assert result3["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_HOST: "192.168.0.2",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: "192.168.1.112",
}
knx_setup.assert_called_once()
async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant) -> None: async def test_tunneling_setup_for_multiple_found_gateways(
hass: HomeAssistant, knx_setup
) -> None:
"""Test tunneling if multiple gateways are found.""" """Test tunneling if multiple gateways are found."""
gateway = _gateway_descriptor("192.168.0.1", 3675) gateway = _gateway_descriptor("192.168.0.1", 3675)
gateway2 = _gateway_descriptor("192.168.1.100", 3675) gateway2 = _gateway_descriptor("192.168.1.100", 3675)
@ -602,32 +579,26 @@ async def test_tunneling_setup_for_multiple_found_gateways(hass: HomeAssistant)
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
await hass.async_block_till_done()
assert tunnel_flow["type"] == FlowResultType.FORM assert tunnel_flow["type"] == FlowResultType.FORM
assert tunnel_flow["step_id"] == "tunnel" assert tunnel_flow["step_id"] == "tunnel"
assert not tunnel_flow["errors"] assert not tunnel_flow["errors"]
with patch( result = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", tunnel_flow["flow_id"],
return_value=True, {CONF_KNX_GATEWAY: str(gateway)},
) as mock_setup_entry: )
result = await hass.config_entries.flow.async_configure( await hass.async_block_till_done()
tunnel_flow["flow_id"], assert result["type"] == FlowResultType.CREATE_ENTRY
{CONF_KNX_GATEWAY: str(gateway)}, assert result["data"] == {
) **DEFAULT_ENTRY_DATA,
await hass.async_block_till_done() CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
assert result["type"] == FlowResultType.CREATE_ENTRY CONF_HOST: "192.168.0.1",
assert result["data"] == { CONF_PORT: 3675,
**DEFAULT_ENTRY_DATA, CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_ROUTE_BACK: False,
CONF_HOST: "192.168.0.1", CONF_KNX_LOCAL_IP: None,
CONF_PORT: 3675, }
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", knx_setup.assert_called_once()
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None,
}
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -660,7 +631,6 @@ async def test_manual_tunnel_step_with_found_gateway(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
await hass.async_block_till_done()
assert tunnel_flow["type"] == FlowResultType.FORM assert tunnel_flow["type"] == FlowResultType.FORM
assert tunnel_flow["step_id"] == "tunnel" assert tunnel_flow["step_id"] == "tunnel"
assert not tunnel_flow["errors"] assert not tunnel_flow["errors"]
@ -671,13 +641,14 @@ async def test_manual_tunnel_step_with_found_gateway(
CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL, CONF_KNX_GATEWAY: OPTION_MANUAL_TUNNEL,
}, },
) )
await hass.async_block_till_done()
assert manual_tunnel_flow["type"] == FlowResultType.FORM assert manual_tunnel_flow["type"] == FlowResultType.FORM
assert manual_tunnel_flow["step_id"] == "manual_tunnel" assert manual_tunnel_flow["step_id"] == "manual_tunnel"
assert not manual_tunnel_flow["errors"] assert not manual_tunnel_flow["errors"]
async def test_form_with_automatic_connection_handling(hass: HomeAssistant) -> None: async def test_form_with_automatic_connection_handling(
hass: HomeAssistant, knx_setup
) -> None:
"""Test we get the form.""" """Test we get the form."""
with patch( with patch(
"homeassistant.components.knx.config_flow.GatewayScanner" "homeassistant.components.knx.config_flow.GatewayScanner"
@ -691,26 +662,20 @@ async def test_form_with_automatic_connection_handling(hass: HomeAssistant) -> N
assert result["type"] == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert not result["errors"] assert not result["errors"]
with patch( result2 = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", result["flow_id"],
return_value=True, {
) as mock_setup_entry: CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
result2 = await hass.config_entries.flow.async_configure( },
result["flow_id"], )
{ await hass.async_block_till_done()
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == CONF_KNX_AUTOMATIC.capitalize() assert result2["title"] == CONF_KNX_AUTOMATIC.capitalize()
assert result2["data"] == { assert result2["data"] == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
} }
knx_setup.assert_called_once()
assert len(mock_setup_entry.mock_calls) == 1
async def _get_menu_step(hass: HomeAssistant) -> FlowResult: async def _get_menu_step(hass: HomeAssistant) -> FlowResult:
@ -737,7 +702,6 @@ async def _get_menu_step(hass: HomeAssistant) -> FlowResult:
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "tunnel" assert result2["step_id"] == "tunnel"
assert not result2["errors"] assert not result2["errors"]
@ -746,7 +710,6 @@ async def _get_menu_step(hass: HomeAssistant) -> FlowResult:
result2["flow_id"], result2["flow_id"],
{CONF_KNX_GATEWAY: str(gateway)}, {CONF_KNX_GATEWAY: str(gateway)},
) )
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.MENU assert result3["type"] == FlowResultType.MENU
assert result3["step_id"] == "secure_key_source" assert result3["step_id"] == "secure_key_source"
return result3 return result3
@ -778,7 +741,6 @@ async def test_get_secure_menu_step_manual_tunnelling(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2["step_id"] == "tunnel" assert result2["step_id"] == "tunnel"
assert not result2["errors"] assert not result2["errors"]
@ -798,12 +760,11 @@ async def test_get_secure_menu_step_manual_tunnelling(
CONF_PORT: 3675, CONF_PORT: 3675,
}, },
) )
await hass.async_block_till_done()
assert result3["type"] == FlowResultType.MENU assert result3["type"] == FlowResultType.MENU
assert result3["step_id"] == "secure_key_source" assert result3["step_id"] == "secure_key_source"
async def test_configure_secure_tunnel_manual(hass: HomeAssistant): async def test_configure_secure_tunnel_manual(hass: HomeAssistant, knx_setup):
"""Test configure tunnelling secure keys manually.""" """Test configure tunnelling secure keys manually."""
menu_step = await _get_menu_step(hass) menu_step = await _get_menu_step(hass)
@ -815,37 +776,32 @@ async def test_configure_secure_tunnel_manual(hass: HomeAssistant):
assert result["step_id"] == "secure_tunnel_manual" assert result["step_id"] == "secure_tunnel_manual"
assert not result["errors"] assert not result["errors"]
with patch( secure_tunnel_manual = await hass.config_entries.flow.async_configure(
"homeassistant.components.knx.async_setup_entry", result["flow_id"],
return_value=True, {
) as mock_setup_entry:
secure_tunnel_manual = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_KNX_SECURE_USER_ID: 2,
CONF_KNX_SECURE_USER_PASSWORD: "password",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth",
},
)
await hass.async_block_till_done()
assert secure_tunnel_manual["type"] == FlowResultType.CREATE_ENTRY
assert secure_tunnel_manual["data"] == {
**DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_KNX_SECURE_USER_ID: 2, CONF_KNX_SECURE_USER_ID: 2,
CONF_KNX_SECURE_USER_PASSWORD: "password", CONF_KNX_SECURE_USER_PASSWORD: "password",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth", CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth",
CONF_HOST: "192.168.0.1", },
CONF_PORT: 3675, )
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", await hass.async_block_till_done()
CONF_KNX_ROUTE_BACK: False, assert secure_tunnel_manual["type"] == FlowResultType.CREATE_ENTRY
CONF_KNX_LOCAL_IP: None, assert secure_tunnel_manual["data"] == {
} **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
assert len(mock_setup_entry.mock_calls) == 1 CONF_KNX_SECURE_USER_ID: 2,
CONF_KNX_SECURE_USER_PASSWORD: "password",
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: "device_auth",
CONF_HOST: "192.168.0.1",
CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None,
}
knx_setup.assert_called_once()
async def test_configure_secure_knxkeys(hass: HomeAssistant): async def test_configure_secure_knxkeys(hass: HomeAssistant, knx_setup):
"""Test configure secure knxkeys.""" """Test configure secure knxkeys."""
menu_step = await _get_menu_step(hass) menu_step = await _get_menu_step(hass)
@ -858,9 +814,6 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant):
assert not result["errors"] assert not result["errors"]
with patch( with patch(
"homeassistant.components.knx.async_setup_entry",
return_value=True,
) as mock_setup_entry, patch(
"homeassistant.components.knx.config_flow.load_keyring", return_value=True "homeassistant.components.knx.config_flow.load_keyring", return_value=True
): ):
secure_knxkeys = await hass.config_entries.flow.async_configure( secure_knxkeys = await hass.config_entries.flow.async_configure(
@ -871,25 +824,24 @@ async def test_configure_secure_knxkeys(hass: HomeAssistant):
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert secure_knxkeys["type"] == FlowResultType.CREATE_ENTRY assert secure_knxkeys["type"] == FlowResultType.CREATE_ENTRY
assert secure_knxkeys["data"] == { assert secure_knxkeys["data"] == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING_TCP_SECURE,
CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys", CONF_KNX_KNXKEY_FILENAME: "knx/testcase.knxkeys",
CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_KNXKEY_PASSWORD: "password",
CONF_KNX_ROUTING_BACKBONE_KEY: None, CONF_KNX_ROUTING_BACKBONE_KEY: None,
CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None, CONF_KNX_ROUTING_SYNC_LATENCY_TOLERANCE: None,
CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None, CONF_KNX_SECURE_DEVICE_AUTHENTICATION: None,
CONF_KNX_SECURE_USER_ID: None, CONF_KNX_SECURE_USER_ID: None,
CONF_KNX_SECURE_USER_PASSWORD: None, CONF_KNX_SECURE_USER_PASSWORD: None,
CONF_HOST: "192.168.0.1", CONF_HOST: "192.168.0.1",
CONF_PORT: 3675, CONF_PORT: 3675,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
CONF_KNX_ROUTE_BACK: False, CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None, CONF_KNX_LOCAL_IP: None,
} }
knx_setup.assert_called_once()
assert len(mock_setup_entry.mock_calls) == 1
async def test_configure_secure_knxkeys_file_not_found(hass: HomeAssistant): async def test_configure_secure_knxkeys_file_not_found(hass: HomeAssistant):
@ -915,7 +867,6 @@ async def test_configure_secure_knxkeys_file_not_found(hass: HomeAssistant):
CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_KNXKEY_PASSWORD: "password",
}, },
) )
await hass.async_block_till_done()
assert secure_knxkeys["type"] == FlowResultType.FORM assert secure_knxkeys["type"] == FlowResultType.FORM
assert secure_knxkeys["errors"] assert secure_knxkeys["errors"]
assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_FILENAME] == "file_not_found" assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_FILENAME] == "file_not_found"
@ -944,20 +895,19 @@ async def test_configure_secure_knxkeys_invalid_signature(hass: HomeAssistant):
CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_KNXKEY_PASSWORD: "password",
}, },
) )
await hass.async_block_till_done()
assert secure_knxkeys["type"] == FlowResultType.FORM assert secure_knxkeys["type"] == FlowResultType.FORM
assert secure_knxkeys["errors"] assert secure_knxkeys["errors"]
assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_PASSWORD] == "invalid_signature" assert secure_knxkeys["errors"][CONF_KNX_KNXKEY_PASSWORD] == "invalid_signature"
async def test_options_flow_connection_type( async def test_options_flow_connection_type(
hass: HomeAssistant, mock_config_entry: MockConfigEntry hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry
) -> None: ) -> None:
"""Test options flow changing interface.""" """Test options flow changing interface."""
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
hass.data[DOMAIN] = Mock() # GatewayScanner uses running XKNX() instance
gateway = _gateway_descriptor("192.168.0.1", 3675) gateway = _gateway_descriptor("192.168.0.1", 3675)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
with patch( with patch(
@ -968,9 +918,8 @@ async def test_options_flow_connection_type(
menu_step["flow_id"], menu_step["flow_id"],
{"next_step_id": "connection_type"}, {"next_step_id": "connection_type"},
) )
assert result["type"] == FlowResultType.FORM
assert result.get("type") == FlowResultType.FORM assert result["step_id"] == "connection_type"
assert result.get("step_id") == "connection_type"
result2 = await hass.config_entries.options.async_configure( result2 = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
@ -978,8 +927,8 @@ async def test_options_flow_connection_type(
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
}, },
) )
assert result2.get("type") == FlowResultType.FORM assert result2["type"] == FlowResultType.FORM
assert result2.get("step_id") == "tunnel" assert result2["step_id"] == "tunnel"
result3 = await hass.config_entries.options.async_configure( result3 = await hass.config_entries.options.async_configure(
result2["flow_id"], result2["flow_id"],
@ -988,9 +937,8 @@ async def test_options_flow_connection_type(
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert result3.get("type") == FlowResultType.CREATE_ENTRY assert result3["type"] == FlowResultType.CREATE_ENTRY
assert not result3.get("data") assert not result3["data"]
assert mock_config_entry.data == { assert mock_config_entry.data == {
CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING, CONF_KNX_CONNECTION_TYPE: CONF_KNX_TUNNELING,
CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240", CONF_KNX_INDIVIDUAL_ADDRESS: "0.0.240",
@ -1003,9 +951,12 @@ async def test_options_flow_connection_type(
CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER, CONF_KNX_STATE_UPDATER: CONF_KNX_DEFAULT_STATE_UPDATER,
CONF_KNX_ROUTE_BACK: False, CONF_KNX_ROUTE_BACK: False,
} }
knx_setup.assert_called_once()
async def test_options_flow_secure_manual_to_keyfile(hass: HomeAssistant) -> None: async def test_options_flow_secure_manual_to_keyfile(
hass: HomeAssistant, knx_setup
) -> None:
"""Test options flow changing secure credential source.""" """Test options flow changing secure credential source."""
mock_config_entry = MockConfigEntry( mock_config_entry = MockConfigEntry(
title="KNX", title="KNX",
@ -1025,7 +976,6 @@ async def test_options_flow_secure_manual_to_keyfile(hass: HomeAssistant) -> Non
CONF_KNX_LOCAL_IP: None, CONF_KNX_LOCAL_IP: None,
}, },
) )
mock_config_entry.add_to_hass(hass)
gateway = _gateway_descriptor( gateway = _gateway_descriptor(
"192.168.0.1", "192.168.0.1",
3675, 3675,
@ -1033,6 +983,8 @@ async def test_options_flow_secure_manual_to_keyfile(hass: HomeAssistant) -> Non
requires_secure=True, requires_secure=True,
) )
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
with patch( with patch(
"homeassistant.components.knx.config_flow.GatewayScanner" "homeassistant.components.knx.config_flow.GatewayScanner"
@ -1042,8 +994,8 @@ async def test_options_flow_secure_manual_to_keyfile(hass: HomeAssistant) -> Non
menu_step["flow_id"], menu_step["flow_id"],
{"next_step_id": "connection_type"}, {"next_step_id": "connection_type"},
) )
assert result.get("type") == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result.get("step_id") == "connection_type" assert result["step_id"] == "connection_type"
result2 = await hass.config_entries.options.async_configure( result2 = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
@ -1080,6 +1032,7 @@ async def test_options_flow_secure_manual_to_keyfile(hass: HomeAssistant) -> Non
CONF_KNX_KNXKEY_PASSWORD: "password", CONF_KNX_KNXKEY_PASSWORD: "password",
}, },
) )
await hass.async_block_till_done()
assert secure_knxkeys["type"] == FlowResultType.CREATE_ENTRY assert secure_knxkeys["type"] == FlowResultType.CREATE_ENTRY
assert mock_config_entry.data == { assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
@ -1097,22 +1050,23 @@ async def test_options_flow_secure_manual_to_keyfile(hass: HomeAssistant) -> Non
CONF_KNX_ROUTE_BACK: False, CONF_KNX_ROUTE_BACK: False,
CONF_KNX_LOCAL_IP: None, CONF_KNX_LOCAL_IP: None,
} }
knx_setup.assert_called_once()
async def test_options_communication_settings( async def test_options_communication_settings(
hass: HomeAssistant, mock_config_entry: MockConfigEntry hass: HomeAssistant, knx_setup, mock_config_entry: MockConfigEntry
) -> None: ) -> None:
"""Test options flow changing communication settings.""" """Test options flow changing communication settings."""
mock_config_entry.add_to_hass(hass) mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id) menu_step = await hass.config_entries.options.async_init(mock_config_entry.entry_id)
result = await hass.config_entries.options.async_configure( result = await hass.config_entries.options.async_configure(
menu_step["flow_id"], menu_step["flow_id"],
{"next_step_id": "communication_settings"}, {"next_step_id": "communication_settings"},
) )
assert result.get("type") == FlowResultType.FORM assert result["type"] == FlowResultType.FORM
assert result.get("step_id") == "communication_settings" assert result["step_id"] == "communication_settings"
result2 = await hass.config_entries.options.async_configure( result2 = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
@ -1121,14 +1075,13 @@ async def test_options_communication_settings(
CONF_KNX_RATE_LIMIT: 40, CONF_KNX_RATE_LIMIT: 40,
}, },
) )
await hass.async_block_till_done() await hass.async_block_till_done()
assert result2.get("type") == FlowResultType.CREATE_ENTRY assert result2["type"] == FlowResultType.CREATE_ENTRY
assert not result2.get("data") assert not result2.get("data")
assert mock_config_entry.data == { assert mock_config_entry.data == {
**DEFAULT_ENTRY_DATA, **DEFAULT_ENTRY_DATA,
CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC, CONF_KNX_CONNECTION_TYPE: CONF_KNX_AUTOMATIC,
CONF_KNX_STATE_UPDATER: False, CONF_KNX_STATE_UPDATER: False,
CONF_KNX_RATE_LIMIT: 40, CONF_KNX_RATE_LIMIT: 40,
} }
knx_setup.assert_called_once()