链式调用shell指令

开发背景

在自动化测试和脚本化任务中，常需按步骤执行 adb shell 命令及处理结果。链式调用方式可将这些步骤用代码呈现，实现自动化设备管理和测试任务。例如自动化测试时，先通过 shell 命令安装应用，再检查安装结果，之后进行其他操作。使用 ADBShell 类的链式调用能将操作相连，利用 Python 脚本实现自动化流程控制。

功能实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class ADBShell:
    def __init__(self, device: str):
        self.shell = None  # shell 进程对象
        self.device = device
        self.stdout_adb = ""
        self.stderr_adb = ""
        self.start_shell()

    def start_shell(self):
        self.stdout_adb = ""
        self.stderr_adb = ""
        self.shell = subprocess.Popen(["adb", "-s", self.device, "shell"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    def write(self, command: str):
        if self.shell.stdin.closed:  # 如果输入流已关闭，则重新启动 ADB shell 进程
            self.start_shell()
        if command[-1]!= '\n':
            command += '\n'
        self.shell.stdin.write(command.encode())
        self.shell.stdin.flush()
        return self

    def read(self):
        if self.shell.stdin:
            self.shell.stdin.flush()
            self.shell.stdin.close()
        stdout, stderr = self.shell.communicate()
        return stdout.decode(), stderr.decode()

    def write_adb(self, adb_command: str):
        full_command = ["adb", "-s", self.device] + adb_command.split()
        process = subprocess.Popen(full_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        stdout, stderr = process.communicate()
        self.stdout_adb += stdout.decode() 
        self.stderr_adb += stderr.decode() 
        return self

    def read_adb(self):
        return self.stdout_adb, self.stderr_adb
    
    def flush(self):
        self.shell.stdin.flush()
        return self

    def close(self):
        if self.shell.stdin:
            self.shell.stdin.close()
        return self

    def sleep(self, sleep_time):
        time.sleep(sleep_time)
        return self

示例

示例场景(获取APP相关信息)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
adbShell = ADBShell("T442A02XNKH0189")
_ = (adbShell.   # 链式执行删除结果，安装APK，启动APK等操作
        write("cd /sdcard/Android/data/com.tt.appinfo/files/").
        write("rm result.txt").
        write_adb("install app-debug.apk").
        write("am start -n com.tt.appinfo/.MainActivity").
        read())
start = time.time()
while((time.time() - start) < 30):    # 等待执行结束
    stdout, stderr = adbShell.write(f"cd /sdcard/Android/data/com.tt.appinfo/files/").write("cat result.txt").read()
    if "1" in stdout:
        break
    time.sleep(1)
else:
    print("timeout")
adbShell.write_adb("pull /sdcard/Android/data/com.tt.appinfo/files/appInfo.db")   # 导出结果
print("finished")