自动化测试框架

自动化测试框架的概念
自动化测试框架就是一个能够进行自动化测试的程序，其本质也是一堆“按照特定结构组织”的代码。自动化测试框架可以提高代码的可复用性、可维护性、可扩展性。

●可复用性：编写的代码不仅适用于一个模块和项目，同时也适用于其他的模块和项目
●可维护性：减少代码维护时的工作量，比如修改某个功能代码时，只需要修改一处即可，不需要每处代码去单独修改
●可扩展性：当要增加或者扩展功能时，可以不影响原有模块

框架设计比较灵活，并没有通用的标准来规定框架必须具备什么结构或功能，框架的目的只有一个，提高测试效率，降低测试成本。

两大核心功能
●驱动测试
○数据驱动（ddt）
○关键字驱动（kdt）
○页面驱动（pom）
●辅助测试
○数据管理
○报告管理
○配置管理
○日志管理等

测试驱动
数据驱动
将测试代码和测试数据分离，一般用于一个测试场景需要多个不同的测试数据的条件下。通过数据驱动测试，有多少条测试数据，测试就会执行多少次。如：unittest的ddt模块

关键字驱动
主打易用性，将测试代码、测试业务和测试数据相分离，灵活性高。
使用时，只需要考虑三个问题，对谁做，做什么，参数
关键字驱动的表达式：object.action(parameters)

data.txt

PowerShell复制代码

# 关键字驱动的脚本
# action;object;parameters
goto;https://segmentfault.com/
delay;2
singleclick;xpath://nav[@class='p-0 navbar navbar-expand navbar-light']
singleclick;xpath://a[@data-mode='password']
delay;2
input;xpath://input[@placeholder="手机号或邮箱"];18423262925
input;xpath://input[@placeholder="密码"];xujiao5770
singleclick;xpath://button[@class='sflex-center btn btn-primary btn-block btn-lg']

util.py

PowerShell复制代码

# 导入Excel的模块
import openpyxl


class Util:

    @classmethod
    def get_test_data_from_txt(cls, path):
        with open(path, 'r', encoding='utf-8') as f:
            cmds = []
            while True:
                # 读取一行：
                line = f.readline()
                # 读取结束
                if not line:
                    break
                cmds.append(line.strip())
            return cmds

    @classmethod
    def get_test_data_from_xlsx(cls, path):
        # 从Excel中读取数据
        workbook = openpyxl.load_workbook(path)
        worksheet = workbook.active
        test_data = []
        for row in range(2, worksheet.max_row + 1):
            cmd = ''
            for column in range(1, worksheet.max_column + 1):
                content = worksheet.cell(row=row, column=column).value
                if content:
                    cmd += str(content)
                    cmd += ';'
            test_data.append(cmd[:-1])
        return test_data


if __name__ == '__main__':
    # cmds = Util.get_test_data_from_txt('data.txt')
    cmds = Util.get_test_data_from_xlsx('data.xlsx')
    print(cmds)

keywords.py

PowerShell复制代码

from selenium import webdriver
import time
from selenium.webdriver.common.action_chains import ActionChains

driver = webdriver.Chrome()
driver.maximize_window()


class KeyWord:

    @classmethod
    def goto(cls, *args):
        print('参数：{}'.format(args))
        if args and len(args) >= 2:
            driver.get(args[1])
            return True
        else:
            print('goto命令不正取，请检查')
            return False

    @classmethod
    def input(cls, *args):
        print('input的参数：{}'.format(*args))
        if args and len(args) >= 3:
            element_split = args[1].split(':')
            # element_split[0]是查找定位的方式
            # element_split[1]是要查找的元素
            driver.find_element(element_split[0], element_split[1]).send_keys(args[2].strip())
            return True
        else:
            print('input命令不正取，请检查')
            return False

    @classmethod
    def singleclick(cls, *args):
        print('singleclick的参数：{}'.format(args[1]))
        if args and len(args) >= 2:
            element_split = args[1].split(':')
            driver.find_element(element_split[0], element_split[1]).click()
            return True
        else:
            print('singleclick命令不正取，请检查')
            return False

    @classmethod
    def delay(cls, *args):
        print('睡眠参数:{}'.format(args[1]))
        if args and len(args) >= 2:
            time.sleep(int(args[1]))
            return True
        else:
            print('delay命令参数不正取，请检查')
            return False

