| # 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 json |
| import logging |
| import os |
| import time |
| |
| from autotest_lib.client.common_lib.cros import arc |
| |
| from autotest_lib.client.common_lib.cros.arc import is_android_container_alive |
| |
| from autotest_lib.client.bin import test |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib.cros import chrome |
| from autotest_lib.client.common_lib.cros import enrollment |
| from autotest_lib.client.common_lib.cros import policy |
| from autotest_lib.client.cros import cryptohome |
| from autotest_lib.client.cros import httpd |
| from autotest_lib.client.cros.input_playback import keyboard |
| from autotest_lib.client.cros.enterprise import enterprise_policy_utils |
| from autotest_lib.client.cros.enterprise import policy_manager |
| from autotest_lib.client.cros.enterprise import enterprise_fake_dmserver |
| from autotest_lib.client.common_lib import ui_utils |
| |
| from telemetry.core import exceptions |
| |
| CROSQA_FLAGS = [ |
| '--gaia-url=https://gaiastaging.corp.google.com', |
| '--lso-url=https://gaiastaging.corp.google.com', |
| '--google-apis-url=https://www-googleapis-test.sandbox.google.com', |
| '--oauth2-client-id=236834563817.apps.googleusercontent.com', |
| '--oauth2-client-secret=RsKv5AwFKSzNgE0yjnurkPVI', |
| ('--cloud-print-url=' |
| 'https://cloudprint-nightly-ps.sandbox.google.com/cloudprint'), |
| '--ignore-urlfetcher-cert-requests'] |
| CROSALPHA_FLAGS = [ |
| ('--cloud-print-url=' |
| 'https://cloudprint-nightly-ps.sandbox.google.com/cloudprint'), |
| '--ignore-urlfetcher-cert-requests'] |
| TESTDMS_FLAGS = [ |
| '--ignore-urlfetcher-cert-requests', |
| '--disable-policy-key-verification'] |
| FLAGS_DICT = { |
| 'prod': [], |
| 'crosman-qa': CROSQA_FLAGS, |
| 'crosman-alpha': CROSALPHA_FLAGS, |
| 'dm-test': TESTDMS_FLAGS, |
| 'dm-fake': TESTDMS_FLAGS |
| } |
| DMS_URL_DICT = { |
| 'prod': 'http://m.google.com/devicemanagement/data/api', |
| 'crosman-qa': |
| 'https://crosman-qa.sandbox.google.com/devicemanagement/data/api', |
| 'crosman-alpha': |
| 'https://crosman-alpha.sandbox.google.com/devicemanagement/data/api', |
| 'dm-test': 'http://chromium-dm-test.appspot.com/d/%s', |
| 'dm-fake': 'http://127.0.0.1:%d/' |
| } |
| DMSERVER = '--device-management-url=%s' |
| # Username and password for the fake dm server can be anything, since |
| # they are not used to authenticate against GAIA. |
| USERNAME = 'fake-user@managedchrome.com' |
| PASSWORD = 'fakepassword' |
| GAIA_ID = 'fake-gaia-id' |
| |
| |
| class EnterprisePolicyTest(arc.ArcTest, test.test): |
| """Base class for Enterprise Policy Tests.""" |
| |
| WEB_PORT = 8080 |
| WEB_HOST = 'http://localhost:%d' % WEB_PORT |
| CHROME_POLICY_PAGE = 'chrome://policy' |
| CHROME_VERSION_PAGE = 'chrome://version' |
| |
| def initialize(self, **kwargs): |
| """ |
| Initialize test parameters. |
| |
| Consume the check_client_result parameter if this test was started |
| from a server test. |
| |
| """ |
| self._initialize_enterprise_policy_test(**kwargs) |
| |
| def _initialize_enterprise_policy_test( |
| self, env='dm-fake', dms_name=None, |
| username=USERNAME, password=PASSWORD, gaia_id=GAIA_ID, |
| set_auto_logout=None, **kwargs): |
| """ |
| Initialize test parameters and fake DM Server. |
| |
| This function exists so that ARC++ tests (which inherit from the |
| ArcTest class) can also initialize a policy setup. |
| |
| @param env: String environment of DMS and Gaia servers. |
| If wanting to point to a 'real' server use 'dm-test' |
| @param username: String user name login credential. |
| @param password: String password login credential. |
| @param gaia_id: String gaia_id login credential. |
| @param dms_name: String name of test DM Server. |
| @param kwargs: Not used. |
| |
| """ |
| self.env = env |
| self.username = username |
| self.password = password |
| self.gaia_id = gaia_id |
| self.set_auto_logout = set_auto_logout |
| self.dms_name = dms_name |
| self.dms_is_fake = (env == 'dm-fake') |
| self.arc_enabled = False |
| self._enforce_variable_restrictions() |
| |
| # Install protobufs and add import path. |
| policy.install_protobufs(self.autodir, self.job) |
| |
| # Initialize later variables to prevent error after an early failure. |
| self._web_server = None |
| self.cr = None |
| |
| # Start AutoTest DM Server if using local fake server. |
| if self.dms_is_fake: |
| self.fake_dm_server = enterprise_fake_dmserver.FakeDMServer() |
| self.fake_dm_server.start(self.tmpdir, self.debugdir) |
| |
| # Get enterprise directory of shared resources. |
| client_dir = os.path.dirname(os.path.dirname(self.bindir)) |
| self.enterprise_dir = os.path.join(client_dir, 'cros/enterprise') |
| |
| if self.set_auto_logout is not None: |
| self._auto_logout = self.set_auto_logout |
| |
| # Log the test context parameters. |
| logging.info('Test Context Parameters:') |
| logging.info(' Environment: %r', self.env) |
| logging.info(' Username: %r', self.username) |
| logging.info(' Password: %r', self.password) |
| logging.info(' Test DMS Name: %r', self.dms_name) |
| |
| def check_page_readiness(self, tab, command): |
| """ |
| Check to see if page has fully loaded. |
| |
| @param tab: chrome tab loading the page. |
| @param command: JS command to be checked in the tab. |
| |
| @returns True if loaded and False if not. |
| |
| """ |
| try: |
| tab.EvaluateJavaScript(command) |
| return True |
| except exceptions.EvaluateException: |
| return False |
| |
| def cleanup(self): |
| """Close out anything used by this test.""" |
| # Clean up AutoTest DM Server if using local fake server. |
| if self.dms_is_fake: |
| self.fake_dm_server.stop() |
| |
| # Stop web server if it was started. |
| if self._web_server: |
| self._web_server.stop() |
| |
| # Close Chrome instance if opened. |
| if self.cr and self._auto_logout: |
| self.cr.close() |
| |
| # Cleanup the ARC test if needed. |
| if self.arc_enabled: |
| super(EnterprisePolicyTest, self).cleanup() |
| |
| def start_webserver(self): |
| """Set up HTTP Server to serve pages from enterprise directory.""" |
| self._web_server = httpd.HTTPListener( |
| self.WEB_PORT, docroot=self.enterprise_dir) |
| self._web_server.run() |
| |
| def _enforce_variable_restrictions(self): |
| """Validate class-level test context parameters. |
| |
| @raises error.TestError if context parameter has an invalid value, |
| or a combination of parameters have incompatible values. |
| """ |
| # Verify |env| is a valid environment. |
| if self.env not in FLAGS_DICT: |
| raise error.TestError('Environment is invalid: %s' % self.env) |
| |
| # Verify test |dms_name| is given if |env| is 'dm-test'. |
| if self.env == 'dm-test' and not self.dms_name: |
| raise error.TestError('dms_name must be given when using ' |
| 'env=dm-test.') |
| if self.env != 'dm-test' and self.dms_name: |
| raise error.TestError('dms_name must not be given when not using ' |
| 'env=dm-test.') |
| if self.env == 'prod' and self.username == USERNAME: |
| raise error.TestError('Cannot user real DMS server when using ' |
| 'a fake test account.') |
| |
| def setup_case(self, |
| user_policies={}, |
| suggested_user_policies={}, |
| device_policies={}, |
| extension_policies={}, |
| skip_policy_value_verification=False, |
| kiosk_mode=False, |
| enroll=False, |
| auto_login=True, |
| auto_logout=True, |
| init_network_controller=False, |
| arc_mode=None, |
| setup_arc=True, |
| use_clouddpc_test=None, |
| disable_default_apps=True, |
| real_gaia=False, |
| extension_paths=[], |
| extra_chrome_flags=[]): |
| """Set up DMS, log in, and verify policy values. |
| |
| If the AutoTest fake DM Server is used, make a JSON policy blob |
| and upload it to the fake DM server. |
| |
| Launch Chrome and sign in to Chrome OS. Examine the user's |
| cryptohome vault, to confirm user is signed in successfully. |
| |
| @param user_policies: dict of mandatory user policies in |
| name -> value format. |
| @param suggested_user_policies: optional dict of suggested policies |
| in name -> value format. |
| @param device_policies: dict of device policies in |
| name -> value format. |
| @param extension_policies: dict of extension policies. |
| @param skip_policy_value_verification: True if setup_case should not |
| verify that the correct policy value shows on policy page. |
| @param enroll: True for enrollment instead of login. |
| @param auto_login: Sign in to chromeos. |
| @param auto_logout: Sign out of chromeos when test is complete. |
| @param init_network_controller: whether to init network controller. |
| @param arc_mode: whether to enable arc_mode on chrome.chrome(). |
| @param setup_arc: whether to run setup_arc in arc.Arctest. |
| @param use_clouddpc_test: whether to run the cloud dpc test. |
| @param extension_paths: list of extensions to install. |
| @param extra_chrome_flags: list of flags to add to Chrome. |
| |
| @raises error.TestError if cryptohome vault is not mounted for user. |
| @raises error.TestFail if |policy_name| and |policy_value| are not |
| shown on the Policies page. |
| """ |
| |
| # Need a real account, for now. Note: Even though the account is 'real' |
| # you can still use a fake DM server. |
| |
| if (arc_mode and self.username == USERNAME) or real_gaia: |
| self.username = 'tester50@managedchrome.com' |
| self.password = 'Test0000' |
| |
| self.pol_manager = policy_manager.Policy_Manager( |
| self.username, |
| self.fake_dm_server if hasattr(self, "fake_dm_server") else None) |
| |
| self._auto_logout = auto_logout |
| self._kiosk_mode = kiosk_mode |
| |
| self.pol_manager.configure_policies(user_policies, |
| suggested_user_policies, |
| device_policies, |
| extension_policies) |
| |
| self._create_chrome(enroll=enroll, |
| auto_login=auto_login, |
| init_network_controller=init_network_controller, |
| extension_paths=extension_paths, |
| arc_mode=arc_mode, |
| disable_default_apps=disable_default_apps, |
| real_gaia=real_gaia, |
| extra_chrome_flags=extra_chrome_flags) |
| |
| self.pol_manager.obtain_policies_from_device(self.cr.autotest_ext) |
| |
| # Skip policy check upon request or if we enroll but don't log in. |
| if not skip_policy_value_verification and auto_login: |
| self.pol_manager.verify_policies() |
| |
| if arc_mode: |
| self.start_arc(use_clouddpc_test, setup_arc) |
| |
| def update_DM_Info(self): |
| """Force an update of the DM server from current policies.""" |
| if not self.pol_manager.fake_dm_server: |
| raise error.TestError('Cannot autoupdate DM sever without the fake' |
| ' DM Sever being configured') |
| self.pol_manager.updateDMServer() |
| |
| def enable_Fake_DM_autoupdate(self): |
| """Enable automatic DM server updates on policy change.""" |
| self.pol_manager.auto_updateDM = True |
| |
| def disable_Fake_DM_autoupdate(self): |
| """Disable automatic DM server updates on policy change.""" |
| self.pol_manager.auto_updateDM = False |
| |
| def remove_policy(self, |
| name, |
| policy_type, |
| extID=None, |
| verify_policies=True): |
| """ |
| Remove a policy from the currently configured policies. |
| |
| @param name: str, the name of the policy |
| @param policy_type: str, the policy type. Must "extension" for |
| extension policies, other for user/device/suggested |
| @param ExtID: str, ID of extension to remove the policy from, if |
| policy_type is extension |
| @param verify_policies: bool, re-verify policies after removal of the |
| specified policy |
| |
| """ |
| self.pol_manager.remove_policy(name, policy_type, extID) |
| self.reload_policies(verify_policies) |
| if verify_policies: |
| self.pol_manager.verify_policies() |
| |
| def start_arc(self, use_clouddpc_test, setup_arc): |
| """ |
| Start ARC when creating the chrome object. |
| |
| Specifically will create the ADB shell container for testing use. |
| |
| We are NOT going to use the arc class initialize, it overwrites the |
| creation of chrome.chrome() in a way which cannot support the DM sever. |
| |
| Instead we check for the android container, and run arc_setup if |
| needed. Note: To use the cloud dpc test, you MUST also run setup_arc |
| |
| @param setup_arc: whether to run setup_arc in arc.Arctest. |
| @param use_clouddpc_test: bool, run_clouddpc_test() or not. |
| |
| """ |
| _APP_FILENAME = 'autotest-deps-cloudpctest-0.4.apk' |
| _DEP_PACKAGE = 'CloudDPCTest-apks' |
| _PKG_NAME = 'com.google.android.apps.work.clouddpc.e2etests' |
| |
| # By default on boot the container is alive, and will not close until |
| # a user with ARC disabled logs in. This wait accounts for that. |
| time.sleep(3) |
| |
| if use_clouddpc_test and not setup_arc: |
| raise error.TestFail('For cloud DPC setup_arc cannot be disabled') |
| |
| if is_android_container_alive(): |
| logging.info('Android Container is alive!') |
| else: |
| logging.error('Android Container is not alive!') |
| |
| # Install the clouddpc test. |
| if use_clouddpc_test: |
| self.arc_setup(dep_packages=_DEP_PACKAGE, |
| apks=[_APP_FILENAME], |
| full_pkg_names=[_PKG_NAME]) |
| self.run_clouddpc_test() |
| else: |
| if setup_arc: |
| self.arc_setup() |
| |
| self.arc_enabled = True |
| |
| def run_clouddpc_test(self): |
| """ |
| Run clouddpc end-to-end test and fail this test if it fails. |
| |
| Assumes start_arc() was run with use_clouddpc_test. |
| |
| Determines the policy values to pass to the test from those set in |
| Chrome OS. |
| |
| @raises error.TestFail if the test does not pass. |
| |
| """ |
| policy_blob = self.pol_manager.getCloudDpc() |
| policy_blob_str = json.dumps(policy_blob, separators=(',', ':')) |
| cmd = ('am instrument -w -e policy "%s" ' |
| 'com.google.android.apps.work.clouddpc.e2etests/' |
| '.ArcInstrumentationTestRunner') % policy_blob_str |
| |
| # Run the command as a shell script so that its length is not capped. |
| temp_shell_script_path = '/sdcard/tmp.sh' |
| arc.write_android_file(temp_shell_script_path, cmd) |
| |
| logging.info('Running clouddpc test with policy: %s', policy_blob_str) |
| results = arc.adb_shell('sh ' + temp_shell_script_path, |
| ignore_status=True).strip() |
| arc.remove_android_file(temp_shell_script_path) |
| if results.find('FAILURES!!!') >= 0: |
| logging.info('CloudDPC E2E Results:\n%s', results) |
| err_msg = results.splitlines()[-1] |
| raise error.TestFail('CloudDPC E2E failure: %s' % err_msg) |
| |
| logging.debug(results) |
| logging.info('CloudDPC E2E test passed!') |
| |
| def _get_policy_value_from_new_tab(self, policy): |
| """ |
| TO BE DEPRICATED DO NOT USE. |
| |
| Gets the value of the policy. |
| |
| """ |
| return self.pol_manager.get_policy_value_from_DUT(policy) |
| |
| def add_policies(self, |
| user={}, |
| suggested_user={}, |
| device={}, |
| extension={}, |
| new=False): |
| """Add policies to the policy rules.""" |
| self.pol_manager.configure_policies(user=user, |
| suggested_user=suggested_user, |
| device=device, |
| extension=extension, |
| new=new) |
| self.reload_policies(True) |
| |
| def update_policies(self, user_policies={}, suggested_user_policies={}, |
| device_policies={}, extension_policies={}): |
| """ |
| Change the policies to the ones provided. |
| |
| NOTE: This will override any current policies configured. |
| |
| @param user_policies: mandatory user policies -> values. |
| @param suggested user_policies: suggested user policies -> values. |
| @param device_policies: mandatory device policies -> values. |
| @param extension_policies: extension policies. |
| |
| """ |
| self.add_policies(user_policies, suggested_user_policies, |
| device_policies, extension_policies, True) |
| |
| def reload_policies(self, wait_for_new=False): |
| """ |
| Force a policy fetch. |
| |
| @param wait_for_new: bool, wait up to 1 second for the policy values |
| from the API to update |
| |
| """ |
| enterprise_policy_utils.refresh_policies(self.cr.autotest_ext, |
| wait_for_new) |
| |
| def verify_extension_stats(self, extension_policies, sensitive_fields=[]): |
| """ |
| Verify the extension policies match what is on chrome://policy. |
| |
| @param extension_policies: the dictionary of extension IDs mapping |
| to download_url and secure_hash. |
| @param sensitive_fields: list of fields that should have their value |
| censored. |
| @raises error.TestError: if the shown values do not match what we are |
| expecting. |
| |
| """ |
| # Refetch the policies from the DUT |
| self.pol_manager.obtain_policies_from_device() |
| |
| # For this method we are expecting the extension policies to be living |
| # within the chrome file system. |
| for ext_id in extension_policies.keys(): |
| filePath = extension_policies[ext_id]['download_url'] |
| actual_extension_policies = ( |
| self._get_extension_policies_from_file(filePath)) |
| |
| # Obfuscate and strip the extra dict level |
| self._configure_extension_file_policies(actual_extension_policies, |
| sensitive_fields) |
| |
| self.pol_manager.configure_extension_visual_policy( |
| {ext_id: actual_extension_policies}) |
| |
| self.pol_manager.verify_policies() |
| |
| def _configure_extension_file_policies(self, policies, sensitive_fields): |
| """ |
| Given policies, change sensitive_fields to "********". |
| |
| @param policies: dict, the extension policies as loaded from the file. |
| @param sensitive_fields: list or set |
| |
| """ |
| for policy_name, value in policies.items(): |
| if policy_name in sensitive_fields: |
| policies[policy_name] = '*' * 8 |
| else: |
| if 'Value' in value: |
| policies[policy_name] = value['Value'] |
| |
| def _get_extension_policies_from_file(self, download_url): |
| """ |
| Get the configured extension policies from file system. |
| |
| Extension tests will store the "actual" extensions as a file within |
| ChromeOS. Open the file within the enterprise_dir, and return the |
| policies from the file. |
| |
| @param download_url: URL of the download location |
| |
| """ |
| policy_file = os.path.join(self.enterprise_dir, |
| download_url.split('/')[-1]) |
| |
| with open(policy_file) as f: |
| policies = json.load(f) |
| return policies |
| |
| def verify_policy_value(self, |
| policy_name, |
| expected_value, |
| extensionID=None): |
| """ |
| Verify the value of a single policy. |
| |
| Note: This will not format the expected value. (Ie it will not |
| automatically change a password/sensitive field to ********) |
| |
| @param policy_name: the policy we are checking. |
| @param expected_value: the expected value for policy_name. |
| |
| @raises error.TestError if value does not match expected. |
| |
| """ |
| self.pol_manager.verify_policy(policy_name, |
| expected_value, |
| extensionID) |
| |
| def _initialize_chrome_extra_flags(self): |
| """ |
| Initialize flags used to create Chrome instance. |
| |
| @returns: list of extra Chrome flags. |
| |
| """ |
| # Construct DM Server URL flags if not using production server. |
| env_flag_list = [] |
| if self.env != 'prod': |
| if self.dms_is_fake: |
| # Use URL provided by the fake AutoTest DM server. |
| dmserver_str = (DMSERVER % self.fake_dm_server.server_url) |
| else: |
| # Use URL defined in the DMS URL dictionary. |
| dmserver_str = (DMSERVER % (DMS_URL_DICT[self.env])) |
| if self.env == 'dm-test': |
| dmserver_str = (dmserver_str % self.dms_name) |
| |
| # Merge with other flags needed by non-prod enviornment. |
| env_flag_list = ([dmserver_str] + FLAGS_DICT[self.env]) |
| |
| return env_flag_list |
| |
| def _enterprise_enroll(self, extra_flags, extension_paths, auto_login): |
| """ |
| Enroll a device, configure self.cr, optionally login. |
| |
| @param extra_chrome_flags: list of flags to add. |
| @param extension_paths: list of extensions to install. |
| @param auto_login: sign in to chromeos. |
| |
| """ |
| self.cr = chrome.Chrome( |
| auto_login=False, |
| autotest_ext=True, |
| extra_browser_args=extra_flags, |
| extension_paths=extension_paths, |
| expect_policy_fetch=True) |
| |
| if self.dms_is_fake: |
| if self._kiosk_mode: |
| enrollment.KioskEnrollment( |
| self.cr.browser, self.username, self.password, |
| self.gaia_id) |
| else: |
| enrollment.EnterpriseFakeEnrollment( |
| self.cr.browser, self.username, self.password, |
| self.gaia_id, auto_login=auto_login) |
| else: |
| enrollment.EnterpriseEnrollment( |
| self.cr.browser, self.username, self.password, |
| auto_login=auto_login) |
| |
| def _create_chrome(self, |
| enroll=False, |
| auto_login=True, |
| arc_mode=None, |
| real_gaia=False, |
| init_network_controller=False, |
| disable_default_apps=True, |
| extension_paths=[], |
| extra_chrome_flags=[]): |
| """ |
| Create a Chrome object. Enroll and/or sign in. |
| |
| Function results in self.cr set as the Chrome object. |
| |
| @param enroll: enroll the device. |
| @param auto_login: sign in to chromeos. |
| @param arc_mode: enable arc mode. |
| @param real_gaia: to login with a real gaia account or not. |
| @param init_network_controller: whether to init network controller. |
| @param disable_default_apps: disables default apps or not |
| (e.g. the Files app). |
| @param extension_paths: list of extensions to install. |
| @param extra_chrome_flags: list of flags to add. |
| """ |
| extra_flags = (self._initialize_chrome_extra_flags() + |
| extra_chrome_flags) |
| |
| logging.info('Chrome Browser Arguments:') |
| logging.info(' extra_browser_args: %s', extra_flags) |
| logging.info(' username: %s', self.username) |
| logging.info(' password: %s', self.password) |
| logging.info(' gaia_login: %s', not self.dms_is_fake) |
| logging.info(' enroll: %s', enroll) |
| |
| if enroll: |
| self._enterprise_enroll(extra_flags, extension_paths, auto_login) |
| elif auto_login: |
| # gaia_login (aka real login) is true on arc_mode, or real_gaia. |
| # otherwise: gaia_login is False when a fake dm server is used. |
| gaia_login = (True if (arc_mode or real_gaia) |
| else not self.dms_is_fake) |
| # Do not disable_gaia_services, or disable_arc_opt_in |
| # if using a gaia_login. |
| disable_gaia_services = disable_arc_opt_in = not gaia_login |
| enterprise_arc_test = gaia_login |
| |
| self.cr = chrome.Chrome( |
| extra_browser_args=extra_flags, |
| username=self.username, |
| password=self.password, |
| gaia_login=gaia_login, |
| arc_mode=arc_mode, |
| disable_arc_opt_in=disable_arc_opt_in, |
| disable_gaia_services=disable_gaia_services, |
| autotest_ext=True, |
| enterprise_arc_test=enterprise_arc_test, |
| init_network_controller=init_network_controller, |
| expect_policy_fetch=True, |
| extension_paths=extension_paths, |
| disable_default_apps=disable_default_apps) |
| |
| else: |
| self.cr = chrome.Chrome( |
| auto_login=False, |
| extra_browser_args=extra_flags, |
| disable_gaia_services=self.dms_is_fake, |
| autotest_ext=True, |
| expect_policy_fetch=True) |
| |
| self.ui = ui_utils.UI_Handler() |
| # Used by arc.py to determine the state of the chrome obj |
| self.initialized = True |
| if auto_login: |
| if not cryptohome.is_vault_mounted(user=self.username, |
| allow_fail=True): |
| raise error.TestError('Expected to find a mounted vault for %s.' |
| % self.username) |
| |
| def navigate_to_url(self, url, tab=None): |
| """Navigate tab to the specified |url|. Create new tab if none given. |
| |
| @param url: URL of web page to load. |
| @param tab: browser tab to load (if any). |
| @returns: browser tab loaded with web page. |
| @raises: telemetry TimeoutException if document ready state times out. |
| """ |
| logging.info('Navigating to URL: %r', url) |
| if not tab: |
| tab = self.cr.browser.tabs.New() |
| tab.Activate() |
| tab.Navigate(url, timeout=8) |
| tab.WaitForDocumentReadyStateToBeComplete() |
| return tab |
| |
| def get_elements_from_page(self, tab, cmd): |
| """Get collection of page elements that match the |cmd| filter. |
| |
| @param tab: tab containing the page to be scraped. |
| @param cmd: JavaScript command to evaluate on the page. |
| @returns object containing elements on page that match the cmd. |
| @raises: TestFail if matching elements are not found on the page. |
| """ |
| try: |
| elements = tab.EvaluateJavaScript(cmd) |
| except Exception as err: |
| raise error.TestFail('Unable to find matching elements on ' |
| 'the test page: %s\n %r' % (tab.url, err)) |
| return elements |
| |
| def log_out_via_keyboard(self): |
| """Log out of the device using the keyboard shortcut.""" |
| _keyboard = keyboard.Keyboard() |
| _keyboard.press_key('ctrl+shift+q') |
| _keyboard.press_key('ctrl+shift+q') |
| _keyboard.close() |
