run_rabbitmq_tests

#!/usr/bin/env python3.10
# (C) British Crown Copyright 2017-2022, Met Office.
"""
Run all the tests using RabbitMQ in the CDDS packages.

Only the server els055 and els056 have RabbitMQ properly installed for CDDS. So,
you can only run these test on these two servers.

For running the test:
1. Connect to one of the server via ssh
2. Go to the workspace directory (the same as on your local VM)
3. Source the environment script
4. Run this script
"""
import argparse
import datetime
import os
import subprocess
import sys

CDDS_DIR = os.path.dirname(os.path.realpath(__file__))
LOG_NAME = os.path.join(CDDS_DIR, 'cdds_rabbitmq_test_failures.log')
ROOT_COMMAND = 'pytest -s'
TESTS_TO_RUN = {
    'cdds': ['-m rabbitMQ'],  # Only run the unit tests using RabbitMQ!
    }


def print_error(text):
    """
    Print error message in red.

    :param text: Text that should be printed in red.
    :type text: str
    """
    print("\033[91m {}\033[00m".format(text))


def print_success(text):
    """
    Print succeed message in green.

    :param text: Text that should be printed in green
    :type text: str
    """
    print("\033[92m {}\033[00m".format(text))


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 main():
    """
    Run all the tests using RabbitMQ in the CDDS packages.
    """
    exit_code = 0
    arguments = parse_arguments()
    for package, tests in TESTS_TO_RUN.items():
        print('\nExecuting tests for {}:'.format(package))
        for test in tests:
            command = ROOT_COMMAND.split()
            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_success('{} success! [{}]'.format(' ' * 3, msg))
            else:
                exit_code = 1
                print_error('{} 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())