Pytest自动化测试

一. Pytest默认测试用例的规则以及基础应用

1.规则

模块名必须以test_开头或者_test结尾
测试类名必须以Test开头，而且不能带有init方法
测试用例必须以test_开头

2.执行方式

通过命令行方式执行：pytest
执行的参数：
     pytest -vs     -v：输出详细信息    -s：输出调试信息
     pytest -vs -n     多线程运行(前提安装插件：pytest-xdist)
     pytest -vs - -reruns=2     失败重跑(前提安装插件：pytest-rerunfailres)
     pytest -vs -x     出现一个用例失败则停止测试
     pytest -vs - -maxfail=2     出现几个失败才终止
     pytest -vs - -html     生成html的简易测试报告(前提是安装插件：pytest-html)
     pytest -vs -k “get_token”     运行测试用例名称中包含某个字符串的测试用例(get_token是用例的类名)
通过主函数main的方式执行

1 2	if __name__ == '__main__': pytest.main(["-vs"])

通过全局配置文件pytest.ini执行
注意：
     1.一般放在项目的根目录下，名称必须是pytest.ini
     2.配置文件中当有中文时需要改变编码格式为GB2312
     3.pytest.ini文件可以改变默认的测试用例规则
     4.不管是命令行运行还是主函数运行，都会加载这个配置文件

[pytest]
# 参数 其中 -m "smoke" 表示只执行被标记的smoke用例，结合下面的[标记]使用
addopts = -vs -m "smoke"
# 指定执行测试用例的文件夹位置 
testpaths = ./test_case
# 执行指定名称开头的测试用例(如下，只执行文件名为ms开头的py文件测试用例)
python_files = ms_*.py
# 执行指定测试类的测试用例(如下，只执行类名为Test开头的测试用例)
python_classes = Test*
# 执行指定函数的测试用例(如下，只执行函数为test开头的测试用例)
python_functions = test_*

# 标记(在函数上方使用@pytest.mark.smoke或@pytest.mark.user_manage来标记用例,可添加多个标记)
markers =
    smoke:冒烟测试
    user_manage:用例管理模块
    ...
    ...

3.pytest跳过测试用例

无条件跳过

1
2
3

@pytest.mark.skip(reason="无理由跳过")
def test_get_token(self):
    print('测试1')

有条件跳过

workage = 8
@pytest.mark.skipif(workage<10,reason="工作经验少于10年跳过")
def test_get_token3(self):
    print("测试2")

4.pytest测试用例的前后置，固件

# 可在每个类中使用，但是代码会比较冗余
def setup_method(self):
    print("每个用例之前执行一次")

def teardown_method(self):
    print("每个用例之后执行一次")

def setup_class(self):
    print("每个类之前执行一次")

def teardown_class(self):
    print('每个类之后执行一次')

# 也可以将他们封装到一个类中，让其他类继承，但是继承的类中所有的测试用例都会添加前后置，在只有一个用例需要前后置时（部分前后置），此方法不适合，则需要用到fixture
# common/common.util.py
class TestQH:
    def setup_method(self):
        print("每个用例之前执行一次")

    def teardown_method(self):
        print("每个用例之后执行一次")

    def setup_class(self):
        print("每个类之前执行一次")

    def teardown_class(self):
        print('每个类之后执行一次')

# test_case/test_case.py
from common.common_util import TestQH
class TestDH(TestQH):
    def test_1(self):
        print("测试111")

    def test_2(self):
        print("测试222")

5.使用fixtrue实现部分前后置

@pytest.fixture(scope=None,autouse=False,params=None,ids=None,name=None)

参数：

scope: 作用域
   scope=“function”   在函数之前和之后执行
        1.手动调用的方式是在测试用例的参数里面加入fixtrue的名称
        2.如果说fixtrue有通过return或yield返回值的话，那么可以把这个值传递到测试用例中，并且值是通过固件的名称传递的

import pytest

@pytest.fixture(scope="function") # function:函数
def exe_database_sql():
    print("执行sql查询语句")
    yield # yield之后是后置
    print('关闭数据库')

class TestDH:
    def test_1(self):
        print("测试111")

    def test_2(self,exe_database_sql): # 在函数的参数中直接调用
        print("测试222")

