我想在python3中使用类似expect的模块.据我所知,无论是pexpect还是fabric都没有使用python3.我可以使用任何类似的包吗？(如果不是,有人知道py3支持是否在任何项目的路线图上？)
不需要完全重叠的功能集.我不认为我的使用情况下,有必要在这里,但我基本上重新实现Linux的期待脚本,做一些配置提供的命令远程登录,但扩展功能.
我试图在从未知状态恢复后清除该行上的任何字符,因为在某些情况下,它们包含我在将来期望方法调用中使用的提示和其他关键字.我已经尝试了多种混合结果的方法,因为我一直在遇到非预期的行为.
看到意外的行为(使用pexpect V3.3和Python 2.7.9):
执行以下代码后,当我随后尝试从缓冲区读取时,偶尔会看到不稳定的行为,其中只有部分累积的字符被清除.这对我的下游逻辑造成了严重破坏.我假设这是因为整个线程被置于休眠状态5秒,因此当它被唤醒时,它没有时间在执行read_nonblocking()命令之前获取完整的传入缓冲区.
time.sleep(5)
flushedStuff += child.read_nonblocking(100000, timeout=0)
当我尝试使用.expect调用以非阻塞方式刷新行时,我发现在TIMEOUT异常之后,传入缓冲区未被清除.它的内容可以在child.before属性中按预期找到,但在下一个.expect方法调用期间也会被解析.所以这根本不冲洗线!我还注意到,read_nonblocking()不从本地缓冲区读取,而是直接通过操作系统从该行读取,因此它没有看到这一点.
try:
child.expect("ZzqQJjSh_Impossible_String", timeout = 5)
except pexpect.TIMEOUT:
pass
flushedStuff = child.before
所以在这之后,我提供一种可靠的方法来刷新行的当前解决方案是扩展spawn类并添加一个执行以下操作的方法...它访问一个未记录的属性:
class ExtendedSpawn(pexpect.spawn):
def flushBuffer(delay)
try:
# Greedily read in all the incoming characters
self.expect("ZzqQJjSh_Impossible_String", timeout = delay)
except pexpect.TIMEOUT:
pass
# Clear local input buffer inside the spawn Class
self.buffer = self.string_type()
return self.before
上述方法也可用于非阻塞睡眠命令.
对于一些应该简单的方法来说,这似乎过于复杂,更不用说我浪费了几天.有没有更好的方法呢？
谢谢!
我正在尝试编写一个运行特定命令的跨平台工具,需要某些输出进行验证,并发送某些输出(如用户名/密码)进行身份验证.
在Unix上,我成功地编写了一个使用pexpect库(via pip install pexpect)的Python工具.这段代码完美无缺,正是我想要做的.我在下面提供了一小段关于概念验证的代码摘录:
self.process = pexpect.spawn('/usr/bin/ctf', env={'HOME':expanduser('~')}, timeout=5)
self.process.expect(self.PROMPT)
self.process.sendline('connect to %s' % server)
sw = self.process.expect(['ERROR', 'Username:', 'Connected to (.*) as (.*)'])
if sw == 0:
pass
elif sw == 1:
asked_for_pw = self.process.expect([pexpect.TIMEOUT, 'Password:'])
if not asked_for_pw:
self.process.sendline(user)
self.process.expect('Password:')
self.process.sendline(passwd)
success = self.process.expect(['Password:', self.PROMPT])
if not success:
self.process.close()
raise CTFError('Invalid password')
elif sw == 2:
self.server = self.process.match.groups()[0]
self.user = self.process.match.groups()[1].strip()
else:
info('Could not match any strings, trying to get server …我正在尝试制作一个以可执行文件名作为参数的程序，运行可执行文件并报告该运行的输入和输出。例如，考虑一个名为“circle”的子程序。我的程序需要运行以下内容：
$ python3 capture_io.py ./circle
输入圆的半径：10
地区：314.158997
[('output', '输入圆的半径：'), ('input', '10\n'), ('output', 'Area: 314.158997\n')]
我决定使用pexpect模块来完成这项工作。它有一个方法interact可以让用户与子程序进行交互，如上所示。它还需要 2 个可选参数：output_filter和input_filter. 从文档：
该
output_filter会通过了所有从子进程的输出。该input_filter会从用户通过了所有的键盘输入。
所以这是我写的代码：
import sys
import pexpect
_stdios = []
def read(data):
_stdios.append(("output", data.decode("utf8")))
return data
def write(data):
_stdios.append(("input", data.decode("utf8")))
return data
def capture_io(argv):
_stdios.clear()
child = pexpect.spawn(argv)
child.interact(input_filter=write, output_filter=read)
child.wait()
return _stdios
if __name__ == '__main__':
stdios_of_child = capture_io(sys.argv[1:])
print(stdios_of_child)
$ python3 capture_io.py ./circle Enter radius of circle: …
我用google搜索"python ssh".有一个很棒的模块pexpect,它可以使用ssh(带密码)访问远程计算机.
连接远程计算机后,我可以执行其他命令.但是我无法再次在python中获得结果.
p = pexpect.spawn("ssh user@remote_computer")
print "connecting..."
p.waitnoecho()
p.sendline(my_password)
print "connected"
p.sendline("ps -ef")
p.expect(pexpect.EOF) # this will take very long time
print p.before
如何得到ps -ef我的结果？
语境:
我有一些使用的代码pexpect,其工作是给出一个命令的"实时"输出.即,当命令生成一些输出时或不久之后打印出一些内容,而不是等到命令完成然后与其输出交互.
我正在做的就是启动和停止服务.我spawn通过该过程执行此操作,然后在打印时输出每一行,如下所示:
def watch(process):
output = ""
while True:
try:
line = process.read_nonblocking(timeout = -1)
print(line, end ="")
output += line
except pexpect.EOF:
break
del process
return output
while True:
print("IN 1")
process = pexpect.spawn("service",["zend-server", "stop"], timeout = None)
watch(process)
print("OUT 1")
print("IN 2")
process = pexpect.spawn("service",["zend-server", "start"], timeout = None)
watch(process)
print("OUT 2")
这段代码应该只循环服务:启动它并反复停止它,打印开始/停止的输出.它打印输出正常.但是,它最终会在"OUT 2"之前挂起.我可以查看输出,并看到service调用停止执行.但是,该watch功能从不会提高EOF并退出.
每项服务都不会发生这种情况.有些服务无限循环.zend-server但是,与其他一些不相关的命令一起,以相同的方式间歇性地失败.
通过"最终挂起",我的意思是它启动/停止服务一些(每次运行变量)次,并挂起.它通常在4-6之后长大,虽然从来没有在第一次通话 - 总是至少在第二次(因此del声明;我认为我会安全地玩).
Python 2.6.6,CentOS(64)6.3,Pexpect 2.3-6,FWIW
题:
为什么pexpect挂在某些命令上？我该如何解决这个问题？使用超时不是一个可行的解决方案,因为其中一些命令实际上可以运行任意长时间. …
我的问题很简单:我expect()可以使用pexpect在stderr上查看某些输出吗？它似乎pexpect.spawn()只能用于期望stdout上的输出.
乌托邦的例子:
import pexpect
child = pexpect.spawn(...)
child.expect("hi", fd=pexpect.STDERR)
或者在散文中,"期待stderr上的字符串'hi'".
我没有在文档中找到任何提及这样的工具,但我注意到该child实例有一个stderr属性......
一个半实现我想要的黑客是将stderr重定向到spawn参数中的stdout,然后我们可以使用常规expect().肯定有更好的办法？
干杯
这是我试图运行的代码:
from pexpect import pxssh
s = pxssh.pxssh()
if not s.login ('myip', 'myusername', 'mypassword'):
print ("SSH session failed on login.")
print (str(s))
else:
print ("SSH session login successful")
s.sendline ('ls -l')
s.prompt() # match the prompt
print (s.before) # print everything before the prompt.
s.logout()
我在运行它时遇到的错误是:
Traceback (most recent call last):
File "test_pexpect.py", line 1, in <module>
from pexpect import pxssh
File "C:\Python35\lib\site-packages\pexpect\pxssh.py", line 23, in <module>
from pexpect import ExceptionPexpect, TIMEOUT, EOF, spawn
ImportError: cannot import name 'spawn'
谁能帮我吗？我在Windows上使用python3.5
我成功打开了与 paramiko 的 ssh 连接。
由于我想使用 pexpect 与远程系统自动交互，我想将 paramiko 连接传递给pexpect.fdpexpect.fdspawn，但两者不适合在一起。
Paramiko 为我提供了一个文件描述符，但文档明确指出它不能用于读取或写入。我需要的是一个用于读写的双向文件描述符，但我很难弄清楚如何连接这些点。
我明白为什么这段代码不起作用，但我不知道如何创建有效的代码。
#!/usr/bin/env python3
import pexpect
import pexpect.fdpexpect
import paramiko
sshc = paramiko.client.SSHClient()
sshc.set_missing_host_key_policy(paramiko.AutoAddPolicy())
sshc.connect('192.0.2.1', username='redacted', password='redacted',
look_for_keys=False, allow_agent=False)
io = pexpect.fdpexpect.fdspawn(sshc.invoke_shell().fileno())
io.sendline('')
当我们docker-compose up-d使用docker-compose.yml文件运行命令来运行 docker 时，它开始构建镜像或从注册表中拉取镜像。我们可以在终端上看到这个命令的每一步。
我正在尝试从 python 脚本运行此命令。命令成功启动，但在命令之后，我不知道该过程完成了多少。有什么方法可以监视docker-compose up -d命令的状态，以便脚本可以让用户（正在使用脚本的人）知道进程完成了多少，或者docker-compose命令是否由于某些原因而失败。？
谢谢
代码：
from pexpect import pxssh
session = pxssh.pxssh()
if not session.login(ip_address,<USERNAME>, <PASSWORD>):
print("SSH session failed on login")
print(str(session))
else:
print("SSH session login successfull")
session.sendline("sudo docker-compose up -d")
session.prompt()
resp = session.before
print(resp)