store-tests

#!/usr/bin/env python3

# This file is part of Cockpit.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <http://www.gnu.org/licenses/>.

import argparse
import logging
import re
import sqlite3
import sys
import urllib.parse
import urllib.request
from collections import defaultdict
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from datetime import datetime

from lib import github, testmap
from lib.aio.jsonutil import get_str
from lib.network import host_ssl_context

# Result line has form like:
# ok 2 test/verify/check-accounts TestAccounts.testExpire [ND@1]
# ok 103 test/verify/check-multi-machine TestMultiMachine.testTroubleshooting
# ok 289 test/verify/check-machines-lifecycle TestMachinesLifecycle.testDelete [ND@1] # SKIP Known issue #1543

# not ok 117 test/verify/check-metrics TestMetrics.testPcp
# not ok 305 test/verify/check-networking-bond TestBonding.testActive [ND@1]
# not ok 24 test/verify/check-foobar TestExample.testBasic [ND] # RETRY 2 (be robust against unstable tests)
re_run = re.compile(r"(not )?ok \d+ (\S+) (\S+)[^#]*(?:# RETRY \d+ \(([^)]+))?")

# duration appears in the line right above the ok/not okTAP result:
# 1 TEST PASSED [26s on centosci-tasks-6w5w6]
re_duration = re.compile(r"# \d TEST (FAILED|PASSED) \[(\d+)s")

DATE_FORMAT = "%Y-%m-%dT%H:%M:%S%z"


@dataclass
class TestRun:
    posted: float
    state: str
    started: float | None = None
    finished: float | None = None
    url: str | None = None
    description: str | None = None


def main() -> None:
    parser = argparse.ArgumentParser(description='Store information about failed tests')
    parser.add_argument("--db", default="test-results.db", help="Database name")
    parser.add_argument("--repo", help="Repository from which to process failures")
    parser.add_argument("--verbose", action="store_true", help="Enable verbose logging")
    parser.add_argument("--test-parser", metavar="LOGFILE",
                        help="Scan given log file for tests and print results; for testing the parser")
    parser.add_argument("revision", help="SHA for which failures should be stored")
    opts = parser.parse_args()

    if opts.verbose:
        logging.basicConfig(level=logging.DEBUG)

    if opts.test_parser:
        test_parser(opts.test_parser)
        return

    db_conn = sqlite3.connect(opts.db)
    cursor = db_conn.cursor()
    init_db(cursor)

    api = github.GitHub(repo=opts.repo)
    statuses = api.all_statuses(opts.revision)
    if not statuses:
        sys.exit(f"Revision {opts.revision} has no statuses")

    contexts: defaultdict[str, list[TestRun]] = defaultdict(list)

    # Read the "status story"
    # First we get 'Not yet tested' or 'Not yet tested (direct trigger)'
    # Then when it is picked up we get "Testing in progress" and finally it is either red or green
    # Putting it all together can give us data like number of retries, how long it waited to get
    # testes, how long testing took...
    # Create map from 'context' to array of 'testrun's where each is a map containing: (order in the
    # array indicates retries, the first is original run, the second is the first retry...)
    # 'repo', 'url', 'state', 'description'
    # 'started' - when this run was picked by bot
    # 'posted' - when this run was triggered
    # 'finished' - when this run finished - 'state' is then either 'success', 'failure' or 'error'
    for status in reversed(statuses):
        if not testmap.is_valid_context(get_str(status, "context"), api.repo):
            continue

        cc = contexts[get_str(status, "context")]

        desc = get_str(status, "description")
        if get_str(status, "state") == "pending":
            if "progress" in desc:
                cc[-1].url = get_str(status, "target_url")
                cc[-1].started = datetime.strptime(get_str(status, "created_at"), DATE_FORMAT).timestamp()  # noqa:DTZ007
            elif "Not yet tested" in desc:
                cc.append(TestRun(
                    posted=datetime.strptime(get_str(status, "created_at"), DATE_FORMAT).timestamp(),  # noqa:DTZ007
                    state="pending",
                ))
            else:
                logging.warning("Pending status has unexpected description '%s', skipping.", desc)
        elif get_str(status, "state") in ["success", "failure", "error"]:
            cc[-1].finished = datetime.strptime(get_str(status, "created_at"), DATE_FORMAT).timestamp()  # noqa:DTZ007
            cc[-1].state = get_str(status, "state")
            if get_str(status, "state") == "error":
                cc[-1].description = desc
        else:
            logging.warning("Status has unexpected state '%s', skipping.", get_str(status, "state"))

    for context, runs in contexts.items():
        repo = api.repo
        if "@" in context:
            parts = context.split("@")
            context = parts[0]
            if not parts[1].startswith("bots"):
                # Can contain branch (e.g `cockpit-project/cockpit/rhel-8`)
                # if not given, defaults to primary branch
                repo = parts[1]

        for retry, run in enumerate(runs):
            started = run.started if run.started is not None else run.posted
            waited = started - run.posted
            took = (run.finished if run.finished is not None else started) - started
            logging.debug(
                "Processing %s%s as %s that waited %ss and took %ss in %s. %s",
                context,
                "(retry no." + str(retry) + ")" if retry else "",
                run.state + (f" ({run.description})" if run.description else ""),
                waited, took, repo, run.url
            )

            # URL is our primary key, so if we don't have one, just fake something unique
            url = run.url or f"queued-{opts.revision}-{context}-{retry}"

            row = cursor.execute("SELECT id FROM TestRuns WHERE url = ?", (url, )).fetchone()
            if row is not None:
                logging.warning("Test run %s already exists as ID %i, skipping", url, row[0])
                continue

            run_id = insert_run(cursor, repo, opts.revision, context, url,
                                run.posted, retry, waited, took, run.state,
                                run.description)

            if run.url:  # If we faked 'url' then don't try to scan it
                raw_log = url[:-5] if url.endswith(".html") else url
                logging.debug("scanning log %s", raw_log)

                # Set headers to avoid getting blocked by Anubis
                req = urllib.request.Request(raw_log, headers={'Accept': 'text/plain'})
                with urllib.request.urlopen(req,
                                            context=host_ssl_context(urllib.parse.urlparse(raw_log).netloc)) as fp:
                    data = fp.readlines()
                    for (testname, failed, retry_reason, seconds) in find_tests(data):
                        insert_test(cursor, run_id, testname, failed, retry_reason, seconds)

                    # Find coverage line - normally second to last line, so check only last 20 lines
                    for line in data[-20:]:
                        if b"Overall line coverage: " in line:
                            coverage_str = line.split(b"Overall line coverage: ")[1].strip()
                            coverage = float(coverage_str.strip(b"%"))
                            insert_coverage(cursor, run_id, coverage)
                            break

    db_conn.commit()
    db_conn.close()


