run

#!/usr/bin/env python3

import argparse
import glob
import os
import re
import shlex
import shutil
import subprocess
import sys

try:
    import pyfloyd
except ModuleNotFoundError:
    pass


class Runner:
    def __init__(self):
        self.arg_parser = None
        self.args = None
        self.grammar = 'grammars/floyd.g'
        self.package = 'pyfloyd'
        self.top = f'src/{self.package}'
        self.parser_file = f'{self.top}/parser.py'
        self.run_cmd = None
        self.subps = None
        self.uv_path = None
        self.version = None

    def add_parser(self, cmd, help):  # pylint: disable=redefined-builtin
        method = getattr(self, 'run_' + cmd.replace('-', '_'))
        subp = self.subps.add_parser(cmd, help=help)
        subp.add_argument(
            '-n',
            '--no-execute',
            action='store_true',
            help="Don't do anything that causes effects.",
        )
        subp.add_argument(
            '-q',
            '--quiet',
            action='store_true',
            help='Suppress output unless something fails.',
        )
        subp.add_argument(
            '-v',
            '--verbose',
            action='store_true',
            help='Echo commands as they are run.',
        )
        subp.set_defaults(func=lambda _: method())
        return subp

    def call(self, *args, **kwargs):
        cmd = shlex.join(*args)
        if self.args.no_execute or self.args.verbose:
            print(f'{cmd}')
        if self.args.no_execute:
            return None
        capture_output = kwargs.get('capture_output', self.args.quiet)
        if 'capture_output' in kwargs:
            del kwargs['capture_output']

        proc = subprocess.run(
            *args, capture_output=capture_output, check=False, **kwargs
        )
        if proc.returncode != 0:
            if self.args.quiet:
                print(proc.stdout)
                print(proc.stderr, file=sys.stderr)
            sys.exit(proc.returncode)
        return proc

    def main(self, argv):
        self.arg_parser = argparse.ArgumentParser(prog='run')
        self.subps = self.arg_parser.add_subparsers(required=True)

        self.add_parser('build', help='Build the package.')
        self.add_parser('check', help='Check the source code with ruff.')
        self.add_parser('checks', help='Same as `run check`.')
        self.add_parser('clean', help='Remove any local files.')

        subp = self.add_parser(
            'coverage', help='Run tests and report code coverage.'
        )
        subp.add_argument(
            '-b',
            '--branch',
            action='store_true',
            help='Report branch coverage.',
        )
        subp.add_argument(
            '--omit',
            help='Omit files whose paths match one of these patterns.',
        )
        subp.add_argument(
            '-u', '--unit', action='store_true', help='Only run unit tests.'
        )
        subp.add_argument('test', nargs='*', action='store')

        self.add_parser(
            'devenv',
            help='Set up a dev venv at //.venv with all the needed packages.',
        )

        subp = self.add_parser(
            'format', help='Format the source code with ruff.'
        )
        subp.add_argument(
            '--check',
            action='store_true',
            help='Just check to see if any files would be modified.',
        )

        subp = self.add_parser('help', help='Get help on a subcommand.')
        subp.add_argument(
            nargs='?',
            action='store',
            dest='subcommand',
            help='The command to get help for.',
        )

        self.add_parser('lint', help='Lint the source code with pylint.')

        self.add_parser('mypy', help='Typecheck the code with mypy.')

        subp = self.add_parser(
            'presubmit',
            help='Run all the steps that should be run prior to commiting.',
        )
        subp.add_argument(
            '-b',
            '--branch',
            action='store_true',
            help='Report branch coverage.',
        )
        subp.add_argument(
            '-f',
            '--failfast',
            action='store_true',
            help='Stop on first fail or error',
        )
        subp.add_argument(
            '--omit',
            help='Omit files whose paths match one of these patterns.',
        )
        subp.add_argument(
            '-u', '--unit', action='store_true', help='Only run unit tests.'
        )
        subp.add_argument('test', nargs='*', action='store')

        subp = self.add_parser('publish', help='Publish packages to PyPI.')
        subp.add_argument(
            '--test',
            action='store_true',
            help='Upload to the PyPI test instance.',
        )
        subp.add_argument(
            '--prod',
            action='store_true',
            help='Upload to the real PyPI instance.',
        )

        subp = self.add_parser('regen', help=f'Regenerate {self.parser_file}.')
        subp.add_argument(
            '--check',
            action='store_true',
            help='Just check to see if any files would be modified.',
        )

        subp = self.add_parser('tests', help='Run the tests.')
        subp.add_argument(
            '-d',
            '--durations',
            metavar='N',
            action='store',
            help='show the N slowed test cases (N=0 for all)',
        )
        subp.add_argument(
            '-f',
            '--failfast',
            action='store_true',
            help='Stop on first fail or error',
        )
        subp.add_argument(
            '-u', '--unit', action='store_true', help='Only run unit tests.'
        )
        subp.add_argument('test', nargs='*', action='store')

        self.args = self.arg_parser.parse_args(self._shuffle_argv(argv))

        self.uv_path = shutil.which('uv')
        if self.uv_path is None:
            print('You need to have `uv` installed to run this script.')
            sys.exit(2)

        if 'VIRTUAL_ENV' in os.environ:
            self.run_cmd = ['python']
        elif self.args.quiet:
            self.run_cmd = [self.uv_path, 'run', '--quiet', 'python']
        else:
            self.run_cmd = [self.uv_path, 'run', 'python']

        self.args.func(self.args)

    def run_build(self):
        cmd = [self.uv_path, 'build']
        if self.args.quiet:
            cmd.append('--quiet')
        self.call(cmd)

    def run_check(self):
        self.call(self.run_cmd + ['-m', 'ruff', 'check'])

    def run_checks(self):
        self.run_check()

    def run_clean(self):
        path = shutil.which('git')
        if path is None:
            print('You must have git installed to clean out the right files.')
            sys.exit(1)

        self.call([path, 'clean', '-fd'])

    def run_coverage(self):
        env = os.environ.copy()
        if self.args.unit:
            env['SKIP'] = 'integration'
            self.args.omit = '*_test.py'
        cmd = self.run_cmd + ['-m', 'coverage', 'run']
        if self.args.branch:
            cmd.append('--branch')
        cmd.extend(
            [
                '-m',
                'unittest',
                'discover',
                '-s',
                'tests',
                '-p',
                '*_test.py',
            ],
        )
        if self.args.verbose:
            cmd.append('-v')
        for test in self.args.test:
            cmd.extend(['-k', self._trim_test_glob(test)])
        self.call(cmd, env=env)
        cmd = self.run_cmd + ['-m', 'coverage', 'report', '--show-missing']
        if self.args.omit:
            cmd.append(f'--omit={self.args.omit}')
        self.call(cmd)

    def run_devenv(self):
        if self.uv_path is None:
            print('You need to have `uv` installed to set up a dev env.')
            sys.exit(2)

        cmd = [self.uv_path, 'sync', '--extra', 'dev']
        if self.args.quiet:
            cmd.append('--quiet')

        in_venv = 'VIRTUAL_ENV' in os.environ
        self.call(cmd)
        if not in_venv:
            print('Run `source .venv/bin/activate` to finish devenv setup.')

    def run_format(self):
        cmd = self.run_cmd + ['-m', 'ruff', 'format']
        if self.args.check:
            cmd.append('--check')
        self.call(cmd)

    def run_help(self):
        if self.args.subcommand:
            self.main([self.args.subcommand, '--help'])
        self.main(['--help'])

    def run_lint(self):
        self.call(self.run_cmd[:-1] + ['pylint', 'run'] + self._files())

    def run_mypy(self):
        self.call(self.run_cmd + ['-m', 'mypy'] + self._files())

    def run_presubmit(self):
        self.args.check = True
        self.run_regen()
        self.run_format()
        self.run_check()
        self.run_lint()
        self.run_mypy()
        self.run_coverage()

    def run_publish(self):
        if not self.args.test and not self.args.prod:
            print('You must specify either --test or --prod to upload.')
            sys.exit(2)

        self._get_version()
        sep = os.path.sep
        tgz = f'dist{sep}{self.package}-{self.version}.tar.gz'
        wheel = f'dist{sep}{self.package}-{self.version}-py3-none-any.whl'
        if not os.path.exists(tgz) or not os.path.exists(wheel):
            print('Run `./run build` first')
            return
        if self.args.test:
            test = ['--publish-url', 'https://test.pypi.org/legacy/']
        else:
            test = []
        self.call([self.uv_path, 'publish'] + test + [tgz, wheel])

    def run_regen(self):
        new_file = f'{self.parser_file}.new'
        with open(self.parser_file, encoding='utf-8') as fp:
            old = fp.read()
        self._gen_parser()
        if self.args.no_execute:
            if self.args.check:
                print(f'diff -q {self.parser_file} {new_file}')
            else:
                print(f'mv {new_file} {self.parser_file}')
            return

        with open(new_file, encoding='utf-8') as fp:
            new = fp.read()
        if self.args.check:
            os.remove(new_file)
            if old != new:
                print('Need to regenerate the parser with `run regen`.')
                sys.exit(1)
            print(f'{self.parser_file} is up to date.')
        elif old == new:
            print(f'{self.parser_file} is up to date.')
        else:
            os.rename(new_file, self.parser_file)
            print(f'{self.parser_file} regenerated.')

    def run_tests(self):
        cmd = self.run_cmd + ['-m', 'unittest', 'discover', '-p', '*_test.py']
        if self.args.quiet:
            cmd.append('-q')
        if self.args.verbose:
            cmd.append('-v')
        if self.args.failfast:
            cmd.append('-f')
        if self.args.durations:
            cmd.extend(['--durations', self.args.durations])

        for test in self.args.test:
            cmd.extend(['-k', self._trim_test_glob(test)])
        env = os.environ.copy()
        if self.args.unit:
            env['SKIP'] = 'integration'
        self.call(cmd, env=env)

    def set_func(self, subp, method):
        subp.set_defaults(func=lambda _: method())

    def _shuffle_argv(self, argv):
        # Take any flags that appear before the subcommand and append
        # them after the subcommand but before any flags following the
        # subcommand.
        leading_args = []
        argc = len(argv)
        i = 0
        while i < argc:
            if argv[i][0] != '-':
                break
            leading_args.append(argv[i])
            i += 1

        return argv[i : i + 1] + leading_args + argv[i + 1 :]

    def _gen_parser(self):
        new_file = f'{self.parser_file}.new'
        self.call(
            self.run_cmd
            + [
                '-m',
                'pyfloyd',
                '-o',
                new_file,
                '--memoize',
                '-c',
                '--python',
                self.grammar,
            ]
        )
        # TODO: Add a command line arg to set the name of the parser file.
        if not self.args.no_execute:
            with open(new_file, encoding='utf8') as fp:
                orig = fp.read()
            new = orig.replace(new_file, self.parser_file)
            if new != orig:
                with open(new_file, 'w', encoding='utf8') as fp:
                    fp.write(new)

    def _files(self):
        return (
            ['run'] + glob.glob(f'{self.top}/*.py') + glob.glob('tests/*.py')
        )

    def _trim_test_glob(self, test):
        if test.startswith('^'):
            test = test[1:]
        else:
            test = '*' + test
        if test.endswith('$'):
            test = test[:-1]
        else:
            test = test + '*'
        return test

    def _get_version(self):
        if self.version is not None:
            return
        if 'pyfloyd' in sys.modules:
            self.version = pyfloyd.__version__
            return

        proc = self.call(
            self.run_cmd + ['-m', 'pyfloyd', '-V'], capture_output=True
        )
        self.version = proc.stdout.decode('utf8').strip()

        m = re.match(r'\d+\.\d+\.\d+(\.dev\d+)?', self.version)
        if not m:
            print(f'Unexpected version format: "{self.version}"')
            sys.exit(1)


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