intel-hard-lab/cts/run_cts.py

#!/usr/bin/env python3
import collections
import glob
import os
import re
import subprocess
import sys
import threading
import time
from datetime import datetime
from pathlib import Path


def now_iso():
    return datetime.now().astimezone().isoformat(timespec="seconds")


def read_lines(path):
    if not os.path.exists(path):
        return []
    with open(path) as f:
        return [l.rstrip("\n") for l in f if l.strip()]


def write_lines(path, lines):
    with open(path, "w") as f:
        for l in lines:
            f.write(l + "\n")


def begin_test_case_names(qpa_path):
    names = set()
    if not os.path.exists(qpa_path):
        return names
    txt = open(qpa_path, errors="replace").read()
    for line in txt.splitlines():
        if line.startswith("#beginTestCaseResult "):
            names.add(line[len("#beginTestCaseResult "):].strip())
    return names


def last_begin_test_case_name(qpa_path):
    last = None
    if not os.path.exists(qpa_path):
        return last
    for line in open(qpa_path, errors="replace"):
        if line.startswith("#beginTestCaseResult "):
            last = line[len("#beginTestCaseResult "):].strip()
    return last


def nproc():
    return os.cpu_count() or 1


def main():
    lab_root = Path(os.environ.get("LAB_ROOT", Path(__file__).resolve().parent.parent))
    drivers_root = Path(os.environ.get("DRIVERS_ROOT", lab_root.parent))
    mesa_src = Path(os.environ.get("MESA_SRC", drivers_root / "intel-hard-graphics"))
    mesa_build = Path(os.environ.get("MESA_BUILD", mesa_src / "build"))
    cts_dir = Path(os.environ.get("CTS_DIR", mesa_src / "VK-GL-CTS/build-cts/external/vulkancts/modules/vulkan"))
    cts_mustpass = Path(os.environ.get("CTS_MUSTPASS", mesa_src / "VK-GL-CTS/external/vulkancts/mustpass/main/vk-default"))
    icd = Path(os.environ.get("VK_DRIVER_FILES", mesa_build / "src/intel/vulkan_hasvk/intel_hasvk13_devenv_icd.x86_64.json"))
    exclude = lab_root / "cts/cts-exclude-gen7.txt"

    scope = sys.argv[1] if len(sys.argv) > 1 else "mustpass"
    shards = int(sys.argv[2]) if len(sys.argv) > 2 else nproc()
    cooldown = int(os.environ.get("COOLDOWN", "60"))
    outdir = Path(os.environ.get(
        "OUTDIR", Path.cwd() / f"cts-run-{datetime.now().strftime('%Y%m%d-%H%M%S')}"))

    deqp = cts_dir / "deqp-vk"
    for p in (deqp, icd):
        if not p.exists():
            print(f"missing: {p}", file=sys.stderr)
            sys.exit(1)

    outdir.mkdir(parents=True, exist_ok=True)
    os.chdir(cts_dir)

    cases_txt = outdir / "cases.txt"
    if scope == "all":
        out = subprocess.run([str(deqp), "--deqp-runmode=stdout-caselist"],
                              capture_output=True, text=True).stdout
        lines = [l[len("TEST: "):] for l in out.splitlines() if l.startswith("TEST: ")]
        src_txt = outdir / "all.txt"
        write_lines(src_txt, lines)
    else:
        lines = []
        for f in sorted(glob.glob(str(cts_mustpass / "*.txt"))):
            lines.extend(l for l in read_lines(f) if l.startswith("dEQP-VK"))
        src_txt = outdir / "mustpass.txt"
        write_lines(src_txt, lines)

    src_lines = read_lines(src_txt)
    if exclude.exists() and exclude.stat().st_size > 0:
        excl = set(read_lines(exclude))
        case_lines = [l for l in src_lines if l not in excl]
    else:
        case_lines = list(src_lines)
    write_lines(cases_txt, case_lines)
    print(f"{len(src_lines)} cases, {len(src_lines) - len(case_lines)} excluded, {len(case_lines)} to run")

    prev = sorted(glob.glob(str(outdir / "*.qpa"))) + sorted(glob.glob(str(outdir / "attempt-*/*.qpa")))
    if prev:
        done = set()
        for f in prev:
            done |= begin_test_case_names(f)
        done_txt = outdir / "done.txt"
        write_lines(done_txt, sorted(done))
        if done:
            case_lines = [l for l in read_lines(cases_txt) if l not in done]
            write_lines(cases_txt, case_lines)
            print(f"resume: {len(done)} already recorded, {len(case_lines)} left")

        n = 1
        while (outdir / f"attempt-{n}").exists():
            n += 1
        attempt_dir = outdir / f"attempt-{n}"
        attempt_dir.mkdir(parents=True, exist_ok=True)
        for f in list(glob.glob(str(outdir / "shard*.qpa"))) + list(glob.glob(str(outdir / "shard*.log"))):
            os.rename(f, attempt_dir / os.path.basename(f))
        print(f"previous attempt parked in attempt-{n}/")

    case_lines = read_lines(cases_txt)
    if not case_lines:
        print("nothing left to run")
        sys.exit(0)

    def run_shard(i):
        list_path = outdir / f"shard{i}.txt"
        shard_lines = [l for j, l in enumerate(case_lines) if (j + 1) % shards == i]
        write_lines(list_path, shard_lines)
        pass_n = 1

        while list_path.exists() and list_path.stat().st_size > 0:
            qpa = outdir / f"shard{i}.pass{pass_n}.qpa"
            log = outdir / f"shard{i}.pass{pass_n}.log"
            env = dict(os.environ, VK_DRIVER_FILES=str(icd))
            with open(log, "w") as lf:
                subprocess.run(
                    [str(deqp), f"--deqp-caselist-file={list_path}",
                     f"--deqp-log-filename={qpa}",
                     "--deqp-log-images=disable", "--deqp-log-shader-sources=disable"],
                    stdout=lf, stderr=subprocess.STDOUT, env=env)

            log_text = open(log, errors="replace").read() if log.exists() else ""
            if "ABORTED" not in log_text:
                break

            last = last_begin_test_case_name(qpa)
            if not last:
                break

            print(f"shard {i} aborted at {last} -- cooling down {cooldown}s", file=sys.stderr)
            with open(outdir / "aborted.txt", "a") as af:
                af.write(last + "\n")
            time.sleep(cooldown)

            shard_lines = read_lines(list_path)
            seen = False
            remaining = []
            for l in shard_lines:
                if seen:
                    remaining.append(l)
                if l == last:
                    seen = True
            write_lines(list_path, remaining)
            pass_n += 1

    threads = [threading.Thread(target=run_shard, args=(i,)) for i in range(shards)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    res = {}
    for f in sorted(glob.glob(str(outdir / "shard*.qpa"))) + sorted(glob.glob(str(outdir / "attempt-*/shard*.qpa"))):
        txt = open(f, errors="replace").read()
        for block in re.split(r"#beginTestCaseResult ", txt)[1:]:
            name = block.split("\n", 1)[0].strip()
            m = re.search(r'<Result StatusCode="(\w+)">', block)
            if m:
                res[name] = m.group(1)

    counts = collections.Counter(res.values())
    print(f"\n{sum(counts.values())} results")
    for k, v in counts.most_common():
        print(f"  {k:16} {v}")

    with open(outdir / "failures.txt", "w") as fh:
        for name, status in sorted(res.items()):
            if status not in ("Pass", "NotSupported"):
                fh.write(f"{status}\t{name}\n")
    print(f"\nfailures -> {outdir}/failures.txt")


if __name__ == "__main__":
    main()