#!/usr/bin/env python3 """Unit tests for machine """ import unittest import os from ctsm import add_cime_to_path # pylint: disable=unused-import from ctsm import unit_testing from ctsm.machine import ( create_machine, get_possibly_overridden_mach_value, CREATE_TEST_QUEUE_UNSPECIFIED, ) from ctsm.machine_utils import get_user from ctsm.machine_defaults import MACHINE_DEFAULTS, MachineDefaults, QsubDefaults from ctsm.joblauncher.job_launcher_no_batch import JobLauncherNoBatch from ctsm.joblauncher.job_launcher_qsub import JobLauncherQsub from ctsm.joblauncher.job_launcher_factory import ( JOB_LAUNCHER_QSUB, JOB_LAUNCHER_NOBATCH, ) # Allow names that pylint doesn't like, because otherwise I find it hard # to make readable unit test names # pylint: disable=invalid-name class TestCreateMachine(unittest.TestCase): """Tests of create_machine""" def assertMachineInfo( self, machine, name, scratch_dir, baseline_dir, account, create_test_retry=0, create_test_queue=CREATE_TEST_QUEUE_UNSPECIFIED, ): """Asserts that the basic machine info is as expected. This does NOT dive down into the job launcher""" self.assertEqual(machine.name, name) self.assertEqual(machine.scratch_dir, scratch_dir) self.assertEqual(machine.baseline_dir, baseline_dir) self.assertEqual(machine.account, account) self.assertEqual(machine.create_test_retry, create_test_retry) self.assertEqual(machine.create_test_queue, create_test_queue) def assertNoBatchInfo(self, machine, nice_level=None): """Asserts that the machine's launcher is of type JobLauncherNoBatch""" launcher = machine.job_launcher self.assertIsInstance(launcher, JobLauncherNoBatch) if nice_level is None: # the default nice level should be 0 nice_level = 0 self.assertEqual(launcher.get_nice_level(), nice_level) def assertQsubInfo(self, machine, queue, walltime, account, required_args, extra_args): """Asserts that the machine's launcher is of type JobLauncherQsub, and has values as expected""" launcher = machine.job_launcher self.assertIsInstance(launcher, JobLauncherQsub) self.assertEqual(launcher.get_queue(), queue) self.assertEqual(launcher.get_walltime(), walltime) self.assertEqual(launcher.get_account(), account) self.assertEqual(launcher.get_required_args(), required_args) self.assertEqual(launcher.get_extra_args(), extra_args) @staticmethod def create_defaults(default_job_launcher=JOB_LAUNCHER_QSUB): """Creates test-specific defaults so we don't tie the tests to changes in the real defaults""" defaults = { "cheyenne": MachineDefaults( job_launcher_type=default_job_launcher, scratch_dir=os.path.join(os.path.sep, "glade", "scratch", get_user()), baseline_dir=os.path.join(os.path.sep, "my", "baselines"), account_required=True, create_test_retry=2, create_test_queue="regular", job_launcher_defaults={ JOB_LAUNCHER_QSUB: QsubDefaults( queue="regular", walltime="06:00:00", extra_args="", required_args="-l select=1:ncpus=36:mpiprocs=1 -r n -l inception=login", ) }, ) } return defaults def test_unknownMachine_defaults(self): """Tests a machine not in the defaults structure, with no overriding arguments""" machine = create_machine("unknown_test_machine", MACHINE_DEFAULTS, account="a123") self.assertMachineInfo( machine=machine, name="unknown_test_machine", scratch_dir=None, baseline_dir=None, account="a123", ) self.assertNoBatchInfo(machine) def test_noBatchMachine_niceLevel(self): """Tests a no-batch machine where the nice level is explicit""" machine = create_machine( "unknown_test_machine", MACHINE_DEFAULTS, job_launcher_type=JOB_LAUNCHER_NOBATCH, scratch_dir="/path/to/scratch", account="a123", job_launcher_nice_level=13, ) self.assertMachineInfo( machine=machine, name="unknown_test_machine", scratch_dir="/path/to/scratch", baseline_dir=None, account="a123", ) self.assertNoBatchInfo(machine, nice_level=13) def test_unknownMachine_argsExplicit(self): """Tests a machine not in the defaults structure, with explicit arguments""" machine = create_machine( "unknown_test_machine", MACHINE_DEFAULTS, scratch_dir="/path/to/scratch", job_launcher_type=JOB_LAUNCHER_QSUB, account="a123", job_launcher_queue="my_queue", job_launcher_walltime="1:23:45", job_launcher_extra_args="--some args", ) self.assertMachineInfo( machine=machine, name="unknown_test_machine", scratch_dir="/path/to/scratch", baseline_dir=None, account="a123", ) self.assertQsubInfo( machine=machine, queue="my_queue", walltime="1:23:45", account="a123", required_args="", extra_args="--some args", ) def test_knownMachine_defaults(self): """Tests a machine known in the defaults structure, with no overriding arguments""" defaults = self.create_defaults() machine = create_machine("cheyenne", defaults, account="a123") self.assertMachineInfo( machine=machine, name="cheyenne", scratch_dir=os.path.join(os.path.sep, "glade", "scratch", get_user()), baseline_dir=os.path.join(os.path.sep, "my", "baselines"), account="a123", create_test_retry=2, create_test_queue="regular", ) self.assertQsubInfo( machine=machine, queue="regular", walltime="06:00:00", account="a123", required_args="-l select=1:ncpus=36:mpiprocs=1 -r n -l inception=login", extra_args="", ) def test_knownMachine_argsExplicit(self): """Tests a machine known in the defaults structure, with explicit arguments""" defaults = self.create_defaults(default_job_launcher=JOB_LAUNCHER_NOBATCH) machine = create_machine( "cheyenne", defaults, job_launcher_type=JOB_LAUNCHER_QSUB, scratch_dir="/custom/path/to/scratch", account="a123", job_launcher_queue="custom_queue", job_launcher_walltime="9:87:65", job_launcher_extra_args="--custom args", ) self.assertMachineInfo( machine=machine, name="cheyenne", scratch_dir="/custom/path/to/scratch", baseline_dir=os.path.join(os.path.sep, "my", "baselines"), account="a123", create_test_retry=2, create_test_queue="regular", ) self.assertQsubInfo( machine=machine, queue="custom_queue", walltime="9:87:65", account="a123", required_args="-l select=1:ncpus=36:mpiprocs=1 -r n -l inception=login", extra_args="--custom args", ) # ------------------------------------------------------------------------ # Tests of get_possibly_overridden_mach_value # ------------------------------------------------------------------------ def test_baselineDir_overridden(self): """Tests get_possibly_overridden_mach_value when baseline_dir is provided""" defaults = self.create_defaults() machine = create_machine("cheyenne", defaults, account="a123") baseline_dir = get_possibly_overridden_mach_value( machine, varname="baseline_dir", value="mypath" ) self.assertEqual(baseline_dir, "mypath") def test_baselineDir_default(self): """Tests get_possibly_overridden_mach_value when baseline_dir is not provided""" defaults = self.create_defaults() machine = create_machine("cheyenne", defaults, account="a123") baseline_dir = get_possibly_overridden_mach_value( machine, varname="baseline_dir", value=None ) self.assertEqual(baseline_dir, os.path.join(os.path.sep, "my", "baselines")) def test_baselineDir_noDefault(self): """Tests get_possibly_overridden_mach_value when baseline_dir is not provided and there is no default""" machine = create_machine("unknown_test_machine", MACHINE_DEFAULTS, account="a123") baseline_dir = get_possibly_overridden_mach_value( machine, varname="baseline_dir", value=None ) self.assertIsNone(baseline_dir) if __name__ == "__main__": unit_testing.setup_for_tests() unittest.main()