Source code for hangout_api.base

"""
Hangout API
===========

Main module of Hangout that provide base HG management and call settings
management.

"""
# pylint can't handle EasyDict properly
# pylint: disable=E1101
import os.path
from pyvirtualdisplay.smartdisplay import SmartDisplay
import seleniumwrapper as selwrap
from chromedriver import CHROMEDRV_PATH
from zope.component import getUtilitiesFor
from retrying import retry
from selenium.webdriver.common.keys import Keys
from time import sleep

from .utils import (
    Utils,
    URLS,
    TIMEOUTS,
    Participant,
    tries_n_time_until_true,
    names_cleaner,
)
from .exceptions import LoginError
from .interfaces import IModule, IOnAirModule

from selenium.webdriver.remote.remote_connection import LOGGER
import logging
LOGGER.setLevel(logging.WARNING)  # reducing selenium verbosity


@retry(stop_max_attempt_number=3)
def _create_hangout_event(browser, name, attendees):
    """
    Creates hangout event on google plus. As arguments takes event name and
    attendees list, also should be provided with 'browser' object where
    visitor is logged in
    """
    browser.get(URLS.onair)
    browser.by_text('Create a Hangout On Air').click(TIMEOUTS.fast)
    # Setting name
    browser.xpath(
        '//input[@aria-label="Give it a name"]').send_keys(name)
    # cleaning 'share' field and send attendees list there
    browser.xpath(
        '//input[@aria-label="Add more people"]').send_keys(
            '\b\b\b' + ','.join(attendees) + ',')
    browser.xpath(
        '//*[@guidedhelpid="shareboxcontrols"]//*[text()="Share"]').click(
            timeout=TIMEOUTS.fast)
    # waiting for redirecting to OnAir event page to be complete
    browser.xpath(
        '//div[@data-tooltip="Start the Hangout On Air"]',
        timeout=TIMEOUTS.extralong)
    return browser.current_url