scope=“class” 在类之前和之后执行
1.手动调用的方式是在类的上面加上@pytest.mark.usefixtures(“exe_database_sql”)装饰器调用

import pytest

@pytest.fixture(scope="class",autouse=False)
def exe_database_sql():
    print("执行sql查询语句")
    yield
    print('关闭数据库')

@pytest.mark.usefixtures("exe_database_sql") # 在此装饰器参数中写入fixture名称调用，在这个类之前和之后执行
class Testjiao:
    def test_1(self):
        print('第1个类')

scope=“package/session” 在整个项目会话之前和之后执行
1.一般会结合conftest.py文件来实现

autouse: 自动执行，默认是False

import pytest
 
@pytest.fixture(scope="function",autouse=True) # 自动在每个函数前后执行
def exe_database_sql():
    print("执行sql查询语句")
    yield # yield之后是后置
    print('关闭数据库')

class TestDH:
    def test_1(self):
        print("测试111")

    def test_2(self):
        print("测试222")

params：实现参数化
1.如何把值传到fixtrue? 通过在fixtrue函数的参数里面加入request来接受参数，然后通过request.param来取值

import pytest

# 读取数据的方法
def read_yaml():
    return ['qq','wx','wb']

@pytest.fixture(scope="function",autouse=False,params=read_yaml()) # 使用params传入读取的list数据
def exe_database_sql(request): # 加入requests接受参数
    print("执行sql查询语句")
    yield request.param # 使用 request.param 返回参数值
    print('关闭数据库')

class TestDH:
    def test_1(self):
        print("测试111")

    def test_2(self,exe_database_sql): # 手动使用fixture
        print(exe_database_sql) # 打印传递后的参数值
        print("测试222")

ids：不能单独使用，必须和params一起使用，作用是对参数起别名（一般不使用）

import pytest

# 读取数据的方法
  def read_yaml():
      return ['qq','wx','wb']

  @pytest.fixture(scope="function",autouse=False,params=read_yaml(),ids=['1','2','3']) # 将读取到的数据换个参数名称
  def exe_database_sql(request):
      print("执行sql查询语句")
      yield request.param
      print('关闭数据库')


  class TestDH:
      def test_1(self):
          print("测试111")

      def test_2(self,exe_database_sql):
          print(exe_database_sql)
          print("测试222")

name：给fixtrue起别名(一般没用)
1.一旦使用了别名，那么fixtrue的名称就不能再使用了，只能使用别名

 import pytest

# 读取数据的方法
def read_yaml():
    return ['qq','wx','wb']

@pytest.fixture(scope="function",autouse=False,params=read_yaml(),ids=['1','2','3'],name="db") # 修改fixtrue别名为db
def exe_database_sql(request):
    print("执行sql查询语句")
    yield request.param
    print('关闭数据库')


class TestDH:
    def test_1(self):
        print("测试111")

    def test_2(self,db): # 此时这里调用的fixtrue别名就只能调用db
        print(db)
        print("测试222")

如果希望在另一个py文件中调用fixtrue需要结合到contest.py文件中使用

二. fixtrue结合conftest.py使用

conftest.py：它是专门用于存放fixtrue的配置文件，名称是固定的，不能变
在conftest.py文件里面所有的方法在调用时都不需要导包
conftest.py文件可以有多个，并且多个conftest.py文件里面的多个fixtrue可以被一个用例调用

 # conftest.py
import pytest

@pytest.fixture(scope="function",autouse=False,name="db")
def exe_database_sql(request):
    print("执行sql查询语句")
    yield "success"
    print('关闭数据库')

# test_api.py
class TestDH:
    def test_1(self,db): # 无需导入包，直接调用fixtrue
        print("测试111")

    def test_2(self,db):
        print("测试222")

setup_method,teardown_method,setup_class,teardown_class,fixtrue,conftest优先级

会话：fixtrue的session级别优先级最高
类：fixtrue的class级别优先级别最高
类：setup_class
函数：fixtrue的function级别优先级别最高
函数：setup_method

三. 总结：pytest执行过程

查询当前目录下的conftest.py文件
查询当前目录下的pytest.ini文件，找到测试用例的位置
查询用例目录下的conftest.py文件
查询py文件中是否有setup_method,teardown_method,setup_class,teardown_class
再根据pytest.ini文件的测试用例的规则去查找用例并执行

