一：框架构思

（此代码只作为简单演示使用，因为好多问题没有考虑到，时间有限，没有做参数化，没有重跑机制，代码规范等等，请各位仅供参考。）

base:是基于seleniium的二次封装的点击、输入、刷新等操作
common:是基于业务的底层公共方法
config:配置文件
log:收集log的方法，以及生成的截图
excute_logs:生成的日志都会打印在一个文件
page_object:webui登录的方法和一些二次封装的方法
testcase:是testcase
reports：是生成reports的方法和生成的报告

image.png

二：框架代码展示

base_page.py文件

  1. # coding=utf-8
  4. import time
  5. from time import sleep
  6. from log.getLogStream import logStream
  8. log = logStream()
  11. # 创建基类
  12. class BasePage:
  13.     # driver = webdriver.Chrome()
  14.     # 构造函数
  15.     def __init__(self, driver):
  16.         log.info('初始化driver{}'.format(driver))
  17.         self.driver = driver
  19.     # 访问URL
  20.     def open(self, url):
  21.         """
  22.         function: 打开浏览器，访问url
  23.         description:
  24.         arg:
  25.         return:
  26.         """
  27.         log.info('访问网址')
  28.         self.driver.get(url)
  29.         self.driver.maximize_window()
  30.         sleep(3)
  32.     # 元素定位
  33.     def locator(self, loc):
  34.         """
  35.         function: 定位元素
  36.         description:
  37.         arg:
  38.         return:
  39.         """
  40.         log.info('正在定位{}元素'.format(loc))
  41.         return self.driver.find_element(*loc)
  43.     # 输入
  44.     def input_(self, loc, txt):
  45.         """
  46.         function: 输入
  47.         description:
  48.         arg:
  49.         return:
  50.         """
  51.         try:
  52.             log.info('正在定位{}元素， 输入{}内容'.format(loc, txt))
  53.             self.locator(loc).send_keys(txt)
  54.             sleep(2)
  55.         except Exception as e:
  56.             self.screenShot()
  57.             log.error('错误日志' % e)
  59.     # 点击
  60.     def click(self, loc):
  61.         """
  62.         function: 点击
  63.         description:
  64.         arg:
  65.         return:
  66.         """
  67.         try:
  68.             log.info('正在点击{}元素'.format(loc))
  69.             self.locator(loc).click()
  70.         except Exception as e:
  71.             self.screenShot()
  72.             log.error('错误日志' % e)
  74.     # 等待
  75.     def wait(self, time_):
  76.         """
  77.         function: 等待
  78.         description:
  79.         arg:
  80.         return:
  81.         """
  82.         log.info('等待时间{}秒'.format(time_))
  83.         sleep(time_)
  85.     # 关闭
  86.     def quit(self):
  87.         """
  88.         function: 退出
  89.         description:
  90.         arg:
  91.         return:
  92.         """
  93.         log.info('退出')
  94.         self.driver.quit()
  96.     # 最大化
  97.     def maxSize(self):
  98.         """
  99.         function: 最大化
  100.         description:
  101.         arg:
  102.         return:
  103.         """
  104.         log.info('最大化')
  105.         self.driver.maximize_window()
  107.     # 截图并保存
  108.     def screenShot(self):
  109.         """
  110.         function: 绑定服务器
  111.         description: bind服务器
  112.         arg:
  113.         return:
  114.         """
  115.         current_time = time.strftime('%Y-%m-%d %H-%M-%S')
  116.         print(current_time)
  117.         pic_path = '../log/screenshot' + '/' + current_time + '.png'
  118.         self.driver.save_screenshot(pic_path)
  120.     # 关闭浏览器
  121.     def close(self):
  122.         """
  123.         function: 关闭当前浏览器
  124.         description:
  125.         arg:
  126.         return:
  127.         """
  128.         self.driver.close()
  130.     # 刷新浏览器
  131.     def refresh(self):
  132.         """
  133.         function: 刷新浏览器
  134.         description:
  135.         arg:
  136.         return:
  137.         """
  138.         self.driver.refresh()
  140.     def waitUntilPageContains(self, message):
  141.         """
  142.         function: 等待界面出现某个字段
  143.         description:
  144.         arg:
  145.         :return:
  146.         example: self.waitUntilPageContains('vsite_automation')
  147.         """
  148.         log.info('正在获取页面%s字段' % message)
  149.         sleep(3)
  150.         msg = self.driver.find_element_by_xpath('//*[contains(text(), "%s")]' % message).text
  151.         print(msg)
  152.         return msg

