python+playwright+页面对象模型Page Object Models

一，项目目录：

image.png

二，安装环境

python3.7 以上
# 安装playwright
pip install playwright -i https://pypi.tuna.tsinghua.edu.cn/simple/ --trusted-host pypi.tuna.tsinghua.edu.cn
playwright install
# 安装插件
pip install pytest
pip install pytest-playwright
pip install allure-pytest
pip install pytest-ordering

allure下载地址：https://repo.maven.apache.org/maven2/io/qameta/allure/allure-commandline/

三，代码实现

pytest.ini

[pytest]
log_cli = 1
log_cli_level = INFO
log_cli_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
log_cli_date_format = %Y-%m-%d %H:%M:%S
log_file = uitest.log
log_file_level = INFO
log_file_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
log_file_date_format=%Y-%m-%d %H:%M:%S

python_files = test_*.py  Test_*.py
python_classes = Test*
python_functions = test_* Test_*


# 设置默认执行命令
addopts = -v --alluredir=./report/xml --clean-alluredir --headed  --browser=webkit

ps：--base-url可添加baseurl，如：--base-url=https://www.baidu.com/

conftest.py

import logging
import os
import time
import pytest
import allure
from playwright.sync_api import BrowserType
from typing import Dict


# 持久上下文，只打开一个上下文
@pytest.fixture(scope="session")
def context(
        browser_type: BrowserType,
        browser_type_launch_args: Dict,
        browser_context_args: Dict
):
    context = browser_type.launch_persistent_context('./auth', **{
        **browser_type_launch_args,
        **browser_context_args,
        "locale": "de-DE",
    })
    yield context
    context.close()


@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):
    """用例执行失败则添加失败截图至allure"""
    print('------------------------------------')
    # 获取钩子方法的调用结果
    outcome = yield
    rep = outcome.get_result()
    # 仅仅获取用例call 执行结果是失败的情况, 不包含 setup/teardown
    if rep.when == "call" and rep.failed:
        page = item.funcargs["page"]
        # 添加allure报告截图
        with allure.step('添加失败截图...'):
            allure.attach(page.screenshot(), "失败截图", allure.attachment_type.PNG)


def pytest_terminal_summary(terminalreporter, exitstatus, config):
    """收集测试结果"""
    # print(terminalreporter.stats)
    total = terminalreporter._numcollected
    passed = len([i for i in terminalreporter.stats.get('passed', []) if i.when != 'teardown'])
    failed = len([i for i in terminalreporter.stats.get('failed', []) if i.when != 'teardown'])
    error = len([i for i in terminalreporter.stats.get('error', []) if i.when != 'teardown'])
    skipped = len([i for i in terminalreporter.stats.get('skipped', []) if i.when != 'teardown'])
    successful = len(terminalreporter.stats.get('passed', [])) / terminalreporter._numcollected * 100

    duration = time.time() - terminalreporter._sessionstarttime
    print('total times: %.2f' % duration, 'seconds')

    with open("result.txt", "w") as fp:
        fp.write(r"测试结果如下：" + "\n")
        fp.write("TOTAL=%s" % total + "\n")
        fp.write("PASSED=%s" % passed + "\n")
        fp.write("FAILED=%s" % failed + "\n")
        fp.write("ERROR=%s" % error + "\n")
        fp.write("SKIPPED=%s" % skipped + "\n")
        fp.write("SUCCESSFUL=%.2f%%" % successful + "\n")
        fp.write("TOTAL_TIMES=%.2fs" % duration)

BasePage.py

import logging
import os
from playwright.sync_api import Page, sync_playwright
from common.conf_yaml import get_conf, root_path
from common.log import setup_rotating_log

BaseUrl = get_conf("baseurl")["baseurl"]


class BasePage:
    _baseurl = ""
    logging = setup_rotating_log()

    def __init__(self, page=None):
        """
        :param page:
        :param on: 是否需要登陆
        """
        self.title = get_conf("title")["title"]
        self._state_file = os.path.join(root_path(), get_conf("state_file")["state_file"])
        if page is None:  # pytest-playwright 执行时不会从这里启动浏览器，这里是为了编写po对象时方便调试
            logging.info("page" + str(page))
            self.p = sync_playwright().start()
            self.browser = self.p.chromium.launch(headless=False)
            if os.path.isfile(self._state_file):
                self.context = self.browser.new_context(storage_state=self._state_file, base_url=BaseUrl)
            else:
                self.context = self.browser.new_context(base_url=BaseUrl)
            self.page: Page = self.context.new_page()
            logging.info(self._baseurl)
            self.page.goto(self._baseurl)
            if self.title not in self.page.title():
                self.login()
                logging.info("现在登陆成功了吗？%s" % (self.page.title()))
                self.context.storage_state(path=self._state_file)
        else:
            self.page: Page = page
        if not self.page.url.lower().startswith('http'):
            # logging.info(self._baseurl)
            self.page.goto(self._baseurl)
        if self.title not in self.page.title():
            self.login()
            logging.info("现在登陆成功了吗？%s" % (self.page.title()))
            # logging.info(self.page)

    def quit(self):
        try:
            self.context.storage_state(path=self._state_file)
            self.context.close()
            self.browser.close()
            self.p.stop()
        except Exception as e:
            print(e)

    def login(self):
        """
        登陆
        :return:
        """
        self.page.locator('#username').fill(get_conf("username")["username"])
        self.page.locator('#password').fill(get_conf("password")["password"])
        self.page.click('button.login-button')
        logging.info("进行登录")

