基於python多線程自制快速冒煙測試工具

一、前言

在項目測試的過程中，發現經常會有下面這些情況：

• 新版本升級后，有時會因為代碼或部署時配置錯誤，某個菜單報類似404或500的錯誤；
• 為避免以上問題，實施部署人員升級完后，需要手動冒煙一遍所有菜單頁面；
• 發布當天還有修改提測，來不及再全部手動回歸一遍；
• 項目周期不長，沒有時間來做業務流程的自動化。

如果所有的菜單都通過手動來冒煙一遍的話，我們假定一個菜單頁面要5秒，那么如果一個項目有600個菜單頁面，那么就需要3000秒，整整一個小時啊，而且是每次升級都要來這么一次。於是我就想，雖然沒時間將所有功能都做成自動化，那能不能實現對所有的菜單頁面都自動去訪問一遍，然后根據訪問結果判斷是不是都是正常的呢？

 

二、實現步驟分析

想法有了，那就得分析該怎么實現了。最大的問題就在於所有菜單對應的url要怎么獲取了，通過各種嘗試之后，發現所測項目由於框架原因，無法直接通過爬蟲來獲取所有的url。

但是在數據庫里，對應前后台都有兩張表，如eclp_A存放有域名，eclp_B則存有各個菜單頁面的具體路徑，那么只需要把兩張表的數據取出來拼接在一起就成了一個完整的url了。

 

具體步驟：

（1）在數據庫里將客戶端和后台管理端（客戶端是uc，后台管理端是eclp）域名和路徑取出來進行拼接放到一個新表，並加入對應斷言關鍵詞；

（2）不同項目的數據庫配置不一樣，所以把數據庫配置放在配置文件，再通過程序獲取；

（3）獲取到數據庫的配置后，連接數據庫，從表里獲取到所有的url；

（4）取到所有的url之后遍歷去request，但是這個需要先登錄獲取到cookie后做個cookie綁定，那么登錄需要用到的用戶密碼也需要放到配置文件；

（5）遍歷訪問所有url，並記錄其訪問結果，成功與失敗個數；

（6）在程序執行訪問頁面時，把成功與失敗的日志都記錄下來，方便后續查找問題；

（7）因為url可能比較多，訪問所有頁面會很慢，所以實現多進程或多線程的方式；

（8）將所有頁面訪問結果利用HTML報告輸出；

 

三、前期准備

1. 在數據庫（此處是oracle）拼接URL

1.1 創建存放url的表

查看后台存放主域名和路徑的表

select t.*,t.rowid from eclp_A t order by id desc;

select t.*,t.rowid from eclp_B t order by id desc;

創建存放完整url的表

drop table eclp_uc_url;

CREATE TABLE eclp_uc_url (

ID NUMBER(20) DEFAULT 0  NOT NULL ,

SUB_SYSTEM_ID  NUMBER(20) DEFAULT 0  NOT NULL ,

SUB_SYSTEM_CODE VARCHAR2(100 BYTE) DEFAULT ''  NULL ,

NAME  VARCHAR2(255 BYTE) DEFAULT ''  NULL , 

DOMAIN VARCHAR2(100 BYTE) DEFAULT ''  NULL ,

PATH VARCHAR2(100 BYTE) DEFAULT ''  NULL ,

URL  VARCHAR2(100 BYTE) DEFAULT ''  NULL ,

ASSERT_WORD  VARCHAR2(255 BYTE) DEFAULT ''  NULL

)

創建主鍵

ALTER TABLE eclp_uc_url ADD PRIMARY KEY (ID);

創建id自增序列

drop SEQUENCE SEQ_ECLP_UC_URL;

CREATE SEQUENCE SEQ_ECLP_UC_URL

INCREMENT BY 1

START WITH 1

MAXVALUE 9999999999999

CYCLE

CACHE 20;

 

1.2 從兩個表分別插入域名和路徑到存放url的新表

eclp的路徑path插入表

INSERT INTO eclp_uc_url(id,SUB_SYSTEM_ID,NAME,PATH) SELECT SEQ_ECLP_UC_URL.NEXTVAL,SUB_SYSTEM_ID,NAME,URL from eclp_A;