run.py

PowerShell复制代码

from selenium import webdriver
from util import Util
from keywords import KeyWord


def run_case_from_txt():
    data = Util.get_test_data_from_txt('data.txt')
    for line in data:
        print('当前执行：{}'.format(line))
        if line.startswith('#'):
            continue
        key_words = line.split(";")
        # 判断是否有这个方法
        if hasattr(KeyWord, key_words[0]):
            # 调用关键字对应的方法
            action = getattr(KeyWord, key_words[0])
            result = action(*key_words)
            if not result:
                print('{}命令执行错误，测试终止'.format(action))
                break
            else:
                print('{}命令执行成功'.format(action))
    pass


if __name__ == '__main__':
    run_case_from_txt()

页面驱动
页面对象（page Object）
属性：所有在该页面的测试操作中所会用到的元素对象。比如在登录测试操作中，我们会用到用户名、密码、验证码以及登录按钮这几个页面元素，这几个页面元素就是登录页面这个页面对象的属性
方法：所有在该页面能执行的跟测试相关的操作，我们会封装为该页面对象的方法。
如果使用传统的测试代码编写方法（不使用PO模型）会导致两个问题：
1代码量大，无法复用代码，代码编写效率很低
2代码维护困难一旦元素属性发生修改，所有涉及该元素的代码都需要改

pages.base_page.py

Python复制代码

class basePage:
    def __init__(self, driver):
        self.driver = driver

    def find_element(self, *loc):
        # loc =(id,username)
        return self.driver.find_element(*loc)

    def find_elements(self, *loc):
        # loc =(id,username)
        return self.driver.find_elements(*loc)

    '''
        如果有其他的基本操作，可以继续封装
    '''

pages.login_page.py

Python复制代码

from pages.base_page import basePage
from selenium.common.exceptions import NoSuchElementException

# 定义登录页面的属性和方法
class LoginPage(basePage):
    username = ('id', 'username')
    password = ('id', 'passeord')
    verifycode = ('id', 'verifycode')

    login_btn = ('xpath', '//button[contains(@onclick,"dologin")]')
    index_msg = ('xpath', '//ul[contains(@class,"navbar-right")]/li[1]/a')

    def login_action(self, un, pw, vc):
        self.find_element(*self.username).send_keys(un)
        self.find_element(*self.password).send_keys(pw)
        self.find_element(*self.verifycode).send_keys(vc)

        self.find_element(*self.login_btn).click()

    def check_login_status(self):
        try:
            index_msg_text = self.find_element(*self.index_msg).text
        except NoSuchElementException:
            print("===========login failed==============")
            return False
        else:
            if 'admin' in index_msg_text:
                print("============login success=============")
                return True
            else:
                print("===========login failed==============")
                return False

testcases.myunit.py

Python复制代码

import unittest
from selenium import webdriver


class MyUnit(unittest.TestCase):
    def setUp(self):
        self.driver = webdriver.Chrome()
        self.driver.get('http://www.baidu.com')

    def tearDown(self):
        self.driver.quit()

testcases.test_login.py

Python复制代码

import unittest
from myunit import MyUnit
from pages.login_page import LoginPage


class LoginTest(MyUnit):

    def test_login_success(self):
        login_page = LoginPage(self.driver)
        login_page.login_action('admin', 'admin123', '123456')

        self.assertTrue(login_page.check_login_status())

    def test_login_failed(self):
        login_page = LoginPage(self.driver)
        login_page.login_action('', 'admin123', '123456')

        self.assertTrue(login_page.check_login_status())

