Home | History | Annotate | Download | only in theme 
      1 #!/usr/bin/python3
      2 #
      3 # Copyright (C) 2015 The Android Open Source Project
      4 #
      5 # Licensed under the Apache License, Version 2.0 (the 'License');
      6 # you may not use this file except in compliance with the License.
      7 # You may obtain a copy of the License at
      8 #
      9 #      http://www.apache.org/licenses/LICENSE-2.0
     10 #
     11 # Unless required by applicable law or agreed to in writing, software
     12 # distributed under the License is distributed on an 'AS IS' BASIS,
     13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     14 # See the License for the specific language governing permissions and
     15 # limitations under the License.
     16 #
     17 
     18 import os
     19 import re
     20 import subprocess
     21 import sys
     22 import threading
     23 import time
     24 
     25 from subprocess import PIPE
     26 
     27 
     28 # class for running android device from python
     29 # it will fork the device processor
     30 class AndroidDevice(object):
     31     def __init__(self, serial):
     32         self._serial = serial
     33 
     34     def run_adb_command(self, cmd, timeout=None):
     35         adb_cmd = "adb -s %s %s" % (self._serial, cmd)
     36         print(adb_cmd)
     37 
     38         adb_process = subprocess.Popen(args=adb_cmd.split(), bufsize=-1, stderr=PIPE, stdout=PIPE)
     39         (out, err) = adb_process.communicate(timeout=timeout)
     40         return out.decode('utf-8').strip(), err.decode('utf-8').strip()
     41 
     42     def run_shell_command(self, cmd):
     43         return self.run_adb_command("shell %s" % cmd)
     44 
     45     def wait_for_device(self, timeout=30):
     46         return self.run_adb_command('wait-for-device', timeout)
     47 
     48     def wait_for_prop(self, key, value, timeout=30):
     49         boot_complete = False
     50         attempts = 0
     51         wait_period = 1
     52         while not boot_complete and (attempts*wait_period) < timeout:
     53             (out, err) = self.run_shell_command("getprop %s" % key)
     54             if out == value:
     55                 boot_complete = True
     56             else:
     57                 time.sleep(wait_period)
     58                 attempts += 1
     59         if not boot_complete:
     60             print("%s not set to %s within timeout!" % (key, value))
     61         return boot_complete
     62 
     63     def wait_for_service(self, name, timeout=30):
     64         service_found = False
     65         attempts = 0
     66         wait_period = 1
     67         while not service_found and (attempts*wait_period) < timeout:
     68             (output, err) = self.run_shell_command("service check %s" % name)
     69             if 'not found' not in output:
     70                 service_found = True
     71             else:
     72                 time.sleep(wait_period)
     73                 attempts += 1
     74         if not service_found:
     75             print("Service '%s' not found within timeout!" % name)
     76         return service_found
     77 
     78     def wait_for_boot_complete(self, timeout=60):
     79         return self.wait_for_prop('dev.bootcomplete', '1', timeout)
     80 
     81     def install_apk(self, apk_path):
     82         self.wait_for_service('package')
     83         (out, err) = self.run_adb_command("install -r -d -g %s" % apk_path)
     84         result = out.split()
     85         return out, err, "Success" in result
     86 
     87     def uninstall_package(self, package):
     88         self.wait_for_service('package')
     89         (out, err) = self.run_adb_command("uninstall %s" % package)
     90         result = out.split()
     91         return "Success" in result
     92 
     93     def run_instrumentation_test(self, option):
     94         self.wait_for_service('activity')
     95         return self.run_shell_command("am instrument -w --no-window-animation %s" % option)
     96 
     97     def is_process_alive(self, process_name):
     98         (out, err) = self.run_shell_command("ps")
     99         names = out.split()
    100         # very lazy implementation as it does not filter out things like uid
    101         # should work mostly unless processName is too simple to overlap with
    102         # uid. So only use name like com.android.xyz
    103         return process_name in names
    104 
    105     def get_version_sdk(self):
    106         return int(self.run_shell_command("getprop ro.build.version.sdk")[0])
    107 
    108     def get_version_codename(self):
    109         return self.run_shell_command("getprop ro.build.version.codename")[0].strip()
    110 
    111     def get_density(self):
    112         if "emulator" in self._serial:
    113             return int(self.run_shell_command("getprop qemu.sf.lcd_density")[0])
    114         else:
    115             return int(self.run_shell_command("getprop ro.sf.lcd_density")[0])
    116 
    117     def get_orientation(self):
    118         return int(self.run_shell_command("dumpsys | grep SurfaceOrientation")[0].split()[1])
    119 
    120 
    121 def enumerate_android_devices(require_prefix=''):
    122     devices = subprocess.check_output(["adb", "devices"])
    123     if not devices:
    124         return []
    125 
    126     devices = devices.decode('UTF-8').split('\n')[1:]
    127     device_list = []
    128 
    129     for device in devices:
    130         if device is not "" and device.startswith(require_prefix):
    131             info = device.split('\t')
    132             if info[1] == "device":
    133                 device_list.append(info[0])
    134 
    135     return device_list
    136