Experiment Unittests

experiment_unittests.py

"""
Test module for the experiment_handler/experiment_prototype code.

This script can be run most simply via 'python3 experiment_unittests.py'. This will run through all
experiments defined in the Borealis experiments top level directory, experiment exception tests
defined in the tests/ directory, as well as any other tests hard coded into this script.
Any experiment that raises an exception when building will show up as a failed test here.

This script can also be run to test individual experiments by using the --experiment flag. For
example: `python3 experiment_unittests.py --experiment normalscan normalsound` will only test the
normalscan and normalsound experiments. Any experiments specified must exist within the Borealis
experiments top level directory (i.e. src/borealis_experiments).

Other command line options include:

- Specifying what radar site to run the tests as

References:
https://stackoverflow.com/questions/32899/how-do-you-generate-dynamic-parameterized-unit-tests-in-python
https://docs.python.org/3/library/unittest.html
https://www.bnmetrics.com/blog/dynamic-import-in-python3

:copyright: 2023 SuperDARN Canada
:author: Kevin Krieger, Theodore Kolkman
"""

import argparse
import random
import string
import unittest
import unittest.mock
import os
import sys
import inspect
import pkgutil
from importlib import import_module
import importlib.util
import json
from typing import Type

# Need the path append to import within this file
BOREALISPATH = os.environ["BOREALISPATH"]
sys.path.append(f"{BOREALISPATH}/src")

from utils.experiment_prototype import (
    ExperimentException,
    ExperimentPrototype,
    retrieve_experiment,
    experiment_handler,
)
from utils.options import Options


def redirect_to_devnull(func, *args, **kwargs):
    with open(os.devnull, "w") as devnull:
        old_stdout = sys.stdout
        old_stderr = sys.stderr
        sys.stdout = devnull
        sys.stderr = devnull
        try:
            result = func(*args, **kwargs)
        finally:
            sys.stdout = old_stdout
            sys.stderr = old_stderr
    return result


def ehmain(
    exp_class: Type[ExperimentPrototype], scheduling_mode="discretionary", **kwargs
):
    """
    Calls the functions within experiment handler that verify an experiment

    :param  exp_class: The class of the experiment to be verified.
    :type   exp_class: Type[ExperimentPrototype]
    :param  scheduling_mode: The scheduling mode to run. Defaults to 'discretionary'
    :type   scheduling_mode: str
    :param  kwargs: The keyword arguments for the experiment
    :type   kwargs: dict
    """
    from utils import log_config

    log_config.log(
        console=False, logfile=False, aggregator=False
    )  # Prevent logging in experiment

    experiment_handler(exp_class, scheduling_mode, embargo=False, **kwargs)


class TestExperimentEnvSetup(unittest.TestCase):
    """
    A unittest class to test the environment setup for the experiment_handler module.
    All test methods must begin with the word 'test' to be run by unittest.
    """

    def setUp(self):
        """
        This function is called before every test_* method within this class (every test case in
        unittest lingo)
        """
        print("\nMethod: ", self._testMethodName)

    @unittest.mock.patch.dict(os.environ, {"BOREALISPATH": ""})
    def test_borealispath(self):
        """
        Test failure to have BOREALISPATH in env
        """
        with self.assertRaisesRegex(ValueError, "BOREALISPATH env variable not set"):
            normalscan = retrieve_experiment("normalscan")
            ehmain(normalscan)

    @unittest.skip("Cannot test this while hdw.dat files are in /usr/local/hdw")
    def test_hdw_file(self):
        """
        Test the code that checks for the hdw.dat file
        """
        options = Options()
        site_name = options.site_id
        hdw_path = options.hdw_path

        # Rename the hdw.dat file temporarily
        os.rename(f"{hdw_path}/hdw.dat.{site_name}", f"{hdw_path}/_hdw.dat.{site_name}")

        with self.assertRaisesRegex(ValueError, "Cannot open hdw.dat.[a-z]{3} file at"):
            normalscan = retrieve_experiment("normalscan")
            ehmain(normalscan)

        # Now rename the hdw.dat file and move on
        os.rename(f"{hdw_path}/_hdw.dat.{site_name}", f"{hdw_path}/hdw.dat.{site_name}")


