• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
18import os
19import re
20import subprocess
21import sys
22import threading
23import time
24
25from subprocess import PIPE
26
27
28# class for running android device from python
29# it will fork the device processor
30class 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
121def 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