allure2markdown.py

import os
import datetime
import csv
import glob
import json
import jinja2
import textwrap
import sys


def transform_allure_report2markdown(allure_results_dir):
    result_files = glob.glob(os.path.join(allure_results_dir, "*.json"))

    # 最新の結果を集める
    results = {"features_map": {}}
    for f in result_files:
        with open(f) as fp:
            result_body = json.loads(fp.read())
        feature = {e["name"]: e["value"] for e in result_body.get("labels")}.get("feature", "null")
        if feature not in results["features_map"]:
            results["features_map"][feature] = {"scenarios_map": {}}
        start_already = results["features_map"][feature]["scenarios_map"].get(result_body.get("name"), {}).get("start")
        if start_already is None or int(result_body.get("start", 0)) >= start_already:
            results["features_map"][feature]["scenarios_map"][result_body.get("name")] = result_body

    # 結果を実施時間順にソート
    for kf in results["features_map"]:
        results["features_map"][kf]["name"] = kf
        results["features_map"][kf]["scenarios"] = sorted(results["features_map"][kf]["scenarios_map"].values(), key=lambda e: e.get("start"))
        del results["features_map"][kf]["scenarios_map"]
    results["features"] = sorted(results["features_map"].values(), key=lambda e: e["scenarios"][0].get("start"))
    del results["features_map"]

    # テスト結果の集計
    stats_status = lambda s: s if s in ["passed", "skipped"] else "failed"
    stats = {
        "features": {"passed":0, "failed":0, "skipped":0},
        "scenarios": {"passed":0, "failed":0, "skipped":0},
        "steps": {"passed":0, "failed":0, "skipped":0},
    }
    results["stats"] = {"passed":0, "failed":0, "skipped":0}
    for feature in results["features"]:
        feature["stats"] = {"passed":0, "failed":0, "skipped":0}
        for scenario in feature.get("scenarios"):
            feature["stats"][stats_status(scenario.get("status"))] += 1
            stats["scenarios"][stats_status(scenario.get("status"))] += 1
            scenario["stats"] = {"passed":0, "failed":0, "skipped":0}
            for step in scenario.get("steps"):
                scenario["stats"][stats_status(step.get("status"))] += 1
                stats["steps"][stats_status(step.get("status"))] += 1
        if feature["stats"]["failed"] == 0:
            if feature["stats"]["passed"] == 0:
                feature_status = "skipped"
            else:
                feature_status = "passed"
        else:
            feature_status = "failed"
        results["stats"][feature_status] += 1
        stats["features"][feature_status] += 1
    for k in ["features", "scenarios", "steps"]:
        stats[k]["tests"] = sum(stats[k].values())

    # 添付ファイルのロード (テキスト形式のみ)・補助情報の追加
    status_icon_map = {"passed": ":white_check_mark:", "failed": ":x:", "broken": ":x:"}
    for feature in results["features"]:
        feature["start"] = feature["scenarios"][0].get("start")
        feature["start_time"] = datetime.datetime.fromtimestamp(int(feature.get("start"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")
        feature["stop"] = feature["scenarios"][-1].get("stop")
        feature["stop_time"] = datetime.datetime.fromtimestamp(int(feature.get("stop"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")
        for scenario in feature["scenarios"]:
            scenario["start_time"] = datetime.datetime.fromtimestamp(int(scenario.get("start"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")
            scenario["stop_time"] = datetime.datetime.fromtimestamp(int(scenario.get("stop"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")
            scenario["status_icon"] = status_icon_map.get(scenario["status"], f'[{scenario["status"]}]')
            scenario["tags"] = [e["value"] for e in scenario.get("labels") if e["name"] == "tag"]
            for step in scenario.get("steps"):
                step["start_time"] = datetime.datetime.fromtimestamp(int(step.get("start"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")
                step["stop_time"] = datetime.datetime.fromtimestamp(int(step.get("stop"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")
                step["status_icon"] = status_icon_map.get(step["status"], f'[{step["status"]}]')
                for attach in step.get("attachments", []):
                    if attach["type"].startswith("text/"):
                        with open(os.path.join(allure_results_dir, attach["source"])) as fp:
                            if attach["type"] == "text/csv":
                                # attach["body"] = csv.reader(fp)
                                attach["body"] = ""
                                for idx, row in enumerate(csv.reader(fp)):
                                    attach["body"] += "| " + " | ".join(row) + " |\n"
                                    if idx == 0:
                                        attach["body"] += "| " + " | ".join(["----" for e in range(len(row))]) + " |\n"
                            else:
                                attach["body"] = fp.read()
    results["start"] = min([e["start"] for e in results["features"]])
    results["start_time"] = datetime.datetime.fromtimestamp(int(results.get("start"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")
    results["stop"] = max([e["stop"] for e in results["features"]])
    results["stop_time"] = datetime.datetime.fromtimestamp(int(results.get("stop"))/1000).astimezone().strftime("%Y/%m/%d %H:%M:%S %Z")

    tpl = """
    Test Result Report
    ---
    
    Date: {{ results["start_time"] }} - {{ results["stop_time"] }}
        
    - {{ stats["features"]["passed"] }} features passed, {{ stats["features"]["failed"] }} failed, {{ stats["features"]["skipped"] }} skipped
    - {{ stats["scenarios"]["passed"] }} scenarios passed, {{ stats["scenarios"]["failed"] }} failed, {{ stats["scenarios"]["skipped"] }} skipped
    - {{ stats["steps"]["passed"] }} steps passed, {{ stats["steps"]["failed"] }} failed, {{ stats["steps"]["skipped"] }} skipped
    
    {% for feature in results["features"] %}
    ## {{ feature["name"] }}  - {{ feature["stats"]["passed"] }} passed, {{ feature["stats"]["failed"] }} failed, {{ feature["stats"]["skipped"] }} skipped

    {% for scenario in feature["scenarios"] %}
    <details>
    <summary>
    {{ scenario["status_icon"] }}
    #{{ loop.index }} {% if scenario["tags"] %}[{% for t in scenario["tags"] %}{{ t }}{% endfor %}]{% endif %}
    {{ scenario["name"] }}
     - {{ scenario["stop"] - scenario["start"] }}ms
    </summary><blockquote>
    
    - Tags: {{ ", ".join(scenario["tags"]) }}
    - Status: {{ scenario["status"] }}
    - Duration: {{ scenario["stop"] - scenario["start"] }}ms ({{ scenario["start_time"] }} - {{ scenario["stop_time"] }})

    {% if scenario["statusDetails"] %}
    Message
    ```
    {{ scenario["statusDetails"]["message"] }}
    ```
    {% if scenario["statusDetails"]["trace"] %}
    Trace 
    ```
    {{ scenario["statusDetails"]["trace"] }}
    ```
    {% endif %}
    {% endif %}
    

    {% for step in scenario["steps"] %}
    <details><summary>{{ step["status_icon"] }} {{ step["name"] }}  - {{ step["stop"] - step["start"] }}ms</summary><blockquote>
    
    - Status: {{ step["status"] }}
    - Duration: {{ step["stop"] - step["start"] }}ms ({{ step["start_time"] }} - {{ step["stop_time"] }})
    
    {% for attach in step["attachments"] %}
    {{ attach["name"] }}
    {% if attach["type"] == "text/csv" %}
    {{ attach["body"] }}
    {% else %}
    ```
    {{ attach["body"] }}
    ```
    {% endif %}
    {% endfor %}
    {% if step["statusDetails"] %}
    Message
    ```
    {{ step["statusDetails"]["message"] }}
    ```
    {% if step["statusDetails"]["trace"] %}
    Trace 
    ```
    {{ step["statusDetails"]["trace"] }}
    ```
    {% endif %}
    {% endif %}
    
    </blockquote></details>
    {% endfor %}
    </blockquote></details>
    {% endfor %}
    
    {% endfor %}
    """
    template = jinja2.Template(source=textwrap.dedent(tpl))
    print(template.render(results=results, stats=stats))


if __name__ == "__main__":
    if len(sys.argv) < 1:
        print("usage: python bin/allure2markdown.py [allure-results dir path]")
        exit(1)
    allure_results_dir = sys.argv[1]
    transform_allure_report2markdown(allure_results_dir)