class TestExperimentRetrieval(unittest.TestCase):
    """
    A class to test that `retrieve_experiment()` works properly.
    """

    def setUp(self):
        """
        This function is called before every test case.
        """
        print("\nRetrieve Experiment test: ", self._testMethodName)

    def test_normalscan(self):
        exp = retrieve_experiment("normalscan")
        assert exp.__name__, "NormalScan"

    def test_two_classes(self):
        with self.assertRaisesRegex(
            ExperimentException,
            "You have more than one experiment class in your experiment file - exiting",
        ):
            retrieve_experiment("tests.beam_order")

    def test_nonexistent_file(self):
        with self.assertRaisesRegex(
            ModuleNotFoundError, "No module named 'borealis_experiments.dummy'"
        ):
            retrieve_experiment("dummy")

    def test_module_with_extra_attributes(self):
        retrieve_experiment("full_fov")


class TestMockExperiments(unittest.TestCase):
    """
    A unittest class to test various ways for an experiment to fail for the experiment_handler
    module. Tests will check that exceptions are correctly thrown for each failure case. All test
    methods must begin with the word 'test' to be run by unittest.
    """

    def setUp(self):
        """
        This function is called before every test_* method (every test case in unittest lingo)
        """
        print("\nException Test: ", self._testMethodName)


class TestActiveExperiments(unittest.TestCase):
    """
    A unittest class to test all Borealis experiments and verify that none of them are built
    incorrectly. Tests are verified using code within experiment handler. All test methods must
    begin with the word 'test' to be run by unittest.
    """

    def setUp(self):
        """
        This function is called before every test_* method (every test case in unittest lingo)
        """
        print("\nExperiment Test: ", self._testMethodName)


def build_unit_tests():
    """
    Create individual unit tests for all test cases specified in tests/ directory of experiments path.
    """

    experiment_package = "tests"
    experiment_path = f"{BOREALISPATH}/src/borealis_experiments/{experiment_package}/"
    if not os.path.exists(experiment_path):
        raise OSError(f"Error: experiment path {experiment_path} is invalid")

    # Iterate through all modules in the borealis_experiments directory
    for _, name, _ in pkgutil.iter_modules([experiment_path]):
        imported_module = import_module(
            "." + name, package=f"borealis_experiments.{experiment_package}"
        )
        # Loop through all attributes of each found module
        for i in dir(imported_module):
            attribute = getattr(imported_module, i)
            # To verify that an attribute is a runnable experiment, check that the attribute is
            # a class and inherits from ExperimentPrototype
            if inspect.isclass(attribute) and issubclass(
                attribute, ExperimentPrototype
            ):
                # Only create a test if the current attribute is the experiment itself
                if "ExperimentPrototype" not in str(attribute):
                    if hasattr(attribute, "error_message"):
                        # If expected to fail, should have a classmethod called "error_message"
                        # that contains the error message raised
                        exp_exception, msg = getattr(attribute, "error_message")()
                        test = exception_test_generator(attribute, exp_exception, msg)
                    else:  # No exception expected - this is a positive test
                        test = experiment_test_generator(attribute)
                    # setattr makes a properly named test method within TestExperimentArchive which
                    # can be run by unittest.main()
                    if hasattr(TestActiveExperiments, f"test_{attribute.__name__}"):
                        raise ValueError(
                            f"Multiple tests have name test_{attribute.__name__}"
                        )
                    setattr(TestMockExperiments, f"test_{attribute.__name__}", test)


def exception_test_generator(
    exp_class: Type[ExperimentPrototype], exception, exception_message
):
    """
    Generate a single test for the given module name and exception message

    :param exp_class:           Experiment class
    :type  exp_class:           Type[ExperimentPrototype]
    :param exception:           Exception that is expected to be raised
    :type  exception:           BaseException
    :param exception_message:   Message from the Exception raised.
    :type  exception_message:   str
    """

    def test(self):
        with self.assertRaisesRegex(exception, exception_message):
            redirect_to_devnull(ehmain, exp_class)

    return test


