RetroArch/tools/ci_paths.py

#!/usr/bin/env python3
"""Write each workflow's push and pull_request paths filters.

A push touching only samples runs only the workflows that build or run
those samples, and a push touching one workflow file runs only that
workflow. Every other change runs everything, as before: the filter is

    '**'                            everything ...
    '!samples/**'                   ... but the samples trees
    '!libretro-common/samples/**'
    '!.github/workflows/*.yml'      ... and the other workflows,
    'samples/gfx/foo/**'            then the sample directories this
    ...                             workflow's jobs name, and those
    '.github/workflows/<self>.yml'  name in turn, and itself.

The sample directories are what tools/ci_local.py's coverage finds in
the jobs' run: steps and the scripts they call, closed over every path
the files in those directories name (a Makefile reaching into a
neighbouring sample, a fixture script another one runs). A workflow
whose jobs walk the samples trees or read the workflow files keeps
them whole. Workflows with a filter of their own are left as they are.

  tools/ci_paths.py           rewrite the generated filters
  tools/ci_paths.py --check   exit 1 if any is out of date
"""
import os
import re
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import ci_local as c  # noqa: E402

BEGIN = '    # ci_paths: generated by tools/ci_paths.py, do not edit'
END = '    # ci_paths: end'
TREES = ('samples/', 'libretro-common/samples/')
# Scripts every job calls for its own setup; they read nothing else
INFRA = {'.github/workflows/scripts/apt-cache-key.sh',
         '.github/workflows/scripts/apt-install.sh',
         '.github/workflows/scripts/use-lld.sh',
         '.github/workflows/scripts/wine-server.sh',
         '.github/workflows/scripts/ccache-make.sh'}
# A walk over a samples tree as a whole: the word with no directory below it
_ALL_SAMPLES = re.compile(r'(?<![\w.-])(?:libretro-common/)?samples'
                          r'(?:/\*|/?["\'\s)]|/?$)', re.M)
# A script that lists directories may reach any sample
_WALK = re.compile(r'os\.walk|os\.listdir|\bglob\b|rglob|iterdir|\bfind\s|ls-files'
                   r'|readdir|\$\(ls\b|\*\*?/|/\*')
_WORKFLOWS = re.compile(r'\.github/workflows(?!/scripts/)(?!/[\w.-]+\.yml)'
                        r'|\bci_local\b|\bload_workflows\b')


def code(text):
    """A script's text less its comments and the lines that only print."""
    return '\n'.join(l for l in text.split('\n')
                     if not re.match(r'\s*(?:#|//|\*|/\*|echo\b|printf\b|print\()', l))


def sample_dir(rel):
    """The sample directory a path lies in, or None outside the trees."""
    for t in TREES:
        if rel.startswith(t):
            parts = rel[len(t):].split('/')
            return t + '/'.join(parts[:2]) if len(parts) >= 2 else t + parts[0]
    return None


def read(rel):
    try:
        with open(os.path.join(c.ROOT, rel), encoding='utf-8', errors='replace') as f:
            return f.read()
    except OSError:
        return ''


def named_paths(text, base):
    """Repository paths a text names, read from base and from the root."""
    out = set()
    for tok in set(c._PATHISH.findall(text)):
        t = tok.strip('"\'(),;:')
        if not t:
            continue
        t = c._strip_vars(t) if '$' in t else t
        for b in (base, c.ROOT):
            p = os.path.normpath(os.path.join(b, t))
            if p.startswith(c.ROOT + os.sep) and os.path.exists(p):
                out.add(os.path.relpath(p, c.ROOT).replace(os.sep, '/'))
                break
    return out