if __name__ == '__main__':
    pg = BasePage()

页面Page（示例）

import logging
import os.path

from playwright.sync_api import expect

from common.conf_yaml import root_path
from page.BasePage import BasePage
from page.HealthDetailPage import HealthDetailPage


class HealthPage(BasePage):
    _baseurl = "/manage/health"

    def add_health(self, data):
        """
        点击接入健康拨测任务
        :param data: 接入拨测任务时需要输入的数据：{"jobName": "测试任务", "systemId": "T-mPTS", "env": "SIT", "probeType": "ICMP", "url": "127.0.0.2", "frequency": "5min",
            "point": ["顺德（内网）"], "remake": "添加健康拨测任务成功"}
        :return:
        """
        logging.info("添加拨测任务")
        self.page.click("button.ant-btn.ant-btn-primary.ant-btn-two-chinese-chars")
        self.page.wait_for_timeout(1000)
        self.page.locator("#jobName").fill(data['jobName'])
        if "systemId" in data and data["systemId"]:
            self.page.locator("#systemId").fill(data['systemId'])
            self.page.get_by_title(data['systemId']).click()
            if "env" in data and data["env"]:
                self.page.locator("#env").fill(data['env'])
                self.page.get_by_title(data['env']).click()
        if "probeType" in data and data["probeType"]:
            self.page.locator("#probeType").fill(data['probeType'])
            self.page.get_by_title(data['probeType']).click()
        self.page.locator("#url").fill(data['url'])
        if "frequency" in data and data["frequency"]:
            self.page.locator("#frequency").fill(data['frequency'])
            self.page.get_by_title(data['frequency'] + ' / 次', exact=True).click()
        if "point" in data and data["point"]:
            self.page.locator("#point").click()
            for e in data['point']:
                self.page.get_by_title(e).click()
        self.page.locator("#jobName").click()
        if "remake" in data and data["remake"]:
            self.page.locator("#remark").fill(data["remake"])
        self.page.locator("div.ant-space-item>button.ant-btn.ant-btn-primary>span").click()

    def add_health_fail(self, data):
        self.add_health(data)
        logging.info("添加拨测任务失败")
        ele_list = self.page.query_selector_all("div.ant-form-item-explain-error")
        errors = [e.text_content() for e in ele_list]
        self.page.get_by_role("button", name="取 消").click()
        return errors

    def get_health_names(self):
        """获取当前列表的监控名称列表"""
        logging.info("获取当前列表的监控名称列表")
        health_name_ele = "td.ant-table-cell.ant-table-cell-fix-left:nth-child(2)>div"
        self.page.wait_for_selector(health_name_ele)
        ele_list = self.page.query_selector_all(health_name_ele)
        # logging.info(ele_list)
        health_names = [e.text_content() for e in ele_list]
        logging.info(health_names)
        return health_names

    def get_health_address(self):
        """获取当前列表的拨测地址列表"""
        logging.info("获取当前列表的拨测地址列表")
        health_address_ele = "td.ant-table-cell.ant-table-cell-fix-left:nth-child(3)"
        self.page.wait_for_selector(health_address_ele)
        ele_list = self.page.query_selector_all(health_address_ele)
        # logging.info(ele_list)
        health_addresses = [e.text_content() for e in ele_list]
        logging.info(health_addresses)
        return health_addresses

    def goto_health_details(self, data):
        """
        点击进入对应健康拨测详情
        :param data:
        :return:
        """
        self.search_health(data)
        logging.info("进入对应健康拨测详情")
        self.redo()
        self.page.wait_for_timeout(1000)
        self.page.get_by_text("详情", exact=True).click()
        return HealthDetailPage(self.page)


if __name__ == '__main__':
    data = {"jobName": "测试任务", "systemId": "T-mPTS", "env": "SIT", "probeType": "ICMP", "url": "127.0.0.2",
            "frequency": "5min",
            "point": ["顺德（内网）"]}
    hp.add_health(data)

用例放在testcase文件夹下：

import logging
import allure
import pytest
from playwright.sync_api import Page

from page.HeathPage import HealthPage


