You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
这个终止进程不同系统平台的实现方式不一样，从代码可以看出，至少windows 32位的实现方式和别的不一样，所以需要单拎出来，先看普遍的实现方法os.kill(pid, sig)吧，它有两个参数，pid指的是进程的pid，sig值signal模块的信号码，这里终止进程的信号码是signal.SIGTERM。对于windows 32位的实现方法我查了一下文档都没有讲这么详细的函数，倒是在stackoverflow的时候查到了一个相关的In Python 2.5, how do I kill a subprocess?
首先就是__init__函数了，这个就是实例化Command类的时候，就会自动调用的，这里就是初始化一些变量。主要就是把参数cmd存到self.cmd中，其他的就不多说了。接下来是一个类的run方法。先把data变量存到self.data中，然后dict(os.environ)把系统的环境变量转成字典形式存到environ中，再调用environ.update(env or {})把env变量更新environ变量，更新的意思是environ没有的项而env有的，就增量加到environ中，如果env和environ中有相同的键，但值不一样，那么就用env的值替换environ对应项的值。update函数里还有个or表达式，意思是如果env是None的话，就用空字典取更新environ，其实就是跟没更新一样。可以看看我电脑里的环境变量
In [4]: print(threading.Thread.__doc__)
A class that represents a thread of control.
This class can be safely subclassed in a limited fashion. There are two ways
to specify the activity: by passing a callable object to the constructor, or
by overriding the run() method in a subclass.
In [5]: print(threading.Thread.start.__doc__)
Start the thread's activity.
It must be called at most once per thread object. It arranges for the
object's run() method to be invoked in a separate thread of control.
This method will raise a RuntimeError if called more than once on the
same thread object.
In [6]: print(threading.Thread.join.__doc__)
Wait until the thread terminates.
This blocks the calling thread until the thread whose join() method is
called terminates -- either normally or through an unhandled exception
or until the optional timeout occurs.
When the timeout argument is present and not None, it should be a
floating point number specifying a timeout for the operation in seconds
(or fractions thereof). As join() always returns None, you must call
isAlive() after join() to decide whether a timeout happened -- if the
thread is still alive, the join() call timed out.
When the timeout argument is not present or None, the operation will
block until the thread terminates.
A thread can be join()ed many times.
join() raises a RuntimeError if an attempt is made to join the current
thread as that would cause a deadlock. It is also an error to join() a
thread before it has been started and attempts to do so raises the same
exception.
@property
def status_code(self):
"""The status code of the process.
If the code is None, assume that it's still running.
"""
return self._status_code
@property
def pid(self):
"""The process' PID."""
return self._process.pid
往下看后三个函数
def expect(self, bytes, stream=None):
"""Block until given bytes appear in the stream."""
if stream is None:
stream = self.std_out
def send(self, str, end='\n'):
"""Sends a line to std_in."""
return self._process.stdin.write(str+end)
def block(self):
"""Blocks until command finishes. Returns Response instance."""
self._status_code = self._process.wait()
def expand_args(command):
"""Parses command strings and returns a Popen-ready list."""
# Prepare arguments.
if isinstance(command, (str, unicode)):
splitter = shlex.shlex(command.encode('utf-8'))
splitter.whitespace = '|'
splitter.whitespace_split = True
command = []
while True:
token = splitter.get_token()
if token:
command.append(token)
else:
break
command = list(map(shlex.split, command))
return command
这个expand_args就是负责解析命令行字符，返回Popen-ready列表。isinstance函数就是判断前者是不是后者的实例，或者是后者的直接或者间接的子类型。具体到当前的函数就是判断command是不是str或者unicode的子类型。官方文档shlex.shlex(instream=None, infile=None, posix=False, punctuation_chars=False)有四个参数，instream指从哪里读取参数，类型必须是a file-/stream-like object with read() and readline() methods(感觉真的是英文说的很直白，翻译成中文倒是很别扭，所以还是放英文吧)这样的文档流或者是字符串。如果该参数为空，默认会从sys.stdin获取；第二个参数infile是一个文件名字符串，这个在调试的时候方便区分。如果第一个参数instream为空，infile默认值就是"stdin"，posix标识操作系统类型，默认指定为False，表明是兼容模式；如果该值为POSIX，则表明是类unix操作系统模式，将按照类unix的解析规则对instream进行解析。punctuation_chars参数的解释The punctuation_chars argument provides a way to make the behaviour even closer to how real shells parse，大约说的是这个参数提供了一个使解析的行为方式更靠近真实的shells parse的方法。它的默认值是False，操作的规则就是在python3.5以前的规则，因为这个参数是在python3.6加的，只有在3.6以后才有意义，如果该值位True，那么在解析中关于();<>|&的解析会有不同，那是怎么样一种规则呢，来看例子
# fxm @ bogon in ~/work/robot-admin on git:master x [12:12:02]
$ python
Python 3.6.1 (default, Jun 1 2017, 10:34:26)
[GCC 4.2.1 Compatible Apple LLVM 8.0.0 (clang-800.0.42.1)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import shlex
>>> text = "a && b; c && d || e; f >'abc'; (def \"ghi\")"
>>> list(shlex.shlex(text))
['a', '&', '&', 'b', ';', 'c', '&', '&', 'd', '|', '|', 'e', ';', 'f', '>', "'abc'", ';', '(', 'def', '"ghi"', ')']
>>> list(shlex.shlex(text, punctuation_chars=True))
['a', '&&', 'b', ';', 'c', '&&', 'd', '||', 'e', ';', 'f', '>', "'abc'", ';', '(', 'def', '"ghi"', ')']
If True, tokens will only be split in whitespaces. This is useful, for example, for parsing command lines with shlex, getting tokens in a similar way to shell arguments. If this attribute is True, punctuation_chars will have no effect, and splitting will happen only on whitespaces. When using punctuation_chars, which is intended to provide parsing closer to that implemented by shells, it is advisable to leave whitespace_split as False (the default value).
# 默认情况下
~/source/backend/python/envoy master
❯ ipython
Python 3.5.2 (default, Nov 17 2016, 17:05:23)
Type "copyright", "credits" or "license" for more information.
IPython 5.1.0 -- An enhanced Interactive Python.
? -> Introduction and overview of IPython's features.
%quickref -> Quick reference.
help -> Python's own help system.
object? -> Details about 'object', use 'object??' for extra details.
In [1]: import shlex
In [2]: command = "cat | tr [:lower:] [:upper:]"
In [3]: splitter = shlex.shlex(command)
In [4]: splitter.get_token()
Out[4]: 'cat'
In [5]: splitter.get_token()
Out[5]: '|'
In [6]: splitter.get_token()
Out[6]: 'tr'
# 加了whitespace和whitespace_split的情况下
~/source/backend/python/envoy master 4m 34s
❯ ipython
Python 3.5.2 (default, Nov 17 2016, 17:05:23)
Type "copyright", "credits" or "license" for more information.
IPython 5.1.0 -- An enhanced Interactive Python.
? -> Introduction and overview of IPython's features.
%quickref -> Quick reference.
help -> Python's own help system.
object? -> Details about 'object', use 'object??' for extra details.
In [1]: import shlex
In [2]: command = "cat | tr [:lower:] [:upper:]"
In [3]: splitter = shlex.shlex(command)
In [4]: splitter.whitespace = '|'
In [5]: splitter.whitespace_split = True
In [6]: splitter.get_token()
Out[6]: 'cat '
In [7]: splitter.get_token()
Out[7]: ' tr [:lower:] [:upper:]'
def run(command, data=None, timeout=None, kill_timeout=None, env=None, cwd=None):
"""Executes a given command and returns Response.
Blocks until process is complete, or timeout is reached.
"""
command = expand_args(command)
history = []
for c in command:
if len(history):
# due to broken pipe problems pass only first 10 KiB
data = history[-1].std_out[0:10*1024]
cmd = Command(c)
try:
out, err = cmd.run(data, timeout, kill_timeout, env, cwd)
status_code = cmd.returncode
except OSError as e:
out, err = '', u"\n".join([e.strerror, traceback.format_exc()])
status_code = 127
r = Response(process=cmd)
r.command = c
r.std_out = out
r.std_err = err
r.status_code = status_code
history.append(r)
r = history.pop()
r.history = history
return r
history = [x, y, z]
r = history.pop() # r = z
r.history = history # z.history = [x, y]
很明显了。当然这个x, y, z代表的是一个个的命令行运行结果的Response对象的实例化结果。
最后还有一个connect函数没看
def connect(command, data=None, env=None, cwd=None):
"""Spawns a new process from the given command."""
# TODO: support piped commands
command_str = expand_args(command).pop()
environ = dict(os.environ)
environ.update(env or {})
process = subprocess.Popen(command_str,
universal_newlines=True,
shell=False,
env=environ,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=0,
cwd=cwd,
)
return ConnectedCommand(process=process)
~/source/backend/python/envoy master*
❯ cat test.py
#!/usr/bin/env python
# encoding: utf-8
import unittest
import envoy
import time
class SimpleTest(unittest.TestCase):
def testInput(self):
r = envoy.run("sed s/i/I/g", "Hi")
self.assertEqual(r.std_out, "HI")
self.assertEqual(r.status_code, 0)
def testPipe(self):
r = envoy.run("echo -n 'hi'| tr [:lower:] [:upper:]")
self.assertEqual(r.std_out, "HI")
self.assertEqual(r.status_code, 0)
def testTimeout(self):
r = envoy.run("yes | head", timeout=1)
self.assertEqual(r.std_out, "y\ny\ny\ny\ny\ny\ny\ny\ny\ny\n")
self.assertEqual(r.status_code, 0)
if __name__ == '__main__':
unittest.main()
有三个测试用例testInput、testPipe、testTimeout，现在来执行代码。
~/source/backend/python/envoy master*
❯ python2 test.py
...
----------------------------------------------------------------------
Ran 3 tests in 2.880s
OK
这三个测试通过了，当然用的是python2.7，如果用python3会报unicode不存在
~/source/backend/python/envoy master*
❯ python3 test.py
EEE
======================================================================
ERROR: testInput (__main__.SimpleTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 11, in testInput
r = envoy.run("sed s/i/I/g", "Hi")
File "/home/mao/source/backend/python/envoy/envoy/core.py", line 205, in run
command = expand_args(command)
File "/home/mao/source/backend/python/envoy/envoy/core.py", line 181, in expand_args
if isinstance(command, (str, unicode)):
NameError: name 'unicode' is not defined
======================================================================
ERROR: testPipe (__main__.SimpleTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 16, in testPipe
r = envoy.run("echo -n 'hi'| tr [:lower:] [:upper:]")
File "/home/mao/source/backend/python/envoy/envoy/core.py", line 205, in run
command = expand_args(command)
File "/home/mao/source/backend/python/envoy/envoy/core.py", line 181, in expand_args
if isinstance(command, (str, unicode)):
NameError: name 'unicode' is not defined
======================================================================
ERROR: testTimeout (__main__.SimpleTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 21, in testTimeout
r = envoy.run("yes | head", timeout=1)
File "/home/mao/source/backend/python/envoy/envoy/core.py", line 205, in run
command = expand_args(command)
File "/home/mao/source/backend/python/envoy/envoy/core.py", line 181, in expand_args
if isinstance(command, (str, unicode)):
NameError: name 'unicode' is not defined
----------------------------------------------------------------------
Ran 3 tests in 0.001s
FAILED (errors=3)
~/source/backend/python/envoy master*
❯ cat test.py
#!/usr/bin/env python
# encoding: utf-8
import unittest
import envoy
import time
class SimpleTest(unittest.TestCase):
# THIS TEST FAILS BECAUSE expand_args DOESN'T HANDLE QUOTES PROPERLY
def test_quoted_args(self):
sentinel = 'quoted_args' * 3
r = envoy.run("python -c 'print \"%s\"'" % sentinel)
self.assertEqual(r.std_out.rstrip(), sentinel)
self.assertEqual(r.status_code, 0)
def test_non_existing_command(self):
r = envoy.run("blah")
self.assertEqual(r.status_code, 127)
if __name__ == '__main__':
unittest.main()
执行测试，查看测试结果
~/source/backend/python/envoy master*
❯ python2 test.py
..
----------------------------------------------------------------------
Ran 2 tests in 0.012s
OK
# 开头导入logging
import logging
logging.basicConfig(level=logging.DEBUG)
# connect函数调试
logging.debug("command: %s" % command_str)
# ConnectedCommand函数调试
@property
def status_code(self):
"""The status code of the process.
If the code is None, assume that it's still running.
"""
logging.debug("status_code: %s" % self._status_code) # 新加调试部分
return self._status_code
def expect(self, bytes, stream=None):
"""Block until given bytes appear in the stream."""
logging.debug("stream: %s" % stream) # 新加调试部分
logging.debug("stream: %s" % self.std_out) # 新加调试部分
if stream is None:
stream = self.std_out
def send(self, str, end='\n'):
"""Sends a line to std_in."""
logging.debug("str: %s" % (str+end)) # 新加调试部分
return self._process.stdin.write(str+end)
def block(self):
"""Blocks until command finishes. Returns Response instance."""
logging.debug("_status_code: %s" % self._process.wait()) # 新加调试部分
self._status_code = self._process.wait()
为了看得更清楚，四个小测试单个进行测试
第一个测试
# 测试用例
def test_status_code_none(self):
c = envoy.connect("sleep 5")
self.assertEqual(c.status_code, None)
# 测试结果
# fxm @ bogon in ~/source/backend/python/envoy_source on git:master x [10:03:14] C:1
$ python2 test.py
DEBUG:root:command: ['sleep', '5']
DEBUG:root:status_code: None
.
----------------------------------------------------------------------
Ran 1 test in 0.036s
OK
测了一个休眠5秒钟的，什么也没执行，返回None，测试通过。
第二个测试
# 测试用例
def test_status_code_success(self):
c = envoy.connect("sleep 1")
time.sleep(2)
self.assertEqual(c.status_code, 0)
# 测试结果
# fxm @ bogon in ~/source/backend/python/envoy_source on git:master x [10:12:24]
$ python2 test.py
DEBUG:root:command: ['sleep', '1']
DEBUG:root:status_code: None
F
======================================================================
FAIL: test_status_code_success (__main__.ConnectedCommandTests)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 17, in test_status_code_success
self.assertEqual(c.status_code, 0)
AssertionError: None != 0
----------------------------------------------------------------------
Ran 1 test in 2.042s
FAILED (failures=1)
# 测试用例
def test_status_code_failure(self):
c = envoy.connect("sleeep 1")
self.assertEqual(c.status_code, 127)
# 测试结果
# fxm @ bogon in ~/source/backend/python/envoy_source on git:master x [10:29:54] C:1
$ python2.7 test.py
DEBUG:root:command: ['sleeep', '1']
E
======================================================================
ERROR: test_status_code_failure (__main__.ConnectedCommandTests)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 20, in test_status_code_failure
c = envoy.connect("sleeep 1")
File "/Users/fxm/source/backend/python/envoy_source/envoy/core.py", line 261, in connect
cwd=cwd,
File "/usr/local/Cellar/python/2.7.13/Frameworks/Python.framework/Versions/2.7/lib/python2.7/subprocess.py", line 390, in __init__
errread, errwrite)
File "/usr/local/Cellar/python/2.7.13/Frameworks/Python.framework/Versions/2.7/lib/python2.7/subprocess.py", line 1024, in _execute_child
raise child_exception
OSError: [Errno 2] No such file or directory
----------------------------------------------------------------------
Ran 1 test in 0.030s
FAILED (errors=1)
就先不深究了，关键还是学习他的代码怎么写
第四个测试
# 测试用例
def test_input(self):
test_string = 'asdfQWER'
r = envoy.connect("cat | tr [:lower:] [:upper:]")
r.send(test_string)
self.assertEqual(r.std_out, test_string.upper())
self.assertEqual(r.status_code, 0)
# 测试结果
# fxm @ bogon in ~/source/backend/python/envoy_source on git:master x [10:33:24] C:1
$ python2.7 test.py
DEBUG:root:command: ['tr', '[:lower:]', '[:upper:]']
DEBUG:root:str: asdfQWER
F
======================================================================
FAIL: test_input (__main__.ConnectedCommandTests)
----------------------------------------------------------------------
Traceback (most recent call last):
File "test.py", line 27, in test_input
self.assertEqual(r.std_out, test_string.upper())
AssertionError: None != 'ASDFQWER'
----------------------------------------------------------------------
Ran 1 test in 0.038s
FAILED (failures=1)
and others
最近深感时间不够用，想做的事情太多，想看的源码太多，而看源码并记录到博客里又是一件很耗费时间的事情。
所以我现在很能理解那些辛辛苦苦码字后被人随意转载且不留作者名字之后作者们愤愤不平想打官司的心情了。这让我想起了一个在微博看到的调侃这方面的一个笑话
笑话是不是这么讲的我已经忘了，照着自己的想法翻译了一遍。扯得有些远了，今天要写的是学习python大神Kenneth Reitz写的envoy库。
我照着一个个commit敲的代码放在envoy_source上。
代码结构如下
核心代码就一个core.py，直接研究core.py里的代码，看看它有多少行代码
也就256行代码，不多。
核心代码
其实还好，算是一个非常小的库了，但这个库在github上的star有3023多个，可见这个库还是很有用的，那么这个库干什么用呢，看看github上的介绍。
Envoy: Python Subprocesses for Humans.
Note: Delegator is a replacement for Envoy.
This is a convenience wrapper around the
subprocessmodule.You don't need this.
But you want it.
Usage
Run a command, get the response:
Pipe stuff around too:
大概的意思就是在python里运行linux命令，可以查看命令运行的状态status_code、输出std_out、错误std_err、命令command和历史history等等。
开头导入后续会用到的库，还有当前的版本、开源许可证和库版本。
然后定义了一个终止进程的函数_terminate_process
这个终止进程不同系统平台的实现方式不一样，从代码可以看出，至少windows 32位的实现方式和别的不一样，所以需要单拎出来，先看普遍的实现方法
os.kill(pid, sig)吧，它有两个参数，pid指的是进程的pid，sig值signal模块的信号码，这里终止进程的信号码是signal.SIGTERM。对于windows 32位的实现方法我查了一下文档都没有讲这么详细的函数，倒是在stackoverflow的时候查到了一个相关的In Python 2.5, how do I kill a subprocess?。就是这里的实现方式，然后在python文档里说到，python3.2以后貌似os.kill已经支持了所有的windows 32位系统。
接下来看终止进程函数
_kill_process这里也是判断系统，如果是windows 32位系统，则调用_terminate_process函数终止，否则，调用标准的os.kill函数。这里可能有人会说在_kill_process函数里已经判断了系统版本了，那在_terminate_process还要判断一次，那不是多此一举吗？其实我觉得不能这么看，这个现在是先调用_kill_process，这在_terminate_process不判断没有问题，但是如果是直接调用_terminate_process的话就有问题了。所以这个也是为了独立函数的完备性吧。
然后是
_is_alive函数这个
_is_alive函数存在的意义是由于不同版本的python实现该方法的名称不同，所以要保证兼容性要做完备判断，如果thread函数is_alive属性，说明该版本对该方法的实现就是thread.is_alive()，所以返回它就是了，否则，统统返回thread.isAlive()。<br><br> 然后是Command`类了首先就是
__init__函数了，这个就是实例化Command类的时候，就会自动调用的，这里就是初始化一些变量。主要就是把参数cmd存到self.cmd中，其他的就不多说了。接下来是一个类的run方法。先把data变量存到self.data中，然后dict(os.environ)把系统的环境变量转成字典形式存到environ中，再调用environ.update(env or {})把env变量更新environ变量，更新的意思是environ没有的项而env有的，就增量加到environ中，如果env和environ中有相同的键，但值不一样，那么就用env的值替换environ对应项的值。update函数里还有个or表达式，意思是如果env是None的话，就用空字典取更新environ，其实就是跟没更新一样。可以看看我电脑里的环境变量当然你自己的环境变量基本上不可能和我的完全一样。
好，现在就轮到
run方法里的target函数了，这里主要就是调用一个subprocess.Popen方法，并将结果存到self.process中。这个函数有很多很多参数，文档里非常详细，但这里就不一一重复了，只需要知道这个方法是创建进程并管理的方法就好。因为如果要研究的话应该在研究python源代码的时候去研究它。接下来进行一个if判断，这个sys.version_info是关于当前python版本的相关信息。它会返回含有5个元素的元组major, minor, micro, releaselevel, and serial。其中除了releaselevel以外都是整型。releaselevel的值有'alpha', 'beta', 'candidate'或者'final'，具体意思的很明显。现在来看这个python3.5.2版本和python2.7.12版本的sys.version_info有什么不同。对比一看就知道major就是主版本号，minor姑且叫副版本号，micro就是小版本的意思了。releaselevel就是版本是最终版还是测试版等，serial这个字面意思是串行的，序列接口的，这个我看不太明白它的意思。但不影响这个条件的判断。
回到条件判断应该就是python主版本号在大于等于3以后的和3之前的处理字符转换这些方面的方法已经不相同了，这又是一个很大的课题，这个据说很多python程序员在这一块上耗费的时间特别长，貌似特别繁琐，据说在3以后已经好很多了。当版本号大于等于3的时候self.process.communicate要用bytes转换成utf-8的编码方式作为参数。而其他的版本则直接以self.data作为参数。
target()函数可能会在运行过程中有异常，如果有，把异常存到self.exc中。然后是一个threading模块的Thread方法，这个方法的
__doc__如下：这是一个实现线程管理的类。这是一个可以在一定的限制条件下被安全继承的类。激活线程活动有两种方法，一种是传一个可调用对象给该类的构造函数，另一个是在子线程中通过
run()方法重写。官网介绍的
threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)方法有六个个参数：run()方法调用的可调用对象，默认为None，表明没什么可以调用的。thread = threading.Thread(target=target)实例化一个threading.Thread()存到thread中，target就是在这之前定义的可调用的target函数，然后调用thread的start方法。start就是启动线程的方法。每个线程最多只能被启动一次，它让对象的run()方法在单独的控制线程中被调用。如果一个线程被调用多次，将会抛出值为RuntimeError的异常。来看看
thread.join(timeout)的_.doc_join方法是线程等待超时时间，也就是说如果在timeout时间内，线程还未结束，则不再阻塞进程。由于join方法的返回值一直是None，所以在调用完join方法后必须得调用isAlive()/is_alive()方法来检测是不是过了超时时间，如果线程timeout未指定或者为None，进程将一直阻塞直到线程运行完成；一个线程可以调用join多次，如果试图加入当前线程会引起死锁的话join会抛出一个值为RuntimeError的异常。如果join方法在线程启动之前调用，也会抛出一样的异常。有了这些基础知识之后，再回头来理解代码，先创建一个以
threading.Thread(target=target)函数为target参数的线程，然后thread.start()启动线程，加上等待超时时间thread.join(timeout)，防止线程一直阻塞进程。如果运行线程过程中有异常，则抛出一个self.exc的异常。通过_is_alive(thread)来检测是否过了超时时间，如果是，就终止self.process，然后再加上一个等待超时时间thread.join(kill_timeout)，这里kill_timeout的意思是，如果在kill_timeout时间内还没运行结束，则完全终止self.process。join完后再进行_is_alive(thread)的判断，如果超时，调用_kill_process(self.process)完全终止。然后再thread.join()（这块看不太懂，为啥还要join一次）。self.returncode = self.process.returncode要把线程运行的returncode返回存到self.returncode中，run方法最终返回self.out和self.err。接下来是一个
ConnectedCommand类同样还是先
__init__初始化一些变量赋值。__enter__返回实例本身。__exit__杀掉进程方法。这里用到了类本身的kill方法。然后是一个返回状态码的方法
这方法就是直接返回
self._status_code，这没什么的，如果这个值是None，说明线程还在运行。主要是来看看这个@property到底是做什么的。看文档说property(fget=None, fset=None, fdel=None, doc=None)是一个类，作用是返回属性。fget是用于返回属性值；fset用于设置属性值；fdel用于删除属性值；doc是属性值的docstring。官网的例子如果
c = C();，那么c.x会调用getx函数；如果c.x = value，那么会调用setx函数赋值；如果del c.x会调用delx删除。doc参数现在是"I'm the 'x' property."，那么属性的docstring就是"I'm the 'x' property."，如果doc未定义，那么属性的docstring默认是fget方法的docstring。上面是
Property类所做的事情，如果在代码中用@property又会怎么样呢。再来看一个例子加了
@property（中文应该叫装饰器）以后，相当于把voltage函数变成了getter，读取属性值。最典型的一个例子如下这个例子和前面的
Class C例子是完全一样的，但这个写比较pythonic。这里有关于装饰器的更详细的资料。回到之前的代码就很容易知道了，两个
@property一个是为了获取状态码_status_code，一个是为了获取进程的pid码_process.pid。往下看后三个函数
根据
expect函数的docstring说法是阻塞直到给定字节出现在流中，这么看不太懂，等看到哪些地方调用再回头看吧。函数里判断stream，如果是None，把self.std_out赋值给stream。send函数作用是发送一行信息到std_in，参数是要发送的信息str和end=\n换行符。这个方法的实现是直接返回self._process.stdin.write(str+end)，str+end保证了发生一行信息。block函数阻塞知道命令行运行结束，然后返回Response实例。直接调用进程的wait()方法返回状态码。往下就是
Response类了这个类就是运行命令行的应答。
__init__先初始化各种参数，主要传入的是process，把它存到_process中。然后定义__repr__，如果command不为None，则输出第一个command，比如返回Response ls这样的。否则只返回Response。往下是
expand_args类这个
expand_args就是负责解析命令行字符，返回Popen-ready列表。isinstance函数就是判断前者是不是后者的实例，或者是后者的直接或者间接的子类型。具体到当前的函数就是判断command是不是str或者unicode的子类型。官方文档shlex.shlex(instream=None, infile=None, posix=False, punctuation_chars=False)有四个参数，instream指从哪里读取参数，类型必须是a file-/stream-like object with read() and readline() methods(感觉真的是英文说的很直白，翻译成中文倒是很别扭，所以还是放英文吧)这样的文档流或者是字符串。如果该参数为空，默认会从sys.stdin获取；第二个参数infile是一个文件名字符串，这个在调试的时候方便区分。如果第一个参数instream为空，infile默认值就是"stdin"，posix标识操作系统类型，默认指定为False，表明是兼容模式；如果该值为POSIX，则表明是类unix操作系统模式，将按照类unix的解析规则对instream进行解析。punctuation_chars参数的解释The punctuation_chars argument provides a way to make the behaviour even closer to how real shells parse，大约说的是这个参数提供了一个使解析的行为方式更靠近真实的shells parse的方法。它的默认值是False，操作的规则就是在python3.5以前的规则，因为这个参数是在python3.6加的，只有在3.6以后才有意义，如果该值位True，那么在解析中关于();<>|&的解析会有不同，那是怎么样一种规则呢，来看例子对比很明显，看起来
punctuation_chars=True更符合我们所需要的解析效果。除此之外，这个参数还可以取字符串作为参数，主要就是特指
();<>|&中的一部分来解析。如仅仅指定"|"和仅仅指定"&"的例子如下对比明显，一眼就能明白，所以有些时候文档看得不太明白，但是看了例子就很容易明白。
另外一点要注意的是由于
punctuation_chars最好的情况是结合posix=True来使用。然后是
shlex.whitespace，官方文档的解释是这里说的意思是把
whitespace指定的当做空格处理，默认的有空格、tab、换行符和回车符。回到代码里splitter.whitespace = '|'就是把'|'当做空格一样来处理的意思，等会讲完whitespace_split会给例子。然后就是
splitter.whitespace_split = True这行了。官方文档关于shlex.whitespace_split的解释是取值为True的时候，仅仅根据
whitespaces来切分，也就是前面shlex.whitespace指定的值，如果没有指定，那就是默认空格，tab等等，而且此时的punctuation_chars将不起作用。上个例子看看没加
whitespace和whitespace_split的时候，是以空格为默认切分的，加了whitespace '|'和whitespace_split = True之后就是以'|'为切分依据的。看了例子就很明显了。例子里已经顺便演示了一下shlex.push_token()的作用，再回到代码中的话，就是做一个while循环，拿到切分的结果存到token中，如果token存在，就把它附加在command列表中。否则结束while循环。然后是
command = list(map(shlex.split, command))，这里是一个map的方法，它的作用就是第一个参数的方法shlex.split分别运用于第二个参数的command列表的每个元素，然后将结果返回一个新的列表。这里返回的结果再用list()方法进行操作后变成了列表的列表覆盖掉原先的command列表。最后的最后return command把command列表返回，作为expand_args函数的返回结果。然后是
run函数run函数执行提供的命令行，阻塞直到进程结束或者是过了超时时间，最后返回Response类。首先调用expand_args(command)把待执行的命令行解析成想要的列表存到command中；接着创建一个history空列表；接着循环command列表，如果history列表长度大于0，把history列表最后一个元素的std_out属性值的[0:10*1024]（指的是命令行管道一次最多只能有10kb的限额），也就是std_out小于等于10kb，将完全存到data中，否则截取前10kb的部分存到data中；然后，以循环项c为参数实例化Command类存到cmd中；接着cmd.run(data, timeout, kill_timeout, env, cwd)以data、timeout、kill_timeout、env和cwd为参数执行Command类的run方法，把结果存到out和err中。status_code = cmd.returncode同样把cmd的returncode存到status_code中。如果执行run方法的过程中有异常，out就是空字符串，err就是用换行符join异常的strerror和traceback.format_exc()获取的堆栈异常信息；接着以cmd为参数实例化一个Response类存到r中，并把cmd.run的运行结果存到r.std_out、r.std_err等中；然后把r附加到history列表中。这个其实很好理解，history就是历史命令行列表，for循环的时候把所有的命令的运行结果r都附加到列表中；for循环结束后r = history.pop()拿出history列表中的最后一项作为当前的项，r.history = history把拿掉最后一项的列表作为最后一项的history属性值。最后把r返回作为run函数的结果。这个很好理解了吧，举个例子很明显了。当然这个x, y, z代表的是一个个的命令行运行结果的
Response对象的实例化结果。最后还有一个
connect函数没看connect函数从给定的命令产生一个新的进程，首先调用command_str = expand_args(command).pop()解析命令command生成的列表取最后一项存到command_str中；然后拿到系统的环境后在根据传入的环境env取更新系统环境；这之前有类似的了。然后把command_str作为subprocess.Popen的参数产生一个进程存到process中，最后调用ConnectedCommand(process=process)生成ConnectedCommand类的一个实例并返回它。到这里为止，core.py算是讲完了；在envoy子文件夹里还有一个和core.py同级的
__.init__.py这个文件做的是一些初始化的工作，比如这里导出了core.py文件里的多个方法以及库的版本号
__version__，一般而言这个文件也可以什么内容都不写。测试
代码写完了，肯定得要测试一翻才能知道它是否能够工作。
先上三个简单测试吧，在
envoy项目根目录下建一个test.py，代码如下有三个测试用例
testInput、testPipe、testTimeout，现在来执行代码。这三个测试通过了，当然用的是
python2.7，如果用python3会报unicode不存在现在这个代码是不兼容python3的，因为作者已经不维护这个代码了，库的介绍里也给了一个新的链接，推荐使用delegator这个库了，当然了学习的意义依然存在。
下一个测试
执行测试，查看测试结果
测试通过，
test_non_existing_command这个测试测的就是一个不存在的命令行blah，如果成功，返回的状态码status_code应该是0，现在返回的状态码等于127，说明这个blah命令行没有正确执行。再来一个测试
执行测试看结果
四个小测试，有一个通过，三个没通过，那还是得具体看看调试信息。在开头导入
logging模块，并配置它的level为DEBUG在，然后在connect函数和ConnectedCommand分别加上如下的调试信息为了看得更清楚，四个小测试单个进行测试
第一个测试
测了一个休眠5秒钟的，什么也没执行，返回
None，测试通过。第二个测试
测试的是休眠1秒，然后
time.sleep(2)休眠两秒，返回的status_code是None而不是0。好吧，结果就是这样，我的测试用例都是从作者的库里直接拿过来的，可能当时作者测试是通过的。可能后来python2的shlex库做了修改了，要让测试通过的话，把status_code改成None就行了。。。现在直接从github拉取的作者代码库做测试也是一样的结果，作者已经有一年半时间不维护了。第三个测试
这个一样，也是返回
None，预设是127当然通不过。这个测试用例我改了，作者的原测试是c = envoy.connect("sleeep 1")，sleep里故意多了一个e，但是测试时报的是这个错就先不深究了，关键还是学习他的代码怎么写
第四个测试
envoy.connect("cat | tr [:lower:] [:upper:]")这个命令解析后拿的是后半部分，在linux命令行里tr [:lower:] [:upper:]作用是把输入流的字符串小写改成大写。r.send(test_string)里做的就是把test_string = 'asdfQWER'作为输入流的字符串，目的是把'asdfQWER'全部变成大写'ASDFQWER'，然而测试的结果并没有，结果的r.std_out是None。总结
这个库到此就算是学习结束了，有几点需要总结一下
The text was updated successfully, but these errors were encountered: