* add a component for hangouts

* add a notify component for hangouts

* add an extra message as title

* add support to listen to all conversations hangouts has

* move hangouts to package and add parameter documentation

* update .coveragerc and requirements_all.txt

* makes linter happy again

* bugfix

* add conversations parameter to command words

* Move the resolution of conversation names to conversations in own a function

* typo

* rename group of exclusion form 'id' to 'id or name'

* refactoring and use config_flow

* makes linter happy again

* remove unused imports

* fix not working regex commands

* fix translations

* cleanup

* remove step_init

* remove logging entry

* clean up events

* move constant

* remove unsed import

* add new files to .converagerc

* isort imports

* add hangouts_utils to ignored packages

* upadte doc and format

* fix I/O not in executor jon

* rename SERVICE_UPDATE_USERS_AND_CONVERSATIONS to SERVICE_UPDATE

* move EVENT_HANGOUTS_{CONNECTED,DISCONNECTED} to dispatcher

* add config flow tests

* Update tox.ini
This commit is contained in:
Marcel Hoppe 2018-08-24 10:39:35 +02:00 committed by Paulus Schoutsen
parent dd9d53c83e
commit ef0eab0f40
16 changed files with 831 additions and 1 deletions

View file

@ -116,6 +116,12 @@ omit =
homeassistant/components/google.py
homeassistant/components/*/google.py
homeassistant/components/hangouts/__init__.py
homeassistant/components/hangouts/const.py
homeassistant/components/hangouts/hangouts_bot.py
homeassistant/components/hangouts/hangups_utils.py
homeassistant/components/*/hangouts.py
homeassistant/components/hdmi_cec.py
homeassistant/components/*/hdmi_cec.py

View file

@ -0,0 +1,31 @@
{
"config": {
"abort": {
"already_configured": "Google Hangouts is already configured",
"unknown": "Unknown error occurred."
},
"error": {
"invalid_2fa": "Invalid 2 Factor Authorization, please try again.",
"invalid_2fa_method": "Invalig 2FA Method (Verify on Phone).",
"invalid_login": "Invalid Login, please try again."
},
"step": {
"2fa": {
"data": {
"2fa": "2FA Pin"
},
"description": "",
"title": "2-Factor-Authorization"
},
"user": {
"data": {
"email": "E-Mail Address",
"password": "Password"
},
"description": "",
"title": "Google Hangouts Login"
}
},
"title": "Google Hangouts"
}
}

View file

@ -0,0 +1,87 @@
"""
The hangouts bot component.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/hangouts/
"""
import logging
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.helpers import dispatcher
from .config_flow import configured_hangouts
from .const import (
CONF_BOT, CONF_COMMANDS, CONF_REFRESH_TOKEN, DOMAIN,
EVENT_HANGOUTS_CONNECTED, EVENT_HANGOUTS_CONVERSATIONS_CHANGED,
MESSAGE_SCHEMA, SERVICE_SEND_MESSAGE,
SERVICE_UPDATE)
REQUIREMENTS = ['hangups==0.4.5']
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
"""Set up the Hangouts bot component."""
config = config.get(DOMAIN, [])
hass.data[DOMAIN] = {CONF_COMMANDS: config[CONF_COMMANDS]}
if configured_hangouts(hass) is None:
hass.async_add_job(hass.config_entries.flow.async_init(
DOMAIN, context={'source': config_entries.SOURCE_IMPORT}
))
return True
async def async_setup_entry(hass, config):
"""Set up a config entry."""
from hangups.auth import GoogleAuthError
try:
from .hangouts_bot import HangoutsBot
bot = HangoutsBot(
hass,
config.data.get(CONF_REFRESH_TOKEN),
hass.data[DOMAIN][CONF_COMMANDS])
hass.data[DOMAIN][CONF_BOT] = bot
except GoogleAuthError as exception:
_LOGGER.error("Hangouts failed to log in: %s", str(exception))
return False
dispatcher.async_dispatcher_connect(
hass,
EVENT_HANGOUTS_CONNECTED,
bot.async_handle_update_users_and_conversations)
dispatcher.async_dispatcher_connect(
hass,
EVENT_HANGOUTS_CONVERSATIONS_CHANGED,
bot.async_update_conversation_commands)
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP,
bot.async_handle_hass_stop)
await bot.async_connect()
hass.services.async_register(DOMAIN, SERVICE_SEND_MESSAGE,
bot.async_handle_send_message,
schema=MESSAGE_SCHEMA)
hass.services.async_register(DOMAIN,
SERVICE_UPDATE,
bot.
async_handle_update_users_and_conversations,
schema=vol.Schema({}))
return True
async def async_unload_entry(hass, _):
"""Unload a config entry."""
bot = hass.data[DOMAIN].pop(CONF_BOT)
await bot.async_disconnect()
return True

View file

@ -0,0 +1,107 @@
"""Config flow to configure Google Hangouts."""
import voluptuous as vol
from homeassistant import config_entries, data_entry_flow
from homeassistant.const import CONF_EMAIL, CONF_PASSWORD
from homeassistant.core import callback
from .const import CONF_2FA, CONF_REFRESH_TOKEN
from .const import DOMAIN as HANGOUTS_DOMAIN
@callback
def configured_hangouts(hass):
"""Return the configures Google Hangouts Account."""
entries = hass.config_entries.async_entries(HANGOUTS_DOMAIN)
if entries:
return entries[0]
return None
@config_entries.HANDLERS.register(HANGOUTS_DOMAIN)
class HangoutsFlowHandler(data_entry_flow.FlowHandler):
"""Config flow Google Hangouts."""
VERSION = 1
def __init__(self):
"""Initialize Google Hangouts config flow."""
self._credentials = None
self._refresh_token = None
async def async_step_user(self, user_input=None):
"""Handle a flow start."""
errors = {}
if configured_hangouts(self.hass) is not None:
return self.async_abort(reason="already_configured")
if user_input is not None:
from hangups import get_auth
from .hangups_utils import (HangoutsCredentials,
HangoutsRefreshToken,
GoogleAuthError, Google2FAError)
self._credentials = HangoutsCredentials(user_input[CONF_EMAIL],
user_input[CONF_PASSWORD])
self._refresh_token = HangoutsRefreshToken(None)
try:
await self.hass.async_add_executor_job(get_auth,
self._credentials,
self._refresh_token)
return await self.async_step_final()
except GoogleAuthError as err:
if isinstance(err, Google2FAError):
return await self.async_step_2fa()
msg = str(err)
if msg == 'Unknown verification code input':
errors['base'] = 'invalid_2fa_method'
else:
errors['base'] = 'invalid_login'
return self.async_show_form(
step_id='user',
data_schema=vol.Schema({
vol.Required(CONF_EMAIL): str,
vol.Required(CONF_PASSWORD): str
}),
errors=errors
)
async def async_step_2fa(self, user_input=None):
"""Handle the 2fa step, if needed."""
errors = {}
if user_input is not None:
from hangups import get_auth
from .hangups_utils import GoogleAuthError
self._credentials.set_verification_code(user_input[CONF_2FA])
try:
await self.hass.async_add_executor_job(get_auth,
self._credentials,
self._refresh_token)
return await self.async_step_final()
except GoogleAuthError:
errors['base'] = 'invalid_2fa'
return self.async_show_form(
step_id=CONF_2FA,
data_schema=vol.Schema({
vol.Required(CONF_2FA): str,
}),
errors=errors
)
async def async_step_final(self):
"""Handle the final step, create the config entry."""
return self.async_create_entry(
title=self._credentials.get_email(),
data={
CONF_EMAIL: self._credentials.get_email(),
CONF_REFRESH_TOKEN: self._refresh_token.get()
})
async def async_step_import(self, _):
"""Handle a flow import."""
return self.async_abort(reason='already_configured')

View file

@ -0,0 +1,78 @@
"""Constants for Google Hangouts Component."""
import logging
import voluptuous as vol
from homeassistant.components.notify import ATTR_MESSAGE, ATTR_TARGET
from homeassistant.const import CONF_NAME
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger('homeassistant.components.hangouts')
DOMAIN = 'hangouts'
CONF_2FA = '2fa'
CONF_REFRESH_TOKEN = 'refresh_token'
CONF_BOT = 'bot'
CONF_CONVERSATIONS = 'conversations'
CONF_DEFAULT_CONVERSATIONS = 'default_conversations'
CONF_COMMANDS = 'commands'
CONF_WORD = 'word'
CONF_EXPRESSION = 'expression'
EVENT_HANGOUTS_COMMAND = 'hangouts_command'
EVENT_HANGOUTS_CONNECTED = 'hangouts_connected'
EVENT_HANGOUTS_DISCONNECTED = 'hangouts_disconnected'
EVENT_HANGOUTS_USERS_CHANGED = 'hangouts_users_changed'
EVENT_HANGOUTS_CONVERSATIONS_CHANGED = 'hangouts_conversations_changed'
CONF_CONVERSATION_ID = 'id'
CONF_CONVERSATION_NAME = 'name'
SERVICE_SEND_MESSAGE = 'send_message'
SERVICE_UPDATE = 'update'
TARGETS_SCHEMA = vol.All(
vol.Schema({
vol.Exclusive(CONF_CONVERSATION_ID, 'id or name'): cv.string,
vol.Exclusive(CONF_CONVERSATION_NAME, 'id or name'): cv.string
}),
cv.has_at_least_one_key(CONF_CONVERSATION_ID, CONF_CONVERSATION_NAME)
)
MESSAGE_SEGMENT_SCHEMA = vol.Schema({
vol.Required('text'): cv.string,
vol.Optional('is_bold'): cv.boolean,
vol.Optional('is_italic'): cv.boolean,
vol.Optional('is_strikethrough'): cv.boolean,
vol.Optional('is_underline'): cv.boolean,
vol.Optional('parse_str'): cv.boolean,
vol.Optional('link_target'): cv.string
})
MESSAGE_SCHEMA = vol.Schema({
vol.Required(ATTR_TARGET): [TARGETS_SCHEMA],
vol.Required(ATTR_MESSAGE): [MESSAGE_SEGMENT_SCHEMA]
})
COMMAND_SCHEMA = vol.All(
# Basic Schema
vol.Schema({
vol.Exclusive(CONF_WORD, 'trigger'): cv.string,
vol.Exclusive(CONF_EXPRESSION, 'trigger'): cv.is_regex,
vol.Required(CONF_NAME): cv.string,
vol.Optional(CONF_CONVERSATIONS): [TARGETS_SCHEMA]
}),
# Make sure it's either a word or an expression command
cv.has_at_least_one_key(CONF_WORD, CONF_EXPRESSION)
)
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_COMMANDS, default=[]): [COMMAND_SCHEMA]
})
}, extra=vol.ALLOW_EXTRA)

View file

@ -0,0 +1,229 @@
"""The Hangouts Bot."""
import logging
import re
from homeassistant.helpers import dispatcher
from .const import (
ATTR_MESSAGE, ATTR_TARGET, CONF_CONVERSATIONS, CONF_EXPRESSION, CONF_NAME,
CONF_WORD, DOMAIN, EVENT_HANGOUTS_COMMAND, EVENT_HANGOUTS_CONNECTED,
EVENT_HANGOUTS_CONVERSATIONS_CHANGED, EVENT_HANGOUTS_DISCONNECTED)
_LOGGER = logging.getLogger(__name__)
class HangoutsBot:
"""The Hangouts Bot."""
def __init__(self, hass, refresh_token, commands):
"""Set up the client."""
self.hass = hass
self._connected = False
self._refresh_token = refresh_token
self._commands = commands
self._word_commands = None
self._expression_commands = None
self._client = None
self._user_list = None
self._conversation_list = None
def _resolve_conversation_name(self, name):
for conv in self._conversation_list.get_all():
if conv.name == name:
return conv
return None
def async_update_conversation_commands(self, _):
"""Refresh the commands for every conversation."""
self._word_commands = {}
self._expression_commands = {}
for command in self._commands:
if command.get(CONF_CONVERSATIONS):
conversations = []
for conversation in command.get(CONF_CONVERSATIONS):
if 'id' in conversation:
conversations.append(conversation['id'])
elif 'name' in conversation:
conversations.append(self._resolve_conversation_name(
conversation['name']).id_)
command['_' + CONF_CONVERSATIONS] = conversations
else:
command['_' + CONF_CONVERSATIONS] = \
[conv.id_ for conv in self._conversation_list.get_all()]
if command.get(CONF_WORD):
for conv_id in command['_' + CONF_CONVERSATIONS]:
if conv_id not in self._word_commands:
self._word_commands[conv_id] = {}
word = command[CONF_WORD].lower()
self._word_commands[conv_id][word] = command
elif command.get(CONF_EXPRESSION):
command['_' + CONF_EXPRESSION] = re.compile(
command.get(CONF_EXPRESSION))
for conv_id in command['_' + CONF_CONVERSATIONS]:
if conv_id not in self._expression_commands:
self._expression_commands[conv_id] = []
self._expression_commands[conv_id].append(command)
try:
self._conversation_list.on_event.remove_observer(
self._handle_conversation_event)
except ValueError:
pass
self._conversation_list.on_event.add_observer(
self._handle_conversation_event)
def _handle_conversation_event(self, event):
from hangups import ChatMessageEvent
if event.__class__ is ChatMessageEvent:
self._handle_conversation_message(
event.conversation_id, event.user_id, event)
def _handle_conversation_message(self, conv_id, user_id, event):
"""Handle a message sent to a conversation."""
user = self._user_list.get_user(user_id)
if user.is_self:
return
_LOGGER.debug("Handling message '%s' from %s",
event.text, user.full_name)
event_data = None
pieces = event.text.split(' ')
cmd = pieces[0].lower()
command = self._word_commands.get(conv_id, {}).get(cmd)
if command:
event_data = {
'command': command[CONF_NAME],
'conversation_id': conv_id,
'user_id': user_id,
'user_name': user.full_name,
'data': pieces[1:]
}
else:
# After single-word commands, check all regex commands in the room
for command in self._expression_commands.get(conv_id, []):
match = command['_' + CONF_EXPRESSION].match(event.text)
if not match:
continue
event_data = {
'command': command[CONF_NAME],
'conversation_id': conv_id,
'user_id': user_id,
'user_name': user.full_name,
'data': match.groupdict()
}
if event_data is not None:
self.hass.bus.fire(EVENT_HANGOUTS_COMMAND, event_data)
async def async_connect(self):
"""Login to the Google Hangouts."""
from .hangups_utils import HangoutsRefreshToken, HangoutsCredentials
from hangups import Client
from hangups import get_auth
session = await self.hass.async_add_executor_job(
get_auth, HangoutsCredentials(None, None, None),
HangoutsRefreshToken(self._refresh_token))
self._client = Client(session)
self._client.on_connect.add_observer(self._on_connect)
self._client.on_disconnect.add_observer(self._on_disconnect)
self.hass.loop.create_task(self._client.connect())
def _on_connect(self):
_LOGGER.debug('Connected!')
self._connected = True
dispatcher.async_dispatcher_send(self.hass, EVENT_HANGOUTS_CONNECTED)
def _on_disconnect(self):
"""Handle disconnecting."""
_LOGGER.debug('Connection lost!')
self._connected = False
dispatcher.async_dispatcher_send(self.hass,
EVENT_HANGOUTS_DISCONNECTED)
async def async_disconnect(self):
"""Disconnect the client if it is connected."""
if self._connected:
await self._client.disconnect()
async def async_handle_hass_stop(self, _):
"""Run once when Home Assistant stops."""
await self.async_disconnect()
async def _async_send_message(self, message, targets):
conversations = []
for target in targets:
conversation = None
if 'id' in target:
conversation = self._conversation_list.get(target['id'])
elif 'name' in target:
conversation = self._resolve_conversation_name(target['name'])
if conversation is not None:
conversations.append(conversation)
if not conversations:
return False
from hangups import ChatMessageSegment, hangouts_pb2
messages = []
for segment in message:
if 'parse_str' in segment and segment['parse_str']:
messages.extend(ChatMessageSegment.from_str(segment['text']))
else:
if 'parse_str' in segment:
del segment['parse_str']
messages.append(ChatMessageSegment(**segment))
messages.append(ChatMessageSegment('',
segment_type=hangouts_pb2.
SEGMENT_TYPE_LINE_BREAK))
if not messages:
return False
for conv in conversations:
await conv.send_message(messages)
async def _async_list_conversations(self):
import hangups
self._user_list, self._conversation_list = \
(await hangups.build_user_conversation_list(self._client))
users = {}
conversations = {}
for user in self._user_list.get_all():
users[str(user.id_.chat_id)] = {'full_name': user.full_name,
'is_self': user.is_self}
for conv in self._conversation_list.get_all():
users_in_conversation = {}
for user in conv.users:
users_in_conversation[str(user.id_.chat_id)] = \
{'full_name': user.full_name, 'is_self': user.is_self}
conversations[str(conv.id_)] = \
{'name': conv.name, 'users': users_in_conversation}
self.hass.states.async_set("{}.users".format(DOMAIN),
len(self._user_list.get_all()),
attributes=users)
self.hass.states.async_set("{}.conversations".format(DOMAIN),
len(self._conversation_list.get_all()),
attributes=conversations)
dispatcher.async_dispatcher_send(self.hass,
EVENT_HANGOUTS_CONVERSATIONS_CHANGED,
conversations)
async def async_handle_send_message(self, service):
"""Handle the send_message service."""
await self._async_send_message(service.data[ATTR_MESSAGE],
service.data[ATTR_TARGET])
async def async_handle_update_users_and_conversations(self, _=None):
"""Handle the update_users_and_conversations service."""
await self._async_list_conversations()

View file

@ -0,0 +1,81 @@
"""Utils needed for Google Hangouts."""
from hangups import CredentialsPrompt, GoogleAuthError, RefreshTokenCache
class Google2FAError(GoogleAuthError):
"""A Google authentication request failed."""
class HangoutsCredentials(CredentialsPrompt):
"""Google account credentials.
This implementation gets the user data as params.
"""
def __init__(self, email, password, pin=None):
"""Google account credentials.
:param email: Google account email address.
:param password: Google account password.
:param pin: Google account verification code.
"""
self._email = email
self._password = password
self._pin = pin
def get_email(self):
"""Return email.
:return: Google account email address.
"""
return self._email
def get_password(self):
"""Return password.
:return: Google account password.
"""
return self._password
def get_verification_code(self):
"""Return the verification code.
:return: Google account verification code.
"""
if self._pin is None:
raise Google2FAError()
return self._pin
def set_verification_code(self, pin):
"""Set the verification code.
:param pin: Google account verification code.
"""
self._pin = pin
class HangoutsRefreshToken(RefreshTokenCache):
"""Memory-based cache for refresh token."""
def __init__(self, token):
"""Memory-based cache for refresh token.
:param token: Initial refresh token.
"""
super().__init__("")
self._token = token
def get(self):
"""Get cached refresh token.
:return: Cached refresh token.
"""
return self._token
def set(self, refresh_token):
"""Cache a refresh token.
:param refresh_token: Refresh token to cache.
"""
self._token = refresh_token

View file

@ -0,0 +1,12 @@
update:
description: Updates the list of users and conversations.
send_message:
description: Send a notification to a specific target.
fields:
target:
description: List of targets with id or name. [Required]
example: '[{"id": "UgxrXzVrARmjx_C6AZx4AaABAagBo-6UCw"}, {"name": "Test Conversation"}]'
message:
description: List of message segments, only the "text" field is required in every segment. [Required]
example: '[{"text":"test", "is_bold": false, "is_italic": false, "is_strikethrough": false, "is_underline": false, "parse_str": false, "link_target": "http://google.com"}, ...]'

View file

@ -0,0 +1,31 @@
{
"config": {
"abort": {
"already_configured": "Google Hangouts is already configured",
"unknown": "Unknown error occurred."
},
"error": {
"invalid_login": "Invalid Login, please try again.",
"invalid_2fa": "Invalid 2 Factor Authorization, please try again.",
"invalid_2fa_method": "Invalig 2FA Method (Verify on Phone)."
},
"step": {
"user": {
"data": {
"email": "E-Mail Address",
"password": "Password"
},
"description": "",
"title": "Google Hangouts Login"
},
"2fa": {
"data": {
"2fa": "2FA Pin"
},
"description": "",
"title": "2-Factor-Authorization"
}
},
"title": "Google Hangouts"
}
}

View file

@ -0,0 +1,66 @@
"""
Hangouts notification service.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/notify.hangouts/
"""
import logging
import voluptuous as vol
from homeassistant.components.notify import (ATTR_TARGET, PLATFORM_SCHEMA,
NOTIFY_SERVICE_SCHEMA,
BaseNotificationService,
ATTR_MESSAGE)
from homeassistant.components.hangouts.const \
import (DOMAIN, SERVICE_SEND_MESSAGE,
TARGETS_SCHEMA, CONF_DEFAULT_CONVERSATIONS)
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = [DOMAIN]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_DEFAULT_CONVERSATIONS): [TARGETS_SCHEMA]
})
NOTIFY_SERVICE_SCHEMA = NOTIFY_SERVICE_SCHEMA.extend({
vol.Optional(ATTR_TARGET): [TARGETS_SCHEMA]
})
def get_service(hass, config, discovery_info=None):
"""Get the Hangouts notification service."""
return HangoutsNotificationService(config.get(CONF_DEFAULT_CONVERSATIONS))
class HangoutsNotificationService(BaseNotificationService):
"""Send Notifications to Hangouts conversations."""
def __init__(self, default_conversations):
"""Set up the notification service."""
self._default_conversations = default_conversations
def send_message(self, message="", **kwargs):
"""Send the message to the Google Hangouts server."""
target_conversations = None
if ATTR_TARGET in kwargs:
target_conversations = []
for target in kwargs.get(ATTR_TARGET):
target_conversations.append({'id': target})
else:
target_conversations = self._default_conversations
messages = []
if 'title' in kwargs:
messages.append({'text': kwargs['title'], 'is_bold': True})
messages.append({'text': message, 'parse_str': True})
service_data = {
ATTR_TARGET: target_conversations,
ATTR_MESSAGE: messages
}
return self.hass.services.call(
DOMAIN, SERVICE_SEND_MESSAGE, service_data=service_data)

View file

@ -136,6 +136,7 @@ HANDLERS = Registry()
# Components that have config flows. In future we will auto-generate this list.
FLOWS = [
'cast',
'hangouts',
'deconz',
'homematicip_cloud',
'hue',

View file

@ -413,6 +413,9 @@ ha-ffmpeg==1.9
# homeassistant.components.media_player.philips_js
ha-philipsjs==0.0.5
# homeassistant.components.hangouts
hangups==0.4.5
# homeassistant.components.sensor.geo_rss_events
haversine==0.4.5

View file

@ -71,6 +71,9 @@ gTTS-token==1.1.1
# homeassistant.components.ffmpeg
ha-ffmpeg==1.9
# homeassistant.components.hangouts
hangups==0.4.5
# homeassistant.components.sensor.geo_rss_events
haversine==0.4.5

View file

@ -51,6 +51,7 @@ TEST_REQUIREMENTS = (
'feedparser',
'foobot_async',
'gTTS-token',
'hangups',
'HAP-python',
'ha-ffmpeg',
'haversine',
@ -105,7 +106,8 @@ TEST_REQUIREMENTS = (
IGNORE_PACKAGES = (
'homeassistant.components.recorder.models',
'homeassistant.components.homekit.*'
'homeassistant.components.homekit.*',
'homeassistant.components.hangouts.hangups_utils'
)
IGNORE_PIN = ('colorlog>2.1,<3', 'keyring>=9.3,<10.0', 'urllib3')

View file

@ -0,0 +1 @@
"""Tests for the Hangouts Component."""

View file

@ -0,0 +1,92 @@
"""Tests for the Google Hangouts config flow."""
from unittest.mock import patch
from homeassistant import data_entry_flow
from homeassistant.components.hangouts import config_flow
async def test_flow_works(hass, aioclient_mock):
"""Test config flow without 2fa."""
flow = config_flow.HangoutsFlowHandler()
flow.hass = hass
with patch('hangups.get_auth'):
result = await flow.async_step_user(
{'email': 'test@test.com', 'password': '1232456'})
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'test@test.com'
async def test_flow_works_with_2fa(hass, aioclient_mock):
"""Test config flow with 2fa."""
from homeassistant.components.hangouts.hangups_utils import Google2FAError
flow = config_flow.HangoutsFlowHandler()
flow.hass = hass
with patch('hangups.get_auth', side_effect=Google2FAError):
result = await flow.async_step_user(
{'email': 'test@test.com', 'password': '1232456'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == '2fa'
with patch('hangups.get_auth'):
result = await flow.async_step_2fa({'2fa': 123456})
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'test@test.com'
async def test_flow_with_unknown_2fa(hass, aioclient_mock):
"""Test config flow with invalid 2fa method."""
from homeassistant.components.hangouts.hangups_utils import GoogleAuthError
flow = config_flow.HangoutsFlowHandler()
flow.hass = hass
with patch('hangups.get_auth',
side_effect=GoogleAuthError('Unknown verification code input')):
result = await flow.async_step_user(
{'email': 'test@test.com', 'password': '1232456'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['errors']['base'] == 'invalid_2fa_method'
async def test_flow_invalid_login(hass, aioclient_mock):
"""Test config flow with invalid 2fa method."""
from homeassistant.components.hangouts.hangups_utils import GoogleAuthError
flow = config_flow.HangoutsFlowHandler()
flow.hass = hass
with patch('hangups.get_auth',
side_effect=GoogleAuthError):
result = await flow.async_step_user(
{'email': 'test@test.com', 'password': '1232456'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['errors']['base'] == 'invalid_login'
async def test_flow_invalid_2fa(hass, aioclient_mock):
"""Test config flow with 2fa."""
from homeassistant.components.hangouts.hangups_utils import Google2FAError
flow = config_flow.HangoutsFlowHandler()
flow.hass = hass
with patch('hangups.get_auth', side_effect=Google2FAError):
result = await flow.async_step_user(
{'email': 'test@test.com', 'password': '1232456'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == '2fa'
with patch('hangups.get_auth', side_effect=Google2FAError):
result = await flow.async_step_2fa({'2fa': 123456})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['errors']['base'] == 'invalid_2fa'