eclp的域名domian插入表

update eclp_uc_url set DOMAIN = (select DOMAIN from eclp_B where eclp_uc_url.SUB_SYSTEM_ID = eclp_B.id)
where exists (select 1 from eclp_B where eclp_uc_url.SUB_SYSTEM_ID = eclp_sub_system.id);

將domain和path拼接成url

MERGE INTO eclp_uc_url A
USING ( select t.id, t.domain || '/' || t.path as urls from eclp_uc_url t) B ON (A.ID = B.ID)
WHEN MATCHED THEN UPDATE SET A.URL = B.URLS;

 

2. 數據庫配置與前后台用戶配置文件

創建一個DbUser.ini文件：

存有數據庫連接配置、前后台用戶信息、瀏覽器選擇（chrome、firefox、ie）

[db]
oracle_ip=127.0.0.1
oracle_account=test_account
oracle_password=test_password

[eclp]
center_url=http://127.0.0.1:8020/system/login.htm
center_account=test01
center_password=123456a
 
[uc]
uc_url=http://127.0.0.1:9001/login.htm
uc_account=test02
uc_password=123456a

[driver]
browser=chrome

 

3. 前后台登錄元素定位方式配置文件

包含用戶名、密碼、登錄按鈕的定位方位，UiObjectMap.ini

[eclp]
LoginAccount=id>account
LoginPassword=id>password
LoginButton=xpath>//input[@class='btn2']

[uc]
LoginAccount=id>account
LoginPassword=id>password
LoginButton=id>button

 

四、代碼實現-封裝

1. 創建包和文件夾目錄

Config：存放配置文件

Driver：存放不同瀏覽器驅動

TestResults：存放測試結果

TestScripts：存放程序腳本

Util：存放封裝方法

2. 將所需用到目錄及文件路徑單獨整理

ProjVar.py

import os

proj_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
conf_path = os.path.join(proj_path,"config")
dbuser_ini_path = os.path.join(proj_path,"config","DbUser.ini")
objectmap_ini_path = os.path.join(proj_path,"config","UiObjectMap.ini")
driver_path = os.path.join(proj_path,"Driver")
logger_path = os.path.join(proj_path,"config","Logger.conf")
result_path = os.path.join(proj_path,"Testresults")
testscript_path = os.path.join(proj_path,"TestScripts")

3. 從ini配置文件讀取數據方法：

ReadConfig.py

# encoding=utf-8
import configparser
import os
import platform
from Config.ProjVar import *

def read_ini_file(ini_file_path, section_name, option_name):
    #創建一個讀取配置文件的實例
    cf = configparser.ConfigParser()
    #將配置文件內容加載到內存
    cf.read(ini_file_path)
    try:
        #根據section和option獲取配置文件中的數據
        value = cf.get(section_name, option_name)
    except:
        print("the specific seciton or the specific option doesn't exit!")
        return None
    else:
        return value

4．獲取元素對象方法

根據元素定位方式獲取到對應元素對象，后面登錄時會用到

ObjectMap.py

#encoding=utf-8
from selenium.webdriver.support.ui import WebDriverWait
import configparser,os
from selenium import webdriver
from Config.ProjVar import *
from Util.ReadConfig import read_ini_file

class ObjectMap(object):
    def __init__(self):
        # 存放頁面元素定位表達方式及定位表達式的配置文件所在絕對路徑
        self.uiObjMapPath = objectmap_ini_path

    def getElementObject(self, driver, webSiteName, elementName):
        try:
            locators = read_ini_file(self.uiObjMapPath, webSiteName, elementName).split(">")
            # 得到定位方式
            locatorMethod = locators[0]
            # 得到定位表達式
            locatorExpression = locators[1]
            print(locatorMethod, locatorExpression)
            # 通過顯示等待方式獲取頁面元素
            element = WebDriverWait(driver, 10).until(lambda x: \
                                    x.find_element(locatorMethod, locatorExpression))
        except Exception as e:
            raise e
        else:
            # 當頁面元素被找到后，將該頁面元素對象返回給調用者
            return element

 

5. 建立一個綁定cookies的session對象