common目录下的业务文件

  1. import paramiko
  2. from time import sleep
  3. import re
  6. class CLI:
  8.     def ssh_ag(self):
  9.         """
  11.         :param self:
  12.         :return:
  13.         """
  14.         # 创建ssh对象
  15.         self.ssh = paramiko.SSHClient()
  16.         # 允许连接不在know_hosts文件中的主机
  17.         self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)
  18.         # 连接AG
  19.         self.ssh.connect(hostname='192.168.120.220', port=22, username='gaojs', password='123')
  20.         sleep(5)
  21.         channel = self.ssh.invoke_shell()
  22.         self.channel = channel
  23.         channel.settimeout(5)
  24.         sleep(5)
  25.         self.cli_cmd('enable')
  26.         self.cli_cmd('')
  27.         self.cli_cmd('config ter')
  29.     def print_step(self):
  30.         """
  32.         :return:
  33.         """
  34.         result = self.channel.recv(2048)
  35.         print(result.decode())

getLogStream.py收集日志文件

  2. import logging
  5. def logStream():
  6.     # 创建一个日志器
  7.     logger = logging.getLogger()
  8.     # 设置日志级别为info
  9.     logger.setLevel(logging.INFO)
  10.     # 日志想要输出到哪，就要制定输出目的地，控制台   要创建一个控制台处理器
  11.     console = logging.StreamHandler()
  12.     logger.addHandler(console)
  14.     # 创建一个格式器
  15.     # 设置日志格式
  16.     fmt = '%(asctime)s %(filename)s %(levelname)s %(module)s %(funcName)s %(message)s'
  17.     # 生成日志信息 生成时间  文件名   日志的状态  类名   方法名  日志内容
  18.     fomator = logging.Formatter(fmt)
  19.     # 优化及控制台格式
  20.     console.setFormatter(fomator)
  22.     # 创建一个处理器   文本处理器   制定日期信息输出到文本处理器
  23.     filehandler = logging.FileHandler('../excute_logs/logger.log', encoding='utf-8')
  24.     logger.addHandler(filehandler)
  26.     # 设置logger.log文本格式
  27.     filehandler.setFormatter(fomator)
  28.     return logger

base_logging.py文件

  1. import logging
  4. def logger():
  5.     # 设置日志格式
  6.     fmt = '%(asctime)s %(filename)s %(levelname)s %(module)s %(funcName)s %(message)s'
  7.     # 生成日志信息 生成时间  文件名   日志的状态  类名   方法名  日志内容
  8.     # 设置日志级别
  9.     logging.basicConfig(level=logging.INFO, format=fmt, filename='../excute_logs/log.log')
  10.     return logging

page_object目录下的login_page.py文件

  1. from time import sleep
  3. from selenium.webdriver.common.by import By
  4. from base.base_page import BasePage
  5. from selenium import webdriver
  8. class LoginPage(BasePage):
  10.     def login(self, username, password):
  11.         """
  12.         description:登录webui界面
  13.         :param username:
  14.         :param password:
  15.         :return:
  16.         example:
  17.             self.login('array', 'admin')
  18.         """
  19.         self.driver = webdriver.Chrome()
  20.         # URL
  21.         # self.url = "https://%s:%s" % ip, port
  22.         self.url = "https://192.168.120.209:8888"
  23.         # 页面元素
  24.         self.user = (By.NAME, "username")
  25.         self.passwd = (By.ID, 'password')
  26.         self.button = (By.ID, 'loginID')
  27.         # 访问URL,最大化
  28.         self.open(self.url)
  29.         # 点击页面上不是隐私连接提示
  30.         try:
  31.             self.waitUntilPageContains('不是私密连接')
  32.             self.driver.find_element_by_xpath('//button[@id="details-button"]').click()
  33.             sleep(1)
  34.             self.driver.find_element_by_xpath('//a[@id="proceed-link"]').click()
  35.         except:
  36.             pass
  37.         # 输入账号
  38.         sleep(5)
  39.         self.input_(self.user, username)
  40.         # 输入密码
  41.         self.input_(self.passwd, password)
  42.         # 点击登录按钮
  43.         self.click(self.button)
  44.         sleep(5)
  46.     def login_L3vpn_test(self, ip, method, username, password, challenge=None, challenge_passwd1=None, challenge_passwd2=None):
  47.         """
  48.         function:登录L3vpn
  49.         :argument:
  50.             ip: 虚拟站点IP
  51.             method:方法名
  52.             username:用户名
  53.             password:密码
  54.         :return:
  55.         examlpe:
  56.             self.login_L3vpn_test('192.168.120.x', 'http_challenge', 'array', 'admin')
  57.             self.login_L3vpn_test(self.vsiteip, 'http_challenge_test', self.username, self.passwd,
  58.                                     challenge=True, challenge_passwd1='chal1', challenge_passwd2='chal2')
  59.         """
  60.         self.driver = webdriver.Chrome()
  61.         # URL
  62.         self.url = "https://%s" % ip
  63.         # 页面元素
  64.         self.user = (By.NAME, "uname")
  65.         self.passwd = (By.NAME, 'pwd')
  66.         self.button = (By.NAME, 'submitbutton')
  67.         self.select_method = (By.NAME, "method")
  68.         self.challenge_signin = (By.NAME, "option")
  69.         # 访问URL,最大化
  70.         self.open(self.url)
  71.         sleep(5)
  72.         try:
  73.             self.driver.find_element_by_xpath('//button[@id="details-button"]').click()
  74.             sleep(1)
  75.             self.driver.find_element_by_xpath('//a[@id="proceed-link"]').click()
  76.         except:
  77.             pass
  78.         # 切换方法
  79.         self.click(self.select_method)
  80.         self.driver.find_element_by_xpath('//option[@value="%s"]' % method).click()
  81.         # 输入账号
  82.         sleep(5)
  83.         self.input_(self.user, username)
  84.         # 输入密码
  85.         self.input_(self.passwd, password)
  86.         # 点击登录按钮
  87.         self.click(self.button)
  88.         sleep(5)
  89.         # 挑战模式
  90.         if challenge:
  91.             try:
  92.                 self.input_(self.passwd, challenge_passwd1)
  93.                 self.click(self.challenge_signin)
  94.                 self.input_(self.passwd, challenge_passwd2)
  95.                 self.click(self.challenge_signin)
  96.                 self.waitUntilPageContains('welcome to the ArrayOS')
  97.             except Exception as e:
  98.                 print(e)
  99.                 print('挑战失败，请重试！')
  100.         else:
  101.             print('不符合挑战条件，请检查配置！')

