1、背景
這篇博客主要是對“數(shù)據(jù)清洗”而言的集歇。數(shù)據(jù)清洗在自動化測試中有什么作用呢。做過接口自動化測試的都知道语淘，測試會造一些測試數(shù)據(jù)驗(yàn)證接口是否正常诲宇。這些測試數(shù)據(jù)最終會去到數(shù)據(jù)庫里。這就是我們所說的測試臟數(shù)據(jù)惶翻。為了保證這些臟數(shù)據(jù)不影響后面的測試姑蓝，在執(zhí)行自動化測試前先要把原有數(shù)據(jù)庫進(jìn)行數(shù)據(jù)備份，在測試完成后對數(shù)據(jù)庫恢復(fù)吕粗，這就做到了數(shù)據(jù)“隔離”纺荧、“備份”、“恢復(fù)”颅筋。測試數(shù)據(jù)不對原有的數(shù)據(jù)造成任何影響宙暇。不知道解釋清楚了沒有~~~
2、最終效果
測試前：數(shù)據(jù)庫備份
測試結(jié)束：數(shù)據(jù)還原
3议泵、安裝第三方庫
pip install paramiko
4占贫、連接、操作Linux先口、
import paramiko
# 創(chuàng)建ssh客戶端
ssh_client = paramiko.SSHClient()
# 自動接受連接
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 連接Linux
# 方式一：用戶名+密碼連接
ssh_client.connect(hostname='ip',port=22,username='用戶名',password='密碼')
# 方式二：用戶名+密鑰(未調(diào)試)
# ssh_client.connect(hostname='ip',port='端口',username='用戶名',password='密碼', pkey=paramiko.RSAKey.from_private_key_file('密鑰文件路徑'))
#操作Linux
"""
stdin :執(zhí)行輸入 stdout：執(zhí)行結(jié)果 stderr：標(biāo)準(zhǔn)錯誤結(jié)果輸出
"""
stdin, stdout, stderr = ssh_client.exec_command('linux命令')
#默認(rèn)返回時<bytes>
#轉(zhuǎn)為字符串輸出
print(stdout.read().decode('utf-8'))
# 關(guān)閉連接
ssh_conn.close()
輸出結(jié)果是/root
5型奥、文件上傳、下載
# 創(chuàng)建連接
ssh_conn = paramiko.Transport('IP',22)
# 連接linux
ssh_conn.connect(username='用戶名',password='密碼')
# 創(chuàng)建sftp客戶端
sftp_client = paramiko.SFTPClient.from_transport(ssh_conn)
# 上傳 （localpath：本地路徑, remotepath：遠(yuǎn)程路徑）
# 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')
# 關(guān)閉連接
ssh_conn.close()
執(zhí)行后碉京，在在服務(wù)器/本地 下載/上傳文件
5厢汹、代碼封裝
class LinuxTools:
# 初始化
def __init__(self,host,port,username,password=None,private_key_file=None):
"""
:param host: 服務(wù)器IP （str）
:param port: 服務(wù)器端口 (int)
:param username: 服務(wù)器登錄用戶名稱 （str）
:param password: 服務(wù)器密碼 （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
# 創(chuàng)建ssh客戶端
ssh_client = paramiko.SSHClient()
# 自動接受服務(wù)器發(fā)過來的密鑰
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())
# 關(guān)閉連接
def ssh_close(self):
self.ssh_client.close()
class SFTPTools:
"""遠(yuǎn)程<-->本地厘灼，上傳夹纫、下載操作"""
def __init__(self,host,port,username,password=None,private_key_file=None):
"""
:param host: 服務(wù)器IP （str）
:param port: 服務(wù)器端口 (int)
:param username: 服務(wù)器登錄用戶名稱 （str）
:param password: 服務(wù)器密碼 （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('連接已關(guān)閉')
補(bǔ)充：
數(shù)據(jù)庫備份：mysqldump -u root -p密碼 數(shù)據(jù)庫名稱 > 備份路徑.sql(備份文件.sql)
數(shù)據(jù)庫還原：mysql -u root -p密碼 數(shù)據(jù)庫名稱 < 備份路徑
數(shù)據(jù)清理大致思路：在測試前備份到本地文件夾 -->在上傳到服務(wù)器 -->測試執(zhí)行完畢 -->在服務(wù)器上執(zhí)行還原命令 月匣。
目前是這么個想法，具體代碼還木有開始寫~~
6奋姿、參考資料
https://www.bilibili.com/video/BV1cQ4y1P7dg?p=5&spm_id_from=pageDriver
https://www.cnblogs.com/zy7y/p/14295902.html
創(chuàng)作不易锄开，點(diǎn)個贊，關(guān)注下唄~~~
