| # Copyright (c) 2012 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 glob, logging, os, re, shutil, subprocess, sys, time |
| |
| import auth_server, common, constants, cros_logging, cros_ui, cryptohome |
| import dns_server, login, ownership, pyauto_test |
| from autotest_lib.client.bin import utils |
| from autotest_lib.client.common_lib import error |
| |
| class UITest(pyauto_test.PyAutoTest): |
| """Base class for tests that drive some portion of the user interface. |
| |
| PYAUTO IS DEPRECATED. PLEASE DO NOT USE THIS CLASS. Use Telemetry instead. |
| |
| By default subclasses will use the default remote credentials before |
| the run_once method is invoked, and will log out at the completion |
| of the test case even if an exception is thrown. |
| |
| Subclasses can opt out of the automatic login by setting the member |
| variable 'auto_login' to False. |
| |
| Subclasses can log in with arbitrary credentials by passing |
| the 'creds' parameter in their control file. See the documentation of |
| UITest.initialize for more details. |
| |
| If your subclass overrides the initialize() or cleanup() methods, it |
| should make sure to invoke this class' version of those methods as well. |
| The standard super(...) function cannot be used for this, since the base |
| test class is not a 'new style' Python class. |
| """ |
| version = 1 |
| |
| skip_oobe = True |
| auto_login = True |
| fake_owner = True |
| username = None |
| password = None |
| |
| # Processes that we know crash and are willing to ignore. |
| crash_blacklist = [] |
| |
| # ftrace-related files. |
| _ftrace_process_fork_event_enable_file = \ |
| '/sys/kernel/debug/tracing/events/sched/sched_process_fork/enable' |
| _ftrace_process_fork_event_filter_file = \ |
| '/sys/kernel/debug/tracing/events/sched/sched_process_fork/filter' |
| _ftrace_signal_generate_event_enable_file = \ |
| '/sys/kernel/debug/tracing/events/signal/signal_generate/enable' |
| _ftrace_signal_generate_event_filter_file = \ |
| '/sys/kernel/debug/tracing/events/signal/signal_generate/filter' |
| _ftrace_trace_file = '/sys/kernel/debug/tracing/trace' |
| |
| _last_chrome_log = '' |
| |
| |
| def start_authserver(self, authenticator=None): |
| """Spin up a local mock of the Google Accounts server, then spin up |
| a local fake DNS server and tell the networking stack to use it. This |
| will trick Chrome into talking to our mock when we login. |
| Subclasses can override this method to change this behavior. |
| """ |
| self._authServer = auth_server.GoogleAuthServer( |
| authenticator=authenticator) |
| self._authServer.run() |
| self._dnsServer = dns_server.LocalDns() |
| self._dnsServer.run() |
| |
| |
| def stop_authserver(self): |
| """Tears down fake dns and fake Google Accounts server. If your |
| subclass does not create these objects, you will want to override this |
| method as well. |
| """ |
| if hasattr(self, '_authServer'): |
| self._authServer.stop() |
| del self._authServer |
| if hasattr(self, '_dnsServer'): |
| try: |
| self._dnsServer.stop() |
| except utils.TimeoutError as err: |
| raise error.TestWarn(err) |
| del self._dnsServer |
| |
| |
| def start_chrome_event_tracing(self): |
| """Start tracing events of a chrome process being created or receiving a |
| signal. |
| """ |
| try: |
| # Clear the trace buffer. |
| utils.open_write_close(self._ftrace_trace_file, '') |
| |
| # Trace only chrome process creation events, which we may later use |
| # to determine if a chrome process is killed by its parent. |
| utils.open_write_close( |
| self._ftrace_process_fork_event_filter_file, |
| 'child_comm==chrome') |
| # Trace only chrome processes receiving any signal except for |
| # the uninteresting SIGPROF (sig 27 on x86 and arm). |
| utils.open_write_close( |
| self._ftrace_signal_generate_event_filter_file, |
| 'comm==chrome && sig!=27') |
| |
| # Enable the process_fork event tracing. |
| utils.open_write_close( |
| self._ftrace_process_fork_event_enable_file, '1') |
| # Enable the signal_generate event tracing. |
| utils.open_write_close( |
| self._ftrace_signal_generate_event_enable_file, '1') |
| except IOError as err: |
| logging.warning('Failed to start chrome signal tracing: %s', err) |
| |
| |
| def stop_chrome_event_tracing(self): |
| """Stop tracing events of a chrome process being created or receiving a |
| signal. |
| """ |
| try: |
| # Disable the process_fork event tracing. |
| utils.open_write_close( |
| self._ftrace_process_fork_event_enable_file, '0') |
| # Disable the signal_generate event tracing. |
| utils.open_write_close( |
| self._ftrace_signal_generate_event_enable_file, '0') |
| |
| # Clear the process_fork event filter. |
| utils.open_write_close( |
| self._ftrace_process_fork_event_filter_file, '0') |
| # Clear the signal_generate event filter. |
| utils.open_write_close( |
| self._ftrace_signal_generate_event_filter_file, '0') |
| |
| # Dump the trace buffer to a log file. |
| trace_file = os.path.join(self.resultsdir, 'chrome_event_trace') |
| trace_data = utils.read_file(self._ftrace_trace_file) |
| utils.open_write_close(trace_file, trace_data) |
| except IOError as err: |
| logging.warning('Failed to stop chrome signal tracing: %s', err) |
| |
| |
| def start_tcpdump(self, iface): |
| """Start tcpdump process, if not running already.""" |
| if not hasattr(self, '_tcpdump'): |
| self._tcpdump = subprocess.Popen( |
| ['tcpdump', '-i', iface, '-vv'], stdout=subprocess.PIPE, |
| stderr=subprocess.STDOUT) |
| |
| |
| def stop_tcpdump(self, fname_prefix): |
| """Stop tcpdump process and save output to a new file.""" |
| if hasattr(self, '_tcpdump'): |
| self._tcpdump.terminate() |
| # Save output to a new file |
| next_index = len(glob.glob( |
| os.path.join(self.resultsdir, '%s-*' % fname_prefix))) |
| tcpdump_file = os.path.join( |
| self.resultsdir, '%s-%d' % (fname_prefix, next_index)) |
| logging.info('Saving tcpdump output to %s' % tcpdump_file) |
| utils.open_write_close(tcpdump_file, self._tcpdump.communicate()[0]) |
| del self._tcpdump |
| |
| |
| def __log_all_processes(self, fname_prefix): |
| """Log all processes to a file. |
| |
| Args: |
| fname_prefix: Prefix of the log file. |
| """ |
| try: |
| next_index = len(glob.glob( |
| os.path.join(self.resultsdir, '%s-*' % fname_prefix))) |
| log_file = os.path.join( |
| self.resultsdir, '%s-%d' % (fname_prefix, next_index)) |
| utils.open_write_close(log_file, utils.system_output('ps -eF')) |
| except (error.CmdError, IOError, OSError) as err: |
| logging.warning('Failed to log all processes: %s', err) |
| |
| |
| def __perform_ui_diagnostics(self): |
| """Save diagnostic logs about UI. |
| |
| This includes the output of: |
| $ initctl status ui |
| $ ps auxwww |
| """ |
| output_file = os.path.join(self.resultsdir, 'ui_diagnostics.txt') |
| with open(output_file, 'w') as output_fd: |
| print >> output_fd, time.asctime(), '\n' |
| cmd = 'initctl status ui' |
| print >> output_fd, '$ %s' % cmd |
| print >> output_fd, utils.system_output(cmd), '\n' |
| cmd = 'ps auxwww' |
| print >> output_fd, '$ %s' % cmd |
| print >> output_fd, utils.system_output(cmd), '\n' |
| logging.info('Saved UI diagnostics to %s' % output_file) |
| |
| |
| def __generate_coredumps(self, names): |
| """Generate core dump files in results dir for given processes. |
| |
| Note that the coredumps are forced via SIGBUS and the processes will be |
| terminated. Ideally we should use gdb gcore to create dumps |
| non-intrusively. However, the current dumps generated by gcore could not |
| be properly read back by gdb, i.e. no reasonable symbolized stack could |
| be generated. |
| |
| Args: |
| names: A list of process names that need to be dumped. |
| """ |
| |
| # Get all pids of named processes. |
| pids = [] |
| for name in names: |
| # Get pids of given name, slice [1:] to skip ps's first line 'PID' |
| pids = pids + [ pid.strip() for pid in utils.system_output( |
| 'ps -C %s -o pid' % name).splitlines()[1:]] |
| logging.info('Will force core dumps for the following pid: %s' % |
| ' '.join(pids)) |
| |
| # Stop all processes so that forcing dump would change their state. |
| for pid in pids: |
| utils.system('kill -STOP %s' % pid) |
| |
| # Force core dump. |
| for pid in pids: |
| utils.system('kill -BUS %s' % pid) |
| |
| # Resume to let the core dump finish. |
| for pid in reversed(pids): |
| utils.system('kill -CONT %s' % pid) |
| |
| |
| def initialize(self, creds=None, is_creating_owner=False, |
| extra_chrome_flags=[], subtract_extra_chrome_flags=[], |
| *args, **kwargs): |
| """Overridden from test.initialize() to log out and (maybe) log in. |
| |
| If self.auto_login is True, this will automatically log in using the |
| credentials specified by 'creds' at startup, otherwise login will not |
| happen. |
| |
| Regardless of the state of self.auto_login, the self.username and |
| self.password properties will be set to the credentials specified |
| by 'creds'. |
| |
| Authentication is not performed against live servers. Instead, we spin |
| up a local DNS server that will lie and say that all sites resolve to |
| 127.0.0.1. The DNS server tells flimflam via DBus that it should be |
| used to resolve addresses. We then spin up a local httpd that will |
| respond to queries at the Google Accounts endpoints. We clear the DNS |
| setting and tear down these servers in cleanup(). |
| |
| Args: |
| creds: String specifying the credentials for this test case. Can |
| be a named set of credentials as defined by |
| constants.CREDENTIALS, or a 'username:password' pair. |
| Defaults to None -- browse without signing-in. |
| is_creating_owner: If the test case is creating a new device owner. |
| extra_chrome_flags: Extra chrome flags to pass to chrome, if any. |
| subtract_extra_chrome_flags: Remove default flags passed to chrome |
| by pyauto, if any. |
| """ |
| # Mark /var/log/messages now; we'll run through all subsequent |
| # log messages at the end of the test and log info about processes that |
| # crashed. |
| self._log_reader = cros_logging.LogReader() |
| self._log_reader.set_start_by_current() |
| |
| # Do not use gaia even for incognito browsing. |
| self.start_authserver() |
| |
| # Run tcpdump on 'lo' interface to investigate network |
| # issues in the lab during login. |
| self.start_tcpdump(iface='lo') |
| |
| # Log all processes so that we can correlate PIDs to processes in |
| # the chrome signal trace. |
| self.__log_all_processes('processes--before-tracing') |
| |
| # Start event tracing related to chrome processes. |
| self.start_chrome_event_tracing() |
| |
| # We yearn for Chrome coredumps... |
| open(constants.CHROME_CORE_MAGIC_FILE, 'w').close() |
| |
| # The UI must be taken down to ensure that no stale state persists. |
| cros_ui.stop() |
| (self.username, self.password) = self.__resolve_creds(creds) |
| # Ensure there's no stale cryptohome from previous tests. |
| try: |
| cryptohome.remove_all_vaults() |
| except cryptohome.ChromiumOSError as err: |
| logging.error(err) |
| |
| # Fake ownership unless the test is explicitly testing owner creation. |
| if not is_creating_owner: |
| logging.info('Faking ownership...') |
| ownership.fake_ownership() |
| self.fake_owner = True |
| else: |
| logging.info('Erasing stale owner state.') |
| ownership.clear_ownership_files() |
| self.fake_owner = False |
| |
| try: |
| cros_ui.start() |
| except: |
| self.__perform_ui_diagnostics() |
| if not login.wait_for_browser_exit('Chrome crashed during login'): |
| self.__generate_coredumps([constants.BROWSER]) |
| raise |
| |
| # Save name of the last chrome log before our test started. |
| log_files = glob.glob(constants.CHROME_LOG_DIR + '/chrome_*') |
| self._last_chrome_log = max(log_files) if log_files else '' |
| |
| pyauto_test.PyAutoTest.initialize( |
| self, auto_login=False, |
| extra_chrome_flags=extra_chrome_flags, |
| subtract_extra_chrome_flags=subtract_extra_chrome_flags, |
| *args, **kwargs) |
| if self.skip_oobe or self.auto_login: |
| self.pyauto.SkipToLogin() |
| if self.auto_login: |
| self.login(self.username, self.password) |
| if is_creating_owner: |
| login.wait_for_ownership() |
| |
| |
| def __resolve_creds(self, creds): |
| """Map credential identifier to username, password and type. |
| Args: |
| creds: credential identifier to resolve. |
| |
| Returns: |
| A (username, password) tuple. |
| """ |
| if not creds: |
| return [None, None] # Browse without signing-in. |
| if creds[0] == '$': |
| if creds not in constants.CREDENTIALS: |
| raise error.TestFail('Unknown credentials: %s' % creds) |
| |
| (name, passwd) = constants.CREDENTIALS[creds] |
| return [cryptohome.canonicalize(name), passwd] |
| |
| (name, passwd) = creds.split(':') |
| return [cryptohome.canonicalize(name), passwd] |
| |
| |
| def take_screenshot(self, fname_prefix, format='png'): |
| """Take screenshot and save to a new file in the results dir. |
| |
| Args: |
| fname_prefix: prefix for the output fname |
| format: string indicating file format ('png', 'jpg', etc) |
| |
| Returns: |
| the path of the saved screenshot file |
| """ |
| next_index = len(glob.glob( |
| os.path.join(self.resultsdir, '%s-*.%s' % (fname_prefix, format)))) |
| screenshot_file = os.path.join( |
| self.resultsdir, '%s-%d.%s' % (fname_prefix, next_index, format)) |
| logging.info('Saving screenshot to %s.' % screenshot_file) |
| |
| old_exc_type = sys.exc_info()[0] |
| try: |
| utils.system('DISPLAY=:0.0 XAUTHORITY=/home/chronos/.Xauthority ' |
| '/usr/local/bin/import -window root -depth 8 %s' % |
| screenshot_file) |
| except Exception as err: |
| # Do not raise an exception if the screenshot fails while processing |
| # another exception. |
| if old_exc_type is None: |
| raise |
| logging.error(err) |
| |
| return screenshot_file |
| |
| |
| def login(self, username=None, password=None): |
| """Log in with a set of credentials. |
| |
| This method is called from UITest.initialize(), so you won't need it |
| unless your testcase has cause to log in multiple times. This |
| DOES NOT affect self.username or self.password. |
| |
| If username and self.username are not defined, logs in as guest. |
| |
| Forces a log out if already logged in. |
| Blocks until login is complete. |
| |
| Args: |
| username: username to log in as, defaults to self.username. |
| password: password to log in with, defaults to self.password. |
| |
| Raises: |
| error.TestError, if login has an error |
| """ |
| if self.logged_in(): |
| self.logout() |
| |
| uname = username or self.username |
| passwd = password or self.password |
| |
| try: |
| screenshot_name = 'login-success-screenshot' |
| if uname: # Regular login |
| login_error = self.pyauto.Login(username=uname, |
| password=passwd) |
| if login_error: |
| screenshot_name = 'login-error-screenshot' |
| raise error.TestFail( |
| 'Error during login (%s, %s): %s. See the file named ' |
| '%s.png in the results folder.' % (uname, passwd, |
| login_error, screenshot_name)) |
| else: # Login as guest |
| self.pyauto.LoginAsGuest() |
| logging.info('Logged in as guest.') |
| if not self.logged_in(): |
| screenshot_name = 'login-bizarre-fail-screenshot' |
| raise error.TestFail('Login was successful, but logged_in() ' |
| 'returned False. This should not happen. ' |
| 'Please check the file named %s.png ' |
| 'located in the results folder.' % |
| screenshot_name) |
| except Exception as err: |
| if isinstance(err, error.AutotestError): |
| raise # Do not modify our own errors. |
| |
| screenshot_name = 'login-fail-screenshot' |
| raise error.TestFail('Exception raised during login: %s. See the ' |
| 'file named %s.png in the results folder.' % |
| (err, screenshot_name)) |
| finally: |
| self.take_screenshot(fname_prefix=screenshot_name) |
| self.stop_tcpdump(fname_prefix='tcpdump-lo--till-login') |
| |
| logging.info('Logged in as %s. You can verify with the ' |
| 'file named %s.png located in the results ' |
| 'folder.' % (uname, screenshot_name)) |
| |
| def logged_in(self): |
| return self.pyauto.GetLoginInfo()['is_logged_in'] |
| |
| |
| def logout(self): |
| """Log out. |
| |
| This method is called from UITest.cleanup(), so you won't need it |
| unless your testcase needs to test functionality while logged out. |
| """ |
| if not self.logged_in(): |
| return |
| self._save_logs_from_cryptohome() |
| |
| try: |
| cros_ui.restart(self.pyauto.Logout) |
| except: |
| self.__perform_ui_diagnostics() |
| if not login.wait_for_browser_exit('Chrome crashed during logout'): |
| self.__generate_coredumps([constants.BROWSER]) |
| raise |
| |
| |
| def _save_logs_from_cryptohome(self): |
| """Recover dirs from cryptohome in case another test run wipes.""" |
| try: |
| for dir in constants.CRYPTOHOME_DIRS_TO_RECOVER: |
| dir_path = os.path.join(constants.CRYPTOHOME_MOUNT_PT, dir) |
| if os.path.isdir(dir_path): |
| target = os.path.join(self.resultsdir, |
| '%s-%f' % (dir, time.time())) |
| logging.debug('Saving %s to %s.', dir_path, target) |
| shutil.copytree(src=dir_path, dst=target, symlinks=True) |
| except (IOError, OSError, shutil.Error) as err: |
| logging.error(err) |
| |
| |
| def validate_basic_policy(self, basic_policy): |
| # Pull in protobuf definitions. |
| sys.path.append(self.srcdir) |
| from device_management_backend_pb2 import PolicyFetchResponse |
| from device_management_backend_pb2 import PolicyData |
| from chrome_device_policy_pb2 import ChromeDeviceSettingsProto |
| from chrome_device_policy_pb2 import UserWhitelistProto |
| |
| response_proto = PolicyFetchResponse() |
| response_proto.ParseFromString(basic_policy) |
| ownership.assert_has_policy_data(response_proto) |
| |
| poldata = PolicyData() |
| poldata.ParseFromString(response_proto.policy_data) |
| ownership.assert_has_device_settings(poldata) |
| ownership.assert_username(poldata, self.username) |
| |
| polval = ChromeDeviceSettingsProto() |
| polval.ParseFromString(poldata.policy_value) |
| ownership.assert_new_users(polval, True) |
| ownership.assert_users_on_whitelist(polval, (self.username,)) |
| |
| |
| def __log_crashed_processes(self, processes): |
| """Runs through the log watched by |watcher| to see if a crash was |
| reported for any process names not listed in |processes|. SIGABRT |
| crashes in chrome or supplied-chrome during ui restart are ignored. |
| """ |
| ui_restart_begin_regex = re.compile(cros_ui.UI_RESTART_ATTEMPT_MSG) |
| crash_regex = re.compile( |
| 'Received crash notification for ([-\w]+).+ (sig \d+)') |
| ui_restart_end_regex = re.compile(cros_ui.UI_RESTART_COMPLETE_MSG) |
| |
| in_restart = False |
| for line in self._log_reader.get_logs().splitlines(): |
| if ui_restart_begin_regex.search(line): |
| in_restart = True |
| elif ui_restart_end_regex.search(line): |
| in_restart = False |
| else: |
| match = crash_regex.search(line) |
| if (match and not match.group(1) in processes and |
| not (in_restart and |
| (match.group(1) == constants.BROWSER or |
| match.group(1) == 'supplied_chrome') and |
| match.group(2) == 'sig 6')): |
| self.job.record('INFO', self.tagged_testname, |
| line[match.start():]) |
| |
| |
| def execute(self, iterations=None, test_length=None, |
| profile_only=None, _get_time=time.time, |
| postprocess_profiled_run=None, constraints=(), *args, **kwargs): |
| """Wrapper around execute to take a screenshot for any exception.""" |
| try: |
| super(UITest, self).execute(iterations=iterations, |
| test_length=test_length, |
| profile_only=profile_only, |
| _get_time=_get_time, |
| postprocess_profiled_run= |
| postprocess_profiled_run, |
| constraints=constraints, |
| *args, **kwargs) |
| except: |
| self.take_screenshot(fname_prefix='test-fail-screenshot') |
| raise |
| |
| |
| def cleanup(self): |
| """Overridden from pyauto_test.cleanup() to log out and restart |
| session_manager when the test is complete. |
| """ |
| try: |
| # Save all chrome logs created during the test. |
| try: |
| for fullpath in glob.glob( |
| constants.CHROME_LOG_DIR + '/chrome_*'): |
| if os.path.isfile(fullpath) and \ |
| not os.path.islink(fullpath) and \ |
| fullpath > self._last_chrome_log: # ignore old logs |
| shutil.copy2(fullpath, self.resultsdir) |
| |
| except (IOError, OSError) as err: |
| logging.error(err) |
| |
| self._save_logs_from_cryptohome() |
| pyauto_test.PyAutoTest.cleanup(self) |
| |
| if os.path.isfile(constants.CRYPTOHOMED_LOG): |
| try: |
| base = os.path.basename(constants.CRYPTOHOMED_LOG) |
| shutil.copy(constants.CRYPTOHOMED_LOG, |
| os.path.join(self.resultsdir, base)) |
| except (IOError, OSError) as err: |
| logging.error(err) |
| |
| if self.fake_owner: |
| logging.info('Erasing fake owner state.') |
| ownership.clear_ownership_files() |
| |
| self.__log_crashed_processes(self.crash_blacklist) |
| |
| if os.path.isfile(constants.CHROME_CORE_MAGIC_FILE): |
| os.unlink(constants.CHROME_CORE_MAGIC_FILE) |
| finally: |
| self.stop_chrome_event_tracing() |
| self.__log_all_processes('processes--after-tracing') |
| self.stop_tcpdump(fname_prefix='tcpdump-lo--till-end') |
| self.stop_authserver() |
| |
| |
| def get_auth_endpoint_misses(self): |
| if hasattr(self, '_authServer'): |
| return self._authServer.get_endpoint_misses() |
| else: |
| return {} |
