import pytest import sys import os import yaml sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from canesm.util import convert_date, year_from_time, month_from_time, previous_month, add_time, is_null from canesm.canesm_ensemble import CanESMensemble setup_folder = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'setup_files') class TestLocalConfig: def test_setup_class(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'develop_canesm.yaml')) assert ens.ensemble_size == 2 assert ens.share_member_code assert ens.start_time == [1990, 1990] assert ens.stop_time == [2014, 2014] assert ens.pp_rdm_num_pert == [0, 10] assert ens.max_threads == 1 def test_setup_table(self, caplog): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'test_config_table.yaml')) assert ens.share_member_code assert ens.phys_parm['pp_solar_const'] == [1360.747, 1361.747, 1362.747, 1363.747, 1364.747] assert ens.phys_parm['scale_ghg'] == [1, 1, 1, 1, 1] assert ens.canesm_cfg['runmode'] == ['AMIP-nudged'] * ens.ensemble_size assert ens.restart_dates == ['4500_m12', '4550_m12', '4600_m12', '4650_m12', '4700_m12'] assert ens.runid == ['run-001', 'run-002', 'run-003', 'run-004', 'run-005'] assert ens.pp_rdm_num_pert == [0, 2, 4, 6, 8] assert ens.run_directory == ['canesm_runs/testing/rlr_test'] * ens.ensemble_size assert 'WARNING' in caplog.text assert 'runid' in caplog.text assert 'canesm_cfg' in caplog.text def test_setup_single_job(self): # make sure broadcasting works properly when only one member is present ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'single_job.yaml')) jobs = ens.jobs job = jobs[0] assert ens.ensemble_size == 1 assert ens.share_member_code assert ens.start_time == [1990] assert ens.stop_time == [2014] # assert ens.pp_rdm_num_pert == [None] assert ens.max_threads == 1 def test_no_tapeload_or_restart(self): # make sure None is passed properly through the yaml setup ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'no_tapeload_or_restart.yaml')) assert ens.tapeload == ['None', 'None'] def test_broadcasting(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'develop_canesm.yaml')) assert ens.canesm_cfg['runmode'] == ['NEMO-AGCM-hist', 'NEMO-AGCM-hist'] def test_verify(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'identical_ensemble.yaml')) with pytest.raises(ValueError): ens.verify_setup() def test_verify_num_pert(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'pp_rdm_num_error.yaml')) with pytest.raises(ValueError): ens.verify_setup() def test_verify_restarts(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'restart_options_error.yaml')) with pytest.raises(ValueError): ens.verify_setup() def test_restart_options(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'develop_canesm.yaml')) job = ens.jobs[0] assert ens.restart_options == {'archive_class': ['crd_cccma', 'crd_cccma']} assert job.restart_files == {'archive_class': 'crd_cccma'} assert job.canesm_cfg['parent_branch_time'] == '1989_m12' assert job.canesm_cfg['parent_runid'] == 'rc3.1-his01' def test_date_conversion(self): assert convert_date(1990) == '1990_m12' assert convert_date(1990, default_month=1) == '1990_m01' assert convert_date('1990_m08') == '1990_m08' assert convert_date('1990m08') == '1990_m08' assert convert_date('1990') == '1990_m12' assert convert_date('1990', default_month=1) == '1990_m01' assert convert_date('1990:08') == '1990_m08' assert convert_date(1990.0) == '1990_m12' assert convert_date('1990-08') == '1990_m08' assert convert_date('1990-8') == '1990_m08' assert previous_month('2012_m02') == '2012_m01' assert previous_month('2012_m01') == '2011_m12' assert add_time('2012_m01', 3) == '2015_m01' assert add_time('2012_m02', 3) == '2015_m02' assert add_time('2012_m02', '1_m01') == '2013_m03' assert add_time('2012_m02', '0_m06') == '2012_m08' assert add_time('2012_m02', '0_m11') == '2013_m01' assert add_time('2012_m02', '0_m10') == '2012_m12' with pytest.raises(ValueError): convert_date('1998-m12') with pytest.raises(ValueError): convert_date(1998.1) with pytest.raises(ValueError): convert_date('four') with pytest.raises(ValueError): convert_date('1990_mO1') with pytest.raises(ValueError): convert_date('199O_m01') def test_incorrect_list_length(self): with pytest.raises(ValueError): CanESMensemble.from_config_file(os.path.join(setup_folder, 'physparm_broadcast_error.yaml')) with pytest.raises(ValueError): CanESMensemble.from_config_file(os.path.join(setup_folder, 'restart_broadcast_error.yaml')) with pytest.raises(ValueError): CanESMensemble.from_config_file(os.path.join(setup_folder, 'canesm_broadcast_error.yaml')) def test_repr(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'develop_canesm.yaml')) assert str(ens) == 'Ensemble object containing 2 members:'\ + '\n\tens_tst-000\n\tens_tst-001\n'\ + 'user: scrd108\nrunid: ens_tst\nmachine: hare' def test_year_from_date(self): assert year_from_time('1990_m12') == 1990 assert year_from_time('1990m12') == 1990 assert year_from_time('1990:12') == 1990 assert year_from_time('1990') == 1990 assert year_from_time(1990) == 1990 with pytest.raises(ValueError): year_from_time('error') def test_month_from_date(self): assert month_from_time('1990_m08') == 8 assert month_from_time('1990m8') == 8 assert month_from_time('1990:8') == 8 assert month_from_time('1990') == 12 assert month_from_time('1990', default_month=7) == 7 assert month_from_time(1990, default_month=7) == 7 with pytest.raises(ValueError): month_from_time('error') def test_restart_dates(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'restart_every_x_years.yaml')) assert ens.restart_dates == ['1989_m12', '1999_m12', '2009_m12', '2019_m12', '2029_m12'] def test_run_directory_list(self): ens = CanESMensemble.from_config_file(os.path.join(setup_folder, 'test_config_table_dates.yaml')) assert ens.run_directory == ['abc', 'abcd', 'abcd', 'abcde', 'abcde'] def test_is_null(self): assert is_null(yaml.load("tapeload: None", Loader=yaml.SafeLoader)['tapeload']) assert is_null(yaml.load("tapeload: none", Loader=yaml.SafeLoader)['tapeload']) assert is_null(yaml.load("tapeload: Null", Loader=yaml.SafeLoader)['tapeload']) assert is_null(yaml.load("tapeload: null", Loader=yaml.SafeLoader)['tapeload']) assert is_null(yaml.load("tapeload: Nil", Loader=yaml.SafeLoader)['tapeload']) assert is_null(yaml.load("tapeload: nil", Loader=yaml.SafeLoader)['tapeload']) if __name__ == '__main__': TestLocalConfig().test_restart_options()