mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 20:57:21 +00:00
Improve BMW test quality (#133704)
This commit is contained in:
parent
b1d8994751
commit
514b74096a
@ -53,6 +53,13 @@ REMOTE_SERVICE_EXC_TRANSLATION = (
|
||||
"Error executing remote service on vehicle. HTTPStatusError: 502 Bad Gateway"
|
||||
)
|
||||
|
||||
BIMMER_CONNECTED_LOGIN_PATCH = (
|
||||
"homeassistant.components.bmw_connected_drive.config_flow.MyBMWAuthentication.login"
|
||||
)
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH = (
|
||||
"homeassistant.components.bmw_connected_drive.coordinator.MyBMWAccount.get_vehicles"
|
||||
)
|
||||
|
||||
|
||||
async def setup_mocked_integration(hass: HomeAssistant) -> MockConfigEntry:
|
||||
"""Mock a fully setup config entry and all components based on fixtures."""
|
||||
|
@ -15,11 +15,13 @@ from homeassistant.components.bmw_connected_drive.const import (
|
||||
CONF_READ_ONLY,
|
||||
CONF_REFRESH_TOKEN,
|
||||
)
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
|
||||
from homeassistant.const import CONF_PASSWORD, CONF_REGION, CONF_USERNAME
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from . import (
|
||||
BIMMER_CONNECTED_LOGIN_PATCH,
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
FIXTURE_CAPTCHA_INPUT,
|
||||
FIXTURE_CONFIG_ENTRY,
|
||||
FIXTURE_GCID,
|
||||
@ -40,97 +42,11 @@ def login_sideeffect(self: MyBMWAuthentication):
|
||||
self.gcid = FIXTURE_GCID
|
||||
|
||||
|
||||
async def test_show_form(hass: HomeAssistant) -> None:
|
||||
"""Test that the form is served with no input."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
|
||||
async def test_authentication_error(hass: HomeAssistant) -> None:
|
||||
"""Test we show user form on MyBMW authentication error."""
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.api.authentication.MyBMWAuthentication.login",
|
||||
side_effect=MyBMWAuthError("Login failed"),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=deepcopy(FIXTURE_USER_INPUT_W_CAPTCHA),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "invalid_auth"}
|
||||
|
||||
|
||||
async def test_connection_error(hass: HomeAssistant) -> None:
|
||||
"""Test we show user form on MyBMW API error."""
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.api.authentication.MyBMWAuthentication.login",
|
||||
side_effect=RequestError("Connection reset"),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=deepcopy(FIXTURE_USER_INPUT_W_CAPTCHA),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
async def test_api_error(hass: HomeAssistant) -> None:
|
||||
"""Test we show user form on general connection error."""
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.api.authentication.MyBMWAuthentication.login",
|
||||
side_effect=MyBMWAPIError("400 Bad Request"),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=deepcopy(FIXTURE_USER_INPUT_W_CAPTCHA),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "cannot_connect"}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("bmw_fixture")
|
||||
async def test_captcha_flow_missing_error(hass: HomeAssistant) -> None:
|
||||
"""Test the external flow with captcha failing once and succeeding the second time."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=deepcopy(FIXTURE_USER_INPUT),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "captcha"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_CAPTCHA_TOKEN: " "}
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "missing_captcha"}
|
||||
|
||||
|
||||
async def test_full_user_flow_implementation(hass: HomeAssistant) -> None:
|
||||
"""Test registering an integration and finishing flow works."""
|
||||
with (
|
||||
patch(
|
||||
"bimmer_connected.api.authentication.MyBMWAuthentication.login",
|
||||
BIMMER_CONNECTED_LOGIN_PATCH,
|
||||
side_effect=login_sideeffect,
|
||||
autospec=True,
|
||||
),
|
||||
@ -155,15 +71,125 @@ async def test_full_user_flow_implementation(hass: HomeAssistant) -> None:
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == FIXTURE_COMPLETE_ENTRY[CONF_USERNAME]
|
||||
assert result["data"] == FIXTURE_COMPLETE_ENTRY
|
||||
assert (
|
||||
result["result"].unique_id
|
||||
== f"{FIXTURE_USER_INPUT[CONF_REGION]}-{FIXTURE_USER_INPUT[CONF_USERNAME]}"
|
||||
)
|
||||
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("side_effect", "error"),
|
||||
[
|
||||
(MyBMWAuthError("Login failed"), "invalid_auth"),
|
||||
(RequestError("Connection reset"), "cannot_connect"),
|
||||
(MyBMWAPIError("400 Bad Request"), "cannot_connect"),
|
||||
],
|
||||
)
|
||||
async def test_error_display_with_successful_login(
|
||||
hass: HomeAssistant, side_effect: Exception, error: str
|
||||
) -> None:
|
||||
"""Test we show user form on MyBMW authentication error and are still able to succeed."""
|
||||
|
||||
with patch(
|
||||
BIMMER_CONNECTED_LOGIN_PATCH,
|
||||
side_effect=side_effect,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=deepcopy(FIXTURE_USER_INPUT_W_CAPTCHA),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": error}
|
||||
|
||||
with (
|
||||
patch(
|
||||
BIMMER_CONNECTED_LOGIN_PATCH,
|
||||
side_effect=login_sideeffect,
|
||||
autospec=True,
|
||||
),
|
||||
patch(
|
||||
"homeassistant.components.bmw_connected_drive.async_setup_entry",
|
||||
return_value=True,
|
||||
) as mock_setup_entry,
|
||||
):
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
deepcopy(FIXTURE_USER_INPUT),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "captcha"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], FIXTURE_CAPTCHA_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == FIXTURE_COMPLETE_ENTRY[CONF_USERNAME]
|
||||
assert result["data"] == FIXTURE_COMPLETE_ENTRY
|
||||
assert (
|
||||
result["result"].unique_id
|
||||
== f"{FIXTURE_USER_INPUT[CONF_REGION]}-{FIXTURE_USER_INPUT[CONF_USERNAME]}"
|
||||
)
|
||||
|
||||
assert len(mock_setup_entry.mock_calls) == 1
|
||||
|
||||
|
||||
async def test_unique_id_existing(hass: HomeAssistant) -> None:
|
||||
"""Test registering an integration and when the unique id already exists."""
|
||||
|
||||
mock_config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
with (
|
||||
patch(
|
||||
BIMMER_CONNECTED_LOGIN_PATCH,
|
||||
side_effect=login_sideeffect,
|
||||
autospec=True,
|
||||
),
|
||||
):
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=deepcopy(FIXTURE_USER_INPUT),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("bmw_fixture")
|
||||
async def test_captcha_flow_missing_error(hass: HomeAssistant) -> None:
|
||||
"""Test the external flow with captcha failing once and succeeding the second time."""
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": config_entries.SOURCE_USER},
|
||||
data=deepcopy(FIXTURE_USER_INPUT),
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "captcha"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_CAPTCHA_TOKEN: " "}
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "missing_captcha"}
|
||||
|
||||
|
||||
async def test_options_flow_implementation(hass: HomeAssistant) -> None:
|
||||
"""Test config flow options."""
|
||||
with (
|
||||
patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
return_value=[],
|
||||
),
|
||||
patch(
|
||||
@ -200,7 +226,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
|
||||
"""Test the reauth form."""
|
||||
with (
|
||||
patch(
|
||||
"bimmer_connected.api.authentication.MyBMWAuthentication.login",
|
||||
BIMMER_CONNECTED_LOGIN_PATCH,
|
||||
side_effect=login_sideeffect,
|
||||
autospec=True,
|
||||
),
|
||||
@ -249,7 +275,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
|
||||
async def test_reconfigure(hass: HomeAssistant) -> None:
|
||||
"""Test the reconfiguration form."""
|
||||
with patch(
|
||||
"bimmer_connected.api.authentication.MyBMWAuthentication.login",
|
||||
BIMMER_CONNECTED_LOGIN_PATCH,
|
||||
side_effect=login_sideeffect,
|
||||
autospec=True,
|
||||
):
|
||||
|
@ -1,7 +1,6 @@
|
||||
"""Test BMW coordinator."""
|
||||
"""Test BMW coordinator for general availability/unavailability of entities and raising issues."""
|
||||
|
||||
from copy import deepcopy
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from bimmer_connected.models import (
|
||||
@ -13,27 +12,56 @@ from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.bmw_connected_drive import DOMAIN as BMW_DOMAIN
|
||||
from homeassistant.components.bmw_connected_drive.const import (
|
||||
CONF_REFRESH_TOKEN,
|
||||
SCAN_INTERVALS,
|
||||
)
|
||||
from homeassistant.const import CONF_REGION
|
||||
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryAuthFailed
|
||||
from homeassistant.helpers import issue_registry as ir
|
||||
from homeassistant.helpers.update_coordinator import UpdateFailed
|
||||
|
||||
from . import FIXTURE_CONFIG_ENTRY
|
||||
from . import BIMMER_CONNECTED_VEHICLE_PATCH, FIXTURE_CONFIG_ENTRY
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
FIXTURE_ENTITY_STATES = {
|
||||
"binary_sensor.m340i_xdrive_door_lock_state": "off",
|
||||
"lock.m340i_xdrive_lock": "locked",
|
||||
"lock.i3_rex_lock": "unlocked",
|
||||
"number.ix_xdrive50_target_soc": "80",
|
||||
"sensor.ix_xdrive50_rear_left_tire_pressure": "2.61",
|
||||
"sensor.ix_xdrive50_rear_right_tire_pressure": "2.69",
|
||||
}
|
||||
FIXTURE_DEFAULT_REGION = FIXTURE_CONFIG_ENTRY["data"][CONF_REGION]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("bmw_fixture")
|
||||
async def test_update_success(hass: HomeAssistant) -> None:
|
||||
"""Test the reauth form."""
|
||||
config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
async def test_config_entry_update(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test if the coordinator updates the refresh token in config entry."""
|
||||
config_entry_fixure = deepcopy(FIXTURE_CONFIG_ENTRY)
|
||||
config_entry_fixure["data"][CONF_REFRESH_TOKEN] = "old_token"
|
||||
config_entry = MockConfigEntry(**config_entry_fixure)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
assert (
|
||||
hass.config_entries.async_get_entry(config_entry.entry_id).data[
|
||||
CONF_REFRESH_TOKEN
|
||||
]
|
||||
== "old_token"
|
||||
)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert config_entry.runtime_data.last_update_success is True
|
||||
assert (
|
||||
hass.config_entries.async_get_entry(config_entry.entry_id).data[
|
||||
CONF_REFRESH_TOKEN
|
||||
]
|
||||
== "another_token_string"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("bmw_fixture")
|
||||
@ -41,125 +69,176 @@ async def test_update_failed(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Test the reauth form."""
|
||||
"""Test a failing API call."""
|
||||
config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
coordinator = config_entry.runtime_data
|
||||
|
||||
assert coordinator.last_update_success is True
|
||||
|
||||
freezer.tick(timedelta(minutes=5, seconds=1))
|
||||
# Test if entities show data correctly
|
||||
for entity_id, state in FIXTURE_ENTITY_STATES.items():
|
||||
assert hass.states.get(entity_id).state == state
|
||||
|
||||
# On API error, entities should be unavailable
|
||||
freezer.tick(SCAN_INTERVALS[FIXTURE_DEFAULT_REGION])
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
side_effect=MyBMWAPIError("Test error"),
|
||||
):
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert coordinator.last_update_success is False
|
||||
assert isinstance(coordinator.last_exception, UpdateFailed) is True
|
||||
for entity_id in FIXTURE_ENTITY_STATES:
|
||||
assert hass.states.get(entity_id).state == "unavailable"
|
||||
|
||||
# And should recover on next update
|
||||
freezer.tick(SCAN_INTERVALS[FIXTURE_DEFAULT_REGION])
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
for entity_id, state in FIXTURE_ENTITY_STATES.items():
|
||||
assert hass.states.get(entity_id).state == state
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("bmw_fixture")
|
||||
async def test_update_reauth(
|
||||
async def test_auth_failed_as_update_failed(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
) -> None:
|
||||
"""Test the reauth form."""
|
||||
"""Test a single auth failure not initializing reauth flow."""
|
||||
config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
coordinator = config_entry.runtime_data
|
||||
# Test if entities show data correctly
|
||||
for entity_id, state in FIXTURE_ENTITY_STATES.items():
|
||||
assert hass.states.get(entity_id).state == state
|
||||
|
||||
assert coordinator.last_update_success is True
|
||||
|
||||
freezer.tick(timedelta(minutes=5, seconds=1))
|
||||
# Due to flaky API, we allow one retry on AuthError and raise as UpdateFailed
|
||||
freezer.tick(SCAN_INTERVALS[FIXTURE_DEFAULT_REGION])
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
side_effect=MyBMWAuthError("Test error"),
|
||||
):
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert coordinator.last_update_success is False
|
||||
assert isinstance(coordinator.last_exception, UpdateFailed) is True
|
||||
for entity_id in FIXTURE_ENTITY_STATES:
|
||||
assert hass.states.get(entity_id).state == "unavailable"
|
||||
|
||||
freezer.tick(timedelta(minutes=5, seconds=1))
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
side_effect=MyBMWAuthError("Test error"),
|
||||
):
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
# And should recover on next update
|
||||
freezer.tick(SCAN_INTERVALS[FIXTURE_DEFAULT_REGION])
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert coordinator.last_update_success is False
|
||||
assert isinstance(coordinator.last_exception, ConfigEntryAuthFailed) is True
|
||||
for entity_id, state in FIXTURE_ENTITY_STATES.items():
|
||||
assert hass.states.get(entity_id).state == state
|
||||
|
||||
# Verify that no issues are raised and no reauth flow is initialized
|
||||
assert len(issue_registry.issues) == 0
|
||||
assert len(hass.config_entries.flow.async_progress_by_handler(BMW_DOMAIN)) == 0
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("bmw_fixture")
|
||||
async def test_init_reauth(
|
||||
async def test_auth_failed_init_reauth(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
) -> None:
|
||||
"""Test the reauth form."""
|
||||
"""Test a two subsequent auth failures initializing reauth flow."""
|
||||
|
||||
config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
# Test if entities show data correctly
|
||||
for entity_id, state in FIXTURE_ENTITY_STATES.items():
|
||||
assert hass.states.get(entity_id).state == state
|
||||
assert len(issue_registry.issues) == 0
|
||||
|
||||
# Due to flaky API, we allow one retry on AuthError and raise as UpdateFailed
|
||||
freezer.tick(SCAN_INTERVALS[FIXTURE_DEFAULT_REGION])
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
side_effect=MyBMWAuthError("Test error"),
|
||||
):
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
for entity_id in FIXTURE_ENTITY_STATES:
|
||||
assert hass.states.get(entity_id).state == "unavailable"
|
||||
assert len(issue_registry.issues) == 0
|
||||
|
||||
# On second failure, we should initialize reauth flow
|
||||
freezer.tick(SCAN_INTERVALS[FIXTURE_DEFAULT_REGION])
|
||||
with patch(
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
side_effect=MyBMWAuthError("Test error"),
|
||||
):
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
for entity_id in FIXTURE_ENTITY_STATES:
|
||||
assert hass.states.get(entity_id).state == "unavailable"
|
||||
assert len(issue_registry.issues) == 1
|
||||
|
||||
reauth_issue = issue_registry.async_get_issue(
|
||||
HOMEASSISTANT_DOMAIN,
|
||||
f"config_entry_reauth_{BMW_DOMAIN}_{config_entry.entry_id}",
|
||||
)
|
||||
assert reauth_issue.active is True
|
||||
|
||||
# Check if reauth flow is initialized correctly
|
||||
flow = hass.config_entries.flow.async_get(reauth_issue.data["flow_id"])
|
||||
assert flow["handler"] == BMW_DOMAIN
|
||||
assert flow["context"]["source"] == "reauth"
|
||||
assert flow["context"]["unique_id"] == config_entry.unique_id
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("bmw_fixture")
|
||||
async def test_captcha_reauth(
|
||||
hass: HomeAssistant,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
) -> None:
|
||||
"""Test the reauth form."""
|
||||
TEST_REGION = "north_america"
|
||||
|
||||
config_entry_fixure = deepcopy(FIXTURE_CONFIG_ENTRY)
|
||||
config_entry_fixure["data"][CONF_REGION] = TEST_REGION
|
||||
config_entry = MockConfigEntry(**config_entry_fixure)
|
||||
"""Test a CaptchaError initializing reauth flow."""
|
||||
config_entry = MockConfigEntry(**FIXTURE_CONFIG_ENTRY)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
coordinator = config_entry.runtime_data
|
||||
# Test if entities show data correctly
|
||||
for entity_id, state in FIXTURE_ENTITY_STATES.items():
|
||||
assert hass.states.get(entity_id).state == state
|
||||
|
||||
assert coordinator.last_update_success is True
|
||||
|
||||
freezer.tick(timedelta(minutes=10, seconds=1))
|
||||
# If library decides a captcha is needed, we should initialize reauth flow
|
||||
freezer.tick(SCAN_INTERVALS[FIXTURE_DEFAULT_REGION])
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
side_effect=MyBMWCaptchaMissingError(
|
||||
"Missing hCaptcha token for North America login"
|
||||
),
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
side_effect=MyBMWCaptchaMissingError("Missing hCaptcha token"),
|
||||
):
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert coordinator.last_update_success is False
|
||||
assert isinstance(coordinator.last_exception, ConfigEntryAuthFailed) is True
|
||||
assert coordinator.last_exception.translation_key == "missing_captcha"
|
||||
for entity_id in FIXTURE_ENTITY_STATES:
|
||||
assert hass.states.get(entity_id).state == "unavailable"
|
||||
assert len(issue_registry.issues) == 1
|
||||
|
||||
reauth_issue = issue_registry.async_get_issue(
|
||||
HOMEASSISTANT_DOMAIN,
|
||||
f"config_entry_reauth_{BMW_DOMAIN}_{config_entry.entry_id}",
|
||||
)
|
||||
assert reauth_issue.active is True
|
||||
|
||||
# Check if reauth flow is initialized correctly
|
||||
flow = hass.config_entries.flow.async_get(reauth_issue.data["flow_id"])
|
||||
assert flow["handler"] == BMW_DOMAIN
|
||||
assert flow["context"]["source"] == "reauth"
|
||||
assert flow["context"]["unique_id"] == config_entry.unique_id
|
||||
|
@ -14,7 +14,7 @@ from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
|
||||
from . import FIXTURE_CONFIG_ENTRY
|
||||
from . import BIMMER_CONNECTED_VEHICLE_PATCH, FIXTURE_CONFIG_ENTRY
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
@ -156,7 +156,7 @@ async def test_migrate_unique_ids(
|
||||
assert entity.unique_id == old_unique_id
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
return_value=[],
|
||||
):
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
@ -212,7 +212,7 @@ async def test_dont_migrate_unique_ids(
|
||||
assert entity.unique_id == old_unique_id
|
||||
|
||||
with patch(
|
||||
"bimmer_connected.account.MyBMWAccount.get_vehicles",
|
||||
BIMMER_CONNECTED_VEHICLE_PATCH,
|
||||
return_value=[],
|
||||
):
|
||||
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
|
Loading…
x
Reference in New Issue
Block a user