四. pytest的断言

就是使用python自己的断言，assert

assert flag is true
assert 1==1
assert “a” in “abc”

五. pytest结合allure-pytest插件生成美观的报告

安装allure-pytest插件
下载allure，下载之后解压，解压之后配置环境变量(把allure目录下的bin目录配置到系统变量的path路径)，下载地址：https://github.com/allure-framework/allure2/releases
验证allure是否安装成功：allure --version ，在pycharm验证
生成allure报告

a.生成临时的json报告，在pytest.ini文件里面加入：addopts = -vs --alluredir=./temps --clean-alluredir（--alluredir=./temps：生成临时报告 --clean-alluredir：清空临时报告）
b.生成正式的allure报告：在run.py添加参数：os.system("allure generate ./temps -o ./reports --clean")

import os
import pytest

if __name__ == '__main__':
    pytest.main()
    os.system("allure generate ./temps -o ./reports --clean")

六. pytest通过parametrize()实现数据驱动

方法：@pytest.mark.parametrize(args_name,args_value)
args_name：参数名称，用于将参数值传递给函数
args_value：参数值：（列表和字典列表，元组和字典元组），有n个值那么用例执行n次

1
2
3

@pytest.mark.parametrize("caseinfo",['孙悟空','猪八戒','唐僧'])
def test_01_get_token(self,caseinfo): # 传入的参数名称必须与上面一致
    print('获取统一接口鉴权码' + caseinfo)

1
2
3

@pytest.mark.parametrize("name,age",[['name','孙悟空'],['age','18']])
    def test_01_get_token(self,name,age):
        print('获取统一接口鉴权码' + str(name)+ "  " + str(age))

七. pytest通过yaml格式实现测试用例读、写、封装

yaml是一种数据格式，扩展名可以使yaml、yml，支持#注释，通过缩进表示层级关系，区分大小写。yaml读取出来是一个字典列表格式

用途：
1.用于做配置文件
2.编写自动化测试用例

数据组成：
1.map对象，键(空格)值

name: 孙悟空

2.数组(list)，使用’-'表示列表

msjy:
    - name1: 孙悟空
    - name2:
        - age1: 18
        - age2: 19
    - name3: 玉皇大帝

封装读取yaml文件代码

# get_token.yaml   yaml文件数据
- name: 获取统一接口鉴权码
request:
    method: post
    url: http://127.0.0.1:5566/overloadcontrolapi/api/v1/login
    headers:
    Authorization: authorization_token
    data:
    userName: donghao
    password: 123456
validate:
    status_code: 200

# yaml.util.py   封装读取yaml
    import os.path
    import yaml

    # 获取项目的根目录
    def get_obj_path():
        return os.path.dirname(__file__).split('common')[0]

    # 读取yaml,需要安装 pip install pyyaml
    def read_yaml(yamlpath):
        with open(get_obj_path() + yamlpath ,mode='r',encoding='utf-8') as f:
            value = yaml.load(stream=f,Loader=yaml.FullLoader)
            return value

    # 下面代码仅做演示测试使用，实际无需下面代码，直接调用上面代码即可
    if __name__ == '__main__':
        # 可在其他调用此文件的地方传入yaml文件路径
        print(read_yaml('test_case/get_token.yaml'))


# test_get_token.py  # 用例读取yaml数据 
    import pytest
    import requests
    from common.yaml_util import read_yaml

    class TestApi:
        # 自动关联cookis
        session = requests.session()
        @pytest.mark.parametrize("caseinfo",read_yaml("test_case/get_token.yaml"))
        def test_01_get_token(self,caseinfo):
            name = caseinfo['name']
            method = caseinfo['request']['method']
            url = caseinfo['request']['url']
            data = caseinfo['request']['data']
            headers = caseinfo['request']['headers']
            validate = caseinfo['validate']['status_code']
            res = TestApi.session.request(method=method,url=url,json=data,headers=headers)
            all_dict = res.json()
            authorization_token = all_dict['data']['token']
            assert res.status_code == validate

八. 统一接口请求封装

# requests_util.py  接口请求封装
import requests

class RequestsUtil:
    session = requests.Session()
    def send_request(self,method,url,data,**kwargs):
        method = str(method).lower()
        res = ""
        if method == "get":
            res = RequestsUtil.session.request(method,url,params=data, **kwargs)
        elif method == "post":
            res = RequestsUtil.session.post(method,url,json=data,**kwargs)
        return res