從元素定位方式配置文件獲取到定位方式，然后用ObjectMap方法獲取到元素對象，用selenium webdriver自動登錄，登錄成功后得到cookies

 GetSessionOfCookie.py

from selenium import webdriver
import requests,time,os
from Config.ProjVar import *
from Util.ReadConfig import read_ini_file
from Util.ObjectMap import *
#from selenium.webdriver.chrome.options import Options

def get_session_of_cookie(domain,url,account,password):
    #從配置文件獲取配置的瀏覽器類型，並對應去登錄獲取cookie
    browser = read_ini_file(dbuser_ini_path, "driver", "browser")
    if browser.lower() == "chrome":
        driverpath = os.path.join(driver_path, "chromedriver.exe")
        driver = webdriver.Chrome(executable_path=driverpath)
    elif browser.lower() == "firefox":
        driverpath = os.path.join(driver_path,"geckodriver.exe")
        driver = webdriver.Firefox(executable_path=driverpath)
    elif browser.lower() == "ie":
        driverpath = os.path.join(driver_path, "IEDriverServer.exe")
        driver = webdriver.Ie(executable_path=driverpath)
    driver.maximize_window()
    time.sleep(1)
    #打開前后台登錄頁面
    driver.get(url)
    driver.implicitly_wait(5)
    # 獲取登錄頁面元素傳值登錄
    objmap = ObjectMap()
    if domain == "eclp":
        objmap.getElementObject(driver, "eclp", "LoginAccount").send_keys(account)
        objmap.getElementObject(driver, "eclp", "LoginPassword").send_keys(password)
        objmap.getElementObject(driver, "eclp", "LoginButton").click()
    elif domain == "uc":
        objmap.getElementObject(driver, "uc", "LoginAccount").send_keys(account)
        objmap.getElementObject(driver, "uc", "LoginPassword").send_keys(password)
        objmap.getElementObject(driver, "uc", "LoginButton").click()
    time.sleep(3)
    #獲取登錄后的cookies
    allcookies = driver.get_cookies()
    print("獲取到登錄后的cookies:%s" % allcookies)
    driver.quit()
    #把上面獲取的的cookies添加到s中
    s = requests.session()
    try:
        # 添加cookies到CookieJar
        c = requests.cookies.RequestsCookieJar()
        for i in allcookies:
            c.set(i["name"], i['value'])
        # 更新session里cookies
        s.cookies.update(c)  
    except Exception as e:
        print(u"添加cookies報錯：%s" %str(e))
    print("查看添加后s的cookies")
    print(s.cookies)
    return s

 

6. 從數據庫表查詢URL

 這里使用的是cx_Oracle連接oracle，注意需要使用與python位數（32或64）對應的數據庫instantclient客戶端：

GetUrlFromOra.py

import cx_Oracle
from Util.ReadConfig import read_ini_file
from Config.ProjVar import *

def get_url_from_oracle(ip,account,password,domain):
    db=cx_Oracle.connect(account+'/'+password+'@'+ip+'/orcl')
    cr = db.cursor()
    sql = ""
    #根據是eclp還是uc來獲取前端還是后端的url
    if domain == "eclp":
        sql = 'select sub_system_id,name,url,assert_word from eclp_uc_url where sub_system_id != 0  order by id desc'
    elif domain == "uc":
        sql = 'select sub_system_id,name,url,assert_word from eclp_uc_url where sub_system_id = 0 order by id desc'
    cr.execute(sql)
    result = cr.fetchall()
    #返回獲取到的所有結果
    return result

 

7. 創建存放測試結果的文件夾方法

先創建一個執行當天日期為名稱的文件夾，若一天執行了多次，在日期文件夾下建一個“第n次”為名稱的文件夾

#encoding = utf-8
import os
import time
from Config.ProjVar import *

#創建日期格式文件夾
def make_date_dir(dir_path):
    if os.path.exists(dir_path):
        #獲取當前時間
        timeTup = time.localtime()
        #轉為xxxx年xx月xx日的格式
        currentDate = str(timeTup.tm_year) + "年" + str(timeTup.tm_mon) + "月" + str(timeTup.tm_mday) + "日"
        #用目標目錄拼接日期得到絕對路徑
        path = os.path.join(dir_path,currentDate)
        if not os.path.exists(path):
            os.mkdir(path)
    else:
        raise Exception("dir_path does not exist!")
    return path