多机兼容测试todo p191-p293

自动化测试的常用文件
 

通常用于移动端测试时,进行测试包的管理

app

防止不同的测试包

如果是web端自动化测试时,不需要该路径

basepage.py:该模块邓庄了一个类basePage.该

TMina

家

类封装最基本的方法

初始化driver

init(self.driver)

basePage类

Find_elemento

pages

基础封装类

Find_elementso

跟driver有关的方法,都可以封装

其他page类,一个页面对应一个page类,存放业

务逻辑操作队形,封装操作相关元素和方法

common

封装公共方法

config

自动化测试结构

用于存放配置文件

XMind

data

试用裤式

数据文件:用于数据驱动

log

用于存放日志文件数据

report

测试报告文件

screenhots

截图

test_case

存放测试类

test_run

执行自动化测试用例,自动化测试用例的入口

接口自动化测试框架
这举个例子，有以下python包
●common
●config
●data
●logging-demo(可以规范一点取名为loggers)
●report
●run
common
common中的内容有：
op_execl.py

Python复制代码

from openpyxl import load_workbook
from common.util import Util
from data.data_config import GlobalVal
import time


class OperateExcel:
    def __init__(self, fn=None):
        if not fn:
            self.fn = Util.get_conf()['testData']
        else:
            self.fn = fn
        self.wb = load_workbook(self.fn)

    # 获取测试用例表格对象
    def get_data(self):
        ws = self.wb.active
        return ws

    # 获取数据的总行数
    def get_lines(self):
        return len(list(self.get_data().rows))

    # 根据行号，列名，获取指定单元格的内容
    def get_value(self, row, col_name):
        col = GlobalVal.get_col_index(col_name)
        return self.get_data().cell(row=row, column=col).value

    # 写入指定内容
    def write_result(self, row, col, value):
        self.get_data().cell(row, col).value = value

    # 保存文件
    def save_file(self):
        now = time.strftime('%Y-%m-%d %H-%M-%S')
        self.wb.save(Util.get_conf()['reportPath'] + now + '.xlsx')


if __name__ == '__main__':
    excel = OperateExcel()
    print(excel.get_lines())
    print(excel.get_value(3, 1))

op_json.py

Python复制代码

import json
from common.util import Util


class OperateJson:

    def __init__(self, json_name):
        self.path = Util.get_conf()[json_name]
        # 以json对象的形式保存测试数据
        self.data = self.read_data()

    def read_data(self):
        with open(self.path, encoding='utf-8') as f:
            return json.load(f)

    # 根据键获取对应的值
    def get_value(self, key):
        return self.data[key]

    # 将字典转化为字符串格式 xxx=yyy&mmmm=nnnn
    def get_value_str(self, dic):
        result = ""
        for k, v in dic.items():
            result += k
            result += '='
            result += v
            result += '&'
        #     去掉最后一个&符号
        return result[:-1]


if __name__ == '__main__':
    op = OperateJson('dataJson')
    print(op.get_value('get_search_success'))
    print(op.get_value_str(op.get_value('get_search_success')))

util.py

Python复制代码

import logging.config

import yaml
from datetime import datetime


class Util:

    @classmethod
    def get_conf(cls):
        with open('../config/conf.yaml', 'r', encoding='UTF8') as f:
            data = yaml.load(f, Loader=yaml.FullLoader)
            return data

    @classmethod
    def get_current_time(cls):
        return datetime.now()


conf_file = Util.get_conf()['loggerConfigPath']
logging.config.fileConfig(conf_file)
logger = logging.getLogger()

if __name__ == '__main__':
    print(Util.get_conf())

get_data.py

Python复制代码

from common.op_execl import OperateExcel
from common.op_json import OperateJson
from common.util import logger


