run_all_tests

#!/usr/bin/env python3
# (C) British Crown Copyright 2017-2026, Met Office.
"""Run all the tests in the CDDS packages."""
import argparse
import datetime
import os
import shutil
import subprocess
import sys

from timeit import default_timer as timer
from datetime import timedelta

CDDS_DIR = os.path.dirname(os.path.realpath(__file__))
LOG_NAME = os.path.join(CDDS_DIR, 'cdds_test_failures.log')
ROOT_COMMAND = 'pytest -s'
TESTS_TO_RUN = {
    'cdds': ['--doctest-modules', '-m slow', '-m integration'],
    'mip_convert': ['--doctest-modules', '-m slow'],
}

ADDITIONAL_OPTS = {
    'cdds': [],
}


def _red(text):
    """Returns text in red"""
    return '\033[31m{}\033[0m'.format(text)


def _green(text):
    """Returns text in green"""
    return '\033[32m{}\033[0m'.format(text)


def du(path):
    """Disk usage in human readable format (e.g. '2,1GB')"""
    out = subprocess.check_output(['quotas']).decode('utf-8')
    try:
        usage = [i for i in out.split("\n") if "spice:scratch" in i][0].split()[1]
    except:
        usage = "<quotas command failed>"
    return usage


def parse_arguments():
    parser = argparse.ArgumentParser(description='Run all tests configuration')
    parser.add_argument('-v', '--stdout_verbose', dest='stdout_verbose', action='store_true', help='Verbose stdout')
    return parser.parse_args()


def check_ruff_compliance(start: float, exit_code: int):
    """Runs 'ruff check' to verify code quality against the rules defined in ruff.toml (replaces the need for
    pycodestyle checks).

    Parameters
    ----------
    start: float
        The start time of the process.
    exit_code: int
        The exit code.
    """
    command = ['ruff', 'check']
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdoutdata, stderrdata = process.communicate()
    if process.returncode == 0:
        info = stdoutdata.decode('utf-8').split('\n')
        msg = info[-2].replace('=', '').strip()
        print(_green('{} success! [{}]'.format(' ' * 3, msg)))
    else:
        exit_code = 1
        print(_red('{} failed!\n!!! Please see "{}" for more details'.format(' ' * 3, LOG_NAME)))
        current_date = datetime.datetime.now()
        msg = '{}: {}: {}\n\n'.format(current_date, 'ruff', command)
        with open(LOG_NAME, 'a') as file_handle:
            file_handle.write(msg)
            file_handle.write(stdoutdata.decode('utf-8'))
            file_handle.write(stderrdata.decode('utf-8'))
            file_handle.write('\n')
    end = timer()
    elapsed_time = timedelta(seconds=end - start)
    print('\n{}'.format('+' * 46))
    print('Elapsed time: {}'.format(elapsed_time))
    if exit_code == 0:
        smiling_face = u'\U0001f604'
        print('{} {}'.format(_green('All tests succeed.'), smiling_face))
    else:
        sad_face = u'\U0001F627'
        print('{} {}'.format(_red('Some tests failed.'), sad_face))


def main():
    """Run all the tests in the CDDS packages."""
    exit_code = 0
    arguments = parse_arguments()
    start = timer()
    print('Using $SCRATCH ({}) for temporary large volume storage'.format(os.environ['SCRATCH']))
    # not yet implemented on Azure
    # print('Current data volume of $SCRATCH: {}'.format(du(os.environ['SCRATCH']+"/testdata")))

    print('Running ruff')
    check_ruff_compliance(start, exit_code)

    for package, tests in TESTS_TO_RUN.items():
        print('\nExecuting tests for {}:'.format(package))
        for test in tests:
            command = ROOT_COMMAND.split()
            if package in ADDITIONAL_OPTS:
                command.extend(ADDITIONAL_OPTS[package])
            command.append(package)
            if test:
                command.extend(test.split(' ', 1))
            command_str = ' '.join(command)
            print('--> {} {}'.format(command_str, (46 - len(command_str)) * '.'), end=''),
            sys.stdout.flush()
            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            stdoutdata, stderrdata = process.communicate()
            if process.returncode == 0:
                info = stdoutdata.decode('utf-8').split('\n')
                msg = info[-2].replace('=', '').strip()
                print(_green('{} success! [{}]'.format(' ' * 3, msg)))

                if package == 'mip_convert' and 'slow' in test:
                    print('Removing test output from $SCRATCH')
                    shutil.rmtree(os.path.join(os.environ['SCRATCH'], 'testdata'))
                    # print('Current data volume of $SCRATCH: {}'.format(du(os.environ['SCRATCH'])))
            else:
                exit_code = 1
                print(_red('{} failed!\n!!! Please see "{}" for more details'.format(' ' * 3, LOG_NAME)))
                current_date = datetime.datetime.now()
                msg = '{}: {}: {}\n\n'.format(current_date, package, command)
                with open(LOG_NAME, 'a') as file_handle:
                    file_handle.write(msg)
                    file_handle.write(stdoutdata.decode('utf-8'))
                    file_handle.write(stderrdata.decode('utf-8'))
                    file_handle.write('\n')
    end = timer()
    elapsed_time = timedelta(seconds=end - start)
    print('\n{}'.format('+' * 46))
    print('Elapsed time: {}'.format(elapsed_time))
    if exit_code == 0:
        smiling_face = u'\U0001f604'
        print('{} {}'.format(_green('All tests succeed.'), smiling_face))
    else:
        sad_face = u'\U0001F627'
        print('{} {}'.format(_red('Some tests failed.'), sad_face))

    print('Running mypy')
    command = [os.path.join(CDDS_DIR, 'run_mypy')]
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    stdoutdata, stderrdata = process.communicate()
    if process.returncode == 0:
        info = stdoutdata.decode('utf-8').split('\n')
        msg = info[-2].replace('=', '').strip()
        print(_green('{} success! [{}]'.format(' ' * 3, msg)))

    else:
        exit_code = 1
        print(_red('{} failed!\n!!! Please see "{}" for more details'.format(' ' * 3, LOG_NAME)))
        current_date = datetime.datetime.now()
        msg = '{}: {}: {}\n\n'.format(current_date, package, command)
        with open(LOG_NAME, 'a') as file_handle:
            file_handle.write(msg)
            file_handle.write(stdoutdata.decode('utf-8'))
            file_handle.write(stderrdata.decode('utf-8'))
            file_handle.write('\n')
    if arguments.stdout_verbose and os.path.exists(LOG_NAME):
        with open(LOG_NAME, 'r') as fh:
            for line in fh.readlines():
                print(line.strip())
    return exit_code


if __name__ == '__main__':
    sys.exit(main())