#創建日期文件夾下多次執行的目錄
def make_report_dir():
    #先創建一個日期格式為名稱的文件夾
    date_path = make_date_dir(result_path)
    #判斷當前目錄已有文件夾數，加1得到新文件夾名並創建
    report_path = os.path.join(date_path, "第" + str(len(os.listdir(date_path)) + 1) + "次測試")
    os.mkdir(report_path)
    #進入到新創建文件夾並獲取當前的絕對路徑，作為后面存放測試結果的文件夾
    os.chdir(report_path)
    result_report_path = os.getcwd()
    return result_report_path

 

8. Log日志設置

使用python的Logger.conf配置文件，設置我們需要的日志級別和存放目錄

 

9. HTML報告模板

創建一個htmlTemplate方法，里面先用HTML語言設置好自己想要的報告模板，后面將測試結果與模板拼接即可。

 

五、主程序實現

從數據庫獲取url方法封裝：斷言關鍵詞等數據放進隊列中，並返回前台客戶端uc和后台管理端eclp綁定cookie的session對象

def get_urls(domain,q):
    #判斷是前端uc還是后端eclp去獲取對應的登錄數據
    if domain == "eclp":
        login_url = read_ini_file(dbuser_ini_path, "eclp", "center_url")
        account = read_ini_file(dbuser_ini_path, "eclp", "center_account")
        password = read_ini_file(dbuser_ini_path, "eclp", "center_password")
    elif domain == "uc":
        login_url = read_ini_file(dbuser_ini_path, "uc", "uc_url")
        account = read_ini_file(dbuser_ini_path, "uc", "uc_account")
        password = read_ini_file(dbuser_ini_path, "uc", "uc_password")
    #獲取eclp或uc的綁定cookie的session對象
    s = get_session_of_cookie(domain,login_url,account,password)
    #從配置文件獲取數據庫的連接配置
    ora_ip = read_ini_file(dbuser_ini_path, "db", "oracle_ip")
    ora_account = read_ini_file(dbuser_ini_path, "db", "oracle_account")
    ora_password = read_ini_file(dbuser_ini_path, "db", "oracle_password")
    #從數據庫獲取所有的url
    url_datas = get_url_from_oracle(ora_ip,ora_account,ora_password,domain)
    #將獲取到的所有url放入隊列中
    for data in url_datas:
        q.put(data)
    return s

執行獲取測試結果方法封裝：

在隊列不為空時，循環取得url、斷言關鍵詞等數據，然后利用綁定了cookied的session對象訪問得到結果；

獲取資源鎖記錄成功與失敗個數，並將結果放入一個共享的html字符串中，以便最后組裝成完成的HTML報告：

