Declared as wrapper=True or hookwrapper=True but function is not a generator function

[KnightBlood]

import asyncio
import glob
import logging
import os
import shutil
import subprocess
import time
from typing import Generator, Optional, List
import allure
import pytest
from playwright.async_api import async_playwright
from playwright.sync_api import (
    Page,
    Playwright,
)

# 配置日志记录
logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 定义项目中使用的目录常量
VIDEO_DIR = os.getenv("VIDEO_DIR", "./video")
SCREENSHOT_DIR = os.getenv("SCREENSHOT_DIR", "./screenshot")
ALLURE_RESULTS_DIR = os.getenv("ALLURE_RESULTS_DIR", "./allure-results")
ALLURE_REPORT_DIR = os.getenv("ALLURE_REPORT_DIR", "./allure-report")
ALLURE_PATH = os.getenv("ALLURE_PATH", "D:/works/PyUIAuto/tools/allure-2.25.0/bin/allure.bat")


# 验证路径的安全性
def validate_path(path: str) -> bool:
    if not os.path.isabs(path):
        logger.error(f"路径不合法: {path}，必须是绝对路径。")
        return False
    return True


# 异步清理指定目录下的所有文件和子目录
async def cleanup_directory(directory: str) -> None:
    if os.path.exists(directory):
        logger.info(f"正在清理 {directory}...")
        for filepath in glob.glob(os.path.join(directory, "*")):
            try:
                if os.path.isdir(filepath):
                    shutil.rmtree(filepath, onerror=handle_error)
                    logger.info(f"已删除目录: {filepath}")
                else:
                    os.remove(filepath, onerror=handle_error)
                    logger.info(f"已删除文件: {filepath}")
            except FileNotFoundError:
                logger.error(f"文件或目录不存在: {filepath}")


async def handle_error(func, path, exc_info):
    logger.error(f"无法删除 {path}: {exc_info}")


# 初始化配置检查
def init_config():
    for dir_name in [VIDEO_DIR, SCREENSHOT_DIR, ALLURE_RESULTS_DIR, ALLURE_REPORT_DIR]:
        if not os.path.isabs(dir_name) or not os.path.exists(dir_name):
            logger.error(f"目录配置错误: {dir_name}")
            exit(1)
    if not validate_path(ALLURE_PATH):
        logger.error(f"Allure路径配置错误: {ALLURE_PATH}")
        exit(1)


# pytest会话开始前清理指定的目录
@pytest.fixture(autouse=True)
async def cleanup_folders():
    folders_to_clean = [
        VIDEO_DIR,
        SCREENSHOT_DIR,
        ALLURE_RESULTS_DIR,
        ALLURE_REPORT_DIR,
    ]
    await asyncio.gather(*[cleanup_directory(folder) for folder in folders_to_clean])


# 初始化Playwright并获取一个Page实例
@pytest.fixture(scope="session")
async def playwright_and_page(base_url: str) -> tuple[Playwright, Page]:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False, slow_mo=50, args=["--start-fullscreen"])
        context = await browser.new_context(record_video_dir=VIDEO_DIR)
        page = await context.new_page()
        await page.goto(base_url)

        yield p, page

        await context.close()
        await browser.close()


# 计算页面渲染时间，并将结果作为附件添加到Allure报告中
@pytest.fixture(autouse=True)
async def measure_render_time() -> None:
    start_time = time.time()
    yield
    end_time = time.time()
    render_time = end_time - start_time
    allure.attach(
        body=f"Render time: {render_time:.2f} seconds",
        name="Render Time",
        attachment_type=allure.attachment_type.TEXT,
    )


# 处理测试结束后的视频和截图，将失败测试的截图和视频添加到Allure报告中
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
async def pytest_runtest_protocol(item: pytest.Item) -> Generator[None, None, None]:
    yield  # 允许内部钩子先执行
    result = item.rep_call.passed
    if not result:
        screenshot_path = create_screenshot_path(item)
        await item.funcargs["page"].screenshot(path=screenshot_path)
        attach_screenshot_to_allure(screenshot_path, item)

        video_file = await process_video(item)
        if video_file:
            attach_video_to_allure(video_file, item)


def create_screenshot_path(item: pytest.Item) -> str:
    return os.path.join(SCREENSHOT_DIR, f"{item.nodeid}.png")


def attach_screenshot_to_allure(screenshot_path: str, item: pytest.Item) -> None:
    allure.attach.file(screenshot_path, attachment_type=allure.attachment_type.PNG, name="Screenshot")


async def process_video(item: pytest.Item) -> Optional[str]:
    video_file_pattern = os.path.join(VIDEO_DIR, f"{item.nodeid}.*.webm")
    video_files = glob.glob(video_file_pattern)
    if video_files:
        video_file = video_files[0]
        new_video_file = os.path.join(VIDEO_DIR, f"FAILED_{item.nodeid}.webm")
        shutil.move(video_file, new_video_file)
        return new_video_file


def attach_video_to_allure(video_file: str, item: pytest.Item) -> None:
    allure.attach.file(video_file, attachment_type=allure.attachment_type.WEBM, name="Video")


# 生成Allure测试报告
@pytest.fixture(autouse=True)
def generate_allure_report():
    def _generate_allure_report() -> None:
        if validate_path(ALLURE_PATH) and validate_path(ALLURE_RESULTS_DIR):
            cmd = [ALLURE_PATH, "generate", "--clean", ALLURE_RESULTS_DIR, "-o", ALLURE_REPORT_DIR]
            try:
                subprocess.run(cmd, check=True)
            except Exception as e:
                logger.error(f"生成Allure报告时遇到错误: {e}")

    request.addfinalizer(_generate_allure_report)

@pytest.hookimpl(tryfirst=True, hookwrapper=True)
async def pytest_runtest_protocol(item: pytest.Item) -> Generator[None, None, None]:
    yield  # 允许内部钩子先执行
    result = item.rep_call.passed
    if not result:
        screenshot_path = create_screenshot_path(item)
        await item.funcargs["page"].screenshot(path=screenshot_path)
        attach_screenshot_to_allure(screenshot_path, item)

        video_file = await process_video(item)
        if video_file:
            attach_video_to_allure(video_file, item)

error with

pluggy._manager.PluginValidationError: Plugin 'D:\\works\\PyUIAuto\\conftest.py' for hook 'pytest_runtest_protocol'
hookimpl definition: pytest_runtest_protocol(item: _pytest.nodes.Item) -> Generator[NoneType, NoneType, NoneType]
Declared as wrapper=True or hookwrapper=True but function is not a generator function

[KnightBlood]

how should i do?

[The-Compiler]

Don't use async def there. Also, please make sure you format your code properly, as this is quite unreadable as-is.