# test_get_token.py  用例中使用封装的接口请求
import pytest
from common.requests_util import RequestsUtil
from common.yaml_util import read_yaml

class TestApi:
    @pytest.mark.parametrize("caseinfo",read_yaml("test_case/get_token.yaml"))
    def test_01_get_token(self,caseinfo):
        name = caseinfo['name']
        method = caseinfo['request']['method']
        url = caseinfo['request']['url']
        data = caseinfo['request']['data']
        headers = caseinfo['request']['headers']
        validate = caseinfo['validate']['status_code']
        res = RequestsUtil.session.request(method=method,url=url,json=data,headers=headers) # 使用封装的接口请求
        all_dict = res.json()
        authorization_token = all_dict['data']['token']
        assert res.status_code == validate
        assert authorization_token is not None and authorization_token != ''

九. Allure测试报告定制

@allure.epic("电商系统") 代表整个产品 / 项目 / 系统层级，通常对应公司的一条产品线或一个完整系统
@allure.feature("用户管理") epic下的子模块，对应产品的核心功能模块（比如用户管理、订单管理、商品管理）
@allure.story("用户登录") feature下的具体业务场景（最小功能单元），对应产品需求中的一个具体场景。“用户登录” 是 “用户管理” 模块下的一个具体场景。
用例标题
- 静态装饰器：@allure.title("测试用户登录功能") 在测试用例加载阶段（代码解析时）就确定了标题，是固定不变的。它的值在定义测试函数时就已经确定，无法在测试用例执行过程中修改。
- 动态装饰器：allure.dynamic.title(casename) 在测试用例执行过程中生效，可以根据测试逻辑、参数、执行结果等动态修改标题，灵活性更高。
@allure.link("https://httpbin.org/post",name="需求文档") 通用链接：关联任意外部链接（需求文档、设计文档、接口文档等）。name是链接的展示名称（报告中显示），第一个参数是实际链接地址。
@allure.issue("BUG-123","登录失败问题") 缺陷 / BUG 链接：专门关联 bug 管理系统（如 Jira、Bugzilla）的 BUG。第一个参数是 BUG 编号 / BUG 地址，name是 BUG 的简要描述。
@allure.testcase("TC-456","测试用例链接") 官方测试用例链接：专门关联测试管理系统（如 TestLink、Zephyr、禅道）的官方用例。用于追溯手工测试用例或官方用例库，方便自动化用例和官方用例对应。
@allure.severity(allure.severity_level.CRITICAL) 用例优先级 / 严重级别：标记用例的重要程度，Allure 内置 5 个级别（从高到低）：
✅ CRITICAL（致命）：核心功能（如支付、登录）
✅ BLOCKER（阻塞）：阻塞后续测试的问题
✅ NORMAL（普通）：常规功能（默认级别）
✅ MINOR（次要）：界面样式、小优化
✅ TRIVIAL（轻微）：文字错别字等
报告中会显示优先级标签（如🔴CRITICAL），可按级别筛选用例
用例失败自动截图+日志记录

import os
import sys
import logging
import threading
from logging.handlers import BaseRotatingHandler
from typing import Optional, Dict, Any, List, Union
import allure
from datetime import datetime, timedelta

# ========== 基础配置（级别文件夹定义） ==========
try:
    PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
except Exception as e:
    PROJECT_ROOT = os.getcwd()
    print(f"自动计算项目根目录失败，使用当前工作目录: {PROJECT_ROOT}，错误原因: {e}", file=sys.stderr)

# 根日志目录（支持手动覆盖）
ROOT_LOG_DIR = os.environ.get("LEOPARD_TEST_LOG_DIR", os.path.join(PROJECT_ROOT, "logs"))
# 各级别专属文件夹（key=级别值，value=文件夹名）
LEVEL_DIR_MAP = {
    logging.DEBUG: "debug_log",
    logging.INFO: "info_log",
    logging.WARNING: "warning_log",
    logging.ERROR: "error_log",
    logging.CRITICAL: "critical_log"
}
# 级别名称映射（用于生成文件名）
LEVEL_NAME_MAP = {
    logging.DEBUG: "debug",
    logging.INFO: "info",
    logging.WARNING: "warning",
    logging.ERROR: "error",
    logging.CRITICAL: "critical"
}