class Getdata:
    def __init__(self):
        self.op_excel = OperateExcel()
        self.data_json = OperateJson('dataJson')
        self.header_json = OperateJson('headerJson')

    # 获取用例行数
    def get_case_lines(self):
        return self.op_excel.get_lines()

    # 获取是否执行
    def get_is_run(self, row):
        value = self.op_excel.get_value(row, 'is_run')
        print(value)
        if value == 'yes':
            return True
        else:
            return False

    # 获取请求类型
    def get_request_method(self, row):
        return self.op_excel.get_value(row, 'request_method')

    # 获取是否有header
    def get_is_header(self, row):
        key = self.op_excel.get_value(row, 'header')
        # print(key)
        if key == 'N/A':
            return None
        else:
            return self.header_json.get_value('httpbin-header')

    # 获取请求地址
    def get_url(self, row):
        return self.op_excel.get_value(row, 'url')

    # 获取请求数据
    def get_request_data(self, row):
        request_key = self.op_excel.get_value(row, 'request_data')
        print('请求数据的key=', request_key)
        if request_key != 'N/A':
            method = self.get_request_method(row)
            request_data = self.data_json.get_value(request_key)
            if method == 'post':
                return request_data
            elif method == 'get':
                return self.data_json.get_value_str(request_data)
        else:
            return None

    # 获取预期结果
    def get_expected_result(self, row):
        return self.op_excel.get_value(row, 'exp_result')

    # 获取用例id
    def get_caseid(self, row):
        return self.op_excel.get_value(row, 'id')

    # 根据caseid来获取指定接口的url，headers,data，method
    def get_case_by_caseid(self, caseid):
        for i in range(1, self.op_excel.get_data().max_row + 1):
            if self.get_caseid(i) == caseid:
                method = self.get_request_method(i)
                url = self.get_url(i)
                headers = self.get_is_header(i)
                request_data = self.get_request_data(i)
                return method, url, headers, request_data
        else:
            logger.error("=====指定的依赖caseid并不存在，请检查======")

    # 获取被依赖的case的id
    def get_depend_caseid(self, row):
        return self.op_excel.get_value(row, 'depend_case')

    # 获取依赖返回字段
    def get_depend_resp_field(self, row):
        return self.op_excel.get_value(row, 'depend_resp_field')

    # 获取数据依赖字段
    def get_depend_field(self, row):
        return self.op_excel.get_value(row, 'depend_field')


if __name__ == '__main__':
    data = GetData()
    print("是否运行=", data.get_is_run(2))
    print("header=", data.get_is_header(2))
    print("url=", data.get_url(2))
    print("请求数据=", data.get_request_data(2))

config
config.yaml

Plain Text复制代码

testdata: ../data/testcases.xlsx
dataJson: ../data/data.json
headerJson: ../data/header.json
reportPath: ../report/
loggerConfigPath: ../config/logger.conf[loggers]

logger.conf

Plain Text复制代码

[loggers]
keys=root,infoLogger

[logger_root]
level=INFO
handlers=consoleHandler,fileHandler

[logger_infoLogger]
handlers=consoleHandler,fileHandler
qualname=infoLogger
propagate=0

[handlers]
keys=consoleHandler,fileHandler

[handler_consoleHandler]
class=StreamHandler
level=INFO
formatter=form01
args=(sys.stdout,)

[handler_fileHandler]
class=FileHandler
level=INFO
formatter=form01
args=('../logging-demo/run2.log','a')

[formatters]
keys=form01

[formatter_form01]
format=%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s

data
data.json

JSON复制代码

{
  "get_search_success": {
    "kw": "福佳白",
    "name": "hali"
  },
  "get_search_failed": {
    "kw": "1664"
  },
  "ip": {
    "ip": "8.8.8.8",
    "key": "c32475bbde59d9a6881b92f22a1fb0d6"
  },
  "get_isp": {
    "isp-no": "",
    "username": "heli"
  }
}

