Add reauth support to LaMetric (#80355)

* Add reauth support to LaMetric

* Adjust docblock
This commit is contained in:
Franck Nijhof 2022-10-14 23:54:14 +02:00 committed by GitHub
parent 3b33e0d832
commit 7bd9ce72f7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 253 additions and 9 deletions

View file

@ -18,7 +18,12 @@ from homeassistant.components.ssdp import (
ATTR_UPNP_SERIAL,
SsdpServiceInfo,
)
from homeassistant.config_entries import SOURCE_DHCP, SOURCE_SSDP, SOURCE_USER
from homeassistant.config_entries import (
SOURCE_DHCP,
SOURCE_REAUTH,
SOURCE_SSDP,
SOURCE_USER,
)
from homeassistant.const import CONF_API_KEY, CONF_DEVICE, CONF_HOST, CONF_MAC
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@ -743,3 +748,173 @@ async def test_dhcp_unknown_device(
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "unknown"
async def test_reauth_cloud_import(
hass: HomeAssistant,
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
aioclient_mock: AiohttpClientMocker,
current_request_with_host: None,
mock_setup_entry: MagicMock,
mock_lametric_cloud_config_flow: MagicMock,
mock_lametric_config_flow: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reauth flow importing api keys from the cloud."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": mock_config_entry.unique_id,
"entry_id": mock_config_entry.entry_id,
},
data=mock_config_entry.data,
)
assert "flow_id" in result
flow_id = result["flow_id"]
await hass.config_entries.flow.async_configure(
flow_id, user_input={"next_step_id": "pick_implementation"}
)
# pylint: disable=protected-access
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": flow_id,
"redirect_uri": "https://example.com/auth/external/callback",
},
)
client = await hass_client_no_auth()
await client.get(f"/auth/external/callback?code=abcd&state={state}")
aioclient_mock.post(
"https://developer.lametric.com/api/v2/oauth2/token",
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
result2 = await hass.config_entries.flow.async_configure(flow_id)
assert result2.get("type") == FlowResultType.ABORT
assert result2.get("reason") == "reauth_successful"
assert mock_config_entry.data == {
CONF_HOST: "127.0.0.1",
CONF_API_KEY: "mock-api-key",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
assert len(mock_lametric_config_flow.device.mock_calls) == 1
assert len(mock_lametric_config_flow.notify.mock_calls) == 1
async def test_reauth_cloud_abort_device_not_found(
hass: HomeAssistant,
hass_client_no_auth: Callable[[], Awaitable[TestClient]],
aioclient_mock: AiohttpClientMocker,
current_request_with_host: None,
mock_setup_entry: MagicMock,
mock_lametric_cloud_config_flow: MagicMock,
mock_lametric_config_flow: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reauth flow importing api keys from the cloud."""
mock_config_entry.unique_id = "UKNOWN_DEVICE"
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": mock_config_entry.unique_id,
"entry_id": mock_config_entry.entry_id,
},
data=mock_config_entry.data,
)
assert "flow_id" in result
flow_id = result["flow_id"]
await hass.config_entries.flow.async_configure(
flow_id, user_input={"next_step_id": "pick_implementation"}
)
# pylint: disable=protected-access
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": flow_id,
"redirect_uri": "https://example.com/auth/external/callback",
},
)
client = await hass_client_no_auth()
await client.get(f"/auth/external/callback?code=abcd&state={state}")
aioclient_mock.post(
"https://developer.lametric.com/api/v2/oauth2/token",
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
result2 = await hass.config_entries.flow.async_configure(flow_id)
assert result2.get("type") == FlowResultType.ABORT
assert result2.get("reason") == "reauth_device_not_found"
assert len(mock_lametric_cloud_config_flow.devices.mock_calls) == 1
assert len(mock_lametric_config_flow.device.mock_calls) == 0
assert len(mock_lametric_config_flow.notify.mock_calls) == 0
async def test_reauth_manual(
hass: HomeAssistant,
mock_setup_entry: MagicMock,
mock_lametric_config_flow: MagicMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reauth flow with manual entry."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": SOURCE_REAUTH,
"unique_id": mock_config_entry.unique_id,
"entry_id": mock_config_entry.entry_id,
},
data=mock_config_entry.data,
)
assert "flow_id" in result
flow_id = result["flow_id"]
await hass.config_entries.flow.async_configure(
flow_id, user_input={"next_step_id": "manual_entry"}
)
result2 = await hass.config_entries.flow.async_configure(
flow_id, user_input={CONF_API_KEY: "mock-api-key"}
)
assert result2.get("type") == FlowResultType.ABORT
assert result2.get("reason") == "reauth_successful"
assert mock_config_entry.data == {
CONF_HOST: "127.0.0.1",
CONF_API_KEY: "mock-api-key",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
}
assert len(mock_lametric_config_flow.device.mock_calls) == 1
assert len(mock_lametric_config_flow.notify.mock_calls) == 1