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 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.
+    managed_commands = {
+        'getprop': 'system_properties[<PROPERTY>]',
+        'setprop': 'system_properties[<PROPERTY>]',
+        'su': 'RunShellCommandWithSU()',
+        }
+
+    base_command = command.split()[0]
+    if base_command in managed_commands:
+      error_msg = ('%s cannot be run directly. Instead use: %s' %
+                   (base_command, managed_commands[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)

[bulach, 2013/12/03 14:50:49]

why not just add this check to RunShellCommand and avoid having this internal
method?

[tonyg, 2013/12/03 15:29:36]

Because RunShellCommandWithSU calls RunShellCommand and CheckCommandIsValid
prohibits su.

[bulach, 2013/12/03 15:39:30]

how about assert that if RunShellCommand's command has "su" then it's caller is
indeed RunShellCommandWithSU?

import inspect

caller = inspect.stack()[1][3]

self._CheckCommandIsValid(command, caller)


my point is that having both RunShellCommand and _RunShellCommand would be prone
to abuse and confusion, so I'd rather keep it simple and even more explicit
about the tests... wdyt?

+    return self._RunShellCommand(command, timeout_time, log_result)
+
+  def _RunShellCommand(self, command, timeout_time, log_result):
     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)
+    self._CheckCommandIsValid(command)

[bulach, 2013/12/03 14:50:49]

see above..

[tonyg, 2013/12/03 15:29:36]

ditto

+    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()