def build_experiment_tests(experiments=None, kwargs=None):
    """
    Create individual unit tests for all experiments within the base borealis_experiments/
    directory. All experiments are run to ensure no exceptions are thrown when they are built.
    """

    experiment_package = "borealis_experiments"
    experiment_path = f"{BOREALISPATH}/src/{experiment_package}/"
    if not os.path.exists(experiment_path):
        raise OSError(f"Error: experiment path {experiment_path} is invalid")

    # parse kwargs and pass to experiment
    kwargs_dict = {}
    if kwargs:
        for element in kwargs:
            if element == "":
                continue
            kwarg = element.split("=")
            if len(kwarg) == 2:
                kwargs_dict[kwarg[0]] = kwarg[1]
            else:
                raise ValueError(f"Bad kwarg: {element}")

    def add_experiment_test(exp_name: str):
        """Add a unit test for a given experiment"""
        imported_module = import_module("." + exp_name, package=experiment_package)
        # Loop through all attributes of each found module
        for i in dir(imported_module):
            attribute = getattr(imported_module, i)
            # To verify that an attribute is a runnable experiment, check that the attribute is
            # a class and inherits from ExperimentPrototype
            if inspect.isclass(attribute) and issubclass(
                attribute, ExperimentPrototype
            ):
                # Only create a test if the current attribute is the experiment itself
                if "ExperimentPrototype" not in str(attribute):
                    test = experiment_test_generator(attribute, **kwargs_dict)
                    # setattr make the "test" function a method within TestActiveExperiments called
                    # "test_[exp_name]" which can be run via unittest.main()
                    test_name = f"test_{exp_name}"
                    if hasattr(TestActiveExperiments, test_name):
                        test_name += "".join(
                            random.choice(string.ascii_lowercase) for _ in range(6)
                        )
                    setattr(TestActiveExperiments, f"test_{exp_name}", test)

    # Grab the experiments specified
    if experiments is not None:
        for name in experiments:
            spec = importlib.util.find_spec("." + name, package=experiment_package)
            if spec is None:
                # Add in a failing test for this experiment name
                setattr(
                    TestActiveExperiments,
                    f"test_{name}",
                    lambda self: self.fail("Experiment not found"),
                )
            else:
                add_experiment_test(name)

    else:
        # Iterate through all modules in the borealis_experiments directory
        for _, name, _ in pkgutil.iter_modules([experiment_path]):
            add_experiment_test(name)


def experiment_test_generator(exp_class: Type[ExperimentPrototype], **kwargs):
    """
    Generate a single test for a given experiment name. The test will try to run the experiment,
    and if any exceptions are thrown (i.e. the experiment is built incorrectly) the test will fail.

    :param exp_class: Experiment class
    :type  exp_class: Type[ExperimentPrototype]
    """

    def test(self):
        try:
            redirect_to_devnull(ehmain, exp_class, **kwargs)
        except Exception as err:
            self.fail(err)

    return test


