<commit_before>from __future__ import annotations

import html
import io
import os
import re
import zipfile

import altair
import altair_saver
import junitparser
import pandas
import requests

TOKEN = os.environ.get("GITHUB_TOKEN")

# Mapping between a symbol (pass, fail, skip) and a color
COLORS = {
    "✓": "#acf2a5",
    "x": "#f2a5a5",
    "s": "#f2ef8f",
}


def get_from_github(url, params={}):
    """
    Make an authenticated request to the GitHub REST API.
    """
    r = requests.get(url, params=params, headers={"Authorization": f"token {TOKEN}"})
    r.raise_for_status()
    return r


def maybe_get_next_page_path(response):
    """
    If a response is paginated, get the url for the next page.
    """
    link_regex = re.compile(r'<([^>]*)>;\s*rel="([\w]*)\"')
    link_headers = response.headers.get("Link")
    next_page_path = None
    if link_headers:
        links = {}
        matched = link_regex.findall(link_headers)
        for match in matched:
            links[match[1]] = match[0]
        next_page_path = links.get("next", None)

    return next_page_path


def get_workflow_listing(repo="dask/distributed", branch="main", event="push"):
    """
    Get a list of workflow runs from GitHub actions.
    """
    params = {"per_page": 100, "branch": branch, "event": event}
    r = get_from_github(
        f"https://api.github.com/repos/{repo}/actions/runs", params=params
    )
    workflows = r.json()["workflow_runs"]
    next_page = maybe_get_next_page_path(r)
    while next_page:
        r = get_from_github(next_page)
        workflows = workflows + r.json()["workflow_runs"]
        next_page = maybe_get_next_page_path(r)

    return workflows


def get_artifacts_for_workflow(run_id, repo="dask/distributed"):
    """
    Get a list of artifacts from GitHub actions
    """
    params = {"per_page": 100}
    r = get_from_github(
        f"https://api.github.com/repos/{repo}/actions/runs/{run_id}/artifacts",
        params=params,
    )
    artifacts = r.json()["artifacts"]
    next_page = maybe_get_next_page_path(r)
    while next_page:
        r = get_from_github(next_page)
        artifacts = workflows + r.json()["workflow_runs"]
        next_page = maybe_get_next_page_path(r)

    return artifacts


def suite_from_name(name: str):
    """
    Get a test suite name from an artifact name. The artifact
    can have matrix partitions, pytest marks, etc. Basically,
    just lop off the front of the name to get the suite.
    """
    return "-".join(name.split("-")[:3])


def download_and_parse_artifact(url):
    """
    Download the artifact at the url parse it.
    """
    try:
        r = get_from_github(url)
        f = zipfile.ZipFile(io.BytesIO(r.content))
        run = junitparser.JUnitXml.fromstring(f.read(f.filelist[0].filename))
        return run
    except Exception:
        print(f"Failed to download/parse {url}")
        return None


def dataframe_from_jxml(run):
    """
    Turn a parsed JXML into a pandas dataframe
    """
    fname = []
    tname = []
    status = []
    message = []
    for suite in run:
        for test in suite:
            fname.append(test.classname)
            tname.append(test.name)
            s = "✓"
            result = test.result

            if len(result) == 0:
                status.append(s)
                message.append("")
                continue
            result = result[0]
            m = result.message if result and hasattr(result, "message") else ""
            if isinstance(result, junitparser.Error):
                s = "x"
            elif isinstance(result, junitparser.Failure):
                s = "x"
            elif isinstance(result, junitparser.Skipped):
                s = "s"
            else:
                s = "x"
            status.append(s)
            message.append(html.escape(m))
    df = pandas.DataFrame(
        {"file": fname, "test": tname, "status": status, "message": message}
    )

    # There are sometimes duplicate tests in the report for some unknown reason.
    # If that is the case, concatenate the messages and prefer to show errors.
    def dedup(group):
        if len(group) > 1:
            if "message" in group.name:
                return group.str.cat(sep="")
            else:
                if (group == "x").any(axis=0):
                    return "x"
                else:
                    return group.iloc[0]
        else:
            return group

    return df.groupby(["file", "test"]).agg(dedup)


if __name__ == "__main__":
    if not TOKEN:
        raise RuntimeError("Failed to find a GitHub Token")
    print("Getting all recent workflows...")
    workflows = get_workflow_listing()

    # Filter the workflows listing to be in the retention period,
    # and only be test runs (i.e., no linting) that completed.
    workflows = [
        w
        for w in workflows
        if (
            pandas.to_datetime(w["created_at"])
            > pandas.Timestamp.now(tz="UTC") - pandas.Timedelta(days=90)
            and w["conclusion"] != "cancelled"
            and w["name"].lower() == "tests"
        )
    ]
    # Each workflow processed takes ~10-15 API requests. To avoid being
    # rate limited by GitHub (1000 requests per hour) we choose just the
    # most recent N runs. This also keeps the viz size from blowing up.
    workflows = sorted(workflows, key=lambda w: w["created_at"])[-50:]
