# 介面自動化測試之資料清洗/隔離/備份/恢復 > 在得到QQ:1301559180 得程式碼貢獻之後，想到了通過ssh連線上伺服器，然後進行資料庫備份，資料庫恢復， 主要使用了 `paramiko`庫 # 最終效果 1. 測試開始前會進行資料備份，並在下面2個路徑儲存對應的資料庫備份(目前備份的資料庫是寫死的為，測試連線的mysql資料庫).sql檔案 > 資料庫伺服器: /root/backup_sql/ > > 本地(執行測試指令碼的機器)當前目錄下的 `backup_sqls` > > **檔案命名方式為: 資料庫名+時間**   # 如何使用 1. 通過在`config.yaml`中配置對應得資料庫資訊，伺服器必要得賬號密碼等資訊(支援私鑰檔案登入，但個人未嘗試)，大致檔案格式如下 ```yaml # 資料庫校驗- mysql database: host: localhost port: 3306 user: root # 不用''會被解析成int型別資料 password: '123456' db_name: test charset: utf8mb4 # 資料庫所在的伺服器配置 ssh_server: port: 22 username: root password: '123456' # 私有金鑰檔案路徑 private_key_file: # 如果使用的docker容器部署mysql服務，需要傳入mysql的容器id/name mysql_container: mysql8 # 資料庫備份檔案匯出的本地路徑, 需要保證存在該資料夾 sql_data_file: backup_sqls/ ``` 2. 然後在`test/conftest.py`做如下操作 ```python #!/usr/bin/env/python3 # -*- coding:utf-8 -*- """ @project: apiAutoTest @author: zy7y @file: conftest.py @ide: PyCharm @time: 2020/12/8 @desc: """ import pytest from tools.data_clearing import DataClearing from tools.db import DB from tools.read_file import ReadFile @pytest.fixture(scope="session") def data_clearing(): """資料清洗""" DataClearing.server_init() # 1. 備份資料庫 DataClearing.backup_mysql() yield # 2. 恢復資料庫 DataClearing.recovery_mysql() DataClearing.close_client() # 若不需要資料清洗功能，請把get_db()入參拿掉 @pytest.fixture(scope="session") def get_db(data_clearing): """關於其作用域請移步檢視官方文件""" try: db = DB() yield db finally: db.close() @pytest.fixture(params=ReadFile.read_testcase()) def cases(request): """用例資料，測試方法引數入參該方法名 cases即可，實現同樣的引數化 目前來看相較於@pytest.mark.parametrize 更簡潔。 """ return request.param ``` # 實現程式碼 ```python #!/usr/bin/env python # -*- coding: utf-8 -*- """ @Time : 2021/1/19 11:44 @Author : zy7y @ProjectName : apiAutoTest @File : data_clearing.py @Software : PyCharm @Github : https://github.com/zy7y @Blog : https://www.cnblogs.com/zy7y """ import os from datetime import datetime import paramiko from tools.read_file import ReadFile from tools import logger class ServerTools: def __init__(self, host: str, port: int = 22, username: str = "root", password: str = None, private_key_file: str = None): # 進行SSH連線 self.trans = paramiko.Transport((host, port)) self.host = host if password is None: self.trans.connect(username=username, pkey=paramiko.RSAKey.from_private_key_file(private_key_file)) else: self.trans.connect(username=username, password=password) # 將sshclient的物件的transport指定為以上的trans self.ssh = paramiko.SSHClient() logger.success("SSH客戶端建立成功.") self.ssh._transport = self.trans # 建立SFTP客戶端 self.ftp_client = paramiko.SFTPClient.from_transport(self.trans) logger.success("SFTP客戶端建立成功.") def execute_cmd(self, cmd: str): """ :param cmd: 伺服器下對應的命令, 可以是list，或者str """ stdin, stdout, stderr = self.ssh.exec_command(cmd) error = stderr.read().decode() logger.info(f"輸入命令: {cmd} -> 輸出結果: {stdout.read().decode()}") logger.error(f"異常資訊: {error}") return error def files_action(self, post: bool, local_path: str = os.getcwd(), remote_path: str = "/root"): """ :param post: 動作 為 True 就是上傳， False就是下載 :param local_path: 本地的檔案路徑， 預設當前指令碼所在的工作目錄 :param remote_path: 伺服器上的檔案路徑，預設在/root目錄下 """ if post: # 上傳檔案 self.ftp_client.put(localpath=local_path, remotepath=f"{remote_path}{os.path.split(local_path)[1]}") logger.info(f"檔案上傳成功: {local_path} -> {self.host}:{remote_path}{os.path.split(local_path)[1]}") else: # 下載檔案 file_path = local_path + os.path.split(remote_path)[1] self.ftp_client.get(remotepath=remote_path, localpath=file_path) logger.info(f"檔案下載成功: {self.host}:{remote_path} -> {file_path}") def ssh_close(self): """關閉連線""" self.trans.close() logger.info("已關閉SSH連線...") class DataClearing: settings = ReadFile.read_config('$.database') server_settings = settings.get('ssh_server') server = None # 匯出的sql檔名稱及字尾 file_name = f"{settings.get('db_name')}_{datetime.now().strftime('%Y-%m-%dT%H_%M_%S')}.sql" @classmethod def server_init(cls, settings=settings, server_settings=server_settings): cls.server = ServerTools(host=settings.get('host'), port=server_settings.get('port'), username=server_settings.get('username'), password=server_settings.get('password'), private_key_file=server_settings.get('private_key_file')) # 新建backup_sql資料夾在伺服器上，存放匯出的sql檔案 cls.server.execute_cmd("mkdir backup_sql") @classmethod def backup_mysql(cls): """ 備份資料庫, 會分別備份在資料庫所在伺服器的/root/backup_sql/目錄下, 與當前專案檔案目錄下的 backup_sqls 每次備份生成一個數據庫名_當前年_月_日T_時_分_秒, 支援linux 伺服器上安裝的mysql服務(本人未除錯),以及linux中docker部署的mysql備份 """ if cls.server_settings.get('mysql_container') is None: cmd = f"mysqldump -h127.0.0.1 -u{cls.settings.get('username')} -p{cls.settings.get('password')} {cls.settings.get('db_name')} > {cls.file_name}" else: # 將mysql服務的容器中的指定資料庫匯出， 參考文章 https://www.cnblogs.com/wangsongbai/p/12666368.html cmd = f"docker exec -i {cls.server_settings.get('mysql_container')} mysqldump -h127.0.0.1 -u{cls.settings.get('user')} -p{cls.settings.get('password')} {cls.settings.get('db_name')} > /root/backup_sql/{cls.file_name}" cls.server.execute_cmd(cmd) cls.server.files_action(0, f"{cls.server_settings.get('sql_data_file')}", f"/root/backup_sql/{cls.file_name}") @classmethod def recovery_mysql(cls, sql_file: str = file_name, database: str = settings.get('db_name')): """ 恢復資料庫, 從伺服器位置(/root/backup_sql/) 或者本地(../backup_sqls)上傳, 傳入的需要是.sql檔案 :param sql_file: .sql資料庫備份檔案, 預設就是匯出的sql檔名稱， 預設檔名稱是匯出的sql檔案 :param database: 恢復的資料庫名稱，預設是備份資料庫(config.yaml中的db_name) """ result = cls.server.execute_cmd(f"ls -l /root/backup_sql/{sql_file}") if "No such file or directory" in result: # 本地上傳 cls.server.files_action(1, f"../backup_sqls/{sql_file}", "/root/backup_sql/") cmd = f"docker exec -i {cls.server_settings.get('mysql_container')} mysql -u{cls.settings.get('user')} -p{cls.settings.get('password')} {database} < /root/backup_sql/{sql_file}" cls.server.execute_cmd(cmd) @classmethod def close_client(cls): cls.server.ssh_close() ``` # 原始碼地址 gitee: https://gitee.com/zy7y/apiAutoTest github: https://github.com/zy7y/apiAutoTest # 參考資料 ``` https://www.cnblogs.com/wangsongbai/p/12666368.html https://www.liujiangblog.com/blog/15/ https://blog.csdn.net/leorx01/article/details/71141643 http://docs.paramiko.org/en/stable/api/client.html https://www.bilibili.com/video/BV1cQ4y1P7dg?p=4 # 詳細的參考資料可以看這裡 https://www.cnblogs.com/zy7y/p/14295902.html ``` # 最後 感謝遇見，歡