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

Add type hints tests MQTT config_flow (#87783)

Browse Source
This commit is contained in:
Jan Bouwhuis 2023-02-09 16:47:28 +01:00 committed by GitHub
parent 07a1a0efa9
commit a244f9b202
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 88 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

170
tests/components/mqtt/test_config_flow.py
View File

@ -1,8 +1,10 @@
"""Test config flow.""" """Test config flow."""
from collections.abc import Generator
from pathlib import Path from pathlib import Path
from random import getrandbits from random import getrandbits
from ssl import SSLError from ssl import SSLError
from unittest.mock import AsyncMock, patch from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from uuid import uuid4 from uuid import uuid4
import pytest import pytest
@ -23,7 +25,7 @@ MOCK_CLIENT_KEY = b"## mock key file ##"
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def mock_finish_setup(): def mock_finish_setup() -> Generator[MagicMock, None, None]:
"""Mock out the finish setup method.""" """Mock out the finish setup method."""
with patch( with patch(
"homeassistant.components.mqtt.MQTT.async_connect", return_value=True "homeassistant.components.mqtt.MQTT.async_connect", return_value=True
@ -32,7 +34,7 @@ def mock_finish_setup():
@pytest.fixture @pytest.fixture
def mock_client_cert_check_fail(): def mock_client_cert_check_fail() -> Generator[MagicMock, None, None]:
"""Mock the client certificate check.""" """Mock the client certificate check."""
with patch( with patch(
"homeassistant.components.mqtt.config_flow.load_pem_x509_certificate", "homeassistant.components.mqtt.config_flow.load_pem_x509_certificate",
@ -42,7 +44,7 @@ def mock_client_cert_check_fail():
@pytest.fixture @pytest.fixture
def mock_client_key_check_fail(): def mock_client_key_check_fail() -> Generator[MagicMock, None, None]:
"""Mock the client key file check.""" """Mock the client key file check."""
with patch( with patch(
"homeassistant.components.mqtt.config_flow.load_pem_private_key", "homeassistant.components.mqtt.config_flow.load_pem_private_key",
@ -52,7 +54,7 @@ def mock_client_key_check_fail():
@pytest.fixture @pytest.fixture
def mock_ssl_context(): def mock_ssl_context() -> Generator[dict[str, MagicMock], None, None]:
"""Mock the SSL context used to load the cert chain and to load verify locations.""" """Mock the SSL context used to load the cert chain and to load verify locations."""
with patch( with patch(
"homeassistant.components.mqtt.config_flow.SSLContext" "homeassistant.components.mqtt.config_flow.SSLContext"
@ -69,7 +71,7 @@ def mock_ssl_context():
@pytest.fixture @pytest.fixture
def mock_reload_after_entry_update(): def mock_reload_after_entry_update() -> Generator[MagicMock, None, None]:
"""Mock out the reload after updating the entry.""" """Mock out the reload after updating the entry."""
with patch( with patch(
"homeassistant.components.mqtt._async_config_entry_updated" "homeassistant.components.mqtt._async_config_entry_updated"
@ -78,14 +80,14 @@ def mock_reload_after_entry_update():
@pytest.fixture @pytest.fixture
def mock_try_connection(): def mock_try_connection() -> Generator[MagicMock, None, None]:
"""Mock the try connection method.""" """Mock the try connection method."""
with patch("homeassistant.components.mqtt.config_flow.try_connection") as mock_try: with patch("homeassistant.components.mqtt.config_flow.try_connection") as mock_try:
yield mock_try yield mock_try
@pytest.fixture @pytest.fixture
def mock_try_connection_success(): def mock_try_connection_success() -> Generator[MqttMockPahoClient, None, None]:
"""Mock the try connection method with success.""" """Mock the try connection method with success."""
_mid = 1 _mid = 1
@ -118,7 +120,7 @@ def mock_try_connection_success():
@pytest.fixture @pytest.fixture
def mock_try_connection_time_out(): def mock_try_connection_time_out() -> Generator[MagicMock, None, None]:
"""Mock the try connection method with a time out.""" """Mock the try connection method with a time out."""
# Patch prevent waiting 5 sec for a timeout # Patch prevent waiting 5 sec for a timeout
@ -130,7 +132,7 @@ def mock_try_connection_time_out():
@pytest.fixture @pytest.fixture
def mock_process_uploaded_file(tmp_path): def mock_process_uploaded_file(tmp_path: Path) -> Generator[MagicMock, None, None]:
"""Mock upload certificate files.""" """Mock upload certificate files."""
file_id_ca = str(uuid4()) file_id_ca = str(uuid4())
file_id_cert = str(uuid4()) file_id_cert = str(uuid4())
@ -170,8 +172,8 @@ def mock_process_uploaded_file(tmp_path):
async def test_user_connection_works( async def test_user_connection_works(
hass: HomeAssistant, hass: HomeAssistant,
mock_try_connection, mock_try_connection: MagicMock,
mock_finish_setup, mock_finish_setup: MagicMock,
mqtt_client_mock: MqttMockPahoClient, mqtt_client_mock: MqttMockPahoClient,
) -> None: ) -> None:
"""Test we can finish a config flow.""" """Test we can finish a config flow."""
@ -201,8 +203,8 @@ async def test_user_connection_works(
async def test_user_v5_connection_works( async def test_user_v5_connection_works(
hass: HomeAssistant, hass: HomeAssistant,
mock_try_connection, mock_try_connection: MagicMock,
mock_finish_setup, mock_finish_setup: MagicMock,
mqtt_client_mock: MqttMockPahoClient, mqtt_client_mock: MqttMockPahoClient,
) -> None: ) -> None:
"""Test we can finish a config flow.""" """Test we can finish a config flow."""
@ -242,7 +244,9 @@ async def test_user_v5_connection_works(
async def test_user_connection_fails( async def test_user_connection_fails(
hass: HomeAssistant, mock_try_connection_time_out, mock_finish_setup hass: HomeAssistant,
mock_try_connection_time_out: MagicMock,
mock_finish_setup: MagicMock,
) -> None: ) -> None:
"""Test if connection cannot be made.""" """Test if connection cannot be made."""
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
@ -265,9 +269,8 @@ async def test_user_connection_fails(
async def test_manual_config_starts_discovery_flow( async def test_manual_config_starts_discovery_flow(
hass: HomeAssistant, hass: HomeAssistant,
mock_try_connection, mock_try_connection: MqttMockPahoClient,
mock_finish_setup, mock_finish_setup: MagicMock,
mqtt_client_mock: MqttMockPahoClient,
) -> None: ) -> None:
"""Test manual config initiates a discovery flow.""" """Test manual config initiates a discovery flow."""
# No flows in progress # No flows in progress
@ -288,9 +291,8 @@ async def test_manual_config_starts_discovery_flow(
async def test_manual_config_set( async def test_manual_config_set(
hass: HomeAssistant, hass: HomeAssistant,
mock_try_connection, mock_try_connection: MqttMockPahoClient,
mock_finish_setup, mock_finish_setup: MagicMock,
mqtt_client_mock: MqttMockPahoClient,
) -> None: ) -> None:
"""Test manual config does not create an entry, and entry can be setup late.""" """Test manual config does not create an entry, and entry can be setup late."""
# MQTT config present in yaml config # MQTT config present in yaml config
@ -381,7 +383,9 @@ async def test_hassio_ignored(hass: HomeAssistant) -> None:
async def test_hassio_confirm( async def test_hassio_confirm(
hass: HomeAssistant, mock_try_connection_success, mock_finish_setup hass: HomeAssistant,
mock_try_connection_success: MqttMockPahoClient,
mock_finish_setup: MagicMock,
) -> None: ) -> None:
"""Test we can finish a config flow.""" """Test we can finish a config flow."""
mock_try_connection.return_value = True mock_try_connection.return_value = True
@ -428,7 +432,9 @@ async def test_hassio_confirm(
async def test_hassio_cannot_connect( async def test_hassio_cannot_connect(
hass: HomeAssistant, mock_try_connection_time_out, mock_finish_setup hass: HomeAssistant,
mock_try_connection_time_out: MagicMock,
mock_finish_setup: MagicMock,
) -> None: ) -> None:
"""Test a config flow is aborted when a connection was not successful.""" """Test a config flow is aborted when a connection was not successful."""
mock_try_connection.return_value = True mock_try_connection.return_value = True
@ -470,7 +476,7 @@ async def test_hassio_cannot_connect(
async def test_option_flow( async def test_option_flow(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator, mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator,
mock_try_connection, mock_try_connection: MagicMock,
) -> None: ) -> None:
"""Test config flow options.""" """Test config flow options."""
with patch( with patch(
@ -567,11 +573,10 @@ async def test_option_flow(
async def test_bad_certificate( async def test_bad_certificate(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator, mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator,
mock_try_connection_success, mock_try_connection_success: MqttMockPahoClient,
tmp_path: Path, mock_ssl_context: dict[str, MagicMock],
mock_ssl_context, mock_process_uploaded_file: MagicMock,
test_error, test_error: str | None,
mock_process_uploaded_file,
) -> None: ) -> None:
"""Test bad certificate tests.""" """Test bad certificate tests."""
# Mock certificate files # Mock certificate files
@ -664,10 +669,10 @@ async def test_bad_certificate(
async def test_keepalive_validation( async def test_keepalive_validation(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator, mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator,
mock_try_connection, mock_try_connection: MagicMock,
mock_reload_after_entry_update, mock_reload_after_entry_update: MagicMock,
input_value, input_value: str,
error, error: bool,
) -> None: ) -> None:
"""Test validation of the keep alive option.""" """Test validation of the keep alive option."""
@ -710,8 +715,8 @@ async def test_keepalive_validation(
async def test_disable_birth_will( async def test_disable_birth_will(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator, mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator,
mock_try_connection, mock_try_connection: MagicMock,
mock_reload_after_entry_update, mock_reload_after_entry_update: MagicMock,
) -> None: ) -> None:
"""Test disabling birth and will.""" """Test disabling birth and will."""
mqtt_mock = await mqtt_mock_entry_no_yaml_config() mqtt_mock = await mqtt_mock_entry_no_yaml_config()
@ -781,8 +786,8 @@ async def test_disable_birth_will(
async def test_invalid_discovery_prefix( async def test_invalid_discovery_prefix(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator, mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator,
mock_try_connection, mock_try_connection: MagicMock,
mock_reload_after_entry_update, mock_reload_after_entry_update: MagicMock,
) -> None: ) -> None:
"""Test setting an invalid discovery prefix.""" """Test setting an invalid discovery prefix."""
mqtt_mock = await mqtt_mock_entry_no_yaml_config() mqtt_mock = await mqtt_mock_entry_no_yaml_config()
@ -836,29 +841,32 @@ async def test_invalid_discovery_prefix(
assert mock_reload_after_entry_update.call_count == 0 assert mock_reload_after_entry_update.call_count == 0
def get_default(schema, key): def get_default(schema: vol.Schema, key: str) -> Any:
"""Get default value for key in voluptuous schema.""" """Get default value for key in voluptuous schema."""
for k in schema.keys(): for schema_key in schema.keys():
if k == key: if schema_key == key:
if k.default == vol.UNDEFINED: if schema_key.default == vol.UNDEFINED:
return None return None
return k.default() return schema_key.default()
def get_suggested(schema, key): def get_suggested(schema: vol.Schema, key: str) -> Any:
"""Get suggested value for key in voluptuous schema.""" """Get suggested value for key in voluptuous schema."""
for k in schema.keys(): for schema_key in schema.keys():
if k == key: if schema_key == key:
if k.description is None or "suggested_value" not in k.description: if (
schema_key.description is None
or "suggested_value" not in schema_key.description
):
return None return None
return k.description["suggested_value"] return schema_key.description["suggested_value"]
async def test_option_flow_default_suggested_values( async def test_option_flow_default_suggested_values(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator, mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator,
mock_try_connection_success, mock_try_connection_success: MqttMockPahoClient,
mock_reload_after_entry_update, mock_reload_after_entry_update: MagicMock,
) -> None: ) -> None:
"""Test config flow options has default/suggested values.""" """Test config flow options has default/suggested values."""
await mqtt_mock_entry_no_yaml_config() await mqtt_mock_entry_no_yaml_config()
@ -895,10 +903,10 @@ async def test_option_flow_default_suggested_values(
mqtt.CONF_USERNAME: "user", mqtt.CONF_USERNAME: "user",
mqtt.CONF_PASSWORD: "pass", mqtt.CONF_PASSWORD: "pass",
} }
for k, v in defaults.items(): for key, value in defaults.items():
assert get_default(result["data_schema"].schema, k) == v assert get_default(result["data_schema"].schema, key) == value
for k, v in suggested.items(): for key, value in suggested.items():
assert get_suggested(result["data_schema"].schema, k) == v assert get_suggested(result["data_schema"].schema, key) == value
result = await hass.config_entries.options.async_configure( result = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
@ -924,10 +932,10 @@ async def test_option_flow_default_suggested_values(
"will_topic": "ha_state/offline", "will_topic": "ha_state/offline",
"will_payload": "offline", "will_payload": "offline",
} }
for k, v in defaults.items(): for key, value in defaults.items():
assert get_default(result["data_schema"].schema, k) == v assert get_default(result["data_schema"].schema, key) == value
for k, v in suggested.items(): for key, value in suggested.items():
assert get_suggested(result["data_schema"].schema, k) == v assert get_suggested(result["data_schema"].schema, key) == value
result = await hass.config_entries.options.async_configure( result = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
@ -957,10 +965,10 @@ async def test_option_flow_default_suggested_values(
mqtt.CONF_USERNAME: "us3r", mqtt.CONF_USERNAME: "us3r",
mqtt.CONF_PASSWORD: "p4ss", mqtt.CONF_PASSWORD: "p4ss",
} }
for k, v in defaults.items(): for key, value in defaults.items():
assert get_default(result["data_schema"].schema, k) == v assert get_default(result["data_schema"].schema, key) == value
for k, v in suggested.items(): for key, value in suggested.items():
assert get_suggested(result["data_schema"].schema, k) == v assert get_suggested(result["data_schema"].schema, key) == value
result = await hass.config_entries.options.async_configure( result = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
@ -981,10 +989,10 @@ async def test_option_flow_default_suggested_values(
"will_topic": "ha_state/offl1ne", "will_topic": "ha_state/offl1ne",
"will_payload": "offl1ne", "will_payload": "offl1ne",
} }
for k, v in defaults.items(): for key, value in defaults.items():
assert get_default(result["data_schema"].schema, k) == v assert get_default(result["data_schema"].schema, key) == value
for k, v in suggested.items(): for key, value in suggested.items():
assert get_suggested(result["data_schema"].schema, k) == v assert get_suggested(result["data_schema"].schema, key) == value
result = await hass.config_entries.options.async_configure( result = await hass.config_entries.options.async_configure(
result["flow_id"], result["flow_id"],
@ -1012,10 +1020,10 @@ async def test_option_flow_default_suggested_values(
async def test_skipping_advanced_options( async def test_skipping_advanced_options(
hass: HomeAssistant, hass: HomeAssistant,
mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator, mqtt_mock_entry_no_yaml_config: MqttMockHAClientGenerator,
mock_try_connection, mock_try_connection: MagicMock,
mock_reload_after_entry_update, mock_reload_after_entry_update: MagicMock,
advanced_options, advanced_options: bool,
step_id, step_id: str,
) -> None: ) -> None:
"""Test advanced options option.""" """Test advanced options option."""
@ -1048,7 +1056,7 @@ async def test_skipping_advanced_options(
async def test_options_user_connection_fails( async def test_options_user_connection_fails(
hass: HomeAssistant, mock_try_connection_time_out hass: HomeAssistant, mock_try_connection_time_out: MagicMock
) -> None: ) -> None:
"""Test if connection cannot be made.""" """Test if connection cannot be made."""
config_entry = MockConfigEntry(domain=mqtt.DOMAIN) config_entry = MockConfigEntry(domain=mqtt.DOMAIN)
@ -1079,7 +1087,7 @@ async def test_options_user_connection_fails(
async def test_options_bad_birth_message_fails( async def test_options_bad_birth_message_fails(
hass: HomeAssistant, mock_try_connection hass: HomeAssistant, mock_try_connection: MqttMockPahoClient
) -> None: ) -> None:
"""Test bad birth message.""" """Test bad birth message."""
config_entry = MockConfigEntry(domain=mqtt.DOMAIN) config_entry = MockConfigEntry(domain=mqtt.DOMAIN)
@ -1117,7 +1125,7 @@ async def test_options_bad_birth_message_fails(
async def test_options_bad_will_message_fails( async def test_options_bad_will_message_fails(
hass: HomeAssistant, mock_try_connection hass: HomeAssistant, mock_try_connection: MagicMock
) -> None: ) -> None:
"""Test bad will message.""" """Test bad will message."""
config_entry = MockConfigEntry(domain=mqtt.DOMAIN) config_entry = MockConfigEntry(domain=mqtt.DOMAIN)
@ -1156,10 +1164,10 @@ async def test_options_bad_will_message_fails(
async def test_try_connection_with_advanced_parameters( async def test_try_connection_with_advanced_parameters(
hass: HomeAssistant, hass: HomeAssistant,
mock_try_connection_success, mock_try_connection_success: MqttMockPahoClient,
tmp_path: Path, tmp_path: Path,
mock_ssl_context, mock_ssl_context: dict[str, MagicMock],
mock_process_uploaded_file, mock_process_uploaded_file: MagicMock,
) -> None: ) -> None:
"""Test config flow with advanced parameters from config.""" """Test config flow with advanced parameters from config."""
@ -1297,10 +1305,9 @@ async def test_try_connection_with_advanced_parameters(
async def test_setup_with_advanced_settings( async def test_setup_with_advanced_settings(
hass: HomeAssistant, hass: HomeAssistant,
mock_try_connection, mock_try_connection: MagicMock,
tmp_path: Path, mock_ssl_context: dict[str, MagicMock],
mock_ssl_context, mock_process_uploaded_file: MagicMock,
mock_process_uploaded_file,
) -> None: ) -> None:
"""Test config flow setup with advanced parameters.""" """Test config flow setup with advanced parameters."""
file_id = mock_process_uploaded_file.file_id file_id = mock_process_uploaded_file.file_id
@ -1456,9 +1463,8 @@ async def test_setup_with_advanced_settings(
async def test_change_websockets_transport_to_tcp( async def test_change_websockets_transport_to_tcp(
hass: HomeAssistant, hass: HomeAssistant,
mock_try_connection, mock_try_connection,
tmp_path: Path, mock_ssl_context: dict[str, MagicMock],
mock_ssl_context, mock_process_uploaded_file: MagicMock,
mock_process_uploaded_file,
) -> None: ) -> None:
"""Test option flow setup with websockets transport settings.""" """Test option flow setup with websockets transport settings."""
config_entry = MockConfigEntry(domain=mqtt.DOMAIN) config_entry = MockConfigEntry(domain=mqtt.DOMAIN)