Python：subprocess模块

简介：subprocess是子流程，即进程的意思，该模块可以启动一个新进程，并连接到它们的输入/输出/错误管道，从而获取返回值。

常用方法和接口：
1、subprocess.run()
2、subprocess.call()
3、subprocess.check_call()
4、subprocess.getstatusoutput()
5、subprocess.getoutput()
6、subprocess.check_output()
7、subprocess.Popen()

subprocess.run()解析：

常用参数：
args：表示要执行的命令。必须是一个字符串，字符串参数列表。
stdin、stdout 和 stderr：子进程的标准输入、输出和错误。其值可以是 subprocess.PIPE、subprocess.DEVNULL、一个已经存在的文件描述符、已经打开的文件对象或者 None。subprocess.PIPE 表示为子进程创建新的管道。subprocess.DEVNULL 表示使用 os.devnull。默认使用的是 None，表示什么都不做。另外，stderr 可以合并到 stdout 里一起输出。
timeout：设置命令超时时间。如果命令执行时间超时，子进程将被杀死，并弹出 TimeoutExpired 异常。
check：如果该参数设置为 True，并且进程退出状态码不是 0，则弹 出 CalledProcessError 异常。
encoding: 如果指定了该参数，则 stdin、stdout 和 stderr 可以接收字符串数据，并以该编码方式编码。否则只接收 bytes 类型的数据。
shell：如果该参数为 True，将通过操作系统的 shell 执行指定的命令。

subprocess.Popen()解析：Popen 是 subprocess的核心，子进程的创建和管理

常用参数：
args：shell命令，可以是字符串或者序列类型（如：list，元组）
bufsize：缓冲区大小。当创建标准流的管道对象时使用，默认-1。
0：不使用缓冲区
1：表示行缓冲，仅当universal_newlines=True时可用，也就是文本模式
正数：表示缓冲区大小
负数：表示使用系统默认的缓冲区大小。
stdin, stdout, stderr：分别表示程序的标准输入、输出、错误句柄
preexec_fn：只在 Unix 平台下有效，用于指定一个可执行对象（callable object），它将在子进程运行之前被调用
shell：如果该参数为 True，将通过操作系统的 shell 执行指定的命令。
cwd：用于设置子进程的当前目录。
env：用于指定子进程的环境变量。如果 env = None，子进程的环境变量将从父进程中继承。

Popen 对象方法：

poll(): 检查进程是否终止，如果终止返回 returncode，否则返回 None。
wait(timeout): 等待子进程终止。
communicate(input,timeout): 和子进程交互，发送和读取数据。
send_signal(singnal): 发送信号到子进程 。
terminate(): 停止子进程,也就是发送SIGTERM信号到子进程。
kill(): 杀死子进程。发送 SIGKILL 信号到子进程。

subprocess.run()案例：

import subprocess

print(subprocess.run(["pip", "list"]))

subprocess.call()案例：执行命令，返回命令的结果和执行状态，0或者非0

import subprocess

result1 = subprocess.call(["pip", "list"])
print("*" * 30)
print(result1)  # 0

result2 = subprocess.call(["pip", "list1"])
print("*" * 30)
print(result2)  # -1

# 运行结果
# ******************************
# 0
# ERROR: unknown command "list1" - maybe you meant "list"
# ******************************
# 1

subprocess.check_call()案例：执行命令，返回结果和状态，正常为0 ，执行错误则抛出异常

import subprocess

result1 = subprocess.check_call(["pip", "list"])
print("*" * 30)
print(result1)

result2 = subprocess.check_call(["pip", "list2"])
print("*" * 30)
print(result2)

# 运行结果，运行错误，抛出异常。
# ******************************
# 0
#    raise CalledProcessError(retcode, cmd)
# subprocess.CalledProcessError: Command '['pip', 'list2']' returned non-zero exit status 1.

subprocess.getstatusoutput()案例：接受字符串形式的命令，返回 一个元组形式的结果，第一个元素是命令执行状态，第二个为执行结果。

import subprocess

result1 = subprocess.getstatusoutput("pip --version")
print(result1)

# 返回：第一个元素0， 第二个为执行结果
# (0, 'pip 21.1.3 from C:\\lib\\site-packages\\pip (python 3.9)\n')