[docs]class Hangouts(object): """ Main class for controlling hangout calls. Initialization does two things: 1. Makes sure that there is active X session. 2. Starts the browser. If 'DISPLAY' can't be found in os.environ than new X session starts. Starting new session handels `PyVirtualDisplay`_. .. _PyVirtualDisplay: http://ponty.github.io/PyVirtualDisplay/ For handling browser used seleniumwrapper library. .. testsetup:: HangoutsBase import os os.environ['DISPLAY'] = '1' from hangout_api import Hangouts from hangout_api.tests.doctests_utils import DummySelenium import seleniumwrapper def get_attribute(self, name): if name == 'aria-label': return ' John Doe ' elif name == 'data-userid': return '108775712935' else: return 'hello' DummySelenium.get_attribute = get_attribute seleniumwrapper.create = DummySelenium .. testsetup:: HangoutsBase2 import os os.environ['DISPLAY'] = '1' from hangout_api import Hangouts from hangout_api.tests.doctests_utils import DummySelenium import seleniumwrapper DummySelenium.location = {'x': 1} seleniumwrapper.create = DummySelenium hangout = Hangouts() .. doctest:: HangoutsBase >>> hangout = Hangouts() """ def __init__(self, executable_path=None, chrome_options=None): self.hangout_id = None self.on_air = None # lets start display in case if no is available self.display = None if not os.environ.get('DISPLAY'): self.display = SmartDisplay() self.display.start() kwargs = {'executable_path': executable_path or CHROMEDRV_PATH} if chrome_options is not None: kwargs['chrome_options'] = chrome_options self.browser = selwrap.create('chrome', **kwargs) self.utils = Utils(self.browser) for name, instance in getUtilitiesFor(IModule): setattr(self, name, instance(self.utils))
[docs] def start(self, on_air=None): """ Start a new hangout. After new hangout is created its id is stored in 'hangout_id' attribure .. doctest:: HangoutsBase >>> hangout.start() >>> hangout.hangout_id 'gs4pp6g62w65moctfqsvihzq2qa' To start OnAir just pass on_air argument to 'start' method. .. doctest:: HangoutsBase >>> hangout.start( ... on_air={'name':'My OnAir', 'attendees':['Friends']}) >>> hangout.start(on_air='https://plus.google.com/events/df34...') """ # onair if on_air is not None: self.on_air = on_air if isinstance(on_air, dict): # in case if on_air is a dict create new hangout event _create_hangout_event( self.browser, on_air['name'], on_air['attendees']) else: # otherwise (hangout is a string) go to event page self.browser.get(on_air) # on event page, redirecting can take some time self.browser.xpath( '//div[@data-tooltip="Start the Hangout On Air"]', timeout=TIMEOUTS.extralong).click(TIMEOUTS.fast) for name, instance in getUtilitiesFor(IOnAirModule): setattr(self, name, instance(self.utils)) else: if not self.browser.current_url.startswith( URLS.hangouts_active_list): self.browser.get(URLS.hangouts_active_list) # G+ opens new window for new hangout, so we need to # switch selenium to it self.browser.by_text( 'Start a video Hangout').click(timeout=TIMEOUTS.fast) # waiting until new window appears tries_n_time_until_true( lambda: len(self.browser.window_handles) <= 1, try_num=100) # self.browser.close() # closing old window self.browser.switch_to_window(self.browser.window_handles[-1]) self.utils.click_cancel_button_if_there_is_one( timeout=TIMEOUTS.extralong) if self.on_air: # waiting for broadcasting to be ready broadcast_button = self.browser.by_text( 'Start broadcast', timeout=TIMEOUTS.long) tries_n_time_until_true(broadcast_button.is_displayed, try_num=600) # setting hangout id property self.hangout_id = self.browser.current_url.replace( URLS.hangout_session_base, '', 1).split('?', 1)[0]
# @retry(stop_max_attempt_number=3)
[docs] def connect(self, hangout_id): """ Connect to an existing hangout. Takes id of targeted hangout as argument. Also it sets hangout_id property: .. doctest:: HangoutsBase >>> hangout.connect('fnar4989hf9834h') >>> hangout.hangout_id 'fnar4989hf9834h' """ self.hangout_id = hangout_id self.browser.get(URLS.hangout_session_base + hangout_id) # there may be a big delay before 'Join' button appears, so there is a # need wait longer than usual join_button = self.browser.xpath( '//*[text()="Join" or text()="Okay, got it!"]', timeout=TIMEOUTS.long) button_text = names_cleaner(join_button.get_attribute('innerText')) if button_text == 'Okay, got it!': # to join hangout we need to set agreement checkbox self.browser.xpath('//*[@role="presentation"]').click( timeout=TIMEOUTS.fast) join_button.click(timeout=TIMEOUTS.average) self.browser.by_text('Join', timeout=TIMEOUTS.long).click( timeout=TIMEOUTS.fast)
@retry(stop_max_attempt_number=3)
[docs] def login(self, username, password, otp=None): """ Log in to google plus. *otp* argument is one time password and it's optional, set it only if you're using 2-factor authorization. .. doctest:: HangoutsBase >>> hangout.login('user@gmail.com', 'password') >>> hangout.login('user_1@gmail.com', 'password', otp='123456') """ # Open login form and sing in with credentials self.browser.get(URLS.service_login) self.browser.by_id('Email').send_keys(username) self.browser.by_id('Passwd').send_keys(password) self.browser.by_id('signIn').click(timeout=TIMEOUTS.fast) # filling up one time password if provides if otp: self.browser.by_id('smsUserPin').send_keys(otp) self.browser.by_id('smsVerifyPin').click(timeout=TIMEOUTS.fast) # checking if log in was successful if not self.utils.is_logged_in: raise LoginError( 'Wasn\'t able to login. Check if credentials are correct' 'and make sure that you have G+ account activated')
@retry(stop_max_attempt_number=3)
[docs] def invite(self, participants): """ Invite person or circle to hangout: .. doctest:: HangoutsBase2 >>> hangout.invite("persona@gmail.com") >>> hangout.invite(["personb@gmail.com", "Public", "Friends"]) """ self.utils.click_cancel_button_if_there_is_one() if not any(isinstance(participants, i) for i in (list, tuple)): participants = [participants, ] # click on Invite People button self.utils.click_menu_element('//div[@aria-label="Invite People"]') input_field = self.browser.xpath( '//input[@placeholder="+ Add names, circles, or email addresses"]') input_field.click() input_field.clear() for participant in participants: input_field.send_keys(participant) sleep(1) # need to wait a bit for HG to make request input_field.send_keys(Keys.RETURN) self.browser.by_text('Invite').click(timeout=TIMEOUTS.fast) # making sure that invitation is posted xpath = '//*[text()="Invitation posted"'\ 'or text()="Waiting for people to join this video call..."]' self.browser.xpath(xpath)
@retry(stop_max_attempt_number=3)
[docs] def participants(self): """ Returns list of namedtuples of current participants: .. doctest:: HangoutsBase >>> hangout.participants() [Participant(name='John Doe', profile_id='108775712935')] """ xpath = '//div[@data-userid]' # for some reason some persons can be listed twice, so let's # filter them with dict participants = { p.get_attribute('data-userid'): p.get_attribute( 'aria-label')[5:-11] for p in self.browser.xpath(xpath, eager=True)} return [Participant(name=pname, profile_id=pid) for pid, pname in participants.items()]
@retry(stop_max_attempt_number=3)
[docs] def disconnect(self): """ Leave hangout (equal on clicking on "Leave call" button). After leaving the call you can create a new one or connect to existing. .. doctest:: HangoutsBase2 >>> hangout.disconnect() """ self.utils.click_cancel_button_if_there_is_one() self.utils.click_menu_element('//div[@aria-label="Leave call"]') self.hangout_id = None if self.on_air is not None: # removing properties that is available only for OnAir for name, _ in getUtilitiesFor(IOnAirModule): delattr(self, name) self.on_air = None # waiting until hangout windows closes tries_n_time_until_true( lambda: len(self.browser.window_handles) == 1, try_num=100) self.browser.switch_to_window(self.browser.window_handles[-1])
def __del__(self): # and quiting browser and display if self.browser: try: if self.hangout_id and self.browser.window_handles: # leaving the call first self.disconnect() finally: self.browser.quit() if self.display is not None: self.display.stop()