474 lines
19 KiB
Python
474 lines
19 KiB
Python
# 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
|
|
|
|
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 import cryptohome
|
|
from autotest_lib.client.cros import enterprise_base
|
|
from autotest_lib.client.cros import httpd
|
|
|
|
CROSDEV_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']
|
|
CROSAUTO_FLAGS = [
|
|
('--cloud-print-url='
|
|
'https://cloudprint-nightly-ps.sandbox.google.com/cloudprint'),
|
|
'--ignore-urlfetcher-cert-requests']
|
|
TESTDMS_FLAGS = [
|
|
'--ignore-urlfetcher-cert-requests',
|
|
'--enterprise-enrollment-skip-robot-auth',
|
|
'--disable-policy-key-verification']
|
|
FLAGS_DICT = {
|
|
'prod': [],
|
|
'cr-dev': CROSDEV_FLAGS,
|
|
'cr-auto': CROSAUTO_FLAGS,
|
|
'dm-test': TESTDMS_FLAGS,
|
|
'dm-fake': TESTDMS_FLAGS
|
|
}
|
|
DMS_URL_DICT = {
|
|
'prod': 'http://m.google.com/devicemanagement/data/api',
|
|
'cr-dev': 'https://cros-dev.sandbox.google.com/devicemanagement/data/api',
|
|
'cr-auto': 'https://cros-auto.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'
|
|
|
|
|
|
class EnterprisePolicyTest(enterprise_base.EnterpriseTest):
|
|
"""Base class for Enterprise Policy Tests."""
|
|
|
|
def setup(self):
|
|
os.chdir(self.srcdir)
|
|
utils.make()
|
|
|
|
def initialize(self, args=()):
|
|
self._initialize_test_context(args)
|
|
|
|
# Start AutoTest DM Server if using local fake server.
|
|
if self.env == 'dm-fake':
|
|
self.import_dmserver(self.srcdir)
|
|
self.start_dmserver()
|
|
self._initialize_chrome_extra_flags()
|
|
self._web_server = None
|
|
|
|
def cleanup(self):
|
|
# Clean up AutoTest DM Server if using local fake server.
|
|
if self.env == 'dm-fake':
|
|
super(EnterprisePolicyTest, self).cleanup()
|
|
|
|
# Stop web server if it was started.
|
|
if self._web_server:
|
|
self._web_server.stop()
|
|
|
|
# Close Chrome instance if opened.
|
|
if self.cr:
|
|
self.cr.close()
|
|
|
|
def start_webserver(self, port):
|
|
"""Set up an HTTP Server to serve pages from localhost.
|
|
|
|
@param port: Port used by HTTP server.
|
|
|
|
"""
|
|
if self.mode != 'list':
|
|
self._web_server = httpd.HTTPListener(port, docroot=self.bindir)
|
|
self._web_server.run()
|
|
|
|
def _initialize_test_context(self, args=()):
|
|
"""Initialize class-level test context parameters.
|
|
|
|
@raises error.TestError if an arg is given an invalid value or some
|
|
combination of args is given incompatible values.
|
|
|
|
"""
|
|
# Extract local parameters from command line args.
|
|
args_dict = utils.args_to_dict(args)
|
|
self.mode = args_dict.get('mode', 'all')
|
|
self.case = args_dict.get('case')
|
|
self.value = args_dict.get('value')
|
|
self.env = args_dict.get('env', 'dm-fake')
|
|
self.username = args_dict.get('username')
|
|
self.password = args_dict.get('password')
|
|
self.dms_name = args_dict.get('dms_name')
|
|
|
|
# If |mode| is 'list', set |env| to generic 'prod', and blank out
|
|
# the other key parameters: case, value.
|
|
if self.mode == 'list':
|
|
self.env = 'prod'
|
|
self.case = None
|
|
self.value = None
|
|
|
|
# If |case| is given then set |mode| to 'single'.
|
|
if self.case:
|
|
self.mode = 'single'
|
|
|
|
# If |mode| is 'all', then |env| must be 'dm-fake', and
|
|
# the |case| and |value| args must not be given.
|
|
if self.mode == 'all':
|
|
if self.env != 'dm-fake':
|
|
raise error.TestError('env must be "dm-fake" '
|
|
'when mode=all.')
|
|
if self.case:
|
|
raise error.TestError('case must not be given '
|
|
'when mode=all.')
|
|
if self.value:
|
|
raise error.TestError('value must not be given '
|
|
'when mode=all.')
|
|
|
|
# If |value| is given, set |is_value_given| flag to True. If it
|
|
# was given as 'none', 'null', or '', then set |value| to 'null'.
|
|
if self.value is not None:
|
|
self.is_value_given = True
|
|
if (self.value.lower() == 'none' or
|
|
self.value.lower() == 'null' or
|
|
self.value == ''):
|
|
self.value = 'null'
|
|
else:
|
|
self.is_value_given = False
|
|
|
|
# Verify |env| is a valid environment.
|
|
if self.env is not None and self.env not in FLAGS_DICT:
|
|
raise error.TestError('env=%s is invalid.' % self.env)
|
|
|
|
# If |env| is 'dm-fake', ensure value and credentials are not given.
|
|
if self.env == 'dm-fake':
|
|
if self.is_value_given:
|
|
raise error.TestError('value must not be given when using '
|
|
'the fake DM Server.')
|
|
if self.username or self.password:
|
|
raise error.TestError('user credentials must not be given '
|
|
'when using the fake DM Server.')
|
|
|
|
# If either credential is not given, set both to default.
|
|
if self.username is None or self.password is None:
|
|
self.username = self.USERNAME
|
|
self.password = self.PASSWORD
|
|
|
|
# Verify |case| is given if |mode|==single.
|
|
if self.mode == 'single' and not self.case:
|
|
raise error.TestError('case must be given when mode is single.')
|
|
|
|
# Verify |case| is given if a |value| is given.
|
|
if self.is_value_given and self.case is None:
|
|
raise error.TestError('value must not be given without also '
|
|
'giving a test case.')
|
|
|
|
# Verify |dms_name| is given iff |env|==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.')
|
|
|
|
# Log the test context parameters.
|
|
logging.info('Test Context Parameters:')
|
|
logging.info(' Run Mode: %r', self.mode)
|
|
logging.info(' Test Case: %r', self.case)
|
|
logging.info(' Expected Value: %r', self.value)
|
|
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 _initialize_chrome_extra_flags(self):
|
|
"""Initialize flags used to create Chrome instance."""
|
|
# Construct DM Server URL flags.
|
|
env_flag_list = []
|
|
if self.env != 'prod':
|
|
if self.env == 'dm-fake':
|
|
# Use URL provided by AutoTest DM server.
|
|
dmserver_str = (DMSERVER % self.dm_server_url)
|
|
else:
|
|
# Use URL defined in 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])
|
|
|
|
self.extra_flags = env_flag_list
|
|
self.cr = None
|
|
|
|
def setup_case(self, policy_name, policy_value, policies_json):
|
|
"""Set up and confirm the preconditions of a test case.
|
|
|
|
If the AutoTest fake DM Server is initialized, make a policy blob
|
|
from |policies_json|, and upload it to the fake server.
|
|
|
|
Launch a chrome browser, and sign in to Chrome OS. Examine the user's
|
|
cryptohome vault, to confirm it signed in successfully.
|
|
|
|
Open the Policies page, and confirm that it shows the specified
|
|
|policy_name| and has the correct |policy_value|.
|
|
|
|
@param policy_name: Name of the policy under test.
|
|
@param policy_value: Expected value to appear on chrome://policy page.
|
|
@param policies_json: JSON string to set up the fake DMS policy value.
|
|
|
|
@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.
|
|
|
|
"""
|
|
# Set up policy on AutoTest DM Server only if initialized.
|
|
if self.env == 'dm-fake':
|
|
self.setup_policy(self._make_json_blob(policies_json))
|
|
|
|
self._launch_chrome_browser()
|
|
tab = self.navigate_to_url('chrome://policy')
|
|
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)
|
|
value_shown = self._get_policy_value_shown(tab, policy_name)
|
|
if not self._policy_value_matches_shown(policy_value, value_shown):
|
|
raise error.TestFail('Policy value shown is not correct: %s '
|
|
'(expected: %s)' %
|
|
(value_shown, policy_value))
|
|
tab.Close()
|
|
|
|
def _launch_chrome_browser(self):
|
|
"""Launch Chrome browser and sign in."""
|
|
logging.info('Chrome Browser Arguments:')
|
|
logging.info(' extra_browser_args: %s', self.extra_flags)
|
|
logging.info(' username: %s', self.username)
|
|
logging.info(' password: %s', self.password)
|
|
|
|
self.cr = chrome.Chrome(extra_browser_args=self.extra_flags,
|
|
username=self.username,
|
|
password=self.password,
|
|
gaia_login=True,
|
|
disable_gaia_services=False,
|
|
autotest_ext=True)
|
|
|
|
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.
|
|
|
|
"""
|
|
logging.info('Navigating to URL: %r', url)
|
|
if not tab:
|
|
tab = self.cr.browser.tabs.New()
|
|
tab.Activate()
|
|
tab.Navigate(url, timeout=5)
|
|
tab.WaitForDocumentReadyStateToBeComplete()
|
|
return tab
|
|
|
|
def _policy_value_matches_shown(self, policy_value, value_shown):
|
|
"""Compare |policy_value| to |value_shown| with whitespace removed.
|
|
|
|
Compare the expected policy value with the value actually shown on the
|
|
chrome://policies page. Before comparing, convert both values to JSON
|
|
formatted strings, and remove all whitespace. Whitespace is removed
|
|
because Chrome processes some policy values to show them in a more
|
|
human readable format.
|
|
|
|
@param policy_value: Expected value to appear on chrome://policy page.
|
|
@param value_shown: Value as it appears on chrome://policy page.
|
|
@param policies_json: JSON string to set up the fake DMS policy value.
|
|
|
|
@returns: True if the strings match after removing all whitespace.
|
|
|
|
"""
|
|
# Convert Python None or '' to JSON formatted 'null' string.
|
|
if value_shown is None or value_shown == '':
|
|
value_shown = 'null'
|
|
if policy_value is None or policy_value == '':
|
|
policy_value = 'null'
|
|
|
|
# Remove whitespace.
|
|
trimmed_value = ''.join(policy_value.split())
|
|
trimmed_shown = ''.join(value_shown.split())
|
|
logging.info('Trimmed policy value shown: %r (expected: %r)',
|
|
trimmed_shown, trimmed_value)
|
|
return trimmed_value == trimmed_shown
|
|
|
|
def _make_json_blob(self, policies_json):
|
|
"""Create policy blob from policies JSON object.
|
|
|
|
@param policies_json: Policies JSON object (name-value pairs).
|
|
@returns: Policy blob to be used to setup the policy server.
|
|
|
|
"""
|
|
policies_json = self._move_modeless_to_mandatory(policies_json)
|
|
policies_json = self._remove_null_policies(policies_json)
|
|
|
|
policy_blob = """{
|
|
"google/chromeos/user": %s,
|
|
"managed_users": ["*"],
|
|
"policy_user": "%s",
|
|
"current_key_index": 0,
|
|
"invalidation_source": 16,
|
|
"invalidation_name": "test_policy"
|
|
}""" % (json.dumps(policies_json), self.USERNAME)
|
|
return policy_blob
|
|
|
|
def _move_modeless_to_mandatory(self, policies_json):
|
|
"""Add the 'mandatory' mode if a policy's mode was omitted.
|
|
|
|
The AutoTest fake DM Server requires that every policy be contained
|
|
within either a 'mandatory' or 'recommended' dictionary, to indicate
|
|
the mode of the policy. This function moves modeless policies into
|
|
the 'mandatory' dictionary.
|
|
|
|
@param policies_json: The policy JSON data (name-value pairs).
|
|
@returns: dict of policies grouped by mode keys.
|
|
|
|
"""
|
|
mandatory_policies = {}
|
|
recommended_policies = {}
|
|
collated_json = {}
|
|
|
|
# Extract mandatory and recommended mode dicts.
|
|
if 'mandatory' in policies_json:
|
|
mandatory_policies = policies_json['mandatory']
|
|
del policies_json['mandatory']
|
|
if 'recommended' in policies_json:
|
|
recommended_policies = policies_json['recommended']
|
|
del policies_json['recommended']
|
|
|
|
# Move any remaining modeless policies into mandatory dict.
|
|
if policies_json:
|
|
mandatory_policies.update(policies_json)
|
|
|
|
# Collate all policies into mandatory & recommended dicts.
|
|
if recommended_policies:
|
|
collated_json.update({'recommended': recommended_policies})
|
|
if mandatory_policies:
|
|
collated_json.update({'mandatory': mandatory_policies})
|
|
|
|
return collated_json
|
|
|
|
def _remove_null_policies(self, policies_json):
|
|
"""Remove policy dict data that is set to None or ''.
|
|
|
|
For the status of a policy to be shown as "Not set" on the
|
|
chrome://policy page, the policy blob must contain no dictionary entry
|
|
for that policy. This function removes policy NVPs from a copy of the
|
|
|policies_json| dictionary that the test case had set to None or ''.
|
|
|
|
@param policies_json: setup policy JSON data (name-value pairs).
|
|
@returns: setup policy JSON data with all 'Not set' policies removed.
|
|
|
|
"""
|
|
policies_json_copy = policies_json.copy()
|
|
for policies in policies_json_copy.values():
|
|
for policy_data in policies.items():
|
|
if policy_data[1] is None or policy_data[1] == '':
|
|
policies.pop(policy_data[0])
|
|
return policies_json_copy
|
|
|
|
def _get_policy_value_shown(self, policy_tab, policy_name):
|
|
"""Get the value shown for the named policy on the Policies page.
|
|
|
|
Takes |policy_name| as a parameter and returns the corresponding
|
|
policy value shown on the chrome://policy page.
|
|
|
|
@param policy_tab: Tab displaying the chrome://policy page.
|
|
@param policy_name: The name of the policy.
|
|
@returns: The value shown for the policy on the Policies page.
|
|
|
|
"""
|
|
row_values = policy_tab.EvaluateJavaScript('''
|
|
var section = document.getElementsByClassName("policy-table-section")[0];
|
|
var table = section.getElementsByTagName('table')[0];
|
|
rowValues = '';
|
|
for (var i = 1, row; row = table.rows[i]; i++) {
|
|
if (row.className !== 'expanded-value-container') {
|
|
var name_div = row.getElementsByClassName('name elide')[0];
|
|
var name = name_div.textContent;
|
|
if (name === '%s') {
|
|
var value_span = row.getElementsByClassName('value')[0];
|
|
var value = value_span.textContent;
|
|
var status_div = row.getElementsByClassName('status elide')[0];
|
|
var status = status_div.textContent;
|
|
rowValues = [name, value, status];
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
rowValues;
|
|
''' % policy_name)
|
|
|
|
value_shown = row_values[1].encode('ascii', 'ignore')
|
|
status_shown = row_values[2].encode('ascii', 'ignore')
|
|
if status_shown == 'Not set.':
|
|
return None
|
|
return value_shown
|
|
|
|
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 json_string(self, policy_value):
|
|
"""Convert policy value to a JSON formatted string.
|
|
|
|
@param policy_value: object containing a policy value.
|
|
@returns: string in JSON format.
|
|
"""
|
|
return json.dumps(policy_value)
|
|
|
|
def _validate_and_run_test_case(self, test_case, run_test):
|
|
"""Validate test case and call the test runner in the test class.
|
|
|
|
@param test_case: name of the test case to run.
|
|
@param run_test: method in test class that runs a test case.
|
|
|
|
"""
|
|
if test_case not in self.TEST_CASES:
|
|
raise error.TestError('Test case is not valid: %s' % test_case)
|
|
logging.info('Running test case: %s', test_case)
|
|
run_test(test_case)
|
|
|
|
def run_once_impl(self, run_test):
|
|
"""Dispatch the common run modes for all child test classes.
|
|
|
|
@param run_test: method in test class that runs a test case.
|
|
|
|
"""
|
|
if self.mode == 'all':
|
|
for test_case in sorted(self.TEST_CASES):
|
|
self._validate_and_run_test_case(test_case, run_test)
|
|
elif self.mode == 'single':
|
|
self._validate_and_run_test_case(self.case, run_test)
|
|
elif self.mode == 'list':
|
|
logging.info('List Test Cases:')
|
|
for test_case, value in sorted(self.TEST_CASES.items()):
|
|
logging.info(' case=%s, value="%s"', test_case, value)
|
|
else:
|
|
raise error.TestError('Run mode is not valid: %s' % self.mode)
|
|
|
|
def run_once(self):
|
|
# The run_once() method is core to all autotest tests. We define it
|
|
# herein to support tests that do not define their own override.
|
|
self.run_once_impl(self.run_test_case)
|