需求:
我们实现了一个telnet客户端的类TelnetClient,调用实例的start()方法启动客户端与服务器交互,交互完毕后需要调用cleanup()方法,关闭已连接的socket,以及将操作历史记录写入文件并关闭。
能否让TelnetClinet的实例支持上下文管理协议,从而替代手工调用cleanup()方法。
思路:
实现上下文管理协议,需要定义实例的__enter__,__exit__方法,它们分别在with开始和结束时被调用
代码:
#!/usr/bin/python
# -*- coding: utf-8 -*-
from telnetlib import Telnet
from sys import stdin ,stdout
from collections import deque
class TelnetClient(object):
def __init__(self,addr,port=23):
self.addr = addr
self.port = port
self.tn = None # 初始化构造的时候,为空
def start(self):
# raise Exception('Text') # 手动产生一个异常,测试在异常的情况下,程序仍然能够正常退出
# 下面开始登录操作
# user,输入用户名
t = self.tn.read_until('login: ')
stdout.write(t)
user = stdin.readline()
#print user
self.tn.write(user)
# print t
# password,输入密码
t = self.tn.read_until('Password: ')
if t.startswith(user[:-1]): t = t[len(user) + 1:] # 实际注释这行也是Ok的
stdout.write(t)
self.tn.write(stdin.readline())
t = self.tn.read_until('$ ')
stdout.write(t) # 在屏幕输出$,可以理解成print
while True:
uinput = stdin.readline() # 获取用户在命令行$后输入的命令,可以理解成input
if not uinput:
break
self.history.append(uinput)
self.tn.write(uinput) # 执行命令
t = self.tn.read_until('$ ')# 读取直到遇到了给定的字符串expected或超时秒数。
# print t t的值为:
# print uinput
stdout.write(t[len(uinput) + 1:])
# stdout.write(t)
def cleanup(self):
pass
def __enter__(self):
self.tn = Telnet(self.addr,self.port) # telnet的连接对象
self.history = deque() # 创建一个队列,来存储telnet操作的历史记录
return self
def __exit__(self,exc_type,exc_val,exc_tb):
# print 'In exit'
self.tn.close()
self.tn = None
with open(self.addr + '_history.txt','w') as f:
f.writelines(self.history) # 写入用户的操作历史记录
# return True # 可以压制with语句中的异常,使在外面捕获不到,即END可以成功打印
# 通过上下文管理器来调用
with TelnetClient('127.0.0.1') as client:
client.start()
#print 'END'
# client = TelnetClient('127.0.0.1')
# print('
start...')
# client.start()
# print('
cleanup')
# client.cleanup()
=========================================================================================
from sys import stdin,stdout
import getpass
import telnetlib
from collections import deque
class TelnetClient:
def __init__(self,host,port=23):
self.host = host
self.port = port
def __enter__(self):
self.tn = telnetlib.Telnet(self.host,self.port)
self.history = deque([])
return self
def __exit__(self,exc_type,exc_value,exc_tb):
print('IN__exit__',exc_type,exc_value,exc_tb)
self.tn.close()
self.tn = None
with open('history.txt' ,'a') as f:
f.writelines(self.history)
return True
def login(self):
# user
self.tn.read_until(b"login: ")
user = input("Enter your remote account: ")
self.tn.write(user.encode('utf8') + b"
")
# password
self.tn.read_until(b"Password: ")
password = getpass.getpass()
self.tn.write(password.encode('utf8') + b"
")
out = self.tn.read_until(b'$ ')
stdout.write(out.decode('utf8'))
def interact(self):
while True:
cmd = stdin.readline()
if not cmd:
break
self.history.append(cmd)
self.tn.write(cmd.encode('utf8'))
out = self.tn.read_until(b'$ ').decode('utf8')
stdout.write(out[len(cmd)+1:])
stdout.flush()
#client = TelnetClient('127.0.0.1')
#client.connect()
#client.login()
#client.interact()
#client.cleanup()
with TelnetClient('127.0.0.1') as client:
raise Exception('TEST')
client.login()
client.interact()
print('END')