blob: 34d574aff552564cb7750079e8caebd7161160c8 [file] [log] [blame]
"""Provides constants and functions needed by tv_testcase classes."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import importlib
import json
import sys
# pylint: disable=C6204
from urllib import urlencode
import urlparse
import container_util
# These are watched for in webdriver_benchmark_test.py
TEST_RESULT = "webdriver_benchmark TEST RESULT"
TEST_COMPLETE = "webdriver_benchmark TEST COMPLETE"
# These are event types that can be injected
EVENT_TYPE_KEY_DOWN = "KeyDown"
EVENT_TYPE_KEY_UP = "KeyUp"
# URL-related constants
BASE_URL = "https://www.youtube.com/"
TV_APP_PATH = "/tv"
BASE_PARAMS = {}
def import_selenium_module(submodule=None):
"""Dynamically imports a selenium.webdriver submodule.
This is done because selenium 3.0 is not commonly pre-installed
on workstations, and we want to have a friendly error message for that
case.
Args:
submodule: module subpath underneath "selenium.webdriver"
Returns:
appropriate module
"""
if submodule:
module_path = ".".join(("selenium", submodule))
else:
module_path = "selenium"
# As of this writing, Google uses selenium 3.0.0b2 internally, so
# thats what we will target here as well.
try:
module = importlib.import_module(module_path)
if submodule is None:
# Only the top-level module has __version__
if not module.__version__.startswith("3.0"):
raise ImportError("Not version 3.0.x")
except ImportError:
sys.stderr.write("Could not import {}\n"
"Please install selenium >= 3.0.0b2.\n"
"Commonly: \"sudo pip install 'selenium>=3.0.0b2'\"\n"
.format(module_path))
sys.exit(1)
return module
def get_url(path, query_params=None):
"""Returns the URL indicated by the path and query parameters."""
parsed_url = list(urlparse.urlparse(BASE_URL))
parsed_url[2] = path
query_dict = BASE_PARAMS.copy()
if query_params:
query_dict.update(urlparse.parse_qsl(parsed_url[4]))
container_util.merge_dict(query_dict, query_params)
parsed_url[4] = urlencode(query_dict, doseq=True)
return urlparse.urlunparse(parsed_url)
def get_tv_url(query_params=None):
"""Returns the tv URL indicated by the query parameters."""
return get_url(TV_APP_PATH, query_params)
def record_test_result(name, result):
"""Records an individual scalar result of a benchmark test.
Args:
name: string name of test case
result: Test result. Must be JSON encodable scalar.
"""
value_to_record = result
string_value_to_record = json.JSONEncoder().encode(value_to_record)
print("{}: {} {}\n".format(TEST_RESULT, name, string_value_to_record))
sys.stdout.flush()
class RecordStrategyMax(object):
""""Records the max of an array of values."""
def run(self, name, values):
"""Records the max of an array of values.
Args:
name: string name of test case
values: must be array of JSON encodable scalar
"""
record_test_result("{}Max".format(name), max(values))
class RecordStrategyMin(object):
""""Records the min of an array of values."""
def run(self, name, values):
"""Records the min of an array of values.
Args:
name: string name of test case
values: must be array of JSON encodable scalar
"""
record_test_result("{}Min".format(name), min(values))
class RecordStrategyMean(object):
""""Records the mean of an array of values."""
def run(self, name, values):
"""Records the mean of an array of values.
Args:
name: string name of test case
values: must be array of JSON encodable scalar
"""
record_test_result("{}Mean".format(name), container_util.mean(values))
class RecordStrategyMedian(object):
""""Records the median of an array of test results."""
def run(self, name, values):
"""Records the median of an array of values.
Args:
name: string name of test case
values: must be array of JSON encodable scalar
"""
record_test_result("{}Median".format(name),
container_util.percentile(values, 50))
class RecordStrategyPercentile(object):
""""Records the specified percentile of an array of test results."""
def __init__(self, percentile):
"""Initializes the record strategy.
Args:
percentile: the percentile to record
"""
self.percentile = percentile
def run(self, name, values):
"""Records the percentile of an array of values.
Args:
name: string name of test case
values: must be array of JSON encodable scalar
Raises:
RuntimeError: Raised on invalid args.
"""
record_test_result("{}Pct{}".format(name, self.percentile),
container_util.percentile(values, self.percentile))
class ResultsRecorder(object):
""""Collects values and records results after a benchmark test ends."""
def __init__(self, name, record_strategies):
"""Initializes the value recorder.
Args:
name: the name to use when recording test results
record_strategies: the strategies to use when the test ends
"""
self.name = name
self.record_strategies = record_strategies
self.values = []
def collect_value(self, value):
self.values.append(value)
def on_end_test(self):
"""Handles logic related to the end of the benchmark test."""
# Only run the strategies if values have been collected.
if self.values:
for record_strategy in self.record_strategies:
record_strategy.run(self.name, self.values)