# 线程锁：保护日志器初始化
logger_lock = threading.Lock()


# 确保所有级别文件夹存在
def _ensure_level_dirs():
    """确保所有日志级别文件夹存在"""
    try:
        # 创建根日志目录
        if not os.path.exists(ROOT_LOG_DIR):
            os.makedirs(ROOT_LOG_DIR, exist_ok=True)
        # 为每个级别创建专属文件夹
        for level_dir in LEVEL_DIR_MAP.values():
            level_dir_path = os.path.join(ROOT_LOG_DIR, level_dir)
            if not os.path.exists(level_dir_path):
                os.makedirs(level_dir_path, exist_ok=True)
    except Exception as e:
        print(f"创建日志目录失败: {e}", file=sys.stderr)


# ========== 1. Allure日志处理器（仅保留基础结构，改为批量添加文件附件） ==========
class AllureLogHandler(logging.Handler):
    """同步日志文件到Allure报告（改为批量添加文件附件）"""
    def emit(self, record):
        """不再单个日志条目生成附件，改为在测试用例结束时批量添加日志文件"""
        pass


# ========== 2. 核心：按级别+日期+文件夹存储的处理器 ==========
class LevelDirDailyFileHandler(BaseRotatingHandler):
    """
    按「级别文件夹+日期」存储日志
    例如：logs/info_log/info_2026-02-27.log、logs/error_log/error_2026-02-27.log
    """

    def __init__(self, target_level, encoding='utf-8', backupCount=7):
        # 获取当前级别对应的文件夹
        self.target_level = target_level
        self.target_level_name = logging.getLevelName(target_level).lower()
        self.level_dir_name = LEVEL_DIR_MAP[target_level]
        self.level_dir_path = os.path.join(ROOT_LOG_DIR, self.level_dir_name)

        self.backupCount = backupCount
        self.current_date = self._get_today()
        # 生成「级别文件夹/级别_日期.log」的完整路径
        self.current_filename = os.path.join(
            self.level_dir_path,
            f"{self.target_level_name}_{self.current_date}.log"
        )
        # 初始化父类（追加模式写入）
        super().__init__(self.current_filename, 'a', encoding=encoding)
        # 显式设置处理器级别（最佳实践）
        self.setLevel(target_level)

    def _get_today(self):
        """获取当天日期（YYYY-MM-DD）"""
        return datetime.now().strftime("%Y-%m-%d")

    def shouldRollover(self, record):
        """仅跨天时切换文件"""
        today = self._get_today()
        if today != self.current_date:
            return True
        return False

    def emit(self, record):
        """严格过滤：仅写入指定级别的日志"""
        if record.levelno != self.target_level:
            return
        super().emit(record)

    def doRollover(self):
        """跨天切换文件，清理当前级别文件夹下的过期日志"""
        # 关闭当前文件
        if self.stream:
            self.stream.close()
            self.stream = None

        # 更新日期和文件名
        self.current_date = self._get_today()
        self.current_filename = os.path.join(
            self.level_dir_path,
            f"{self.target_level_name}_{self.current_date}.log"
        )

        # 清理当前级别文件夹下超过保留天数的日志
        self._clean_old_logs()

        # 打开新文件
        self.stream = self._open()

    def _clean_old_logs(self):
        """清理当前级别文件夹下的过期日志"""
        try:
            # 筛选当前级别文件夹下的日志文件：如 info_*.log
            log_files = [
                f for f in os.listdir(self.level_dir_path)
                if f.startswith(f"{self.target_level_name}_") and f.endswith('.log')
            ]
            if not log_files:
                return

            # 按日期排序（旧→新）
            def _parse_log_date(filename):
                """解析日志文件名中的日期"""
                date_str = filename.replace(f"{self.target_level_name}_", "").replace(".log", "")
                return datetime.strptime(date_str, "%Y-%m-%d")

            log_files.sort(key=_parse_log_date)

            # 删除超过保留天数的文件
            if len(log_files) > self.backupCount:
                for old_file in log_files[:-self.backupCount]:
                    old_file_path = os.path.join(self.level_dir_path, old_file)
                    os.remove(old_file_path)
                    # 记录清理日志
                    logging.getLogger("LeopardTestLogger").info(
                        f"清理过期日志文件：{old_file_path}"
                    )
        except Exception as e:
            # 使用自身日志器记录错误，而非仅打印到stderr
            logging.getLogger("LeopardTestLogger").error(
                f"清理{self.target_level_name}级别过期日志失败: {e}"
            )


