Update cloud auth (#9357)
* Update cloud logic * Lint * Update test requirements * Address commments, fix tests * Add credentials
This commit is contained in:
Paulus Schoutsen authored and 
Pascal Vizeli committed
1 parent
90f9a6bc0a
commit
c9fc3fae6e
12 files changed
+1047
-822
No files matched your search
@@ -0,0 +1,271 @@
|
||||
"""Tests for the tools to communicate with the cloud."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from botocore.exceptions import ClientError
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.cloud import DOMAIN, auth_api
|
||||
|
||||
|
||||
MOCK_AUTH = {
|
||||
"id_token": "fake_id_token",
|
||||
"access_token": "fake_access_token",
|
||||
"refresh_token": "fake_refresh_token",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cloud_hass(hass):
|
||||
"""Fixture to return a hass instance with cloud mode set."""
|
||||
hass.data[DOMAIN] = {'mode': 'development'}
|
||||
return hass
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_write():
|
||||
"""Mock reading authentication."""
|
||||
with patch.object(auth_api, '_write_info') as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_read():
|
||||
"""Mock writing authentication."""
|
||||
with patch.object(auth_api, '_read_info') as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_cognito():
|
||||
"""Mock warrant."""
|
||||
with patch('homeassistant.components.cloud.auth_api._cognito') as mock_cog:
|
||||
yield mock_cog()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_auth():
|
||||
"""Mock warrant."""
|
||||
with patch('homeassistant.components.cloud.auth_api.Auth') as mock_auth:
|
||||
yield mock_auth()
|
||||
|
||||
|
||||
def aws_error(code, message='Unknown', operation_name='fake_operation_name'):
|
||||
"""Generate AWS error response."""
|
||||
response = {
|
||||
'Error': {
|
||||
'Code': code,
|
||||
'Message': message
|
||||
}
|
||||
}
|
||||
return ClientError(response, operation_name)
|
||||
|
||||
|
||||
def test_load_auth_with_no_stored_auth(cloud_hass, mock_read):
|
||||
"""Test loading authentication with no stored auth."""
|
||||
mock_read.return_value = None
|
||||
auth = auth_api.load_auth(cloud_hass)
|
||||
assert auth.cognito is None
|
||||
|
||||
|
||||
def test_load_auth_with_invalid_auth(cloud_hass, mock_read, mock_cognito):
|
||||
"""Test calling load_auth when auth is no longer valid."""
|
||||
mock_cognito.get_user.side_effect = aws_error('SomeError')
|
||||
auth = auth_api.load_auth(cloud_hass)
|
||||
|
||||
assert auth.cognito is None
|
||||
|
||||
|
||||
def test_load_auth_with_valid_auth(cloud_hass, mock_read, mock_cognito):
|
||||
"""Test calling load_auth when valid auth."""
|
||||
auth = auth_api.load_auth(cloud_hass)
|
||||
|
||||
assert auth.cognito is not None
|
||||
|
||||
|
||||
def test_auth_properties():
|
||||
"""Test Auth class properties."""
|
||||
auth = auth_api.Auth(None, None)
|
||||
assert not auth.is_logged_in
|
||||
auth.account = {}
|
||||
assert auth.is_logged_in
|
||||
|
||||
|
||||
def test_auth_validate_auth_verification_fails(mock_cognito):
|
||||
"""Test validate authentication with verify request failing."""
|
||||
mock_cognito.get_user.side_effect = aws_error('UserNotFoundException')
|
||||
|
||||
auth = auth_api.Auth(None, mock_cognito)
|
||||
assert auth.validate_auth() is False
|
||||
|
||||
|
||||
def test_auth_validate_auth_token_refresh_needed_fails(mock_cognito):
|
||||
"""Test validate authentication with refresh needed which gets 401."""
|
||||
mock_cognito.get_user.side_effect = aws_error('NotAuthorizedException')
|
||||
mock_cognito.renew_access_token.side_effect = \
|
||||
aws_error('NotAuthorizedException')
|
||||
|
||||
auth = auth_api.Auth(None, mock_cognito)
|
||||
assert auth.validate_auth() is False
|
||||
|
||||
|
||||
def test_auth_validate_auth_token_refresh_needed_succeeds(mock_write,
|
||||
mock_cognito):
|
||||
"""Test validate authentication with refresh."""
|
||||
mock_cognito.get_user.side_effect = [
|
||||
aws_error('NotAuthorizedException'),
|
||||
MagicMock(email='hello@home-assistant.io')
|
||||
]
|
||||
|
||||
auth = auth_api.Auth(None, mock_cognito)
|
||||
assert auth.validate_auth() is True
|
||||
assert len(mock_write.mock_calls) == 1
|
||||
|
||||
|
||||
def test_auth_login_invalid_auth(mock_cognito, mock_write):
|
||||
"""Test trying to login with invalid credentials."""
|
||||
mock_cognito.authenticate.side_effect = aws_error('NotAuthorizedException')
|
||||
auth = auth_api.Auth(None, None)
|
||||
with pytest.raises(auth_api.Unauthenticated):
|
||||
auth.login('user', 'pass')
|
||||
|
||||
assert not auth.is_logged_in
|
||||
assert len(mock_cognito.get_user.mock_calls) == 0
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
def test_auth_login_user_not_found(mock_cognito, mock_write):
|
||||
"""Test trying to login with invalid credentials."""
|
||||
mock_cognito.authenticate.side_effect = aws_error('UserNotFoundException')
|
||||
auth = auth_api.Auth(None, None)
|
||||
with pytest.raises(auth_api.UserNotFound):
|
||||
auth.login('user', 'pass')
|
||||
|
||||
assert not auth.is_logged_in
|
||||
assert len(mock_cognito.get_user.mock_calls) == 0
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
def test_auth_login_user_not_confirmed(mock_cognito, mock_write):
|
||||
"""Test trying to login without confirming account."""
|
||||
mock_cognito.authenticate.side_effect = \
|
||||
aws_error('UserNotConfirmedException')
|
||||
auth = auth_api.Auth(None, None)
|
||||
with pytest.raises(auth_api.UserNotConfirmed):
|
||||
auth.login('user', 'pass')
|
||||
|
||||
assert not auth.is_logged_in
|
||||
assert len(mock_cognito.get_user.mock_calls) == 0
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
def test_auth_login(cloud_hass, mock_cognito, mock_write):
|
||||
"""Test trying to login without confirming account."""
|
||||
mock_cognito.get_user.return_value = \
|
||||
MagicMock(email='hello@home-assistant.io')
|
||||
auth = auth_api.Auth(cloud_hass, None)
|
||||
auth.login('user', 'pass')
|
||||
assert auth.is_logged_in
|
||||
assert len(mock_cognito.authenticate.mock_calls) == 1
|
||||
assert len(mock_write.mock_calls) == 1
|
||||
result_hass, result_auth = mock_write.mock_calls[0][1]
|
||||
assert result_hass is cloud_hass
|
||||
assert result_auth is auth
|
||||
|
||||
|
||||
def test_auth_renew_access_token(mock_write, mock_cognito):
|
||||
"""Test renewing an access token."""
|
||||
auth = auth_api.Auth(None, mock_cognito)
|
||||
assert auth.renew_access_token()
|
||||
assert len(mock_write.mock_calls) == 1
|
||||
|
||||
|
||||
def test_auth_renew_access_token_fails(mock_write, mock_cognito):
|
||||
"""Test failing to renew an access token."""
|
||||
mock_cognito.renew_access_token.side_effect = aws_error('SomeError')
|
||||
auth = auth_api.Auth(None, mock_cognito)
|
||||
assert not auth.renew_access_token()
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
def test_auth_logout(mock_write, mock_cognito):
|
||||
"""Test renewing an access token."""
|
||||
auth = auth_api.Auth(None, mock_cognito)
|
||||
auth.account = MagicMock()
|
||||
auth.logout()
|
||||
assert auth.account is None
|
||||
assert len(mock_write.mock_calls) == 1
|
||||
|
||||
|
||||
def test_auth_logout_fails(mock_write, mock_cognito):
|
||||
"""Test error while logging out."""
|
||||
mock_cognito.logout.side_effect = aws_error('SomeError')
|
||||
auth = auth_api.Auth(None, mock_cognito)
|
||||
auth.account = MagicMock()
|
||||
with pytest.raises(auth_api.CloudError):
|
||||
auth.logout()
|
||||
assert auth.account is not None
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
def test_register(mock_cognito):
|
||||
"""Test registering an account."""
|
||||
auth_api.register(None, 'email@home-assistant.io', 'password')
|
||||
assert len(mock_cognito.register.mock_calls) == 1
|
||||
result_email, result_password = mock_cognito.register.mock_calls[0][1]
|
||||
assert result_email == 'email@home-assistant.io'
|
||||
assert result_password == 'password'
|
||||
|
||||
|
||||
def test_register_fails(mock_cognito):
|
||||
"""Test registering an account."""
|
||||
mock_cognito.register.side_effect = aws_error('SomeError')
|
||||
with pytest.raises(auth_api.CloudError):
|
||||
auth_api.register(None, 'email@home-assistant.io', 'password')
|
||||
|
||||
|
||||
def test_confirm_register(mock_cognito):
|
||||
"""Test confirming a registration of an account."""
|
||||
auth_api.confirm_register(None, '123456', 'email@home-assistant.io')
|
||||
assert len(mock_cognito.confirm_sign_up.mock_calls) == 1
|
||||
result_code, result_email = mock_cognito.confirm_sign_up.mock_calls[0][1]
|
||||
assert result_email == 'email@home-assistant.io'
|
||||
assert result_code == '123456'
|
||||
|
||||
|
||||
def test_confirm_register_fails(mock_cognito):
|
||||
"""Test an error during confirmation of an account."""
|
||||
mock_cognito.confirm_sign_up.side_effect = aws_error('SomeError')
|
||||
with pytest.raises(auth_api.CloudError):
|
||||
auth_api.confirm_register(None, '123456', 'email@home-assistant.io')
|
||||
|
||||
|
||||
def test_forgot_password(mock_cognito):
|
||||
"""Test starting forgot password flow."""
|
||||
auth_api.forgot_password(None, 'email@home-assistant.io')
|
||||
assert len(mock_cognito.initiate_forgot_password.mock_calls) == 1
|
||||
|
||||
|
||||
def test_forgot_password_fails(mock_cognito):
|
||||
"""Test failure when starting forgot password flow."""
|
||||
mock_cognito.initiate_forgot_password.side_effect = aws_error('SomeError')
|
||||
with pytest.raises(auth_api.CloudError):
|
||||
auth_api.forgot_password(None, 'email@home-assistant.io')
|
||||
|
||||
|
||||
def test_confirm_forgot_password(mock_cognito):
|
||||
"""Test confirming forgot password."""
|
||||
auth_api.confirm_forgot_password(
|
||||
None, '123456', 'email@home-assistant.io', 'new password')
|
||||
assert len(mock_cognito.confirm_forgot_password.mock_calls) == 1
|
||||
result_code, result_password = \
|
||||
mock_cognito.confirm_forgot_password.mock_calls[0][1]
|
||||
assert result_code == '123456'
|
||||
assert result_password == 'new password'
|
||||
|
||||
|
||||
def test_confirm_forgot_password_fails(mock_cognito):
|
||||
"""Test failure when confirming forgot password."""
|
||||
mock_cognito.confirm_forgot_password.side_effect = aws_error('SomeError')
|
||||
with pytest.raises(auth_api.CloudError):
|
||||
auth_api.confirm_forgot_password(
|
||||
None, '123456', 'email@home-assistant.io', 'new password')
|
||||
@@ -1,352 +0,0 @@
|
||||
"""Tests for the tools to communicate with the cloud."""
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import aiohttp
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.cloud import DOMAIN, cloud_api, const
|
||||
import homeassistant.util.dt as dt_util
|
||||
|
||||
from tests.common import mock_coro
|
||||
|
||||
|
||||
MOCK_AUTH = {
|
||||
"access_token": "jvCHxpTu2nfORLBRgQY78bIAoK4RPa",
|
||||
"expires_at": "2017-08-29T05:33:28.266048+00:00",
|
||||
"expires_in": 86400,
|
||||
"refresh_token": "C4wR1mgb03cs69EeiFgGOBC8mMQC5Q",
|
||||
"scope": "",
|
||||
"token_type": "Bearer"
|
||||
}
|
||||
|
||||
|
||||
def url(path):
|
||||
"""Create a url."""
|
||||
return urljoin(const.SERVERS['development']['host'], path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cloud_hass(hass):
|
||||
"""Fixture to return a hass instance with cloud mode set."""
|
||||
hass.data[DOMAIN] = {'mode': 'development'}
|
||||
return hass
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_write():
|
||||
"""Mock reading authentication."""
|
||||
with patch.object(cloud_api, '_write_auth') as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_read():
|
||||
"""Mock writing authentication."""
|
||||
with patch.object(cloud_api, '_read_auth') as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_async_login_invalid_auth(cloud_hass, aioclient_mock, mock_write):
|
||||
"""Test trying to login with invalid credentials."""
|
||||
aioclient_mock.post(url('o/token/'), status=401)
|
||||
with pytest.raises(cloud_api.Unauthenticated):
|
||||
yield from cloud_api.async_login(cloud_hass, 'user', 'pass')
|
||||
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_async_login_cloud_error(cloud_hass, aioclient_mock, mock_write):
|
||||
"""Test exception in cloud while logging in."""
|
||||
aioclient_mock.post(url('o/token/'), status=500)
|
||||
with pytest.raises(cloud_api.UnknownError):
|
||||
yield from cloud_api.async_login(cloud_hass, 'user', 'pass')
|
||||
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_async_login_client_error(cloud_hass, aioclient_mock, mock_write):
|
||||
"""Test client error while logging in."""
|
||||
aioclient_mock.post(url('o/token/'), exc=aiohttp.ClientError)
|
||||
with pytest.raises(cloud_api.UnknownError):
|
||||
yield from cloud_api.async_login(cloud_hass, 'user', 'pass')
|
||||
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_async_login(cloud_hass, aioclient_mock, mock_write):
|
||||
"""Test logging in."""
|
||||
aioclient_mock.post(url('o/token/'), json={
|
||||
'expires_in': 10
|
||||
})
|
||||
now = dt_util.utcnow()
|
||||
with patch('homeassistant.components.cloud.cloud_api.utcnow',
|
||||
return_value=now):
|
||||
yield from cloud_api.async_login(cloud_hass, 'user', 'pass')
|
||||
|
||||
assert len(mock_write.mock_calls) == 1
|
||||
result_hass, result_data = mock_write.mock_calls[0][1]
|
||||
assert result_hass is cloud_hass
|
||||
assert result_data == {
|
||||
'expires_in': 10,
|
||||
'expires_at': (now + timedelta(seconds=10)).isoformat()
|
||||
}
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_with_no_stored_auth(cloud_hass, mock_read):
|
||||
"""Test loading authentication with no stored auth."""
|
||||
mock_read.return_value = None
|
||||
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_timeout_during_verification(cloud_hass, mock_read):
|
||||
"""Test loading authentication with timeout during verification."""
|
||||
mock_read.return_value = MOCK_AUTH
|
||||
|
||||
with patch.object(cloud_api.Cloud, 'async_refresh_account_info',
|
||||
side_effect=asyncio.TimeoutError):
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_verification_failed_500(cloud_hass, mock_read,
|
||||
aioclient_mock):
|
||||
"""Test loading authentication with verify request getting 500."""
|
||||
mock_read.return_value = MOCK_AUTH
|
||||
aioclient_mock.get(url('account.json'), status=500)
|
||||
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_token_refresh_needed_401(cloud_hass, mock_read,
|
||||
aioclient_mock):
|
||||
"""Test loading authentication with refresh needed which gets 401."""
|
||||
mock_read.return_value = MOCK_AUTH
|
||||
aioclient_mock.get(url('account.json'), status=403)
|
||||
aioclient_mock.post(url('o/token/'), status=401)
|
||||
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_token_refresh_needed_500(cloud_hass, mock_read,
|
||||
aioclient_mock):
|
||||
"""Test loading authentication with refresh needed which gets 500."""
|
||||
mock_read.return_value = MOCK_AUTH
|
||||
aioclient_mock.get(url('account.json'), status=403)
|
||||
aioclient_mock.post(url('o/token/'), status=500)
|
||||
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_token_refresh_needed_timeout(cloud_hass, mock_read,
|
||||
aioclient_mock):
|
||||
"""Test loading authentication with refresh timing out."""
|
||||
mock_read.return_value = MOCK_AUTH
|
||||
aioclient_mock.get(url('account.json'), status=403)
|
||||
aioclient_mock.post(url('o/token/'), exc=asyncio.TimeoutError)
|
||||
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_token_refresh_needed_succeeds(cloud_hass, mock_read,
|
||||
aioclient_mock):
|
||||
"""Test loading authentication with refresh timing out."""
|
||||
mock_read.return_value = MOCK_AUTH
|
||||
aioclient_mock.get(url('account.json'), status=403)
|
||||
|
||||
with patch.object(cloud_api.Cloud, 'async_refresh_access_token',
|
||||
return_value=mock_coro(True)) as mock_refresh:
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is None
|
||||
assert len(mock_refresh.mock_calls) == 1
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_load_auth_token(cloud_hass, mock_read, aioclient_mock):
|
||||
"""Test loading authentication with refresh timing out."""
|
||||
mock_read.return_value = MOCK_AUTH
|
||||
aioclient_mock.get(url('account.json'), json={
|
||||
'first_name': 'Paulus',
|
||||
'last_name': 'Schoutsen'
|
||||
})
|
||||
|
||||
result = yield from cloud_api.async_load_auth(cloud_hass)
|
||||
|
||||
assert result is not None
|
||||
assert result.account == {
|
||||
'first_name': 'Paulus',
|
||||
'last_name': 'Schoutsen'
|
||||
}
|
||||
assert result.auth == MOCK_AUTH
|
||||
|
||||
|
||||
def test_cloud_properties():
|
||||
"""Test Cloud class properties."""
|
||||
cloud = cloud_api.Cloud(None, MOCK_AUTH)
|
||||
assert cloud.access_token == MOCK_AUTH['access_token']
|
||||
assert cloud.refresh_token == MOCK_AUTH['refresh_token']
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_refresh_account_info(cloud_hass, aioclient_mock):
|
||||
"""Test refreshing account info."""
|
||||
aioclient_mock.get(url('account.json'), json={
|
||||
'first_name': 'Paulus',
|
||||
'last_name': 'Schoutsen'
|
||||
})
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
assert cloud.account is None
|
||||
result = yield from cloud.async_refresh_account_info()
|
||||
assert result
|
||||
assert cloud.account == {
|
||||
'first_name': 'Paulus',
|
||||
'last_name': 'Schoutsen'
|
||||
}
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_refresh_account_info_500(cloud_hass, aioclient_mock):
|
||||
"""Test refreshing account info and getting 500."""
|
||||
aioclient_mock.get(url('account.json'), status=500)
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
assert cloud.account is None
|
||||
result = yield from cloud.async_refresh_account_info()
|
||||
assert not result
|
||||
assert cloud.account is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_refresh_token(cloud_hass, aioclient_mock, mock_write):
|
||||
"""Test refreshing access token."""
|
||||
aioclient_mock.post(url('o/token/'), json={
|
||||
'access_token': 'refreshed',
|
||||
'expires_in': 10
|
||||
})
|
||||
now = dt_util.utcnow()
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
with patch('homeassistant.components.cloud.cloud_api.utcnow',
|
||||
return_value=now):
|
||||
result = yield from cloud.async_refresh_access_token()
|
||||
assert result
|
||||
assert cloud.auth == {
|
||||
'access_token': 'refreshed',
|
||||
'expires_in': 10,
|
||||
'expires_at': (now + timedelta(seconds=10)).isoformat()
|
||||
}
|
||||
assert len(mock_write.mock_calls) == 1
|
||||
write_hass, write_data = mock_write.mock_calls[0][1]
|
||||
assert write_hass is cloud_hass
|
||||
assert write_data == cloud.auth
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_refresh_token_unknown_error(cloud_hass, aioclient_mock,
|
||||
mock_write):
|
||||
"""Test refreshing access token."""
|
||||
aioclient_mock.post(url('o/token/'), status=500)
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
result = yield from cloud.async_refresh_access_token()
|
||||
assert not result
|
||||
assert cloud.auth == MOCK_AUTH
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_revoke_token(cloud_hass, aioclient_mock, mock_write):
|
||||
"""Test revoking access token."""
|
||||
aioclient_mock.post(url('o/revoke_token/'))
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
yield from cloud.async_revoke_access_token()
|
||||
assert cloud.auth is None
|
||||
assert len(mock_write.mock_calls) == 1
|
||||
write_hass, write_data = mock_write.mock_calls[0][1]
|
||||
assert write_hass is cloud_hass
|
||||
assert write_data is None
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_revoke_token_invalid_client_creds(cloud_hass, aioclient_mock,
|
||||
mock_write):
|
||||
"""Test revoking access token with invalid client credentials."""
|
||||
aioclient_mock.post(url('o/revoke_token/'), status=401)
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
with pytest.raises(cloud_api.UnknownError):
|
||||
yield from cloud.async_revoke_access_token()
|
||||
assert cloud.auth is not None
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_revoke_token_request_error(cloud_hass, aioclient_mock,
|
||||
mock_write):
|
||||
"""Test revoking access token with invalid client credentials."""
|
||||
aioclient_mock.post(url('o/revoke_token/'), exc=aiohttp.ClientError)
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
with pytest.raises(cloud_api.UnknownError):
|
||||
yield from cloud.async_revoke_access_token()
|
||||
assert cloud.auth is not None
|
||||
assert len(mock_write.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_request(cloud_hass, aioclient_mock):
|
||||
"""Test making request to the cloud."""
|
||||
aioclient_mock.post(url('some_endpoint'), json={'hello': 'world'})
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
request = yield from cloud.async_request('post', 'some_endpoint')
|
||||
assert request.status == 200
|
||||
data = yield from request.json()
|
||||
assert data == {'hello': 'world'}
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_request_requiring_refresh_fail(cloud_hass, aioclient_mock):
|
||||
"""Test making request to the cloud."""
|
||||
aioclient_mock.post(url('some_endpoint'), status=403)
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
with patch.object(cloud_api.Cloud, 'async_refresh_access_token',
|
||||
return_value=mock_coro(False)) as mock_refresh:
|
||||
request = yield from cloud.async_request('post', 'some_endpoint')
|
||||
assert request.status == 403
|
||||
assert len(mock_refresh.mock_calls) == 1
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_cloud_request_requiring_refresh_success(cloud_hass, aioclient_mock):
|
||||
"""Test making request to the cloud."""
|
||||
aioclient_mock.post(url('some_endpoint'), status=403)
|
||||
cloud = cloud_api.Cloud(cloud_hass, MOCK_AUTH)
|
||||
with patch.object(cloud_api.Cloud, 'async_refresh_access_token',
|
||||
return_value=mock_coro(True)) as mock_refresh, \
|
||||
patch.object(cloud_api.Cloud, 'async_refresh_account_info',
|
||||
return_value=mock_coro()) as mock_account_info:
|
||||
request = yield from cloud.async_request('post', 'some_endpoint')
|
||||
assert request.status == 403
|
||||
assert len(mock_refresh.mock_calls) == 1
|
||||
assert len(mock_account_info.mock_calls) == 1
|
||||
@@ -5,9 +5,7 @@ from unittest.mock import patch, MagicMock
|
||||
import pytest
|
||||
|
||||
from homeassistant.bootstrap import async_setup_component
|
||||
from homeassistant.components.cloud import DOMAIN, cloud_api
|
||||
|
||||
from tests.common import mock_coro
|
||||
from homeassistant.components.cloud import DOMAIN, auth_api
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -21,6 +19,20 @@ def cloud_client(hass, test_client):
|
||||
return hass.loop.run_until_complete(test_client(hass.http.app))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_auth(cloud_client, hass):
|
||||
"""Fixture to mock authentication."""
|
||||
auth = hass.data[DOMAIN]['auth'] = MagicMock()
|
||||
return auth
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_cognito():
|
||||
"""Mock warrant."""
|
||||
with patch('homeassistant.components.cloud.auth_api._cognito') as mock_cog:
|
||||
yield mock_cog()
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_account_view_no_account(cloud_client):
|
||||
"""Test fetching account if no account available."""
|
||||
@@ -29,129 +41,300 @@ def test_account_view_no_account(cloud_client):
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_account_view(hass, cloud_client):
|
||||
def test_account_view(mock_auth, cloud_client):
|
||||
"""Test fetching account if no account available."""
|
||||
cloud = MagicMock(account={'test': 'account'})
|
||||
hass.data[DOMAIN]['cloud'] = cloud
|
||||
mock_auth.account = MagicMock(email='hello@home-assistant.io')
|
||||
req = yield from cloud_client.get('/api/cloud/account')
|
||||
assert req.status == 200
|
||||
result = yield from req.json()
|
||||
assert result == {'test': 'account'}
|
||||
assert result == {'email': 'hello@home-assistant.io'}
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_login_view(hass, cloud_client):
|
||||
def test_login_view(mock_auth, cloud_client):
|
||||
"""Test logging in."""
|
||||
cloud = MagicMock(account={'test': 'account'})
|
||||
cloud.async_refresh_account_info.return_value = mock_coro(None)
|
||||
|
||||
with patch.object(cloud_api, 'async_login',
|
||||
MagicMock(return_value=mock_coro(cloud))):
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'username': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
mock_auth.account = MagicMock(email='hello@home-assistant.io')
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'email': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
|
||||
assert req.status == 200
|
||||
|
||||
result = yield from req.json()
|
||||
assert result == {'test': 'account'}
|
||||
assert hass.data[DOMAIN]['cloud'] is cloud
|
||||
assert result == {'email': 'hello@home-assistant.io'}
|
||||
assert len(mock_auth.login.mock_calls) == 1
|
||||
result_user, result_pass = mock_auth.login.mock_calls[0][1]
|
||||
assert result_user == 'my_username'
|
||||
assert result_pass == 'my_password'
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_login_view_invalid_json(hass, cloud_client):
|
||||
def test_login_view_invalid_json(mock_auth, cloud_client):
|
||||
"""Try logging in with invalid JSON."""
|
||||
req = yield from cloud_client.post('/api/cloud/login', data='Not JSON')
|
||||
assert req.status == 400
|
||||
assert 'cloud' not in hass.data[DOMAIN]
|
||||
assert len(mock_auth.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_login_view_invalid_schema(hass, cloud_client):
|
||||
def test_login_view_invalid_schema(mock_auth, cloud_client):
|
||||
"""Try logging in with invalid schema."""
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'invalid': 'schema'
|
||||
})
|
||||
assert req.status == 400
|
||||
assert 'cloud' not in hass.data[DOMAIN]
|
||||
assert len(mock_auth.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_login_view_request_timeout(hass, cloud_client):
|
||||
def test_login_view_request_timeout(mock_auth, cloud_client):
|
||||
"""Test request timeout while trying to log in."""
|
||||
with patch.object(cloud_api, 'async_login',
|
||||
MagicMock(side_effect=asyncio.TimeoutError)):
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'username': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
mock_auth.login.side_effect = asyncio.TimeoutError
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'email': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
|
||||
assert req.status == 502
|
||||
assert 'cloud' not in hass.data[DOMAIN]
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_login_view_invalid_credentials(hass, cloud_client):
|
||||
def test_login_view_invalid_credentials(mock_auth, cloud_client):
|
||||
"""Test logging in with invalid credentials."""
|
||||
with patch.object(cloud_api, 'async_login',
|
||||
MagicMock(side_effect=cloud_api.Unauthenticated)):
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'username': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
mock_auth.login.side_effect = auth_api.Unauthenticated
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'email': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
|
||||
assert req.status == 401
|
||||
assert 'cloud' not in hass.data[DOMAIN]
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_login_view_unknown_error(hass, cloud_client):
|
||||
def test_login_view_unknown_error(mock_auth, cloud_client):
|
||||
"""Test unknown error while logging in."""
|
||||
with patch.object(cloud_api, 'async_login',
|
||||
MagicMock(side_effect=cloud_api.UnknownError)):
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'username': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
mock_auth.login.side_effect = auth_api.UnknownError
|
||||
req = yield from cloud_client.post('/api/cloud/login', json={
|
||||
'email': 'my_username',
|
||||
'password': 'my_password'
|
||||
})
|
||||
|
||||
assert req.status == 500
|
||||
assert 'cloud' not in hass.data[DOMAIN]
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_logout_view(hass, cloud_client):
|
||||
def test_logout_view(mock_auth, cloud_client):
|
||||
"""Test logging out."""
|
||||
cloud = MagicMock()
|
||||
cloud.async_revoke_access_token.return_value = mock_coro(None)
|
||||
hass.data[DOMAIN]['cloud'] = cloud
|
||||
|
||||
req = yield from cloud_client.post('/api/cloud/logout')
|
||||
assert req.status == 200
|
||||
data = yield from req.json()
|
||||
assert data == {'result': 'ok'}
|
||||
assert 'cloud' not in hass.data[DOMAIN]
|
||||
assert data == {'message': 'ok'}
|
||||
assert len(mock_auth.logout.mock_calls) == 1
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_logout_view_request_timeout(hass, cloud_client):
|
||||
def test_logout_view_request_timeout(mock_auth, cloud_client):
|
||||
"""Test timeout while logging out."""
|
||||
cloud = MagicMock()
|
||||
cloud.async_revoke_access_token.side_effect = asyncio.TimeoutError
|
||||
hass.data[DOMAIN]['cloud'] = cloud
|
||||
|
||||
mock_auth.logout.side_effect = asyncio.TimeoutError
|
||||
req = yield from cloud_client.post('/api/cloud/logout')
|
||||
assert req.status == 502
|
||||
assert 'cloud' in hass.data[DOMAIN]
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_logout_view_unknown_error(hass, cloud_client):
|
||||
def test_logout_view_unknown_error(mock_auth, cloud_client):
|
||||
"""Test unknown error while loggin out."""
|
||||
cloud = MagicMock()
|
||||
cloud.async_revoke_access_token.side_effect = cloud_api.UnknownError
|
||||
hass.data[DOMAIN]['cloud'] = cloud
|
||||
|
||||
mock_auth.logout.side_effect = auth_api.UnknownError
|
||||
req = yield from cloud_client.post('/api/cloud/logout')
|
||||
assert req.status == 502
|
||||
assert 'cloud' in hass.data[DOMAIN]
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_register_view(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post('/api/cloud/register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'password': 'falcon42'
|
||||
})
|
||||
assert req.status == 200
|
||||
assert len(mock_cognito.register.mock_calls) == 1
|
||||
result_email, result_pass = mock_cognito.register.mock_calls[0][1]
|
||||
assert result_email == 'hello@bla.com'
|
||||
assert result_pass == 'falcon42'
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_register_view_bad_data(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post('/api/cloud/register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'not_password': 'falcon'
|
||||
})
|
||||
assert req.status == 400
|
||||
assert len(mock_cognito.logout.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_register_view_request_timeout(mock_cognito, cloud_client):
|
||||
"""Test timeout while logging out."""
|
||||
mock_cognito.register.side_effect = asyncio.TimeoutError
|
||||
req = yield from cloud_client.post('/api/cloud/register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'password': 'falcon42'
|
||||
})
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_register_view_unknown_error(mock_cognito, cloud_client):
|
||||
"""Test unknown error while loggin out."""
|
||||
mock_cognito.register.side_effect = auth_api.UnknownError
|
||||
req = yield from cloud_client.post('/api/cloud/register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'password': 'falcon42'
|
||||
})
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_register_view(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post('/api/cloud/confirm_register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'confirmation_code': '123456'
|
||||
})
|
||||
assert req.status == 200
|
||||
assert len(mock_cognito.confirm_sign_up.mock_calls) == 1
|
||||
result_code, result_email = mock_cognito.confirm_sign_up.mock_calls[0][1]
|
||||
assert result_email == 'hello@bla.com'
|
||||
assert result_code == '123456'
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_register_view_bad_data(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post('/api/cloud/confirm_register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'not_confirmation_code': '123456'
|
||||
})
|
||||
assert req.status == 400
|
||||
assert len(mock_cognito.confirm_sign_up.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_register_view_request_timeout(mock_cognito, cloud_client):
|
||||
"""Test timeout while logging out."""
|
||||
mock_cognito.confirm_sign_up.side_effect = asyncio.TimeoutError
|
||||
req = yield from cloud_client.post('/api/cloud/confirm_register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'confirmation_code': '123456'
|
||||
})
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_register_view_unknown_error(mock_cognito, cloud_client):
|
||||
"""Test unknown error while loggin out."""
|
||||
mock_cognito.confirm_sign_up.side_effect = auth_api.UnknownError
|
||||
req = yield from cloud_client.post('/api/cloud/confirm_register', json={
|
||||
'email': 'hello@bla.com',
|
||||
'confirmation_code': '123456'
|
||||
})
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_forgot_password_view(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post('/api/cloud/forgot_password', json={
|
||||
'email': 'hello@bla.com',
|
||||
})
|
||||
assert req.status == 200
|
||||
assert len(mock_cognito.initiate_forgot_password.mock_calls) == 1
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_forgot_password_view_bad_data(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post('/api/cloud/forgot_password', json={
|
||||
'not_email': 'hello@bla.com',
|
||||
})
|
||||
assert req.status == 400
|
||||
assert len(mock_cognito.initiate_forgot_password.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_forgot_password_view_request_timeout(mock_cognito, cloud_client):
|
||||
"""Test timeout while logging out."""
|
||||
mock_cognito.initiate_forgot_password.side_effect = asyncio.TimeoutError
|
||||
req = yield from cloud_client.post('/api/cloud/forgot_password', json={
|
||||
'email': 'hello@bla.com',
|
||||
})
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_forgot_password_view_unknown_error(mock_cognito, cloud_client):
|
||||
"""Test unknown error while loggin out."""
|
||||
mock_cognito.initiate_forgot_password.side_effect = auth_api.UnknownError
|
||||
req = yield from cloud_client.post('/api/cloud/forgot_password', json={
|
||||
'email': 'hello@bla.com',
|
||||
})
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_forgot_password_view(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post(
|
||||
'/api/cloud/confirm_forgot_password', json={
|
||||
'email': 'hello@bla.com',
|
||||
'confirmation_code': '123456',
|
||||
'new_password': 'hello2',
|
||||
})
|
||||
assert req.status == 200
|
||||
assert len(mock_cognito.confirm_forgot_password.mock_calls) == 1
|
||||
result_code, result_new_password = \
|
||||
mock_cognito.confirm_forgot_password.mock_calls[0][1]
|
||||
assert result_code == '123456'
|
||||
assert result_new_password == 'hello2'
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_forgot_password_view_bad_data(mock_cognito, cloud_client):
|
||||
"""Test logging out."""
|
||||
req = yield from cloud_client.post(
|
||||
'/api/cloud/confirm_forgot_password', json={
|
||||
'email': 'hello@bla.com',
|
||||
'not_confirmation_code': '123456',
|
||||
'new_password': 'hello2',
|
||||
})
|
||||
assert req.status == 400
|
||||
assert len(mock_cognito.confirm_forgot_password.mock_calls) == 0
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_forgot_password_view_request_timeout(mock_cognito,
|
||||
cloud_client):
|
||||
"""Test timeout while logging out."""
|
||||
mock_cognito.confirm_forgot_password.side_effect = asyncio.TimeoutError
|
||||
req = yield from cloud_client.post(
|
||||
'/api/cloud/confirm_forgot_password', json={
|
||||
'email': 'hello@bla.com',
|
||||
'confirmation_code': '123456',
|
||||
'new_password': 'hello2',
|
||||
})
|
||||
assert req.status == 502
|
||||
|
||||
|
||||
@asyncio.coroutine
|
||||
def test_confirm_forgot_password_view_unknown_error(mock_cognito,
|
||||
cloud_client):
|
||||
"""Test unknown error while loggin out."""
|
||||
mock_cognito.confirm_forgot_password.side_effect = auth_api.UnknownError
|
||||
req = yield from cloud_client.post(
|
||||
'/api/cloud/confirm_forgot_password', json={
|
||||
'email': 'hello@bla.com',
|
||||
'confirmation_code': '123456',
|
||||
'new_password': 'hello2',
|
||||
})
|
||||
assert req.status == 502
|
||||
Reference in new issue
Block a user