| OLD | NEW |
| (Empty) | |
| 1 # Copyright 2016 The Chromium Authors. All rights reserved. |
| 2 # Use of this source code is governed by a BSD-style license that can be |
| 3 # found in the LICENSE file. |
| 4 |
| 5 """Chrome Autofill Task Flow |
| 6 |
| 7 Execute a set of autofill tasks in a fresh ChromeDriver instance that has been |
| 8 pre-loaded with some default profile. |
| 9 |
| 10 Requires: |
| 11 - Selenium python bindings |
| 12 http://selenium-python.readthedocs.org/ |
| 13 |
| 14 - ChromeDriver |
| 15 https://sites.google.com/a/chromium.org/chromedriver/downloads |
| 16 The ChromeDriver executable must be available on the search PATH. |
| 17 |
| 18 - Chrome |
| 19 """ |
| 20 |
| 21 import abc |
| 22 from urlparse import urlparse |
| 23 import os |
| 24 import shutil |
| 25 from random import choice |
| 26 from string import ascii_lowercase |
| 27 |
| 28 from selenium import webdriver |
| 29 from selenium.common.exceptions import TimeoutException, WebDriverException |
| 30 from selenium.webdriver.chrome.options import Options |
| 31 |
| 32 |
| 33 class TaskFlow(object): |
| 34 """Represents an executable set of Autofill Tasks. |
| 35 |
| 36 Attributes: |
| 37 profile: Dict of profile data that acts as the master source for |
| 38 validating autofill behaviour. |
| 39 debug: Whether debug output should be printed (False if not specified). |
| 40 """ |
| 41 __metaclass__ = abc.ABCMeta |
| 42 def __init__(self, profile, debug=False): |
| 43 self.set_profile(profile) |
| 44 self._debug = debug |
| 45 self._running = False |
| 46 |
| 47 self._tasks = self._generate_task_sequence() |
| 48 |
| 49 def set_profile(self, profile): |
| 50 """Validates |profile| before assigning it as the source of user data. |
| 51 |
| 52 Args: |
| 53 profile: Dict of profile data that acts as the master source for |
| 54 validating autofill behaviour. |
| 55 |
| 56 Raises: |
| 57 ValueError: The |profile| dict provided is missing required keys |
| 58 """ |
| 59 if not isinstance(profile, dict): |
| 60 raise ValueError('profile must be a a valid dictionary'); |
| 61 |
| 62 self._profile = profile |
| 63 |
| 64 def run(self, user_data_dir, chrome_binary=None): |
| 65 """Generates and executes a sequence of chrome driver tasks. |
| 66 |
| 67 Args: |
| 68 user_data_dir: Path string for the writable directory in which profiles |
| 69 should be stored. |
| 70 chrome_binary: Path string to the Chrome binary that should be used by |
| 71 ChromeDriver. |
| 72 |
| 73 If None then it will use the PATH to find a binary. |
| 74 |
| 75 Raises: |
| 76 RuntimeError: Running the TaskFlow was attempted while it's already |
| 77 running. |
| 78 Exception: Any failure encountered while running the tests |
| 79 """ |
| 80 if self._running: |
| 81 raise RuntimeError('Cannot run TaskFlow when already running') |
| 82 |
| 83 self._running = True |
| 84 |
| 85 self._run_tasks(user_data_dir, chrome_binary=chrome_binary) |
| 86 |
| 87 self._running = False |
| 88 |
| 89 @abc.abstractmethod |
| 90 def _generate_task_sequence(self): |
| 91 """Generates a set of executable tasks that will be run in ChromeDriver. |
| 92 |
| 93 Note: Subclasses must implement this method. |
| 94 |
| 95 Raises: |
| 96 NotImplementedError: Subclass did not implement the method |
| 97 |
| 98 Returns: |
| 99 A list of AutofillTask instances that are to be run in ChromeDriver. |
| 100 |
| 101 These tasks are to be run in order. |
| 102 """ |
| 103 raise NotImplementedError() |
| 104 |
| 105 def _run_tasks(self, user_data_dir, chrome_binary=None): |
| 106 """Runs the internal set of tasks in a fresh ChromeDriver instance. |
| 107 |
| 108 Args: |
| 109 user_data_dir: Path string for the writable directory in which profiles |
| 110 should be stored. |
| 111 chrome_binary: Path string to the Chrome binary that should be used by |
| 112 ChromeDriver. |
| 113 |
| 114 If None then it will use the PATH to find a binary. |
| 115 |
| 116 Raises: |
| 117 Exception: Any failure encountered while running the tests |
| 118 """ |
| 119 driver = self._get_driver(user_data_dir, chrome_binary=chrome_binary) |
| 120 try: |
| 121 for task in self._tasks: |
| 122 task.run(driver) |
| 123 finally: |
| 124 driver.quit() |
| 125 shutil.rmtree(self._profile_dir_dst) |
| 126 |
| 127 def _get_driver(self, user_data_dir, profile_name=None, chrome_binary=None, |
| 128 chromedriver_binary='chromedriver'): |
| 129 """Spin up a ChromeDriver instance that uses a given set of user data. |
| 130 |
| 131 Generates a temporary profile data directory using a local set of test data. |
| 132 |
| 133 Args: |
| 134 user_data_dir: Path string for the writable directory in which profiles |
| 135 should be stored. |
| 136 profile_name: Name of the profile data directory to be created/used in |
| 137 user_data_dir. |
| 138 |
| 139 If None then an eight character name will be generated randomly. |
| 140 |
| 141 This directory will be removed after the task flow completes. |
| 142 chrome_binary: Path string to the Chrome binary that should be used by |
| 143 ChromeDriver. |
| 144 |
| 145 If None then it will use the PATH to find a binary. |
| 146 |
| 147 Returns: The generated Chrome Driver instance. |
| 148 """ |
| 149 options = Options() |
| 150 |
| 151 if profile_name is None: |
| 152 profile_name = ''.join(choice(ascii_lowercase) for i in range(8)) |
| 153 |
| 154 options.add_argument('--profile-directory=%s' % profile_name) |
| 155 |
| 156 full_path = os.path.realpath(__file__) |
| 157 path, filename = os.path.split(full_path) |
| 158 profile_dir_src = os.path.join(path, 'testdata', 'Default') |
| 159 self._profile_dir_dst = os.path.join(user_data_dir, profile_name) |
| 160 self._copy_tree(profile_dir_src, self._profile_dir_dst) |
| 161 |
| 162 if chrome_binary is not None: |
| 163 options.binary_location = chrome_binary |
| 164 |
| 165 options.add_argument('--user-data-dir=%s' % user_data_dir) |
| 166 options.add_argument('--show-autofill-type-predictions') |
| 167 |
| 168 service_args = [] |
| 169 |
| 170 driver = webdriver.Chrome(executable_path=chromedriver_binary, |
| 171 chrome_options=options, |
| 172 service_args=service_args) |
| 173 driver.set_page_load_timeout(15) # seconds |
| 174 return driver |
| 175 |
| 176 def _copy_tree(self, src, dst): |
| 177 """Recursively copy a directory tree. |
| 178 |
| 179 If the destination directory does not exist then it will be created for you. |
| 180 Doesn't overwrite newer existing files. |
| 181 |
| 182 Args: |
| 183 src: Path to the target source directory. It must exist. |
| 184 dst: Path to the target destination directory. Permissions to create the |
| 185 the directory (if necessary) and modify it's contents. |
| 186 """ |
| 187 if not os.path.exists(dst): |
| 188 os.makedirs(dst) |
| 189 for item in os.listdir(src): |
| 190 src_item = os.path.join(src, item) |
| 191 dst_item = os.path.join(dst, item) |
| 192 if os.path.isdir(src_item): |
| 193 self._copy_tree(src_item, dst_item) |
| 194 elif (not os.path.exists(dst_item) or |
| 195 os.stat(src_item).st_mtime - os.stat(dst_item).st_mtime > 1): |
| 196 # Copy a file if it doesn't already exist, or if existing one is older. |
| 197 shutil.copy2(src_item, dst_item) |
| OLD | NEW |