paramiko-tools

import paramiko
import os
import logging
import json
import unittest
from stat import S_ISDIR, S_ISREG

__author__ = ‘Chen Quan‘

logging.basicConfig(level=logging.ERROR,
                    format=‘%(asctime)s  - %(levelname)s -->%(funcName)s  at line %(lineno)d: \n %(message)s‘)
log = logging.getLogger()


class Paramiko_Sftp(object):

    def __init__(self, ip, port, user, pwd):
        self.port = port
        self.pwd = pwd
        self.user = user
        self.ip = ip
        self.sftp =self.sftp_client()
        self.ssh = self.ssh_client()

    def ssh_client(self):
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(self.ip, self.port, self.user, self.pwd)
        if client:
            return client


    def sftp_client(self):
        tp = paramiko.Transport(self.ip,self.port)
        tp.connect(username=self.user, password=self.pwd)
        sftp = paramiko.SFTPClient.from_transport(tp)
        return sftp


    def get_dir(self, dir):

        shell = "ls -l  %s | awk ‘{print $9}‘" % dir
        input, out, err = self.ssh.exec_command(shell)
        res, err = out.readlines(), err.read()
        if res:
            files_list = [i.strip(‘\n‘) for i in res if i.strip(‘\n‘)]
            return files_list
        if err:
            raise FileExistsError(err.decode("utf-8"))

    def cmd(self, shell: str):

        try:
            input, out, err =self.ssh.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 close(self):
        self.sftp.close()
        self.ssh.close()



    def  is_file(self,filepath):
        try:
            result = S_ISREG(self.sftp.stat(filepath).st_mode)
        except IOError:  # no such file
            result = False
        return result


    def  is_dir(self,dir):

        try:
            result = S_ISDIR(self.sftp.stat(dir).st_mode)
        except IOError:     # no such file
            result = False
        return result

    def exists(self, remote_path):

        return True if self.sftp.stat(remote_path) else False

    def touch(self,file):
        if self.exists(file):
            raise FileExistsError("%s already exist"%file)
        else:
            self.ssh.exec_command("touch %s"%file)

    def remove(self,file):
        if self.exists(file):
            self.sftp.remove(file)
        else:
            raise FileNotFoundError("NO %s SUCH FILE"%file)

    def rm(self,dir):
        if self.exists(dir):
            files=self.sftp.listdir(dir)
            for file  in files:
              fp=os.path.join(dir,file)
              if self.is_dir(fp):
                  self.rm(fp)
              else:
                  self.remove(fp)
            self.sftp.rmdir(dir)
        else:
            raise FileNotFoundError("no such %s dir "% dir )


    def list_dir(self,dir):
        return self.sftp.listdir(dir)

    def mkdir(self,dir):
        if self.exists(dir):
            self.rm(dir)
            self.mkdir(dir)
        else:
            self.mkdir(dir)


    def makedirs(self,remotedir,mode=777):
        if self.is_dir(remotedir):
            pass

        elif self.is_file(remotedir):
            raise OSError("a file with the same name as the remotedir, "
                          "‘%s‘, already exists." % remotedir)
        else:

            head, tail = os.path.split(remotedir)
            if head and not self.is_dir(head):
                self.makedirs(head, mode)

            if tail:
                self.sftp.mkdir(remotedir, mode=mode)

    def open(self,filename,mode,data=None):
        """read or write file on remote server,check if file not exists it will occur"""
        with  paramiko.Transport(self.ip,self.port).connect(username=self.user, password=self.pwd)as f:
            sftp = paramiko.SFTPClient.from_transport(f)
            if mode==‘r‘ or mode=="r+":
                with  sftp.open(filename,mode) as fp:
                    readlines=fp.readlines()
                return readlines
            elif mode==‘w‘ or mode =="w+":
                with  sftp.open(filename, mode) as fp:
                    fp.write(data)
            else:
                raise ValueError("check mode in [‘r‘, ‘r+‘,‘w‘, ‘w+‘] or filename exists!")

class TestSftp(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        ip = "192.168.110.151"
        port = 22
        user = "root"
        pwd = "admin"
        cls.client =Paramiko_Sftp(ip, port, user, pwd)
        log.info("start selfcheck method of sftp ")

    @classmethod
    def tearDownClass(cls):
        try:
            cls.client.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.list_dir("/usr/local/listdir")
        self.assertIn(‘list.txt‘, files)

    def test_call_backstdout(self):
        shell = "ls -l /usr/local"
        readlines = self.client.cmd(shell)
        self.assertIn("redisdb", readlines)

    def test_exsitdir(self):
        a = self.client.exists("/usr/local")
        assert a == True

    def test_exsistfile(self):
        b = self.client.exists("/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(path)
        self.assertEqual(a, True)
        self.client.remove(path)

    def test_userremove(self):
        """remove file """
        path = "/usr/local/tou.txt"
        self.client.touch(path)
        self.client.remove(path)
        a = self.client.exists(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()