/
conftest.py
95 lines (74 loc) · 3.07 KB
/
conftest.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
"""Fixtures for component testing."""
from unittest.mock import patch
import pytest
from homeassistant.auth.const import GROUP_ID_ADMIN, GROUP_ID_READ_ONLY
from homeassistant.setup import async_setup_component
from homeassistant.components.websocket_api.http import URL
from homeassistant.components.websocket_api.auth import (
TYPE_AUTH, TYPE_AUTH_OK, TYPE_AUTH_REQUIRED)
from tests.common import MockUser, CLIENT_ID
@pytest.fixture
def hass_ws_client(aiohttp_client):
"""Websocket client fixture connected to websocket server."""
async def create_client(hass, access_token=None):
"""Create a websocket client."""
assert await async_setup_component(hass, 'websocket_api')
client = await aiohttp_client(hass.http.app)
patches = []
if access_token is None:
patches.append(patch(
'homeassistant.auth.AuthManager.active', return_value=False))
patches.append(patch(
'homeassistant.auth.AuthManager.support_legacy',
return_value=True))
patches.append(patch(
'homeassistant.components.websocket_api.auth.'
'validate_password', return_value=True))
else:
patches.append(patch(
'homeassistant.auth.AuthManager.active', return_value=True))
patches.append(patch(
'homeassistant.components.http.auth.setup_auth'))
for p in patches:
p.start()
try:
websocket = await client.ws_connect(URL)
auth_resp = await websocket.receive_json()
assert auth_resp['type'] == TYPE_AUTH_REQUIRED
if access_token is None:
await websocket.send_json({
'type': TYPE_AUTH,
'api_password': 'bla'
})
else:
await websocket.send_json({
'type': TYPE_AUTH,
'access_token': access_token
})
auth_ok = await websocket.receive_json()
assert auth_ok['type'] == TYPE_AUTH_OK
finally:
for p in patches:
p.stop()
# wrap in client
websocket.client = client
return websocket
return create_client
@pytest.fixture
def hass_access_token(hass, hass_admin_user):
"""Return an access token to access Home Assistant."""
refresh_token = hass.loop.run_until_complete(
hass.auth.async_create_refresh_token(hass_admin_user, CLIENT_ID))
yield hass.auth.async_create_access_token(refresh_token)
@pytest.fixture
def hass_admin_user(hass):
"""Return a Home Assistant admin user."""
admin_group = hass.loop.run_until_complete(hass.auth.async_get_group(
GROUP_ID_ADMIN))
return MockUser(groups=[admin_group]).add_to_hass(hass)
@pytest.fixture
def hass_read_only_user(hass):
"""Return a Home Assistant read only user."""
read_only_group = hass.loop.run_until_complete(hass.auth.async_get_group(
GROUP_ID_READ_ONLY))
return MockUser(groups=[read_only_group]).add_to_hass(hass)