def init_db(cursor: sqlite3.Cursor) -> None:
    cursor.execute("""CREATE TABLE if not exists TestRuns
                    (id INTEGER PRIMARY KEY,
                     project TEXT,
                     revision TEXT,
                     context TEXT,
                     url TEXT,
                     time TIMESTAMP,
                     retry INTEGER,
                     wait_seconds INTEGER,
                     run_seconds INTEGER,
                     state TEXT,
                     description TEXT,
                     UNIQUE (url))
                   """)
    cursor.execute("""CREATE TABLE if not exists Tests
                    (testname TEXT NOT NULL,
                     retry_reason TEXT,
                     failed INTEGER,
                     run INTEGER NOT NULL,
                     seconds INTEGER NOT NULL,
                     FOREIGN KEY (run) REFERENCES TestRuns(id))
                   """)
    cursor.execute("""CREATE TABLE if not exists TestCoverage
                    (coverage REAL,
                     run INTEGER NOT NULL,
                     FOREIGN KEY (run) REFERENCES TestRuns(id))
                   """)


def insert_run(
    cursor: sqlite3.Cursor,
    project: str,
    revision: str,
    context: str,
    url: str,
    time: float,
    retry: int,
    waited: float,
    took: float,
    state: str,
    desc: str | None,
) -> int | None:
    return cursor.execute("INSERT INTO TestRuns VALUES (null, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                          (project, revision, context, url, time, retry, waited, took, state, desc)).lastrowid


def insert_test(
    cursor: sqlite3.Cursor,
    run_id: int | None,
    testname: str,
    failed: bool,
    retry_reason: str | None,
    seconds: str | int,
) -> None:
    cursor.execute("INSERT INTO Tests VALUES (?, ?, ?, ?, ?)",
                   (testname, retry_reason, int(failed), run_id, int(seconds)))


def insert_coverage(cursor: sqlite3.Cursor, run_id: int | None, coverage: float) -> None:
    cursor.execute("INSERT INTO TestCoverage VALUES (?, ?)", (coverage, run_id))


def find_tests(fp: Iterable[bytes]) -> Iterator[tuple[str, bool, str | None, str | int]]:
    last_msg = ""
    prev_msg = ""
    save_message = False
    for raw_line in fp:
        line = raw_line.decode('utf-8')

        if save_message:
            last_msg = line.strip()
            save_message = False
            continue

        # If the issue is known, don't save unexpected messages
        if line.startswith("ok ") and "SKIP Known issue" in line:
            save_message = False
            last_msg = ""

        if "FAIL: Test completed, but found unexpected" in line and "raise" not in line:
            save_message = True

        m = re_run.match(line)
        if m:
            # sometimes line breaks are missing:
            # not ok 117 test/verify/check-metrics TestMetrics.testPcp
            testname = f"{m.group(2)} {m.group(3)}".rstrip("#")
            if "make-checkout-workdir" in testname:
                testname = testname.split("make-checkout-workdir")[1]
            t = re_duration.match(prev_msg)
            yield (testname, bool(m.group(1)), last_msg or (m.group(4) or None), (t and t.group(2)) or 0)

            last_msg = ""

        prev_msg = line.strip()


def test_parser(logfile: str) -> None:
    with open(logfile, 'rb') as f:
        for (testname, failed, retry_reason, seconds) in find_tests(f):
            print(f"{'FAIL' if failed else 'PASS'} {testname} "
                  f"{('RETRY:' + retry_reason) if retry_reason else ''} time {seconds}s")


if __name__ == '__main__':
    main()