<commit_msg>Show scheduled test runs in report (#5812)<commit_after>from __future__ import annotations

import html
import io
import os
import re
import zipfile

import altair
import altair_saver
import junitparser
import pandas
import requests

TOKEN = os.environ.get("GITHUB_TOKEN")

# Mapping between a symbol (pass, fail, skip) and a color
COLORS = {
    "✓": "#acf2a5",
    "x": "#f2a5a5",
    "s": "#f2ef8f",
}


def get_from_github(url, params={}):
    """
    Make an authenticated request to the GitHub REST API.
    """
    r = requests.get(url, params=params, headers={"Authorization": f"token {TOKEN}"})
    r.raise_for_status()
    return r


def maybe_get_next_page_path(response):
    """
    If a response is paginated, get the url for the next page.
    """
    link_regex = re.compile(r'<([^>]*)>;\s*rel="([\w]*)\"')
    link_headers = response.headers.get("Link")
    next_page_path = None
    if link_headers:
        links = {}
        matched = link_regex.findall(link_headers)
        for match in matched:
            links[match[1]] = match[0]
        next_page_path = links.get("next", None)

    return next_page_path


def get_workflow_listing(repo="dask/distributed", branch="main", event="push"):
    """
    Get a list of workflow runs from GitHub actions.
    """
    since = str((pandas.Timestamp.now(tz="UTC") - pandas.Timedelta(days=90)).date())
    params = {"per_page": 100, "branch": branch, "event": event, "created": f">{since}"}
    r = get_from_github(
        f"https://api.github.com/repos/{repo}/actions/runs", params=params
    )
    workflows = r.json()["workflow_runs"]
    next_page = maybe_get_next_page_path(r)
    while next_page:
        r = get_from_github(next_page)
        workflows = workflows + r.json()["workflow_runs"]
        next_page = maybe_get_next_page_path(r)

    return workflows


def get_artifacts_for_workflow(run_id, repo="dask/distributed"):
    """
    Get a list of artifacts from GitHub actions
    """
    params = {"per_page": 100}
    r = get_from_github(
        f"https://api.github.com/repos/{repo}/actions/runs/{run_id}/artifacts",
        params=params,
    )
    artifacts = r.json()["artifacts"]
    next_page = maybe_get_next_page_path(r)
    while next_page:
        r = get_from_github(next_page)
        artifacts = workflows + r.json()["workflow_runs"]
        next_page = maybe_get_next_page_path(r)

    return artifacts


def suite_from_name(name: str):
    """
    Get a test suite name from an artifact name. The artifact
    can have matrix partitions, pytest marks, etc. Basically,
    just lop off the front of the name to get the suite.
    """
    return "-".join(name.split("-")[:3])


def download_and_parse_artifact(url):
    """
    Download the artifact at the url parse it.
    """
    try:
        r = get_from_github(url)
        f = zipfile.ZipFile(io.BytesIO(r.content))
        run = junitparser.JUnitXml.fromstring(f.read(f.filelist[0].filename))
        return run
    except Exception:
        print(f"Failed to download/parse {url}")
        return None


def dataframe_from_jxml(run):
    """
    Turn a parsed JXML into a pandas dataframe
    """
    fname = []
    tname = []
    status = []
    message = []
    for suite in run:
        for test in suite:
            fname.append(test.classname)
            tname.append(test.name)
            s = "✓"
            result = test.result

            if len(result) == 0:
                status.append(s)
                message.append("")
                continue
            result = result[0]
            m = result.message if result and hasattr(result, "message") else ""
            if isinstance(result, junitparser.Error):
                s = "x"
            elif isinstance(result, junitparser.Failure):
                s = "x"
            elif isinstance(result, junitparser.Skipped):
                s = "s"
            else:
                s = "x"
            status.append(s)
            message.append(html.escape(m))
    df = pandas.DataFrame(
        {"file": fname, "test": tname, "status": status, "message": message}
    )

    # There are sometimes duplicate tests in the report for some unknown reason.
    # If that is the case, concatenate the messages and prefer to show errors.
    def dedup(group):
        if len(group) > 1:
            if "message" in group.name:
                return group.str.cat(sep="")
            else:
                if (group == "x").any(axis=0):
                    return "x"
                else:
                    return group.iloc[0]
        else:
            return group

    return df.groupby(["file", "test"]).agg(dedup)


if __name__ == "__main__":
    if not TOKEN:
        raise RuntimeError("Failed to find a GitHub Token")
    print("Getting all recent workflows...")
    workflows = get_workflow_listing(event="push") + get_workflow_listing(
        event="schedule"
    )

    # Filter the workflows listing to be in the retention period,
    # and only be test runs (i.e., no linting) that completed.
    workflows = [
        w
        for w in workflows
        if (
            pandas.to_datetime(w["created_at"])
            > pandas.Timestamp.now(tz="UTC") - pandas.Timedelta(days=90)
            and w["conclusion"] != "cancelled"
            and w["name"].lower() == "tests"
        )
    ]
    # Each workflow processed takes ~10-15 API requests. To avoid being
    # rate limited by GitHub (1000 requests per hour) we choose just the
    # most recent N runs. This also keeps the viz size from blowing up.
    workflows = sorted(workflows, key=lambda w: w["created_at"])[-50:]

{'fraction_non_alphanumeric': 0.1197317139490619, 'fraction_numerical': 0.00670715127345237, 'mean_word_length': 2.330516148505886, 'pattern_counts': {'":': 27, '<': 5, '<?xml version=': 0, '>': 12, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1360502', 'n_tokens_mistral': 3553, 'n_tokens_neox': 3407, 'n_words': 923}