data-config.py

Python复制代码

class GlobalVal:

    # 定义每列的下标
    @classmethod
    def get_col_index(cls, col):
        dic = {
            "id": 1,
            "module": 2,
            "url": 3,
            "is_run": 4,
            "request_method": 5,
            "header": 6,
            "depend_case": 7,
            "depend_resp_field": 8,
            "depend_field": 9,
            "request_data": 10,
            "exp_result": 11,
            "act_result": 12,
            "test_result": 13,
            "exec_time": 14
        }
        return dic[col]

header.json

JSON复制代码

{
  "httpbin-header": {
    "User-Agent": "Chrom",
    "Original": "http://www.baidu.com"
  }
}

testcases.xlsx

testcases.xlsx(10 KB)

logging-demo
存放日志文件的

Python复制代码

import logging

# DEBUG/INFO/WARNNING/ERROR/CRITICAL 常见的日志级别
# 只显示日志级别以上的日志，如果不配置，默认是warning及以上的日志
logging.basicConfig(level=logging.INFO, filename='run.log',
                    format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
logging.info('hello')

# handler处理器

上面的logger1.py文件，这个是跑日志的demo，这种只能在当前文件中使用logger的指定格式。需要全局使用，还是得用logger.conf的使用。使用import方式，导入logger，这种方式已经在run_main.py文件中使用。

Python复制代码

conf_file = Util.get_conf()['loggerConfigPath']
logging.config.fileConfig(conf_file)
logger = logging.getLogger()

report
这个文件存放，接口测试用例跑完过后，生成的测试结果的用例Excel文件

run
run_method.py

Python复制代码

import requests
import re


class RunMethod:

    @classmethod
    def get_method(cls, url, data=None, headers=None):
        if data:
            # 如果指定了参数，则将参数加在url后面，?连接
            url = url + '?' + data
        if headers:
            return requests.get(url, headers=headers)
        else:
            return requests.get(url)

    @classmethod
    def post_method(cls, url, data, headers=None):
        if headers:
            return requests.post(url, data, headers)
        else:
            return requests.post(url, data)

    # 从返回内容中，获取指定字段的值
    @classmethod
    def get_spec_field_from_resp(cls, field, text):
        result = re.findall('"' + field + '":"(.*?)"', text)
        if result:
            return result[0]


if __name__ == '__main__':
    text = '{"resultcode":"200","reason":"查询成功","result":{"Country":"中国","Province":"江苏省","City":"无锡市","Isp":"电信"},"error_code":0}'
    print(RunMethod.get_spec_field_from_resp('Isp', text))

run_main.py

Python复制代码

from common.get_data import GetData
from run.run_method import RunMethod
from common.op_execl import OperateExcel
from data.data_config import GlobalVal
from common.util import Util
from common.util import logger


class RunMain:

    def __init__(self):
        self.data = GetData()
        self.op_execl = OperateExcel()

    def run_test(self):
        # 获取用例总行数
        row_count = self.data.get_case_lines()
        print(row_count)
        act_result_col = GlobalVal.get_col_index('act_result')
        test_result_col = GlobalVal.get_col_index('test_result')
        exec_time_col = GlobalVal.get_col_index('exec_time')
        # 第一行是列名，就不遍历了，从第二行开始遍历
        for i in range(2, row_count + 1):
            logger.info('=====当前执行第{}条用例========'.format(i - 1))
            is_run = self.data.get_is_run(i)
            if is_run:
                # 依赖的用例id
                depend_caseid = self.data.get_depend_caseid(i)
                if depend_caseid:
                    depend_method, depend_url, depend_headers, depend_request_data = self.data.get_case_by_caseid(
                        depend_caseid)
                    if depend_method == 'get':
                        resp = RunMethod.get_method(depend_url, depend_request_data, depend_headers)
                    else:
                        resp = RunMethod.get_method(depend_url, depend_request_data, depend_headers)
                    # 从用例中获取依赖的返回值的字段名
                    depend_resp_field = self.data.get_depend_resp_field(i)
                    # 从依赖用例返回值中提取指定字段的值
                    depend_resp_field_value = RunMethod.get_spec_field_from_resp(depend_resp_field, resp.text)
                    # 将上步获取的值添加到请求数据的json中
                    request_data = self.data.get_request_data(i)
                    depend_field = self.data.get_depend_field(i)
                    request_data[depend_field] = depend_resp_field_value
                else:
                    request_data = self.data.get_request_data(i)
                url = self.data.get_url(i)
                header = self.data.get_is_header(i)
                request_method = self.data.get_request_method(i)
                expected_result = self.data.get_expected_result(i)
                result = None
                if request_method == 'get':
                    # 请求的实际结果
                    result = RunMethod.get_method(url, request_data, header)
                else:
                    result = RunMethod.post_method(url, request_data, header)
                try:
                    text = result.text
                except Exception:
                    text = 'None'

                # 写入实际测试结果
                self.op_execl.write_result(i, act_result_col, text)
                # 判断测试结果
                if ',' in expected_result:
                    exp_result_list = expected_result.split(',')
                    for item in exp_result_list:
                        if '{' in item:
                            # 期望结果中有变量的情况，将变量赋值
                            var_field = item[item.index('{') + 1:item.index('}')]
                            item = item.split(':')[0] + ': "' + request_data[var_field] + '"'
                        if item not in text:
                            print('测试失败，item:{}没在实际结果:{}里'.format(item, result))
                            self.op_execl.write_result(i, test_result_col, '测试失败')
                            break
                    else:
                        self.op_execl.write_result(i, test_result_col, '测试通过')
                else:
                    if '{' in expected_result:
                        # 期望结果中有变量的情况，将变量赋值
                        var_field = expected_result[expected_result.index('{') + 1:expected_result.index('}')]
                        expected_result = expected_result.split(':')[0] + ': "' + request_data[var_field] + '"'
                    if expected_result not in text:
                        print('测试失败，expected_result:{}没在实际结果:{}里'.format(expected_result, result))
                        self.op_execl.write_result(i, test_result_col, '测试失败')
                    else:
                        self.op_execl.write_result(i, test_result_col, '测试通过')
            else:
                self.op_execl.write_result(i, act_result_col, 'N/A')
                self.op_execl.write_result(i, test_result_col, 'N/A')
            self.op_execl.write_result(i, exec_time_col, Util.get_current_time())
        # 保存文件
        self.op_execl.save_file()


if __name__ == '__main__':
    run = RunMain()
    print(run.run_test())

temp.py
手动的临时测试代码

Python复制代码

import requests

headers = {
    'Token': 'gkhodjsgk1'
}
resp = requests.get('http://httpbin.org/headers', headers=headers)
print(resp.text)

url = 'http://apis.juhe.cn/ip/ipNew'

params = {
    "ip": "58.215.15.114",  # 查询的IP地址
    "key": "c32475bbde59d9a6881b92f22a1fb0d6",  # 您申请的接口API接口请求Key
}

resp = requests.get(url, params=params)
print(resp.text)

data = {
    'isp-no': 'Level3',
    'username': 'heli'
}
resp = requests.post('https://httpbin.org/post', data=data)
print(resp.text)

 如果对python自动化测试、web自动化、接口自动化、移动端自动化、大型互联网架构技术、面试经验交流等等感兴趣的老铁们，可以关注我。我会在公众号（程序员阿沐）/群里（810119819）不定期的发放免费的资料链接，这些资料都是从各个技术网站搜集、整理出来的，如果你有好的学习资料可以私聊发我，我会注明出处之后分享给大家。欢迎分享，欢迎评论，欢迎转发。需要资料的同学可以关注我获取资料链接。