Factor out a system_properties interface for interacting with getprop/setprop.

build/android/pylib/android_commands.py

 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Provides an interface to communicate with the device via the adb command.
 
 Assumes adb binary is currently on system path.
 """
 
 import collections
 import datetime
+import inspect
 import logging
 import os
 import re
 import shlex
 import signal
 import subprocess
 import sys
 import tempfile
 import time
 
 import cmd_helper
 import constants
 import screenshot
+import system_properties
 
 from utils import host_path_finder
 
 try:
   from pylib import pexpect
 except:
   pexpect = None
 
 sys.path.append(os.path.join(
     constants.DIR_SOURCE_ROOT, 'third_party', 'android_testrunner'))
@@ skipping 186 matching lines @@
   """Returns the timestamp of the given |log_line| in the given year."""
   try:
     return datetime.datetime.strptime('%s-%s' % (year, log_line[:18]),
                                       '%Y-%m-%d %H:%M:%S.%f')
   except (ValueError, IndexError):
     logging.critical('Error reading timestamp from ' + log_line)
     return None
 
 
 class AndroidCommands(object):
-  """Helper class for communicating with Android device via adb.
+  """Helper class for communicating with Android device via adb."""
 
-  Args:
-    device: If given, adb commands are only send to the device of this ID.
-        Otherwise commands are sent to all attached devices.
-  """
+  def __init__(self, device=None, api_strict_mode=False):
+    """Constructor.
 
-  def __init__(self, device=None):
+    Args:
+      device: If given, adb commands are only send to the device of this ID.
+          Otherwise commands are sent to all attached devices.
+      api_strict_mode: A boolean indicating whether fatal errors should be
+          raised if this API is used improperly.
+    """
     adb_dir = os.path.dirname(constants.ADB_PATH)
     if adb_dir and adb_dir not in os.environ['PATH'].split(os.pathsep):
       # Required by third_party/android_testrunner to call directly 'adb'.
       os.environ['PATH'] += os.pathsep + adb_dir
     self._adb = adb_interface.AdbInterface()
     if device:
       self._adb.SetTargetSerial(device)
     self._device = device
     self._logcat = None
     self.logcat_process = None
     self._logcat_tmpoutfile = None
     self._pushed_files = []
     self._device_utc_offset = None
     self._potential_push_size = 0
     self._actual_push_size = 0
     self._external_storage = ''
     self._util_wrapper = ''
+    self._api_strict_mode = api_strict_mode
+    self._system_properties = system_properties.SystemProperties(self.Adb())
+
+    if not self._api_strict_mode:
+      logging.warning(
+          'API STRICT MODE IS DISABLED.\n'
+          'It should be enabled as soon as possible as it will eventually '
+          'become the default.')
+
+  @property
+  def system_properties(self):
+    return self._system_properties
 
   def _LogShell(self, cmd):
     """Logs the adb shell command."""
     if self._device:
       device_repr = self._device[-4:]
     else:
       device_repr = '????'
     logging.info('[%s]> %s', device_repr, cmd)
 
   def Adb(self):
     """Returns our AdbInterface to avoid us wrapping all its methods."""
+    # TODO(tonyg): Disable this method when in _api_strict_mode.
     return self._adb
 
   def GetDevice(self):
     """Returns the device serial."""
     return self._device
 
   def IsOnline(self):
     """Checks whether the device is online.
 
     Returns:
@@ skipping 68 matching lines @@
     Args:
       full_reboot: Whether to fully reboot the device or just restart the shell.
     """
     # TODO(torne): hive can't reboot the device either way without breaking the
     # connection; work out if we can handle this better
     if os.environ.get('USING_HIVE'):
       logging.warning('Ignoring reboot request as we are on hive')
       return
     if full_reboot or not self.IsRootEnabled():
       self._adb.SendCommand('reboot')
+      self._system_properties = system_properties.SystemProperties(self.Adb())
       timeout = 300
       retries = 1
       # Wait for the device to disappear.
       while retries < 10 and self.IsOnline():
         time.sleep(1)
         retries += 1
     else:
       self.RestartShell()
       timeout = 120
     # To run tests we need at least the package manager and the sd card (or
     # other external storage) to be ready.
     self.WaitForDevicePm()
     self.WaitForSdCardReady(timeout)
 
   def Shutdown(self):
     """Shuts down the device."""
     self._adb.SendCommand('reboot -p')
+    self._system_properties = system_properties.SystemProperties(self.Adb())
 
   def Uninstall(self, package):
     """Uninstalls the specified package from the device.
 
     Args:
       package: Name of the package to remove.
 
     Returns:
       A status string returned by adb uninstall
     """
@@ skipping 147 matching lines @@
       set.
     """
     logging.info('Waiting for system boot completed...')
     self._adb.SendCommand('wait-for-device')
     # Now the device is there, but system not boot completed.
     # Query the sys.boot_completed flag with a basic command
     boot_completed = False
     attempts = 0
     wait_period = 5
     while not boot_completed and (attempts * wait_period) < wait_time:
