[autotest] More ownership/device policy tests

client/cros/cros_ui_test.py

 # Copyright (c) 2010 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 dbus, logging, os, re, shutil, socket, sys
 import common
-import auth_server, constants as chromeos_constants, cryptohome, dns_server
+import auth_server, constants, cryptohome, dns_server
 import cros_logging, cros_ui, login, ownership
 from autotest_lib.client.bin import test, utils
 from autotest_lib.client.common_lib import error
 from dbus.mainloop.glib import DBusGMainLoop
 
 from autotest_lib.client.cros import flimflam_test_path
 import flimflam
 
 
 class UITest(test.test):
@@ skipping 61 matching lines @@
         for device in self._flim.GetObjectList('Device'):
             properties = device.GetProperties()
             interface = properties['Interface']
             logging.debug("Considering " + interface)
             for path in properties['IPConfigs']:
                 ipconfig = self._flim.GetObjectInterface('IPConfig', path)
 
                 servers = ipconfig.GetProperties().get('NameServers', None)
                 if servers != None:
                     self._dns[path] = ','.join(servers)
-                    logging.debug("Stored DNS for "  + interface)
+                    logging.debug("Stored %s for %s" % (self._dns[path],
+                                                        interface))
                 ipconfig.SetProperty('NameServers', '127.0.0.1')
                 logging.debug("Using local DNS for "  + interface)
 
         utils.poll_for_condition(
             lambda: self.__attempt_resolve('www.google.com', '127.0.0.1'),
             login.TimeoutError('Timed out waiting for DNS changes.'),
             10)
 
 
     def revert_dns(self):
@@ skipping 45 matching lines @@
         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.  We use DBus to tell flimflam to use this DNS server 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
-                chromeos_constants.CREDENTIALS, or a 'username:password' pair.
+                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.
 
         """
 
         # 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()
 
         if creds:
             self.start_authserver()
 
         # Fake ownership unless the test is explicitly testing owner creation.
         if not is_creating_owner and not os.access(
-            chromeos_constants.OWNER_KEY_FILE, os.F_OK):
+            constants.OWNER_KEY_FILE, os.F_OK):
             logging.info('Owner credentials not found. Faking ownership...')
             self.__fake_ownership()
             self.fake_owner = True
 
         if login.logged_in():
             login.attempt_logout()
 
         (self.username, self.password) = self.__resolve_creds(creds)
         # Ensure there's no stale cryptohome from previous tests.
         try:
             cryptohome.remove_vault(self.username)
         except cryptohome.ChromiumOSError as err:
             logging.error(err)
 
         if is_creating_owner:
             logging.info('Erasing stale owner state.')
-            # Ensure there's no stale owner state from previous tests.
-            try:
-                os.unlink(chromeos_constants.OWNER_KEY_FILE)
-                os.unlink(chromeos_constants.SIGNED_PREFERENCES_FILE)
-            except (IOError, OSError) as err:
-                logging.info(err)
+            ownership.clear_ownership()
 
         login.refresh_login_screen()
         if self.auto_login:
             self.login(self.username, self.password)
             if is_creating_owner:
                 login.wait_for_ownership()
 
     def __fake_ownership(self):
         """Fake ownership by generating the necessary magic files."""
         # Determine the module directory.
         dirname = os.path.dirname(__file__)
         mock_certfile = os.path.join(dirname, 'mock_owner_cert.pem')
         mock_signedprefsfile = os.path.join(dirname, 'mock_owner.preferences')
         utils.open_write_close(
-            chromeos_constants.OWNER_KEY_FILE,
+            constants.OWNER_KEY_FILE,
             ownership.cert_extract_pubkey_der(mock_certfile))
         shutil.copy(mock_signedprefsfile,
-                    chromeos_constants.SIGNED_PREFERENCES_FILE)
+                    constants.SIGNED_PREFERENCES_FILE)
 
 
     def __canonicalize(self, credential):
         """Perform basic canonicalization of |email_address|
 
         Perform basic canonicalization of |email_address|, taking
         into account that gmail does not consider '.' or caps inside a
         username to matter.  It also ignores everything after a '+'.
         For example, c.masone+abc@gmail.com == cMaSone@gmail.com, per
         http://mail.google.com/support/bin/answer.py?hl=en&ctx=mail&answer=10313
         """
         if not credential:
           return None
 
         parts = credential.split('@')
         if len(parts) != 2:
           raise error.TestError("Malformed email: " + credential)
 
         (name, domain) = parts
         name = name.partition('+')[0]
-        if (domain == chromeos_constants.SPECIAL_CASE_DOMAIN):
+        if (domain == constants.SPECIAL_CASE_DOMAIN):
             name = name.replace('.', '')
         return '@'.join([name, domain]).lower()
 
 
     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 chromeos_constants.CREDENTIALS:
+            if creds not in constants.CREDENTIALS:
                 raise error.TestFail('Unknown credentials: %s' % creds)
 
-            (name, passwd) = chromeos_constants.CREDENTIALS[creds]
+            (name, passwd) = constants.CREDENTIALS[creds]
             return [self.__canonicalize(name), passwd]
 
         (name, passwd) = creds.split(':')
         return [self.__canonicalize(name), passwd]
 
 
     def ensure_login_complete(self):
         """Wait for authentication to complete.  If you want a different
         termination condition, override this method.
         """
@@ skipping 71 matching lines @@
         in_logout = False
         for line in self._log_reader.get_logs().splitlines():
             if logout_start_regex.search(line):
                 in_logout = True
             elif logout_complete_regex.search(line):
                 in_logout = False
             else:
                 match = crash_regex.search(line)
                 if (match and not match.group(1) in processes and
                     not (in_logout and
-                         (match.group(1) == chromeos_constants.BROWSER or
+                         (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 cleanup(self):
         """Overridden from test.cleanup() to log out when the test is complete.
         """
-        logpath = chromeos_constants.CHROME_LOG_DIR
+        logpath = constants.CHROME_LOG_DIR
 
         try:
             for filename in os.listdir(logpath):
                 fullpath = os.path.join(logpath, filename)
                 if os.path.isfile(fullpath):
                     shutil.copy(fullpath, os.path.join(self.resultsdir,
                                                        filename))
 
         except (IOError, OSError) as err:
             logging.error(err)
 
         if login.logged_in():
             try:
                 shutil.copy(
-                    os.path.join(chromeos_constants.CRYPTOHOME_MOUNT_PT,
+                    os.path.join(constants.CRYPTOHOME_MOUNT_PT,
                                  'log', 'chrome'),
                     self.resultsdir+'/chrome_postlogin_log')
             except (IOError, OSError) as err:
                 logging.error(err)
             self.logout()
 
-        if os.path.isfile(chromeos_constants.CRYPTOHOMED_LOG):
+        if os.path.isfile(constants.CRYPTOHOMED_LOG):
             try:
-                base = os.path.basename(chromeos_constants.CRYPTOHOMED_LOG)
-                shutil.copy(chromeos_constants.CRYPTOHOMED_LOG,
+                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.')
-            try:
-                os.unlink(chromeos_constants.OWNER_KEY_FILE)
-                os.unlink(chromeos_constants.SIGNED_PREFERENCES_FILE)
-            except (IOError, OSError) as err:
-                logging.info(err)
+            ownership.clear_ownership()
 
         self.stop_authserver()
         self.__log_crashed_processes(self.crash_blacklist)
 
 
     def get_auth_endpoint_misses(self):
         if hasattr(self, '_authServer'):
             return self._authServer.get_endpoint_misses()
         else:
             return {}