def run_tests(raw_args=None, buffer=True, print_results=True):
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--experiments",
        required=False,
        nargs="+",
        default=None,
        help="Only run the experiments specified after this option. Experiments \
                            specified must exist within the top-level Borealis experiments directory.",
    )
    parser.add_argument(
        "--kwargs",
        required=False,
        nargs="+",
        default=list(),
        help="Keyword arguments to pass to the experiments. Note that kwargs are passed to all "
        "experiments specified.",
    )
    parser.add_argument(
        "--module",
        required=False,
        default="__main__",
        help="If calling from another python file, this should be set to "
        "'experiment_unittests' in order to properly work.",
    )
    parser.add_argument(
        "--no-tests",
        required=False,
        action="store_true",
        help="Only test the main experiments, not those in tests/",
    )
    verbose_group = parser.add_mutually_exclusive_group()
    verbose_group.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Increase verbosity",
    )
    verbose_group.add_argument(
        "-q",
        "--quiet",
        action="store_true",
        help="Decrease verbosity",
    )
    args = parser.parse_args(raw_args)

    # Read in config.ini file for current site to make necessary directories
    path = (
        f"{os.environ['BOREALISPATH']}/config/"
        f"{os.environ['RADAR_ID']}/"
        f"{os.environ['RADAR_ID']}_config.ini"
    )
    try:
        with open(path, "r") as data:
            raw_config = json.load(data)
    except OSError:
        errmsg = f"Cannot open config file at {path}"
        raise ValueError(errmsg)

    # These directories are required for ExperimentHandler to run
    data_directory = raw_config["data_directory"]
    log_directory = raw_config["log_handlers"]["logfile"]["directory"]
    hdw_path = raw_config["hdw_path"]
    hdw_dat_file = f"{hdw_path}/hdw.dat.{os.environ['RADAR_ID']}"
    if not os.path.exists(data_directory):
        os.makedirs(data_directory)
    if not os.path.exists(log_directory):
        os.makedirs(log_directory)
    if not os.path.exists(hdw_path):
        os.makedirs(hdw_path)
    if not os.path.exists(hdw_dat_file):
        open(hdw_dat_file, "w")

    experiments = args.experiments
    if experiments is None:  # Run all unit tests and experiment tests
        print("Running tests on all experiments")
        if not args.no_tests:
            build_unit_tests()
        build_experiment_tests()
        argv = [sys.argv[0]]
    else:  # Only test specified experiments
        print(f"Running tests on experiments {experiments}")
        build_experiment_tests(experiments, args.kwargs)
        exp_tests = []
        for exp in experiments:
            # Check experiment exists
            if hasattr(TestActiveExperiments, f"test_{exp}"):
                # Create correct string to test the experiment with unittest
                exp_tests.append(f"TestActiveExperiments.test_{exp}")
            else:
                print(f"Could not find experiment {exp}. Exiting...")
                exit(1)
        argv = [parser.prog] + exp_tests
    if print_results:
        if args.quiet:
            verbosity = 0
        if args.verbose:
            verbosity = 2
        else:
            verbosity = 1
        result = unittest.main(
            module=args.module,
            argv=argv,
            exit=False,
            buffer=buffer,
            verbosity=verbosity,
        )
    else:
        result = redirect_to_devnull(
            unittest.main, module=args.module, argv=argv, exit=False, buffer=buffer
        )

    # Clean up the directories/files we created
    try:
        os.removedirs(data_directory)
    except OSError:  # If directories not empty, this will fail. That is fine.
        pass
    try:
        os.removedirs(log_directory)
    except OSError:  # If directories not empty, this will fail. That is fine.
        pass
    if os.path.getsize(hdw_dat_file) == 0:
        try:
            os.remove(hdw_dat_file)
        except OSError:  # Path is a directory
            os.removedirs(hdw_dat_file)
        else:  # File removed, now clean up directories
            try:
                os.removedirs(hdw_path)
            except OSError:  # If directories not empty, this will fail. That is fine.
                pass

    if isinstance(result, unittest.TestProgram):
        result = result.result

    return result


if __name__ == "__main__":
    from utils import log_config

    log = log_config.log(console=False, logfile=False, aggregator=False)

    run_tests(sys.argv[1:])

This script tests both runnable experiments (those in borealis/src/borealis_experiments/) and a set of unit tests (those in borealis/src/borealis_experiments/tests/). Some unit tests are meant to raise an exception; these tests have an extra method defined which returns the expected exception and a regex of the expected error message. An example unit test is shown below.

Example Unit Test file

import borealis_experiments.superdarn_common_fields as scf
from utils.experiment_prototype import ExperimentPrototype
from pydantic import ValidationError


class AveMethodDNE(ExperimentPrototype):
    cpid = 1

    def __init__(self):
        super().__init__()
        self.add_slice(
            {
                "pulse_sequence": scf.SEQUENCE_7P,
                "tau_spacing": scf.TAU_SPACING_7P,
                "pulse_len": scf.PULSE_LEN_45KM,
                "num_ranges": scf.STD_NUM_RANGES,
                "first_range": scf.STD_FIRST_RANGE,
                "intt": scf.INTT_MS,
                "beam_angle": scf.STD_BEAM_ANGLES,
                "rx_beam_order": scf.STD_BEAM_ORDER,
                "tx_beam_order": scf.STD_BEAM_ORDER,
                "freq": scf.COMMON_MODE_FREQ_1,
                "acf": True,
                "averaging_method": "not_a_method",  # This is not a valid method
            }
        )

    @classmethod
    def error_message(cls):
        return (
            ValidationError,
            "averaging_method\n" "  Input should be 'mean' or 'median'",
        )