result2 = subprocess.getstatusoutput("pip -vesi")
print(result2)

# 执行失败的案例
# (2, '\nUsage:   \n  pip <command> [options]\n\nno such option: -e')

subprocess.getoutput()案例：接受字符串形式的命令，放回执行结果

import subprocess

result1 = subprocess.getoutput("pip --version")
print(result1)
# 返回
# pip 21.1.3 from D:\\lib\site-packages\pip (python 3.9)


result2 = subprocess.getoutput("pip -vesi")
print(result2)
# 返回
# Usage:   
#   pip <command> [options]
# 
# no such option: -e

subprocess.check_output()案例：执行命令，返回执行的结果，而不是打印

import subprocess

result1 = subprocess.check_output("pip --version")
print(result1)
# 返回
# b'pip 21.1.3 from D:\\lib\\site-packages\\pip (python 3.9)\r\r\n'


result2 = subprocess.check_output("pip -vesi")
print(result2)
# 返回
#     raise CalledProcessError(retcode, process.args,
# subprocess.CalledProcessError: Command 'pip -vesi' returned non-zero exit status 2.

subprocess.Popen()综合案例：案例中视频长度代码，请参考 - Python：获取视频时长 - 实践场景

# coding:utf-8
import time
import datetime
import subprocess
from analysis_video import video_duration

file = r"C:\Users\videos\duck.mp4"  # 请根据实际情况填写视频路径
video_time = video_duration(file)
print(video_time)
video_pid = subprocess.Popen(f'C:\Program Files\Windows Media Player\wmplayer.exe {file}')


# poll(): 检查进程是否终止，如果终止返回 returncode，否则返回 None。
print(f"now:{datetime.datetime.now()}. - 当前返回状态：{video_pid.poll()}")
# now:2021-11-28 11:20:48.504796. - 当前返回状态：None

# 获取当前执行子shell的程序的进程号
print("当前执行进程号为：{}".format(video_pid.pid))

sleeps = int(video_time) + 2
print(f"根据视频时长等待:{sleeps}秒")
time.sleep(sleeps)

# kill(): 杀死子进程。发送 SIGKILL 信号到子进程。
video_pid.kill()
time.sleep(1)
print(f"now:{datetime.datetime.now()}. - 当前返回状态：{video_pid.poll()}")
# now:2021-11-28 11:21:05.520822. - 当前返回状态：1

# wait()
# 等待命令执行完成，并且返回结果状态
print(video_pid.wait())  # 1

运行结果：

image.png

附，各种信号总结：

SIGHUP     终止进程     终端线路挂断
SIGINT     终止进程     中断进程
SIGQUIT   建立CORE文件终止进程，并且生成core文件
SIGILL   建立CORE文件       非法指令
SIGTRAP   建立CORE文件       跟踪自陷
SIGBUS   建立CORE文件       总线错误
SIGSEGV   建立CORE文件       段非法错误
SIGFPE   建立CORE文件       浮点异常
SIGIOT   建立CORE文件       执行I/O自陷
SIGKILL   终止进程     杀死进程
SIGPIPE   终止进程     向一个没有读进程的管道写数据
SIGALARM   终止进程     计时器到时
SIGTERM   终止进程     软件终止信号
SIGSTOP   停止进程     非终端来的停止信号
SIGTSTP   停止进程     终端来的停止信号
SIGCONT   忽略信号     继续执行一个停止的进程
SIGURG   忽略信号     I/O紧急信号
SIGIO     忽略信号     描述符上可以进行I/O
SIGCHLD   忽略信号     当子进程停止或退出时通知父进程
SIGTTOU   停止进程     后台进程写终端
SIGTTIN   停止进程     后台进程读终端
SIGXGPU   终止进程     CPU时限超时
SIGXFSZ   终止进程     文件长度过长
SIGWINCH   忽略信号     窗口大小发生变化
SIGPROF   终止进程     统计分布图用计时器到时
SIGUSR1   终止进程     用户定义信号1
SIGUSR2   终止进程     用户定义信号2
SIGVTALRM 终止进程     虚拟计时器到时

图片

微信公众号：玩转测试开发
欢迎关注，共同进步，谢谢！