-      output = self._adb.SendShellCommand('getprop sys.boot_completed',
-                                          retry_count=1)
+      output = self.system_properties['sys.boot_completed']
       output = output.strip()
       if output == '1':
         boot_completed = True
       else:
         # If 'error: xxx' returned when querying the flag, it means
         # adb server lost the connection to the emulator, so restart the adb
         # server.
         if 'error:' in output:
           self.RestartAdbServer()
         time.sleep(wait_period)
@@ skipping 13 matching lines @@
       output = self.RunShellCommand('ls ' + external_storage)
       if output:
         sdcard_ready = True
       else:
         time.sleep(wait_period)
         attempts += 1
     if not sdcard_ready:
       raise errors.WaitForResponseTimedOutError(
           'SD card not ready after %s seconds' % timeout_time)
 
+  def _CheckCommandIsValid(self, command):
+    """Raises a ValueError if the command is not valid."""
+
+    # A dict of commands the user should not run directly and a mapping to the
+    # API they should use instead.
+    preferred_apis = {
+        'getprop': 'system_properties[<PROPERTY>]',
+        'setprop': 'system_properties[<PROPERTY>]',
+        'su': 'RunShellCommandWithSU()',
+        }
+
+    # A dict of commands to methods that may call them.
+    whitelisted_callers = {
+        'su': 'RunShellCommandWithSU',
+        }
+
+    base_command = shlex.split(command)[0]
+    if (base_command in preferred_apis and
+        (base_command not in whitelisted_callers or
+         whitelisted_callers[base_command] not in [
+          f[3] for f in inspect.stack()])):
+      error_msg = ('%s cannot be run directly. Instead use: %s' %
+                   (base_command, preferred_apis[base_command]))
+      if self._api_strict_mode:
+        raise ValueError(error_msg)
+      else:
+        logging.warning(error_msg)
+
   # It is tempting to turn this function into a generator, however this is not
   # possible without using a private (local) adb_shell instance (to ensure no
   # other command interleaves usage of it), which would defeat the main aim of
   # being able to reuse the adb shell instance across commands.
   def RunShellCommand(self, command, timeout_time=20, log_result=False):
     """Send a command to the adb shell and return the result.
 
     Args:
       command: String containing the shell command to send. Must not include
                the single quotes as we use them to escape the whole command.
       timeout_time: Number of seconds to wait for command to respond before
         retrying, used by AdbInterface.SendShellCommand.
       log_result: Boolean to indicate whether we should log the result of the
                   shell command.
 
     Returns:
       list containing the lines of output received from running the command
     """
+    self._CheckCommandIsValid(command)
     self._LogShell(command)
     if "'" in command: logging.warning(command + " contains ' quotes")
     result = self._adb.SendShellCommand(
         "'%s'" % command, timeout_time).splitlines()
     if ['error: device not found'] == result:
       raise errors.DeviceUnresponsiveError('device not found')
     if log_result:
       self._LogShell('\n'.join(result))
     return result
 
@@ skipping 391 matching lines @@
   _TEMP_SCRIPT_FILE_BASE_FMT = 'temp_script_file_%d.sh'
 
   def _GetDeviceTempFileName(self, base_name):
     i = 0
     while self.FileExistsOnDevice(
         self.GetExternalStorage() + '/' + base_name % i):
       i += 1
     return self.GetExternalStorage() + '/' + base_name % i
 
   def RunShellCommandWithSU(self, command, timeout_time=20, log_result=False):
-    return self.RunShellCommand('su -c %s' % command,
-                                timeout_time=timeout_time,
-                                log_result=log_result)
+    return self.RunShellCommand('su -c %s' % command, timeout_time, log_result)
 
   def CanAccessProtectedFileContents(self):
     """Returns True if Get/SetProtectedFileContents would work via "su".
 
     Devices running user builds don't have adb root, but may provide "su" which
     can be used for accessing protected files.
     """
     r = self.RunShellCommandWithSU('cat /dev/null')
     return r == [] or r[0].strip() == ''
 
