forked from home-assistant/core
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_config_flow.py
163 lines (138 loc) · 4.93 KB
/
test_config_flow.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
"""Test the myUplink config flow."""
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.myuplink.const import (
DOMAIN,
OAUTH2_AUTHORIZE,
OAUTH2_TOKEN,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from .const import CLIENT_ID
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
REDIRECT_URL = "https://example.com/auth/external/callback"
CURRENT_SCOPE = "WRITESYSTEM READSYSTEM offline_access"
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth,
aioclient_mock,
current_request_with_host,
setup_credentials,
) -> None:
"""Check full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT_URL,
},
)
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URL}"
f"&state={state}"
f"&scope={CURRENT_SCOPE.replace(' ', '+')}"
)
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(
f"homeassistant.components.{DOMAIN}.async_setup_entry", return_value=True
) as mock_setup:
await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
async def test_flow_reauth(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
current_request_with_host: None,
setup_credentials: None,
mock_config_entry: MockConfigEntry,
expires_at: float,
) -> None:
"""Test reauth step."""
OLD_SCOPE = "READSYSTEM offline_access"
OLD_SCOPE_TOKEN = {
"auth_implementation": DOMAIN,
"token": {
"access_token": "Fake_token",
"scope": OLD_SCOPE,
"expires_in": 86399,
"refresh_token": "3012bc9f-7a65-4240-b817-9154ffdcc30f",
"token_type": "Bearer",
"expires_at": expires_at,
},
}
assert mock_config_entry.data["token"]["scope"] == CURRENT_SCOPE
assert hass.config_entries.async_update_entry(
mock_config_entry, data=OLD_SCOPE_TOKEN
)
assert mock_config_entry.data["token"]["scope"] == OLD_SCOPE
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": mock_config_entry.entry_id,
},
data=mock_config_entry.data,
)
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT_URL,
},
)
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URL}"
f"&state={state}"
f"&scope={CURRENT_SCOPE.replace(' ', '+')}"
)
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": "updated-refresh-token",
"access_token": "updated-access-token",
"type": "Bearer",
"expires_in": "60",
"scope": CURRENT_SCOPE,
},
)
with patch(
f"homeassistant.components.{DOMAIN}.async_setup_entry", return_value=True
) as mock_setup:
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result.get("type") == FlowResultType.ABORT
assert result.get("reason") == "reauth_successful"
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert len(mock_setup.mock_calls) == 1
assert mock_config_entry.data["token"]["scope"] == CURRENT_SCOPE