import paramiko
import os
import logging
import json
import unittest
from stat import S_ISDIR,S_ISREG
logging.basicConfig(level = logging.ERROR,format = ‘%(asctime)s - %(levelname)s -->%(funcName)s at line %(lineno)d: \n %(message)s‘)
log= logging.getLogger()
"""
version 0.1:
fit initial client can not be shut down ,submit a func close client auto after
called paramiko func,support remote upload and download func extend WebankSftp class
"""
# paramiko client
def WebankClients(ip,port,user,pwd):
return WebankSftp(ip,port,user,pwd)
class WebankSftp(object):
"""
how to use it fast:
>>ip = ‘192.168.110.137‘
>>port = 22
>>pwd = ‘admin‘
>>user = ‘root‘
>>client=WebankClients(ip,port,user,pwd)
>>client.query_dir("/usr/local/")
2019-10-27 12:44:41,191 - INFO -->query_dir at line 58:
[‘python36‘, ‘Python-3.6.6‘, ‘Python-3.6.6.tgz‘, ‘test6‘]
>> client.query_dir("/usr/local/test677")
2019-10-27 12:56:22,275 - INFO -->_log at line 1819:
Authentication (password) successful!
Traceback (most recent call last):
File "D:/workspace/ReportProject/wbsftp/minisftp.py", line 223, in <module>
client.query_dir("/usr/local/test677")
File "D:/workspace/ReportProject/wbsftp/minisftp.py", line 64, in query_dir
raise FileExistsError(err.decode("utf-8"))
FileExistsError: ls: cannot access /usr/local/test677: No such file or directory
"""
def __init__(self,ip,port,user,pwd):
self.port = port
self.pwd = pwd
self.user = user
self.ip = ip
self.cli=SftpOpen(self.ip, self.port, self.user, self.pwd).sftp
self.para_cli=self.initial()
self.para_sftp=self.para_cli.open_sftp()
def initial(self):
client = paramiko.SSHClient()
try:
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(self.ip, self.port, self.user, self.pwd)
if client:
return client
except Exception as e:
log.error(e)
# paramiko function
def query_dir(self,dir)->list:
# connection=self.initial()
connection = self.para_cli
shell="ls -l %s | awk ‘{print $9}‘" % dir
input, out, err = connection.exec_command(shell)
res, err = out.readlines(), err.read()
if res:
files_list = [i.strip(‘\n‘) for i in res if i.strip(‘\n‘)]
# log.info(files_list)
return files_list
if err:
# log.info(err)
raise FileExistsError(err.decode("utf-8"))
# paramiko function
def callback_stdout(self,shell:str):
""" this is method for execute Shell COMMAND and CallBack Result"""
#connection = self.initial()
connection=self.para_cli
try:
input, out, err = connection.exec_command(shell,get_pty=True)
res, err = out.readlines(), err.read()
if res:
i=‘‘
for j in res:
i =i+j
log.info("call_back res:%s" % i)
return i
if err:
log.info("call_back err: %s" % err.decode("utf-8"))
return err.decode("utf-8")
except Exception as f:
log.error(f)
def touch(self,filepath):
# client = self.initial()
client =self.para_cli
def create_file(filepath):
client.exec_command("touch %s"%filepath)
# create_file(filepath)
# try:
a=self.exists_file(filepath)
if a:
self.removes(filepath)
create_file(filepath)
info={"msg": "File {} found and have remove it ,success create {}".format(filepath,filepath),"cdde":200,"status":"success"}
log.info(json.dumps(info,indent=4,ensure_ascii=False) )
else:
create_file(filepath)
# if self.exists_dir(os.path.dirname(filepath)):
# create_file(filepath)
# info={‘msg‘: ‘{} exists but {} not found,so directly create {} sucess‘.format(os.path.dirname(filepath),filepath,filepath)
# ,‘code‘:200,‘status‘:‘success‘}
# log.info(json.dumps(info,indent=4,ensure_ascii=False) )
# else:
# # create pdir of filename
# self.mkdir(os.path.dirname(filepath))
# create_file(filepath)
# log.info("{‘msg‘: ‘mkdir %s success,touch %s success ,status:200‘}"%(os.path.dirname(filepath),filepath))
# except Exception as e:
# log.error(e)
#sftp func
def mkdir(self,dir):
""" if linux have exists filename the same with dir ,
dir will create failed because it have exists else it will create success """
#client=SftpOpen(self.ip,self.port,self.user,self.pwd).sftp
client=self.cli
# os.path.isdir() just for win not work for linux, will always return False
try:
bool=self.exists_dir(dir)
if bool:
client.rmdir(dir)
client.mkdir(dir)
else:
client.mkdir(dir)
except Exception as e:
#log.error(e)
log.error("{‘msg‘: ‘mkdir %s failed maybe it have exists same name file or dir ,‘code‘:100,‘status‘: ‘failed‘}"%dir)
else:
log.info("{‘msg‘: ‘mkdir %s success‘,‘code‘: 200,‘status‘:‘success‘}"% dir)
#sftp func
def removes(self,filepath):
""" this is a method for delete file"""
#client=SftpOpen(self.ip,self.port,self.user,self.pwd).sftp
client=self.cli
try:
if self.exists_file(filepath):
client.remove(filepath)
else:
pass
except FileNotFoundError as e:
info={‘msg‘: ‘File %s Not Found Error‘%filepath, ‘retcode‘:100,‘status‘: ‘failed‘}
log.error(json.dumps(info,ensure_ascii=False,indent=4),exc_info=False)
else:
log.info("{‘msg‘:‘remove file %s success‘,‘recode‘:200,‘status‘: ‘success‘}"% filepath)
# sftp func
def list_dir(self,dir):
#client = SftpOpen(self.ip, self.port, self.user, self.pwd).sftp
client=self.cli
try:
res= client.listdir(dir)
information=json.dumps({"files": res},indent=4,ensure_ascii=False)
log.info("%s contains files:" %dir+information)
return res
except FileNotFoundError:
log.error("{‘msg‘: ‘%s Not Found Error‘, ‘retcode‘:100,‘status‘: ‘failed‘}"%dir)
# sftp function
def rm(self,absdir):
def isdirs(filepath, sftp):
"""this is a method for recursion and delete files and dir
,os.path.isdir() it doesn‘t work for linux
"""
return S_ISDIR(sftp.stat(filepath).st_mode)
def subsftp(absdir,sftp):
files = sftp.listdir(absdir)
try:
for file in files:
filepath = os.path.join(absdir, file)
fps = filepath.replace("\\", "/")
if isdirs(fps, sftp):
# recur util del all files under file path is dir
self.rm(fps)
else:
sftp.remove(fps)
log.info("{‘msg‘: ‘remove file %s success,‘retcode‘: 200}" %(fps))
# rmdir empty dir
sftp.rmdir(absdir)
except Exception as e:
log.error(e)
else:
log.info("{‘msg‘: ‘recursion delete %s whole dir,‘recode‘:200}" % (absdir))
#sftp = SftpOpen(self.ip, self.port, self.user, self.pwd).sftp
sftp=self.cli
try:
subsftp(absdir, sftp)
except Exception as e:
log.error(e)
# sftp func
def exists_dir(self,dir):
# client=SftpOpen(self.ip, self.port, self.user, self.pwd).sftp
client=self.cli
def isdirs(dir, sftp):
return S_ISDIR(sftp.stat(dir).st_mode)
try:
isdirs(dir,client)
log.info("exists %s " % dir)
return True
except FileNotFoundError as f:
# info={"msg": "Not Found DIR %s"%dir,"retcode":100,"status": "failed"}
# log.error(json.dumps(info,indent=4,ensure_ascii=False))
return False
#sftp func
def exists_file(self,filepath):
client=self.para_sftp
try:
client.stat(filepath)
except Exception as e:
# info = {"msg": "Not Found FileName %s" % filepath, "retcode": 100, "status": "failed"}
# log.error(info)
return False
else:
log.info("exists %s file "%filepath)
return True
def sftp_close(self):
self.cli.close()
self.para_cli.close()
self.para_sftp.close()
class SftpOpen(object):
def __init__(self,ip,port,user,pwd):
self.tp= paramiko.Transport(ip, port)
self.tp.connect(username=user, password=pwd)
self.sftp = paramiko.SFTPClient.from_transport(self.tp)
class TestSftp(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
ip = "192.168.110.151"
port = 22
user = "root"
pwd = "admin"
cls.client = WebankClients(ip, port, user, pwd)
log.info("start selfcheck method of sftp ")
@classmethod
def tearDownClass(cls) -> None:
try:
cls.client.sftp_close()
except Exception as e:
log.error(e)
else:
log.info("close all connections")
log.info("finish all unittest selfcheck method of sftp ")
def test_query_dir(self):
"""测试查询dir下dirs and files"""
files=self.client.query_dir("/usr/local/listdir")
self.assertIn(‘list.txt‘,files)
def test_call_backstdout(self):
shell="ls -l /usr/local"
readlines=self.client.callback_stdout(shell)
self.assertIn("redisdb",readlines)
def test_exsitdir(self):
a=self.client.exists_dir("/usr/local")
assert a==True
def test_exsistfile(self):
b=self.client.exists_file("/usr/local/redisdb/logs/redis.log")
assert b==True
def test_touch(self):
"""create file """
path="/usr/local/toutest.txt"
self.client.touch(path)
a=self.client.exists_file(path)
self.assertEqual(a,True)
self.client.removes(path)
def test_userremove(self):
"""remove file """
path="/usr/local/tou.txt"
self.client.touch(path)
self.client.removes(path)
a=self.client.exists_file(path)
self.assertEqual(a,False)
def test_mkandrm(self):
"""bug 已经存在目录无法直接mkdir"""
self.client.mkdir("/usr/local/test1")
self.client.rm("/usr/local/test1")
if __name__ == ‘__main__‘:
unittest.main()
.......
----------------------------------------------------------------------
Ran 7 tests in 1.030s
OK
Process finished with exit code 0
原文地址：https://www.cnblogs.com/SunshineKimi/p/11823647.html
时间： 2024-08-01 14:40:00