MyTest/TestCore/result_parser.py

import re
import json
from pathlib import Path
from Utils.Log import log_for_api
logger = log_for_api()


class TestResultParser:
    """测试结果解析器"""

    def __init__(self, results_dir):
        self.results_dir = Path(results_dir)

    def parse_pytest_output(self, output):
        """从pytest输出中解析测试结果"""
        results = {
            "total": 0,
            "passed": 0,
            "failed": 0,
            "skipped": 0,
            "errors": 0,
            "duration": 0,
            "pass_rate": 0
        }

        # 尝试从输出中解析结果
        try:
            # 匹配 pytest 结果摘要行
            pattern = r"(\d+) passed.*?(\d+) failed.*?(\d+) skipped"
            match = re.search(pattern, output)

            if match:
                results["passed"] = int(match.group(1))
                results["failed"] = int(match.group(2))
                results["skipped"] = int(match.group(3))
                results["total"] = results["passed"] + results["failed"] + results["skipped"]

                # 计算通过率
                if results["total"] > 0:
                    results["pass_rate"] = round((results["passed"] / results["total"]) * 100, 2)

            # 尝试解析持续时间
            duration_pattern = r"in (\d+\.\d+)s"
            duration_match = re.search(duration_pattern, output)
            if duration_match:
                results["duration"] = float(duration_match.group(1))

        except Exception as e:
            logger.error(f"解析pytest输出时出错: {e}")

        return results

    def parse_allure_results(self):
        """从Allure结果文件中解析更详细的信息"""
        results = {
            "suites": [],
            "categories": [],
            "status_counts": {"passed": 0, "failed": 0, "broken": 0, "skipped": 0, "unknown": 0}
        }

        try:
            # 查找所有result.json文件
            result_files = list(self.results_dir.glob("*-result.json"))

            for result_file in result_files:
                with open(result_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)

                    # 统计状态
                    status = data.get("status", "unknown")
                    if status in results["status_counts"]:
                        results["status_counts"][status] += 1

                    # 收集套件信息
                    suite_name = data.get("labels", {}).get("suite", "Unknown Suite")
                    if suite_name not in results["suites"]:
                        results["suites"].append(suite_name)

        except Exception as e:
            logger.error(f"解析Allure结果时出错: {e}")

        return results