# ========== 3. 核心日志类 ==========
class Logger:
    _instance: Optional['Logger'] = None
    _logger: Optional[logging.Logger] = None

    # 单例模式（加线程锁）
    def __new__(cls):
        with logger_lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance._init_logger()
        return cls._instance

    def _init_logger(self):
        if Logger._logger is not None:
            return

        # 确保所有级别文件夹存在
        _ensure_level_dirs()

        # 基础日志器配置
        Logger._logger = logging.getLogger("LeopardTestLogger")
        Logger._logger.setLevel(logging.DEBUG)  # 总级别设为最低
        Logger._logger.propagate = False

        # 统一日志格式
        formatter = logging.Formatter(
            '[%(asctime)s] [%(process)d-%(threadName)s] [%(filename)s:%(lineno)d] - %(levelname)s: %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )

        # ========== 为每个级别创建独立处理器（按文件夹存储） ==========  backupCount为日志保存天数
        # 1. DEBUG级别：logs/debug_log/debug_2026-02-27.log
        debug_handler = LevelDirDailyFileHandler(logging.DEBUG, backupCount=7)
        debug_handler.setFormatter(formatter)

        # 2. INFO级别：logs/info_log/info_2026-02-27.log
        info_handler = LevelDirDailyFileHandler(logging.INFO, backupCount=7)
        info_handler.setFormatter(formatter)

        # 3. WARNING级别：logs/warning_log/warning_2026-02-27.log
        warning_handler = LevelDirDailyFileHandler(logging.WARNING, backupCount=7)
        warning_handler.setFormatter(formatter)

        # 4. ERROR级别：logs/error_log/error_2026-03-01.log
        error_handler = LevelDirDailyFileHandler(logging.ERROR, backupCount=7)
        error_handler.setFormatter(formatter)

        # 5. CRITICAL级别：logs/critical_log/critical_2026-02-27.log
        critical_handler = LevelDirDailyFileHandler(logging.CRITICAL, backupCount=7)
        critical_handler.setFormatter(formatter)

        # 6. 彩色控制台处理器（所有级别都输出）
        class ColoredStreamHandler(logging.StreamHandler):
            COLORS = {
                logging.DEBUG: '\033[0;36m',  # 青色
                logging.INFO: '\033[0;32m',  # 绿色
                logging.WARNING: '\033[0;33m',  # 黄色
                logging.ERROR: '\033[0;31m',  # 红色
                logging.CRITICAL: '\033[0;35m'  # 紫色
            }
            RESET = '\033[0m'

            def format(self, record):
                msg = super().format(record)
                color = self.COLORS.get(record.levelno, self.RESET)
                return f"{color}{msg}{self.RESET}"

        console_handler = ColoredStreamHandler(stream=sys.stdout)
        console_handler.setLevel(logging.DEBUG)
        console_handler.setFormatter(formatter)

        # 7. Allure报告处理器
        allure_handler = AllureLogHandler()
        allure_handler.setFormatter(formatter)

        # ========== 添加所有处理器 ==========
        if not Logger._logger.handlers:
            Logger._logger.addHandler(debug_handler)
            Logger._logger.addHandler(info_handler)
            Logger._logger.addHandler(warning_handler)
            Logger._logger.addHandler(error_handler)
            Logger._logger.addHandler(critical_handler)
            Logger._logger.addHandler(console_handler)
            Logger._logger.addHandler(allure_handler)

    # ========== 封装日志方法 ==========
    def debug(self, message: str) -> None:
        """DEBUG → logs/debug_log/debug_2026-02-27.log"""
        try:
            self._logger.debug(message.strip())
        except Exception as e:
            print(f"[日志错误] 记录DEBUG日志失败: {str(e)}", file=sys.stderr)

    def info(self, message: str) -> None:
        """INFO → logs/info_log/info_2026-02-27.log"""
        try:
            self._logger.info(message.strip())
        except Exception as e:
            print(f"[日志错误] 记录INFO日志失败: {str(e)}", file=sys.stderr)

    def warning(self, message: str) -> None:
        """WARNING → logs/warning_log/warning_2026-02-27.log"""
        try:
            self._logger.warning(message.strip())
        except Exception as e:
            print(f"[日志错误] 记录WARNING日志失败: {str(e)}", file=sys.stderr)

    def error(self, message: str, exc_info: bool = False) -> None:
        """ERROR → logs/error_log/error_2026-02-27.log"""
        try:
            self._logger.error(message.strip(), exc_info=exc_info)
        except Exception as e:
            print(f"[日志错误] 记录ERROR日志失败: {str(e)}", file=sys.stderr)

    def critical(self, message: str, exc_info: bool = True) -> None:
        """CRITICAL → logs/critical_log/critical_2026-02-27.log"""
        try:
            self._logger.critical(message.strip(), exc_info=exc_info)
        except Exception as e:
            print(f"[日志错误] 记录CRITICAL日志失败: {str(e)}", file=sys.stderr)

    def api_logs(
            self,
            url: str,
            data: Union[Dict[str, Any], str, None] = None,
            codes: Union[int, str, None] = None,
            responses: Union[Dict[str, Any], str, None] = None,
            data_name: Union[str, None] = None,
            data_list: Union[List[Any], None] = None
    ) -> None:
        """
        接口日志 → logs/info_log/info_2026-02-27.log
        优化：参数支持空值，日志展示更友好
        """
        # 空值处理
        data = data or "无"
        codes = codes or "无"
        responses = responses or "无"
        data_name = data_name or "无"
        data_list = data_list or "无"

        log_msg = (
                "\n" + "=" * 80 +
                f"\n【请求URL】：{url}"
                f"\n【请求参数】：{data}"
                f"\n【返回状态码】：{codes}"
                f"\n【返回结果】：{responses}"
                f"\n【数据库验证字段】：{data_name}"
                f"\n【数据库查询数据】：{data_list}"
                + "\n" + "=" * 80 + "\n"
        )
        self.info(log_msg)

    def get_today_log_files(self) -> Dict[str, str]:
        """
        获取当天所有级别的日志文件路径
        返回格式：{"info": "/logs/info_log/info_2026-03-01.log", ...}
        """
        today = datetime.now().strftime("%Y-%m-%d")
        log_files = {}
        for level, level_name in LEVEL_NAME_MAP.items():
            level_dir = LEVEL_DIR_MAP[level]
            log_file_path = os.path.join(ROOT_LOG_DIR, level_dir, f"{level_name}_{today}.log")
            log_files[level_name] = log_file_path
        return log_files

    def add_log_files_to_allure(self):
        """
        将当天有内容的日志文件作为附件添加到Allure报告（对应当前测试用例）
        仅添加文件大小>0的日志文件，避免空文件占用空间
        """
        try:
            today_log_files = self.get_today_log_files()
            for level_name, file_path in today_log_files.items():
                # 检查文件是否存在且有内容
                if os.path.exists(file_path) and os.path.getsize(file_path) > 0:
                    # 读取文件内容
                    with open(file_path, 'r', encoding='utf-8') as f:
                        log_content = f.read()
                    # 添加到Allure报告，附件名称清晰标识级别和日期
                    allure.attach(
                        body=log_content,
                        name=f"{level_name}_log_{datetime.now().strftime('%Y-%m-%d')}.log",
                        attachment_type=allure.attachment_type.TEXT,
                        extension="log"
                    )
            self.info("已将当天有内容的日志文件添加到Allure报告")
        except Exception as e:
            self.error(f"添加日志文件到Allure报告失败: {e}", exc_info=True)


# 全局实例
log = Logger()

import pytest
import allure
import os
import re
from selenium import webdriver
from selenium.common.exceptions import WebDriverException
from leopardv5_Test.page.log_util import log

# 全局driver变量初始化
driver = None

# UI测试fixture名白名单（用于匹配driver）
UI_FIXTURE_NAMES = ["browser", "driver", "chrome", "firefox", "web_driver"]


@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
    # 执行其他钩子获取报告对象
    outcome = yield
    rep = outcome.get_result()

    # 只要用例失败（无论setup/call/teardown阶段），都会进入失败截图逻辑
    if rep.failed:
    # 可选择：仅在call阶段（用例核心执行阶段）失败时才触发截图
    # if rep.failed and rep.when == "call":
        # UI测试用例：执行截图逻辑（现在所有失败用例都会走到这里）
        current_driver = None
        try:
            # 兼容多种UI fixture名
            for fixture_name in UI_FIXTURE_NAMES:
                if fixture_name in item.funcargs:
                    current_driver = item.funcargs[fixture_name]
                    break
            # 备用方案：使用全局driver变量
            if not current_driver and driver is not None:
                current_driver = driver

            # 执行截图（增加更严谨的driver有效性判断）
            if current_driver is not None:
                try:
                    # 检查driver是否有效（兼容不同Selenium版本）
                    current_driver.current_url  # 触发可能的WebDriverException
                    # 优化截图名称：彻底清理特殊字符
                    clean_nodeid = re.sub(r'[^\w\-_.]', '_', item.nodeid)
                    screenshot_name = f"{clean_nodeid}_{rep.when}_失败截图"
                    # 附加截图到allure报告
                    allure.attach(
                        current_driver.get_screenshot_as_png(),
                        name=screenshot_name,
                        attachment_type=allure.attachment_type.PNG
                    )
                    log.info(f"【截图成功】用例「{item.nodeid}」已生成失败截图并附加到Allure报告")
                except WebDriverException as e:
                    error_msg = f"【截图失败】用例「{item.nodeid}」，driver已失效：{str(e)}"
                    log.error(error_msg)
                except Exception as e:
                    error_msg = f"【截图失败】用例「{item.nodeid}」，原因：{str(e)}"
                    log.error(error_msg)
            else:
                warning_msg = f"【截图警告】用例「{item.nodeid}」无可用的driver对象，跳过截图"
                log.warning(warning_msg)
        except Exception as e:
            error_msg = f"【截图异常】用例「{item.nodeid}」处理截图逻辑时出错：{str(e)}"
            log.error(error_msg)


@pytest.fixture(scope="session")  # scope为session级，实现浏览器复用
def browser():
    global driver
    # 检查是否启用多进程（pytest-xdist），给出警告
    if "PYTEST_XDIST_WORKER" in os.environ:
        warning_msg = "【警告】检测到pytest-xdist多进程模式，全局driver变量可能存在冲突，建议使用fixture传递driver"
        log.warning(warning_msg)

    if driver is None:
        try:
            # 配置Chrome选项，绕过反自动化检测
            options = webdriver.ChromeOptions()
            options.add_experimental_option("excludeSwitches", ["enable-automation"])
            options.add_experimental_option("useAutomationExtension", False)
            options.add_argument("--disable-blink-features=AutomationControlled")
            # 可选：无头模式
            # options.add_argument("--headless=new")
            # 可选：禁用GPU加速
            # options.add_argument("--disable-gpu")

            driver = webdriver.Chrome(options=options)
            driver.maximize_window()
            driver.implicitly_wait(10)
            log.info("【浏览器初始化成功】已启动Chrome浏览器（session级别）")
        except Exception as e:
            log.error(f"【浏览器初始化失败】原因：{str(e)}")
            raise  # 抛出异常，终止测试

    yield driver

    # 浏览器退出逻辑
    try:
        if driver is not None:
            # 检查driver是否仍有效
            driver.session_id
            driver.quit()
            log.info("【浏览器退出成功】已关闭Chrome浏览器（session级别）")
    except WebDriverException:
        log.warning("【浏览器退出提示】driver已提前关闭，无需重复退出")
    except Exception as e:
        log.error(f"【浏览器退出失败】原因：{str(e)}")
    finally:
        # 重置全局driver
        driver = None

# 全局自动生效的fixture：每个用例执行完都添加日志文件到Allure
@pytest.fixture(scope="function", autouse=True)
def add_log_to_allure():
    yield  # 执行用例逻辑
    # 用例执行完成后（无论成功/失败），自动添加当天有内容的日志文件到Allure
    log.add_log_files_to_allure()

# test_a_login_logout.py  # 测试用例
from leopardv5_Test.page.log_util import log

log = Logger()
def test_01_login():
    log.info("日志内容" )  # 这样也能同时将日志写入报告中