site_tests/desktopui_SonicExtension/test_utils.py

# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.


import json
import os
import time

import selenium

from autotest_lib.client.bin import utils
from extension_pages import e2e_test_utils
from extension_pages import options


class TestUtils(object):
    """Contains all the helper functions for Chrome mirroring automation."""

    short_wait_secs = 3
    step_timeout_secs = 60
    cpu_fields = ['user', 'nice', 'system', 'idle', 'iowait', 'irq',
                  'softirq', 'steal', 'guest', 'guest_nice']
    cpu_idle_fields = ['idle', 'iowait']

    def __init__(self):
        """Constructor"""


    def set_mirroring_options(self, driver, extension_id, settings):
        """Apply all the settings given by the user to the option page.

        @param driver: The chromedriver instance of the test
        @param extension_id: The id of the Cast extension
        @param settings: The settings and information about the test
        """
        options_page = options.OptionsPage(driver, extension_id)
        options_page.open_hidden_options_menu()
        time.sleep(self.short_wait_secs)
        for key in settings.keys():
            options_page.set_value(key, settings[key])


    def start_v2_mirroring_test_utils(
            self, driver, extension_id, receiver_ip, url, fullscreen,
            udp_proxy_server=None, network_profile=None):
        """Use test util page to start mirroring session on specific device.

        @param driver: The chromedriver instance
        @param extension_id: The id of the Cast extension
        @param receiver_ip: The ip of the Eureka dongle to launch the activity
        @param url: The URL to navigate to
        @param fullscreen: click the fullscreen button or not
        @param udp_proxy_server: the address of udp proxy server,
            it should be a http address, http://<ip>:<port>
        @param network_profile: the network profile,
            it should be one of wifi, bad and evil.
        @return True if the function finishes
        """
        e2e_test_utils_page = e2e_test_utils.E2ETestUtilsPage(
                driver, extension_id)
        time.sleep(self.short_wait_secs)
        tab_handles = driver.window_handles
        e2e_test_utils_page.receiver_ip_or_name_v2_text_box().set_value(
                receiver_ip)
        e2e_test_utils_page.url_to_open_v2_text_box().set_value(url)
        if udp_proxy_server:
            e2e_test_utils_page.udp_proxy_server_text_box().set_value(
                    udp_proxy_server)
        if network_profile:
            e2e_test_utils_page.network_profile_text_box().set_value(
                    network_profile)
        time.sleep(self.short_wait_secs)
        e2e_test_utils_page.open_then_mirror_v2_button().click()
        time.sleep(self.short_wait_secs)
        all_handles = driver.window_handles
        video_handle = [x for x in all_handles if x not in tab_handles].pop()
        driver.switch_to_window(video_handle)
        self.navigate_to_test_url(driver, url, fullscreen)
        return True


    def stop_v2_mirroring_test_utils(self, driver, extension_id):
      """Use test util page to stop a mirroring session on a specific device.

      @param driver: The chromedriver instance
      @param extension_id: The id of the Cast extension
      @param activity_id: The id of the mirroring activity
      """
      e2e_test_utils_page = e2e_test_utils.E2ETestUtilsPage(driver,
                                                            extension_id)
      e2e_test_utils_page.go_to_page()
      time.sleep(self.short_wait_secs)
      e2e_test_utils_page.stop_v2_mirroring_button().click()


    def start_v2_mirroring_sdk(self, driver, device_ip, url, extension_id):
        """Use SDK to start a mirroring session on a specific device.

        @param driver: The chromedriver instance
        @param device_ip: The IP of the Eureka device
        @param url: The URL to navigate to
        @param extension_id: The id of the Cast extension
        @return True if the function finishes
        @raise RuntimeError for timeouts
        """
        self.set_auto_testing_ip(driver, extension_id, device_ip)
        self.nagviate(driver, url, False)
        time.sleep(self.short_wait_secs)
        driver.execute_script('loadScript()')
        self._wait_for_result(
                lambda: driver.execute_script('return isSuccessful'),
                'Timeout when initiating mirroring... ...')
        driver.execute_script('startV2Mirroring()')
        self._wait_for_result(
                lambda: driver.execute_script('return isSuccessful'),
                'Timeout when triggering mirroring... ...')
        return True


    def stop_v2_mirroring_sdk(self, driver, activity_id=None):
        """Use SDK to stop the mirroring activity in Chrome.

        @param driver: The chromedriver instance
        @param activity_id: The id of the mirroring activity
        @raise RuntimeError for timeouts
        """
        driver.execute_script('stopV2Mirroring()')
        self._wait_for_result(
                lambda: driver.execute_script('return isSuccessful'),
                self.step_timeout_secs)


    def set_auto_testing_ip(self, driver, extension_id, device_ip):
        """Set the auto testing IP on the extension page.

        @param driver: The chromedriver instance
        @param extension_id: The id of the Cast extension
        @param device_ip: The IP of the device to test against
        """
        e2e_test_utils_page = e2e_test_utils.E2ETestUtilsPage(
                driver, extension_id)
        e2e_test_utils_page.execute_script(
                'localStorage["AutoTestingIp"] = "%s";' % device_ip)


    def upload_v2_mirroring_logs(self, driver, extension_id):
        """Upload v2 mirroring logs for the latest mirroring session.

        @param driver: The chromedriver instance of the browser
        @param extension_id: The extension ID of the Cast extension
        @return The report id in crash staging server.
        @raises RuntimeError if an error occurred during uploading
        """
        e2e_test_utils_page = e2e_test_utils.E2ETestUtilsPage(
                driver, extension_id)
        e2e_test_utils_page.go_to_page()
        time.sleep(self.short_wait_secs)
        e2e_test_utils_page.upload_v2_mirroring_logs_button().click()
        report_id = self._wait_for_result(
            e2e_test_utils_page.v2_mirroring_logs_scroll_box().get_value,
            'Failed to get v2 mirroring logs')
        if 'Failed to upload logs' in report_id:
          raise RuntimeError('Failed to get v2 mirroring logs')
        return report_id


    def get_chrome_version(self, driver):
        """Return the Chrome version that is being used for running test.

        @param driver: The chromedriver instance
        @return The Chrome version
        """
        get_chrome_version_js = 'return window.navigator.appVersion;'
        app_version = driver.execute_script(get_chrome_version_js)
        for item in app_version.split():
           if 'Chrome/' in item:
              return item.split('/')[1]
        return None


    def get_chrome_revision(self, driver):
        """Return Chrome revision number that is being used for running test.

        @param driver: The chromedriver instance
        @return The Chrome revision number
        """
        get_chrome_revision_js = ('return document.getElementById("version").'
                                  'getElementsByTagName("span")[2].innerHTML;')
        driver.get('chrome://version')
        return driver.execute_script(get_chrome_revision_js)


    def get_extension_id_from_flag(self, extra_flags):
        """Gets the extension ID based on the whitelisted extension id flag.

        @param extra_flags: A string which contains all the extra chrome flags
        @return The ID of the extension. Return None if nothing is found.
        """
        extra_flags_list = extra_flags.split()
        for flag in extra_flags_list:
            if 'whitelisted-extension-id=' in flag:
                return flag.split('=')[1]
        return None


    def navigate_to_test_url(self, driver, url, fullscreen):
        """Navigate to a given URL. Click fullscreen button if needed.

        @param driver: The chromedriver instance
        @param url: The URL of the site to navigate to
        @param fullscreen: True and the video will play in full screen mode.
                           Otherwise, set to False
        """
        driver.get(url)
        driver.refresh()
        if fullscreen:
          self.request_full_screen(driver)


    def request_full_screen(self, driver):
        """Request full screen.

        @param driver: The chromedriver instance
        """
        try:
            time.sleep(self.short_wait_secs)
            driver.find_element_by_id('fsbutton').click()
        except selenium.common.exceptions.NoSuchElementException as error_message:
            print 'Full screen button is not found. ' + str(error_message)


    def set_focus_tab(self, driver, tab_handle):
      """Set the focus on a tab.

      @param driver: The chromedriver instance
      @param tab_handle: The chrome driver handle of the tab
      """
      driver.switch_to_window(tab_handle)
      driver.get_screenshot_as_base64()


    def block_setup_dialog(self, driver, extension_id):
        """Tab cast through the extension.

        @param driver: A chromedriver instance that has the extension loaded.
        @param extension_id: Id of the extension to use.
        """
        e2e_test_utils_page = e2e_test_utils.E2ETestUtilsPage(
                driver, extension_id)
        e2e_test_utils_page.go_to_page()
        time.sleep(self.short_wait_secs)
        driver.execute_script(
            'localStorage["blockChromekeySetupAutoLaunchOnInstall"] = "true"')


    def close_popup_tabs(self, driver):
        """Close any popup windows the extension might open by default.

        Since we're going to handle the extension ourselves all we need is
        the main browser window with a single tab. The safest way to handle
        the popup however, is to close the currently active tab, so we don't
        mess with chromedrivers ui debugger.

        @param driver: Chromedriver instance.
        @raises Exception If you close the tab associated with
            the ui debugger.
        """
        # TODO: There are several, albeit hacky ways, to handle this popup
        # that might need to change with different versions of the extension
        # until the core issue is resolved. See crbug.com/338399.
        current_tab_handle = driver.current_window_handle
        for handle in driver.window_handles:
            if current_tab_handle != handle:
                try:
                    time.sleep(self.short_wait_secs)
                    driver.switch_to_window(handle)
                    driver.close()
                except:
                    pass
        driver.switch_to_window(current_tab_handle)


    def output_dict_to_file(self, dictionary, file_name,
                            path=None, sort_keys=False):
        """Output a dictionary into a file.

        @param dictionary: The dictionary to be output as JSON
        @param file_name: The name of the file that is being output
        @param path: The path of the file. The default is None
        @param sort_keys: Sort dictionary by keys when output. False by default
        """
        if path is None:
            path = os.path.abspath(os.path.dirname(__file__))
        # if json file exists, read the existing one and append to it
        json_file = os.path.join(path, file_name)
        if os.path.isfile(json_file) and dictionary:
            with open(json_file, 'r') as existing_json_data:
                json_data = json.load(existing_json_data)
            dictionary = dict(json_data.items() + dictionary.items())
        output_json = json.dumps(dictionary, sort_keys=sort_keys)
        with open(json_file, 'w') as file_handler:
            file_handler.write(output_json)


    def compute_cpu_utilization(self, cpu_dict):
        """Generate the upper/lower bound and the average CPU consumption.

        @param cpu_dict: The dictionary that contains CPU usage every sec.
        @returns A dict that contains upper/lower bound and average cpu usage.
        """
        cpu_bound = {}
        cpu_usage = sorted(cpu_dict.values())
        cpu_bound['lower_bound'] = (
                '%.2f' % cpu_usage[int(len(cpu_usage) * 10.0 / 100.0)])
        cpu_bound['upper_bound'] = (
                '%.2f' % cpu_usage[int(len(cpu_usage) * 90.0 / 100.0)])
        cpu_bound['average'] = '%.2f' % (sum(cpu_usage) / float(len(cpu_usage)))
        return cpu_bound


    def cpu_usage_interval(self, duration, interval=1):
        """Get the CPU usage over a period of time based on interval.

        @param duration: The duration of getting the CPU usage.
        @param interval: The interval to check the CPU usage. Default is 1 sec.
        @return A dict that contains CPU usage over different time intervals.
        """
        current_time = 0
        cpu_usage = {}
        while current_time < duration:
            pre_times = self._get_system_times()
            time.sleep(interval)
            post_times = self._get_system_times()
            cpu_usage[current_time] = self._get_avg_cpu_usage(
                    pre_times, post_times)
            current_time += interval
        return cpu_usage


    def _get_avg_cpu_usage(self, pre_times, post_times):
        """Calculate the average CPU usage of two different periods of time.

        @param pre_times: The CPU usage information of the start point.
        @param post_times: The CPU usage information of the end point.
        @return Average CPU usage over a time period.
        """
        diff_times = {}
        for field in self.cpu_fields:
            diff_times[field] = post_times[field] - pre_times[field]

        idle_time = sum(diff_times[field] for field in self.cpu_idle_fields)
        total_time = sum(diff_times[field] for field in self.cpu_fields)
        return float(total_time - idle_time) / total_time * 100.0


    def _get_system_times(self):
        """Get the CPU information from the system times.

        @return An list with CPU usage of different processes.
        """
        proc_stat = utils.read_file('/proc/stat')
        for line in proc_stat.split('\n'):
            if line.startswith('cpu '):
                times = line[4:].strip().split(' ')
                times = [int(jiffies) for jiffies in times]
                return dict(zip(self.cpu_fields, times))


    def _wait_for_result(self, get_result, error_message):
        """Wait for the result.

        @param get_result: the function to get result.
        @param error_message: the error message in the exception
            if it is failed to get result.
        @return The result.
        @raises RuntimeError if it is failed to get result within
            self.step_timeout_secs.
        """
        start = time.time()
        while (((time.time() - start) < self.step_timeout_secs)
               and not get_result()):
            time.sleep(self.step_timeout_secs/10.0)
        if not get_result():
            raise RuntimeError(error_message)
        return get_result()