reports目录下的testReports.py文件

import time
import unittest
from BeautifulReport import BeautifulReport


# 找到用例defaultTestLoader默认加载
# import HTMLTestRunner


def testReports():
    """
    function: 生成测试报告方法
    description: 生成测试报告
    arg:
    return:
    """
    case_dir = '../testcase/aaa_http/'
    discover = unittest.defaultTestLoader.discover(case_dir, 'test*.py')

    # 用时间命名测试报告   测试报告生成时间  +  后缀名   2021-11-20 14-49-30test_report.html
    report_dir = '../reports/'
    now = time.strftime('%Y-%m-%d %H-%M-%S')
    report_name = report_dir + '/' +now+ '_test_report.html'

    with open(report_name, 'wb') as f:
        # 执行用例
        # HTMLTestRunner.HTMLTestRunner(stream=f, verbosity=2, title='unittest测试报告练习', description='练习HTMLTestRunner使用').run(discover)
        BeautifulReport(discover).report(description=u'UAG每日构建测试报告', filename=report_name, report_dir='../reports/')

run.py文件：

from reports.testReports import testReports


if __name__ == "__main__":
    testReports()

三：用例代码

testcase目录下的test_01文件

import unittest
from common.agCli import *
from page_object.login_page import *


class Testcase(unittest.TestCase, LoginPage, CLI):

    def setUp(self) -> None:
        self.ssh_ag()
        self.vsitename = 'vsite_automation'
        self.username = 'array'
        self.passwd = 'admin'
        self.message = 'vsite_automation'

    def test_01(self):
        """
        description:  cli创建虚拟站点，登录webui去查看是否创建成功，是否有vsite_automation虚拟站点信息
        :return:
        date: 2022/02/17
        author: gaojs
        """
        self.cli_cmd('virtual site name vsite_automation')
        self.login(self.username, self.passwd)
        self.switch_vsite(self.vsitename)
        msg = self.waitUntilPageContains(self.message)
        print(msg)
        if msg not in(self.message):
            raise Exception('切换虚拟站点失败，请重试！')

    # 恢复环境
    def tearDown(self) -> None:
        self.cli_cmd('no virtual site name vsite_automation')
        self.cli_cmd('YES')
        self.quit_enable()
        self.close()

压力测试用例test_02：

from locust import HttpUser, between, task, TaskSet
import os
from common.agCli import *
import logging


class TaskTest(TaskSet, CLI):

    # 执行并发前置动作，比如清理当前所有session
    def on_start(self):
        """
        description:登录ag, 清理log
        :return:
        """
        self.ssh_ag()
        self.clear_log()
        logging.info('清理log结束，压测开始！！！')

    # 压测任务，也可以是@task(10)啥的，这个数字是代表权重,数值越大,执行的频率就越高
    @task
    def login(self):
        url = '/prx/000/http/localh/login'
        data = {
            "method": "http1",
            "uname": "gaojs",
            "pwd1": "",
            "pwd2": "",
            "pwd": "admin",
            "submitbutton": "Sign"
        }
        header = {"Content-Type": "application/json;charset=UTF-8"}
        self.client.request(method='POST', url=url, data=data, headers=header, name='登录虚拟站点', verify=False, allow_redirects=False)

    # 执行并发测试后执行的动作，比如保存log等操作,查看报告http://localhost:8089/
    def on_stop(self):
        self.ssh_ag()
        self.cli_cmd('switch vsite')
        self.cli_cmd('session kill all')
        logging.info('清理session结束，压测结束，请查看report, http://localhost:8089！！！')


class Login(HttpUser):
    host = 'https://192.168.120.206'
    # 每次请求停顿时间
    wait_time = between(1, 3)
    tasks = [TaskTest]


if __name__ == "__main__":
    os.system("locust -f locust_test.py --host=https://192.168.120.206")

四：测试报告

生成报告展示：
image.png

image.png