def plan(wf):
    """The generated filter for a workflow, or None to leave it alone."""
    trig = wf.triggers()
    events = [e for e in ('push', 'pull_request') if e in trig]
    if not events:
        return None
    for e in events:
        spec = trig.get(e)
        if isinstance(spec, dict) and (spec.get('paths') or spec.get('paths-ignore')) \
                and BEGIN not in wf.text:
            return None
    cov = c.Coverage()
    texts = []
    for job in wf.jobs.values():
        if not isinstance(job, dict) or 'steps' not in job:
            return None
        for step in job['steps']:
            if str(step.get('uses') or '').startswith('./'):
                return None
        for step, text, wd in c.job_texts(job):
            base = os.path.join(c.ROOT, wd) if wd and '${{' not in wd else c.ROOT
            cov.add_text(text + '\n' + wd, base, 'run')
            texts.append(text + '\n' + wd)
    scripts = [f for f in cov.files if f.endswith(('.sh', '.py')) and f not in INFRA]
    # A tool or CI script may take its directories from a list or a walk
    # no reading of its text can be sure of: it keeps the trees whole
    tools = [f for f in cov.files if f.startswith(('tools/', '.github/workflows/scripts/'))
             and f not in INFRA]
    every = code('\n'.join(texts + [read(f) for f in scripts]))
    keep_samples = bool(tools) or bool(_ALL_SAMPLES.search(every)) \
        or any(_WALK.search(read(f)) for f in scripts)
    keep_workflows = bool(_WORKFLOWS.search(every))
    if 'ci_paths' in every:
        # The check reads every sample directory and workflow
        return []

    dirs = set()
    for rel in list(cov.files) + list(cov.dirs):
        d = sample_dir(rel)
        if d:
            dirs.add(d)
    for src in cov.suffixes:
        d = sample_dir(src)
        if d:
            dirs.add(d)
    todo = sorted(dirs)
    while todo:
        d = todo.pop()
        top = os.path.join(c.ROOT, d)
        if not os.path.isdir(top):
            continue
        for at, subdirs, names in os.walk(top):
            subdirs[:] = [s for s in subdirs if not s.startswith(('.', 'obj'))]
            for n in names:
                full = os.path.join(at, n)
                if os.path.getsize(full) > 1 << 20:
                    continue
                rel = os.path.relpath(full, c.ROOT).replace(os.sep, '/')
                text = read(rel)
                if '\0' in text:
                    continue
                if _ALL_SAMPLES.search(code(text)):
                    keep_samples = True
                for p in named_paths(text, at):
                    nd = sample_dir(p)
                    if nd and nd not in dirs:
                        dirs.add(nd)
                        todo.append(nd)

    paths = ['**']
    if not keep_samples:
        paths += ['!samples/**', '!libretro-common/samples/**']
    if not keep_workflows:
        paths += ['!.github/workflows/*.yml']
    if not keep_samples:
        # A file at a tree's top (a stub several samples link) is
        # filtered by its own path, not as a directory
        glob = lambda d: d if os.path.isfile(os.path.join(c.ROOT, d)) else d + '/**'
        paths += [glob(d) for d in sorted(dirs) if d.count('/') >= 2]
        paths += [glob(d) for d in sorted(dirs) if d.count('/') < 2]
    if not keep_workflows:
        paths.append('.github/workflows/' + wf.file)
    if paths == ['**']:
        return []
    return paths


def rewrite(wf, paths):
    lines = wf.text.split('\n')
    # drop the block written before
    out, skip = [], False
    for l in lines:
        if l == BEGIN:
            skip = True
            continue
        if skip:
            if l == END:
                skip = False
            continue
        out.append(l)
    lines = out
    if not paths:
        return '\n'.join(lines)
    block = [BEGIN, '    paths:'] + ["      - '%s'" % p for p in paths] + [END]
    on = lines.index('on:')
    i = on + 1
    result = lines[:i]
    while i < len(lines) and (lines[i].startswith(' ') or lines[i] == ''):
        m = re.match(r'  (push|pull_request):\s*$', lines[i])
        result.append(lines[i])
        i += 1
        if m:
            while i < len(lines) and lines[i].startswith('    '):
                result.append(lines[i])
                i += 1
            result += block
    return '\n'.join(result + lines[i:])


def main(argv):
    check = '--check' in argv
    stale = []
    for wf in c.load_workflows():
        paths = plan(wf)
        if paths is None:
            continue
        new = rewrite(wf, paths)
        if new != wf.text:
            stale.append(wf.file)
            if not check:
                with open(wf.path, 'w', encoding='utf-8') as f:
                    f.write(new)
    if check:
        if stale:
            for f in stale:
                print('ci_paths: %s: the paths filter is out of date' % f)
            print('ci_paths: run tools/ci_paths.py and commit the result')
            return 1
        print('ci_paths: every generated paths filter is current')
        return 0
    for f in stale:
        print('ci_paths: wrote %s' % f)
    return 0


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))