@allure.feature("健康拨测模块")
@allure.story("测试健康拨测模块")
class Test_HealthCheck:

    @pytest.fixture(autouse=True)
    def setup(self, page: Page):
        self.page = page
        yield
        self.page.close()

    @pytest.mark.run(order=0)
    @allure.title("用例标题：test_add_health_success")
    @pytest.mark.parametrize('data', [
        {"jobName": "UI自动化测试任务", "systemId": "T-mPTS", "env": "SIT", "probeType": "ICMP", "url": "127.0.0.2",
         "frequency": "5min", "point": ["顺德（内网）"], "status": "下发中"}])
    def test_add_health_success(self, data):
        """
        添加拨测任务成功
        :param data: 用例数据
        :return:
        """
        healthpage = HealthPage(self.page)
        with allure.step("1.添加健康拨测任务成功;2.点击刷新；3.验证添加信息匹配(断言拨测名称，拨测地址，添加成功任务状态为下发中)"):
            healthpage.add_health(data)
            healthpage.redo()
            assert data['jobName'] in healthpage.get_health_names()
            assert data['url'] in healthpage.get_health_address()
            assert data['status'] in healthpage.get_health_status()

    # @pytest.mark.skip
    @pytest.mark.run(order=2)
    @allure.title("用例标题：test_add_health_fail")
    @pytest.mark.parametrize('data', [
        {"jobName": "", "systemId": "T-mPTS", "env": "SIT", "probeType": "ICMP", "url": "127.0.0.1",
         "frequency": "5min", "point": ["顺德（内网）"], "error": "监控名称不能为空"},
        {"jobName": "UI自动化测试任务", "systemId": "", "env": "SIT", "probeType": "ICMP", "url": "127.0.0.1",
         "frequency": "5min", "point": ["顺德（内网）"], "error": "归属系统不能为空"},
        {"jobName": "UI自动化测试任务", "systemId": "T-mPTS", "env": "", "probeType": "ICMP", "url": "127.0.0.1",
         "frequency": "5min", "point": ["顺德（内网）"], "error": "环境不能为空"},
        {"jobName": "UI自动化测试任务", "systemId": "T-mPTS", "env": "SIT", "probeType": "", "url": "127.0.0.1",
         "frequency": "5min", "point": ["顺德（内网）"], "error": "拨测类型不能为空"},
        {"jobName": "UI自动化测试任务", "systemId": "T-mPTS", "env": "SIT", "probeType": "ICMP", "url": "",
         "frequency": "5min", "point": ["顺德（内网）"], "error": "例如：10.1.1.1或onemonitorsit.midea.com"},
        {"jobName": "UI自动化测试任务", "systemId": "T-mPTS", "env": "SIT", "probeType": "ICMP", "url": "127.0.0.1",
         "frequency": "5min", "error": "拨测点不能为空"}
    ])
    def test_add_health_fail(self, data):
        """
        验证添加用例健康拨测失败场景
        :param data:
        :return:
        """
        healthpage = HealthPage(self.page)
        with allure.step("1.添加健康拨测任务失败;2.提示验证"):
            assert data['error'] in healthpage.add_health_fail(data)

common/log.py

import logging
import logging.handlers
from common import conf_yaml
import os


def setup_rotating_log(log_file="uitest.log"):
    """用于调试页面page时打印日志"""
    log_path_file = os.path.join(conf_yaml.root_path(), log_file)
    # 定义日志格式
    log_formatter = logging.Formatter('%(asctime)s [%(levelname)6s] %(message)s (%(filename)s:%(lineno)s)',
                                      datefmt='%Y-%m-%d %H:%M:%S')

    max_log_size = 1 * 1024 * 1024 * 1024  # 1G
    backup_count = 2

    # 创建滚动文件处理器
    rotating_handler = logging.handlers.RotatingFileHandler(
        filename=log_path_file, maxBytes=max_log_size, backupCount=backup_count)
    rotating_handler.setFormatter(log_formatter)

    # 创建控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(log_formatter)

    # 创建日志记录器
    logger = logging.getLogger()
    logger.addHandler(rotating_handler)  # 添加滚动文件处理器
    logger.addHandler(console_handler)  # 添加控制台处理器
    logger.setLevel(logging.INFO)

    return logger

common/conf_yaml.py

import os
import yaml

def get_conf(*key):
    """
    获取配置
    :param key:需要获取的配置名称
    :return:
    """
    my_conf = os.path.join(root_path(), r'auth\my_conf.yml')
    # print(my_conf)
    with open(my_conf, encoding='utf-8') as f:
        res = yaml.load(f.read(), Loader=yaml.FullLoader)
        res_dict = {}
        for e in key:
            if e in res:
                res_dict[e] = res[e]
        return res_dict


def root_path():
    """返回工程主路径"""
    if str(os.path.dirname(os.path.realpath(__file__))).endswith('common'):
        path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
    else:
        path = os.path.dirname(os.path.realpath(__file__))
    # print(path)
    return path

配置文件放在auth下：

image.png

四，执行用例后查看
执行结果：

image.png

五，查看allure测试报告：

image.png