#!/usr/bin/env python3
# Copyright 2018 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Tests for the client_replay.py script.
To run tests, just call this script with the intended chrome, chromedriver
binaries and an optional test filter. This file interfaces with the
client_replay mainly using the CommandSequence class. Each of the test cases
matches a test case from chromedriver/test/run_py_tests.py; they run the same
case from run_py_tests.py, then replay the log and check that the behavior
matches.
"""
# pylint: disable=g-import-not-at-top, g-bad-import-order
import json
import optparse
import os
import re
import shutil
import subprocess
import sys
import tempfile
import traceback
import unittest
import client_replay
_THIS_DIR = os.path.abspath(os.path.dirname(__file__))
_PARENT_DIR = os.path.join(_THIS_DIR, os.pardir)
_TEST_DIR = os.path.join(_PARENT_DIR, "test")
_PY_TESTS = os.path.join(_TEST_DIR, "run_py_tests.py")
sys.path.insert(1, _PARENT_DIR)
import chrome_paths
import util
sys.path.remove(_PARENT_DIR)
sys.path.insert(1, _TEST_DIR)
import unittest_util
import webserver
sys.path.remove(_TEST_DIR)
# pylint: enable=g-import-not-at-top, g-bad-import-order
_NEGATIVE_FILTER = []
def SubstituteVariableEntries(s):
"""Identifies and removes items that can legitimately vary between runs."""
white_list = r'(("(id|userDataDir|frameId|version' \
r'|element-6066-11e4-a52e-4f735466cecf|message|timestamp' \
r'|expiry|chromedriverVersion|sessionId)": ' \
r'("[0-9]\.[0-9]*(\.[0-9]*)? \([a-f0-9]*\)"|[^\s},]*))' \
r'|[A-F0-9]{32}|cd_frame_id_="[a-f0-9]*")'
return re.sub(white_list, "<variable_item>", s)
def ClearPort(s):
"""Removes port numbers from urls in the given string."""
s = re.sub(r":([0-9]){5}/", "<port>", s)
return re.sub(r"localhost:([0-9]*)", "localhost:<port>", s)
def GenerateTestLog(test_name, chromedriver_path, chrome_path, log_dir):
"""Place the ChromeDriver log from running |test_name| in |log_dir|.
The log file is put in |log_dir| and named |test_name|.log.
Args:
test_name: test name from run_py_tests.py. Example: testGetTitle
chromedriver_path: path to ChromeDriver binary
chrome_path: path to Chrome binary
log_dir: directory in which to put the ChromeDriver log file.
Raises:
RuntimeError: run_py_tests.py had a test failure or other error.
"""
args = [
sys.executable,
_PY_TESTS,
"--chromedriver=%s" % chromedriver_path,
"--chrome=%s" % chrome_path,
"--replayable=true",
"--log-path=%s" % log_dir,
"--filter=%s" % ("*" + test_name)
]
result = subprocess.call(args)
if result != 0:
raise RuntimeError("run_py_tests.py could not be run or failed.")
class ChromeDriverClientReplayTest(unittest.TestCase):
"""Base class for test cases."""
def __init__(self, *args, **kwargs):
super(ChromeDriverClientReplayTest, self).__init__(*args, **kwargs)
@classmethod
def setUpClass(cls):
"""Starts the server for the necessary pages for testing."""
# make a temp dir to place test logs into
cls.log_dir = tempfile.mkdtemp()
try:
cls.http_server = webserver.WebServer(chrome_paths.GetTestData())
cls.server_url = cls.http_server.GetUrl()
except Exception:
cls.tearDownClass()
raise
@classmethod
def tearDownClass(cls):
"""Tears down the server."""
shutil.rmtree(cls.log_dir)
if getattr(cls, 'http_server', False):
cls.http_server.Shutdown()
def CheckResponsesMatch(self, real, logged):
"""Asserts that the given pair of responses match.
These are usually the replay response and the logged response.
Checks that they match, up to differences in session, window, element
IDs, timestamps, etc.
Args:
real: actual response from running the command
logged: logged response, taken from the log file
"""
if not real and not logged:
return
if isinstance(real, dict) and "message" in real:
real = "ERROR " + real["message"].split("\n")[0]
# pylint: disable=unidiomatic-typecheck
self.assertTrue(type(logged) == type(real)
or (isinstance(real, str)
and isinstance(logged, str)))
# pylint: enable=unidiomatic-typecheck
if isinstance(real, str) \
and (real[:14] == "<!DOCTYPE html" or real[:5] == "<html"):
real = "".join(real.split())
logged = "".join(logged.split())
if not isinstance(real, str):
real = json.dumps(real)
logged = json.dumps(logged)
real = ClearPort(real)
logged = ClearPort(logged)
real = SubstituteVariableEntries(real)
logged = SubstituteVariableEntries(logged)
self.assertEqual(real, logged)
def runTest(self, test_name):
"""Runs the test. Compares output from Chrome to the output in the log file.
Args:
test_name: name of the test to run from run_py_tests.py.
"""
log_file = os.path.join(ChromeDriverClientReplayTest.log_dir,
test_name + ".log")
GenerateTestLog(test_name, _CHROMEDRIVER, _CHROME, log_file)
with open(log_file) as lf:
replay_path = log_file if _OPTIONS.devtools_replay else ""
server = client_replay.StartChromeDriverServer(_CHROMEDRIVER,
_OPTIONS.output_log_path,
replay_path)
chrome_binary = (util.GetAbsolutePathOfUserPath(_CHROME)
if _CHROME else None)
replayer = client_replay.Replayer(lf, server, chrome_binary,
self.server_url)
real_response = None
while True:
command = replayer.command_sequence.NextCommand(real_response)
if not command:
break
logged_response = replayer.command_sequence._last_response
real_response = replayer.executor.Execute(
client_replay._COMMANDS[command.name],
command.GetPayloadPrimitive())
self.CheckResponsesMatch(real_response["value"],
logged_response.GetPayloadPrimitive())
server.Kill()
def GetFunctionName(self):
"""Get the name of the function that calls this one."""
# https://stackoverflow.com/questions/251464/
# how-to-get-a-function-name-as-a-string-in-python
return traceback.extract_stack(None, 2)[0][2]
def testGetPageSource(self):
self.runTest(self.GetFunctionName())
def testCloseWindow(self):
self.runTest(self.GetFunctionName())
def testIFrameWithExtensionsSource(self):
self.runTest(self.GetFunctionName())
def testUnexpectedAlertBehaviour(self):
self.runTest(self.GetFunctionName())
def testFileDownloadWithClick(self):
self.runTest(self.GetFunctionName())
def testCanSwitchToPrintPreviewDialog(self):
self.runTest(self.GetFunctionName())
def testClearElement(self):
self.runTest(self.GetFunctionName())
def testEmulateNetworkConditions(self):
self.runTest(self.GetFunctionName())
def testSwitchToWindow(self):
self.runTest(self.GetFunctionName())
def testEvaluateScript(self):
self.runTest(self.GetFunctionName())
def testEvaluateInvalidScript(self):
self.runTest(self.GetFunctionName())
def testGetTitle(self):
self.runTest(self.GetFunctionName())
def testSendCommand(self):
self.runTest(self.GetFunctionName())
def testGetSessions(self):
self.runTest(self.GetFunctionName())
def testQuitASessionMoreThanOnce(self):
self.runTest(self.GetFunctionName())
def GetNegativeFilter(chrome_version):
"""Construct the appropriate negative test filter for the chrome ."""
if _NEGATIVE_FILTER:
return "*-" + ":__main__.".join([""] + _NEGATIVE_FILTER)
return "*"
def main():
usage = "usage: %prog <chromedriver binary> <chrome binary> [options]"
parser = optparse.OptionParser(usage=usage)
parser.add_option(
"", "--output-log-path",
help="Output verbose server logs to this file")
parser.add_option(
"", "--filter", type="string", default="*",
help="Filter for specifying what tests to run, \"*\" will run all,"
"including tests excluded by default. E.g., *testRunMethod")
parser.add_option(
"", "--devtools-replay", help="Replay DevTools instead of using\n"
"real Chrome.")
# Need global to access these from the test runner.
# pylint: disable=global-variable-undefined
global _OPTIONS, _CHROMEDRIVER, _CHROME
# pylint: enable=global-variable-undefined
_OPTIONS, args = parser.parse_args()
_CHROMEDRIVER = util.GetAbsolutePathOfUserPath(args[0])
_CHROME = util.GetAbsolutePathOfUserPath(args[1])
if not os.path.exists(_CHROMEDRIVER):
parser.error("Path given for chromedriver is invalid.\n"
'Please run "%s --help" for help' % __file__)
if not os.path.exists(_CHROME):
parser.error("Path given for chrome is invalid.\n"
'Please run "%s --help" for help' % __file__)
all_tests_suite = unittest.defaultTestLoader.loadTestsFromModule(
sys.modules[__name__])
test_filter = (GetNegativeFilter()
if not _OPTIONS.filter else _OPTIONS.filter)
tests = unittest_util.FilterTestSuite(all_tests_suite, test_filter)
result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(tests)
sys.exit(len(result.failures) + len(result.errors))
if __name__ == "__main__":
main()
Codebase Browser
chromium