@@ skipping 80 matching lines @@
                               r'\s*=\s*\w+\s*$', re.MULTILINE)
       properties = re.sub(re_replace, '', properties)
       if enable:
         properties += '\n%s=all\n' % JAVA_ASSERT_PROPERTY
 
       file(temp_props_file.name, 'w').write(properties)
       self._adb.Push(temp_props_file.name, LOCAL_PROPERTIES_PATH)
 
     # Next, check the current runtime value is what we need, and
     # if not, set it and report that a reboot is required.
-    was_set = 'all' in self.RunShellCommand('getprop ' + JAVA_ASSERT_PROPERTY)
+    was_set = 'all' in self.system_properties[JAVA_ASSERT_PROPERTY]
     if was_set == enable:
       return False
-
-    self.RunShellCommand('setprop %s "%s"' % (JAVA_ASSERT_PROPERTY,
-                                              enable and 'all' or ''))
+    self.system_properties[JAVA_ASSERT_PROPERTY] = enable and 'all' or ''
     return True
 
   def GetBuildId(self):
     """Returns the build ID of the system (e.g. JRM79C)."""
-    build_id = self.RunShellCommand('getprop ro.build.id')[0]
+    build_id = self.system_properties['ro.build.id']
     assert build_id
     return build_id
 
   def GetBuildType(self):
     """Returns the build type of the system (e.g. eng)."""
-    build_type = self.RunShellCommand('getprop ro.build.type')[0]
+    build_type = self.system_properties['ro.build.type']
     assert build_type
     return build_type
 
   def GetBuildProduct(self):
     """Returns the build product of the device (e.g. maguro)."""
-    build_product = self.RunShellCommand('getprop ro.build.product')[0]
+    build_product = self.system_properties['ro.build.product']
     assert build_product
     return build_product
 
   def GetProductName(self):
     """Returns the product name of the device (e.g. takju)."""
-    name = self.RunShellCommand('getprop ro.product.name')[0]
+    name = self.system_properties['ro.product.name']
     assert name
     return name
 
   def GetBuildFingerprint(self):
     """Returns the build fingerprint of the device."""
-    build_fingerprint = self.RunShellCommand('getprop ro.build.fingerprint')[0]
+    build_fingerprint = self.system_properties['ro.build.fingerprint']
     assert build_fingerprint
     return build_fingerprint
 
   def GetDescription(self):
     """Returns the description of the system.
 
     For example, "yakju-userdebug 4.1 JRN54F 364167 dev-keys".
     """
-    description = self.RunShellCommand('getprop ro.build.description')[0]
+    description = self.system_properties['ro.build.description']
     assert description
     return description
 
   def GetProductModel(self):
     """Returns the name of the product model (e.g. "Galaxy Nexus") """
-    model = self.RunShellCommand('getprop ro.product.model')[0]
+    model = self.system_properties['ro.product.model']
     assert model
     return model
 
   def GetWifiIP(self):
     """Returns the wifi IP on the device."""
-    wifi_ip = self.RunShellCommand('getprop dhcp.wlan0.ipaddress')[0]
+    wifi_ip = self.system_properties['dhcp.wlan0.ipaddress']
     # Do not assert here. Devices (e.g. emulators) may not have a WifiIP.
     return wifi_ip
 
   def GetSubscriberInfo(self):
     """Returns the device subscriber info (e.g. GSM and device ID) as string."""
     iphone_sub = self.RunShellCommand('dumpsys iphonesubinfo')
     assert iphone_sub
     return '\n'.join(iphone_sub)
 
   def GetBatteryInfo(self):
     """Returns the device battery info (e.g. status, level, etc) as string."""
     battery = self.RunShellCommand('dumpsys battery')
     assert battery
     return '\n'.join(battery)
 
   def GetSetupWizardStatus(self):
     """Returns the status of the device setup wizard (e.g. DISABLED)."""
-    status = self.RunShellCommand('getprop ro.setupwizard.mode')[0]
+    status = self.system_properties['ro.setupwizard.mode']
     # On some devices, the status is empty if not otherwise set. In such cases
     # the caller should expect an empty string to be returned.
     return status
 
   def StartMonitoringLogcat(self, clear=True, logfile=None, filters=None):
     """Starts monitoring the output of logcat, for use with WaitForLogMatch.
 
     Args:
       clear: If True the existing logcat output will be cleared, to avoiding
              matching historical output lurking in the log.
@@ skipping 523 matching lines @@
   """
   def __init__(self, output):
     self._output = output
 
   def write(self, data):
     data = data.replace('\r\r\n', '\n')
     self._output.write(data)
 
   def flush(self):
     self._output.flush()