1、背景

这篇博客主要是对“数据清洗”而言的。数据清洗在自动化测试中有什么作用呢。做过接口自动化测试的都知道，测试会造一些测试数据验证接口是否正常。这些测试数据最终会去到数据库里。这就是我们所说的测试脏数据。为了保证这些脏数据不影响后面的测试，在执行自动化测试前先要把原有数据库进行数据备份，在测试完成后对数据库恢复，这就做到了数据“隔离”、“备份”、“恢复”。测试数据不对原有的数据造成任何影响。不知道解释清楚了没有~~~

2、最终效果

测试前：数据库备份
测试结束：数据还原

3、安装第三方库

  pip install paramiko

4、连接、操作Linux、

import paramiko


# 创建ssh客户端
ssh_client = paramiko.SSHClient()

# 自动接受连接
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 连接Linux
# 方式一：用户名+密码连接
ssh_client.connect(hostname='ip',port=22,username='用户名',password='密码')

# 方式二：用户名+密钥(未调试)
# ssh_client.connect(hostname='ip',port='端口',username='用户名',password='密码', pkey=paramiko.RSAKey.from_private_key_file('密钥文件路径'))

#操作Linux
"""
    stdin :执行输入     stdout：执行结果     stderr：标准错误结果输出
"""
stdin, stdout, stderr = ssh_client.exec_command('linux命令')

#默认返回时<bytes>
#转为字符串输出
print(stdout.read().decode('utf-8'))

# 关闭连接
ssh_conn.close()

输出结果是/root

5、文件上传、下载

# 创建连接
ssh_conn = paramiko.Transport('IP',22)

# 连接linux
ssh_conn.connect(username='用户名',password='密码')

# 创建sftp客户端
sftp_client = paramiko.SFTPClient.from_transport(ssh_conn)

# 上传 （localpath：本地路径, remotepath：远程路径）
# sftp_client.put(localpath=r'C:\Users\Mr.he\Desktop\apiAutoTest\log\run2021-07-03_16-43-58_518374.log', remotepath='/root/mylog.log')

# 下载
sftp_client.get(remotepath='/root/mylog.log', localpath=r'C:\Users\Mr.he\Desktop\apiAutoTest\log\mylog.log')

# 关闭连接
ssh_conn.close()

执行后，在在服务器/本地 下载/上传文件

5、代码封装

class LinuxTools:
    # 初始化
    def __init__(self,host,port,username,password=None,private_key_file=None):
        """
        :param host: 服务器IP （str）
        :param port: 服务器端口 (int)
        :param username: 服务器登录用户名称 （str）
        :param password: 服务器密码 （str）
        :param private_key_file: 私钥文件路劲 （str）
        """
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.private_key_file = private_key_file


        # 创建ssh客户端
        ssh_client = paramiko.SSHClient()
        # 自动接受服务器发过来的密钥
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            # 判断是密码方式登录还是私钥方式登录
            if self.password is None:
                private = paramiko.RSAKey.from_private_key_file(self.private_key_file)
                ssh_client.connect(hostname=self.host, port=self.port, username=self.username, pkey=private)

            else:
                ssh_client.connect(hostname=self.host, port=self.port, username=self.username, password=self.password)
            self.ssh_client = ssh_client
        except Exception as e:
            print('连接失败！！！{}'.format(e))

    # 操作linux
    def handler_cmd(self,cmd):
        """
        :param cmd: linux操作命令，（str,list）
        """
        if isinstance(cmd, list):
            for c in cmd:
                stdin, stdout, stderr = self.ssh_client.exec_command(c)
        else:
            stdin, stdout, stderr = self.ssh_client.exec_command(cmd)
            print(stdout.read().decode())

    # 关闭连接
    def ssh_close(self):
        self.ssh_client.close()


class SFTPTools:
    """远程<-->本地，上传、下载操作"""
    def __init__(self,host,port,username,password=None,private_key_file=None):
        """
        :param host: 服务器IP （str）
        :param port: 服务器端口 (int)
        :param username: 服务器登录用户名称 （str）
        :param password: 服务器密码 （str）
        :param private_key_file: 私钥文件路劲 （str）
        """
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.private_key_file = private_key_file

        ssh_client = paramiko.Transport((self.host,self.port))
        try:
            # 判断是密码方式登录还是私钥方式登录
            if self.password is None:
                private = paramiko.RSAKey.from_private_key_file(self.private_key_file)
                ssh_client.connect(username=self.username, pkey=private)

            else:
                ssh_client.connect( username=self.username, password=self.password)
            # 这个地方我也不知道为什么，照抄吧~~~
            self.ssh_client = ssh_client
        except Exception as e:
            print('连接失败！！！{}'.format(e))

        self.ftp_client = paramiko.SFTPClient.from_transport(ssh_client)

    def handler_file(self,local_path,remote_path,is_upload=True):
    # try:
        if is_upload:
            self.ftp_client.put(localpath=local_path,remotepath=remote_path)
            print('上传成功')
        else:
            self.ftp_client.get(remotepath=remote_path,localpath=local_path)
            print('下载成功')
    # except Exception as e:
    #     print(e)

    def ssh_close(self):
        self.ssh_client.close()
        print('连接已关闭')

补充：
数据库备份：mysqldump -u root -p密码 数据库名称 > 备份路径.sql(备份文件.sql)
数据库还原：mysql -u root -p密码 数据库名称 < 备份路径

数据清理大致思路：在测试前备份到本地文件夹 -->在上传到服务器 -->测试执行完毕 -->在服务器上执行还原命令 。
目前是这么个想法，具体代码还木有开始写~~

6、参考资料

https://www.bilibili.com/video/BV1cQ4y1P7dg?p=5&spm_id_from=pageDriver
https://www.cnblogs.com/zy7y/p/14295902.html

创作不易，点个赞，关注下呗~~~