starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-03 10:05:58 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add reauth for Netatmo when token or token scope is invalid (#57487)

Browse Source
This commit is contained in:
Tobias Sauerwein
2021-10-26 16:09:10 +02:00
committed by GitHub
parent c9966a3b04
commit 3970a50553
9 changed files with 313 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/netatmo/conftest.py
+2 -2
View File
@@ -22,7 +22,7 @@ def mock_config_entry_fixture(hass):
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": " ".join(ALL_SCOPES),
"scope": ALL_SCOPES,
},
},
options={
@@ -53,7 +53,7 @@ def mock_config_entry_fixture(hass):
return mock_entry
@pytest.fixture
@pytest.fixture(name="netatmo_auth")
def netatmo_auth():
"""Restrict loaded platforms to list given."""
with patch(
tests/components/netatmo/test_config_flow.py
+110 -15
View File
@@ -13,6 +13,8 @@ from homeassistant.components.netatmo.const import (
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
from homeassistant.helpers import config_entry_oauth2_flow
from .common import ALL_SCOPES
from tests.common import MockConfigEntry
CLIENT_ID = "1234"
@@ -67,21 +69,7 @@ async def test_full_flow(
},
)
scope = "+".join(
[
"access_camera",
"access_presence",
"read_camera",
"read_homecoach",
"read_presence",
"read_smokedetector",
"read_station",
"read_thermostat",
"write_camera",
"write_presence",
"write_thermostat",
]
)
scope = "+".join(sorted(ALL_SCOPES))
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
@@ -227,3 +215,110 @@ async def test_option_flow_wrong_coordinates(hass):
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
for k, v in expected_result.items():
assert config_entry.options[CONF_WEATHER_AREAS]["Home"][k] == v
async def test_reauth(
hass, hass_client_no_auth, aioclient_mock, current_request_with_host
):
"""Test initialization of the reauth flow."""
assert await setup.async_setup_component(
hass,
"netatmo",
{
"netatmo": {CONF_CLIENT_ID: CLIENT_ID, CONF_CLIENT_SECRET: CLIENT_SECRET},
"http": {"base_url": "https://example.com"},
},
)
result = await hass.config_entries.flow.async_init(
"netatmo", context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
scope = "+".join(sorted(ALL_SCOPES))
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}&scope={scope}"
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"homeassistant.components.netatmo.async_setup_entry", return_value=True
) as mock_setup:
await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
new_entry = hass.config_entries.async_entries(DOMAIN)[0]
assert new_entry.state == config_entries.ConfigEntryState.LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
# Should show form
result = await hass.config_entries.flow.async_init(
"netatmo", context={"source": config_entries.SOURCE_REAUTH}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "reauth_confirm"
# Confirm reauth flow
result2 = await hass.config_entries.flow.async_configure(result["flow_id"], {})
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
# Update entry
with patch(
"homeassistant.components.netatmo.async_setup_entry", return_value=True
) as mock_setup:
result3 = await hass.config_entries.flow.async_configure(result2["flow_id"])
await hass.async_block_till_done()
new_entry2 = hass.config_entries.async_entries(DOMAIN)[0]
assert result3["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result3["reason"] == "reauth_successful"
assert new_entry2.state == config_entries.ConfigEntryState.LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
tests/components/netatmo/test_init.py
+104 -52
View File
@@ -4,6 +4,7 @@ from datetime import timedelta
from time import time
from unittest.mock import AsyncMock, patch
import aiohttp
import pyatmo
from homeassistant import config_entries
@@ -14,6 +15,7 @@ from homeassistant.setup import async_setup_component
from homeassistant.util import dt
from .common import (
ALL_SCOPES,
FAKE_WEBHOOK_ACTIVATION,
fake_post_request,
selected_platforms,
@@ -49,24 +51,8 @@ FAKE_WEBHOOK = {
}
async def test_setup_component(hass):
async def test_setup_component(hass, config_entry):
"""Test setup and teardown of the netatmo component."""
config_entry = MockConfigEntry(
domain="netatmo",
data={
"auth_implementation": "cloud",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": "read_station",
},
},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
@@ -248,7 +234,7 @@ async def test_setup_with_cloudhook(hass):
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": "read_station",
"scope": ALL_SCOPES,
},
},
)
@@ -298,24 +284,8 @@ async def test_setup_with_cloudhook(hass):
assert not hass.config_entries.async_entries(DOMAIN)
async def test_setup_component_api_error(hass):
async def test_setup_component_api_error(hass, config_entry):
"""Test error on setup of the netatmo component."""
config_entry = MockConfigEntry(
domain="netatmo",
data={
"auth_implementation": "cloud",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": "read_station",
},
},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
@@ -337,24 +307,8 @@ async def test_setup_component_api_error(hass):
mock_impl.assert_called_once()
async def test_setup_component_api_timeout(hass):
async def test_setup_component_api_timeout(hass, config_entry):
"""Test timeout on setup of the netatmo component."""
config_entry = MockConfigEntry(
domain="netatmo",
data={
"auth_implementation": "cloud",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": "read_station",
},
},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
@@ -429,3 +383,101 @@ async def test_setup_component_with_delay(hass, config_entry):
await hass.async_stop()
mock_dropwebhook.assert_called_once()
async def test_setup_component_invalid_token_scope(hass):
"""Test handling of invalid token scope."""
config_entry = MockConfigEntry(
domain="netatmo",
data={
"auth_implementation": "cloud",
"token": {
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
"expires_at": time() + 1000,
"scope": " ".join(
[
"read_smokedetector",
"read_thermostat",
"write_thermostat",
]
),
},
},
options={},
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
) as mock_impl, patch(
"homeassistant.components.webhook.async_generate_url"
) as mock_webhook:
mock_auth.return_value.async_post_request.side_effect = fake_post_request
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
assert await async_setup_component(hass, "netatmo", {})
await hass.async_block_till_done()
mock_auth.assert_not_called()
mock_impl.assert_called_once()
mock_webhook.assert_not_called()
assert config_entry.state is config_entries.ConfigEntryState.SETUP_ERROR
assert hass.config_entries.async_entries(DOMAIN)
assert len(hass.states.async_all()) > 0
for config_entry in hass.config_entries.async_entries("netatmo"):
await hass.config_entries.async_remove(config_entry.entry_id)
async def test_setup_component_invalid_token(hass, config_entry):
"""Test handling of invalid token."""
async def fake_ensure_valid_token(*args, **kwargs):
print("fake_ensure_valid_token")
raise aiohttp.ClientResponseError(
request_info=aiohttp.client.RequestInfo(
url="http://example.com",
method="GET",
headers={},
real_url="http://example.com",
),
code=400,
history=(),
)
with patch(
"homeassistant.components.netatmo.api.AsyncConfigEntryNetatmoAuth",
) as mock_auth, patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation",
) as mock_impl, patch(
"homeassistant.components.webhook.async_generate_url"
) as mock_webhook, patch(
"homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session"
) as mock_session:
mock_auth.return_value.async_post_request.side_effect = fake_post_request
mock_auth.return_value.async_addwebhook.side_effect = AsyncMock()
mock_auth.return_value.async_dropwebhook.side_effect = AsyncMock()
mock_session.return_value.async_ensure_token_valid.side_effect = (
fake_ensure_valid_token
)
assert await async_setup_component(hass, "netatmo", {})
await hass.async_block_till_done()
mock_auth.assert_not_called()
mock_impl.assert_called_once()
mock_webhook.assert_not_called()
assert config_entry.state is config_entries.ConfigEntryState.SETUP_ERROR
assert hass.config_entries.async_entries(DOMAIN)
assert len(hass.states.async_all()) > 0
for config_entry in hass.config_entries.async_entries("netatmo"):
await hass.config_entries.async_remove(config_entry.entry_id)