client/site_tests/video_WebRtcPeerConnectionWithCamera/video_WebRtcPeerConnectionWithCamera.py

# Copyright 2015 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 logging
import os
import time

from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros import chrome
from autotest_lib.client.cros.video import device_capability
from autotest_lib.client.cros.video import helper_logger

EXTRA_BROWSER_ARGS = ['--use-fake-ui-for-media-stream']

# Variables from the getusermedia.html page.
IS_VIDEO_INPUT_FOUND = 'isVideoInputFound'

# Polling timeout.
TIMEOUT = 90


class video_WebRtcPeerConnectionWithCamera(test.test):
    """Tests a full WebRTC call with a real webcam."""
    version = 1

    def cleanup(self):
        """Autotest cleanup function

        It is run by common_lib/test.py.
        """
        if utils.is_virtual_machine():
            try:
                utils.run('sudo modprobe -r vivid')
            except Exception as e:
                raise error.TestFail('Failed to unload vivid', e)

    def start_loopback(self, cr, video_codec):
        """Opens WebRTC loopback page.

        @param cr: Autotest Chrome instance.
        @param video_codec: video codec to use.
        """
        cr.browser.platform.SetHTTPServerDirectories(self.bindir)

        self.tab = cr.browser.tabs[0]
        self.tab.Navigate(cr.browser.platform.http_server.UrlOf(
                os.path.join(self.bindir, 'loopback.html')))
        self.tab.WaitForDocumentReadyStateToBeComplete()

        if utils.is_virtual_machine():
            # Make sure if Chrome has already recognized vivid as an
            # external camera before calling 'getUserMedia()'.
            self.wait_camera_detected()

        self.tab.EvaluateJavaScript(
                "testWebRtcLoopbackCall('%s')" % video_codec)

    def wait_camera_detected(self):
        """Waits until a camera is detected.
        """
        for _ in range(10):
            self.tab.EvaluateJavaScript('checkVideoInput()')
            if self.tab.EvaluateJavaScript(IS_VIDEO_INPUT_FOUND):
                return
            time.sleep(1)

        raise error.TestFail('Can not find video input device')

    def wait_test_completed(self, timeout_secs):
        """Waits until the test is done.

        @param timeout_secs Max time to wait in seconds.

        @raises TestError on timeout, or javascript eval fails.
        """
        def _test_done():
            status = self.tab.EvaluateJavaScript('getStatus()')
            logging.debug(status);
            return status != 'running'

        utils.poll_for_condition(
            _test_done, timeout=timeout_secs, sleep_interval=1,
            desc='loopback.html reports itself as finished')

    @helper_logger.video_log_wrapper
    def run_once(self, video_codec, capability):
        """Runs the video_WebRtcPeerConnectionWithCamera test.

        @param video_codec: video codec to use.
        @param capability: Capability required for executing this test.
        """
        device_capability.DeviceCapability().ensure_capability(capability)
        with chrome.Chrome(extra_browser_args=EXTRA_BROWSER_ARGS +\
                           [helper_logger.chrome_vmodule_flag()],
                           init_network_controller=True) as cr:

            # TODO(keiichiw): vivid should be loaded in self.setup() after
            # crbug.com/871185 is fixed
            if utils.is_virtual_machine():
                try:
                    utils.run('sudo modprobe vivid n_devs=1 node_types=0x1')
                except Exception as e:
                    raise error.TestFail('Failed to load vivid', e)

            self.start_loopback(cr, video_codec)
            self.wait_test_completed(TIMEOUT)
            self.print_loopback_result(video_codec)

    def print_loopback_result(self, video_codec):
        """Prints loopback results (unless we failed to retrieve them).

        @param video_codec: video codec to use.
        @raises TestError if the test failed outright.
        """
        status = self.tab.EvaluateJavaScript('getStatus()')
        if status != 'ok-done':
            raise error.TestFail('Failed: %s' % status)

        results = self.tab.EvaluateJavaScript('getResults()')
        logging.info('Camera Type: %s', results['cameraType'])
        logging.info('PeerConnectionstats: %s', results['peerConnectionStats'])
        logging.info('FrameStats: %s', results['frameStats'])

        pc_stats = results.get('peerConnectionStats')
        if not pc_stats:
            raise error.TestFail('Peer Connection Stats is empty')
        self.output_perf_value(
                description='max_input_fps_%s' % video_codec,
                value=pc_stats[1], units='fps', higher_is_better=True)
        self.output_perf_value(
                description='max_sent_fps_%s' % video_codec,
                value=pc_stats[4], units='fps', higher_is_better=True)

        frame_stats = results.get('frameStats')
        if not frame_stats:
            raise error.TestFail('Frame Stats is empty')

        self.output_perf_value(
                description='black_frames_%s' % video_codec,
                value=frame_stats['numBlackFrames'],
                units='frames', higher_is_better=False)
        self.output_perf_value(
                description='frozen_frames_%s' % video_codec,
                value=frame_stats['numFrozenFrames'],
                units='frames', higher_is_better=False)
        self.output_perf_value(
                description='total_num_frames_%s' % video_codec,
                value=frame_stats['numFrames'],
                units='frames', higher_is_better=True)