def get_urls_info():
    #設置測試結果為全局變量
    global share_result
    global success_case_num
    global fail_case_num
    flagDict = {0: 'red', 1: '#00AC4E'}
    #隊列不為空時，循環獲取數據
    while not q.empty():
        #每個測試數據的測試結果，success或fail
        test_case_result = ""
        start = time.time()
        start_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        #取隊列
        data = q.get()
        #分別取出要用到的菜單名稱、url、斷言關鍵詞等數據
        sub_system_id = data[0]
        url_name = data[1]
        url = data[2]
        assert_word = data[3]
        #存放斷言失敗的菜單的源碼文件路徑
        html_file_path = os.path.join(result_report_path, url_name + ".html")
        #訪問菜單頁面並記錄結果
        try:
            判斷是eclp的還是uc的菜單，用不同的session對象去訪問
            if sub_system_id == 0:
                r = uc_s.get(url)
            else:
                r = eclp_s.get(url)
            #斷言
            assert assert_word in r.text
            flag = 1
            debug("菜單： " + url_name + " 斷言成功：")
            #記錄當次是否訪問成功
            test_case_result = "Success"
            #獲得資源鎖，記錄成功個數
            lock.acquire()
            success_case_num += 1
            lock.release()
        except AssertionError as e:
            flag = 0
            error("菜單： " + url_name+" 關鍵詞斷言失敗，錯誤信息：" + str(traceback.format_exc()))
            test_case_result = "Fail"
            #如果失敗，獲取資源鎖，記錄失敗數
            lock.acquire()
            fail_case_num += 1
            lock.release()
            #將失敗的菜單頁面的源碼保存為一個html文件，存到測試結果目錄
            with open(html_file_path, "w", encoding="utf-8") as fp:
                fp.write(r.text)
        except Exception as e:
            flag = 0
            error("菜單： " + url_name+" 發生未知異常，錯誤信息：" + str(traceback.format_exc()))
            test_case_result = "Fail"
            lock.acquire()
            fail_case_num += 1
            lock.release()
            with open(html_file_path, "w", encoding="utf-8") as fp:
                fp.write(r.text)
        finally:
            end = time.time()
            waste_time = int((end - start) * 1000)
            #獲取資源鎖，將當次訪問的菜單名稱、url斷言關鍵詞、時間等添加到html字符串
            lock.acquire()
            # 每一組數據測試結束后，都將其測試結果信息插入表格行的HTML代碼中
            share_result += '''
                  <tr>
                      <td>%s</td>
                      <td>%s</td>
                      <td>%s</td>
                      <td>%s</td>
                      <td>%s</td>
                      <td style="color:%s">%s</td>
                  </tr>''' % (url_name, url, assert_word, waste_time, start_time, flagDict[flag], test_case_result)
            lock.release()

 

由於這里更多的是文本、多IO的操作，因此我們選擇多線程來實現：

#定義一個繼承threading.Thread線程的類
class MyThread(threading.Thread):
    def __init__(self, func):
        threading.Thread.__init__(self)
        self.func = func
    def run(self):
        self.func()

main程序：

if __name__ == "__main__":
    # 創建用來存放日志和報告的文件夾
    result_report_path = make_report_dir()
    #建一個線程共享資源鎖
    lock = threading.Lock()
    #定義一個隊列
    q = Queue()
    #定義測試結果變量
    share_result = ""
    success_case_num = 0
    fail_case_num = 0
    #獲取前后台的綁定了cookie的session對象
    eclp_s = get_urls("eclp",q)
    uc_s = get_urls("uc",q)

    #創建多個線程，並將get_urls_info任務傳入處理
    threads = []
    for i in range(4):
        thread = MyThread(get_urls_info)
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()
    #將測試結果與模板拼接得到HTML報告
    htmlTemplate(share_result,result_report_path,success_case_num,fail_case_num)

 

六、測試結果

在測試結果目錄Testresults中有不同日期的文件夾：

下層是一天中多次執行的文件夾，要看最新的測試結果進入“第n次”中n最大的文件夾即可：

 

進入后可以查看到訪問失敗菜單的html源碼，若執行后有失敗的菜單，可以打開查看里面源碼是什么樣的；

另外還有當次執行的《冒煙測試報告》，打開可以查看總個數、成功個數、失敗個數以及每個菜單的執行情況，

 失敗的菜單頁面對應測試結果會顯示紅色：

 

在程序腳本目錄下會生成一個日志文件SmokingTestLog.log，每次執行的日志都會增量記錄在這里：

 

 

七、總結

至此快速的冒煙測試就完成了，基本實現了可以從數據庫獲取所有菜單頁面URL，並利用多線程遍歷訪問，並記錄結果生成HTML報告的功能。

可應用場景：

（1）開發人員開發完成，在開發環境執行快速冒煙，檢查是否有異常頁面；

（2）新功能提測后，測試人眼執行快速冒煙，檢查是否有異常頁面再深入測試；

（3）發布UAT之前，執行快速冒煙，回歸是否有頁面異常

（4）實施人員升級后，執行快速冒煙，檢驗是否部署、配置正確，避免線上出現異常

不同的項目可能模式有所不同，如果菜單未放在表里，則需要看是否可以直接爬取或是其他方式實現了。

若文中有不正確的地方，或有更好的實現方法，歡迎指出並提寶貴意見。