run

#!/usr/bin/env python3

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

import json5


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

    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('pylint', 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(
            '--check',
            action='store_true',
            help='Check the package intead of publishing.',
        )
        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):
        self._check_version()
        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'])
        self.call(
            ['rm', '-fr', '.coverage', 'build', 'dist', 'json5.egg-info']
        )

    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',
                '-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_pylint(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_pylint()
        self.run_mypy()
        self.run_coverage()

        # Build is way too noisy by default
        quiet = self.args.quiet
        self.args.quiet = True
        self.run_build()
        self.args.quiet = quiet

        self.run_publish()

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

        self._check_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 = ['--repository', 'testpypi']
        else:
            test = []
        if self.args.check:
            self.call(self.run_cmd + ['-m', 'twine', 'check'] + [tgz, wheel])
        else:
            self.call(
                self.run_cmd + ['-m', 'twine', 'upload'] + test + [tgz, wheel]
            )

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

        with open(self.parser_file, encoding='utf-8') as fp:
            new = fp.read()
        if self.args.check:
            os.remove(self.parser_file)
            os.rename(orig_file, self.parser_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:
            print(f'{self.parser_file} regenerated.')

    def run_tests(self):
        self.call(self.run_cmd + ['-m', 'doctest', 'json5/lib.py'])

        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):
        tool = '../glop/glop/tool.py'
        if not os.path.exists(tool):
            print('Need `glop` repo to be at ../glop.')
            sys.exit(1)

        self.call(
            [
                sys.executable,
                tool,
                '-o',
                self.parser_file,
                '--no-main',
                '--no-memoize',
                '-c',
                self.grammar,
            ]
        )
        self.call(self.run_cmd + ['-m', 'ruff', 'format', self.parser_file])

    def _files(self):
        return (
            ['run']
            + glob.glob(f'{self.package}/*.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 _check_version(self):
        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:]))