From 8c957f193c900db8b15341bc291c0d1f07e8ffd8 Mon Sep 17 00:00:00 2001 From: Mostafa Farrag Date: Tue, 14 Jan 2025 14:33:10 +0100 Subject: [PATCH] reformat bc test files --- tests/dflowfm/bc/test_forcing_model.py | 290 +++++++++++++++++++++++++ 1 file changed, 290 insertions(+) create mode 100644 tests/dflowfm/bc/test_forcing_model.py diff --git a/tests/dflowfm/bc/test_forcing_model.py b/tests/dflowfm/bc/test_forcing_model.py new file mode 100644 index 000000000..d88b3c144 --- /dev/null +++ b/tests/dflowfm/bc/test_forcing_model.py @@ -0,0 +1,290 @@ +from pathlib import Path +import numpy as np +import pytest +from pydantic.v1.error_wrappers import ValidationError + +from hydrolib.core.dflowfm.bc.models import ( + T3D, + Astronomic, + AstronomicCorrection, + Constant, + ForcingBase, + ForcingModel, + Harmonic, + HarmonicCorrection, + QHTable, + QuantityUnitPair, + TimeInterpolation, + TimeSeries +) +from tests.utils import ( + assert_files_equal, + invalid_test_data_dir, + test_data_dir, + test_output_dir, + test_reference_dir, +) + +TEST_BC_FILE = "test.bc" + +def quantityunitpair(quantity, unit, verticalpositionindex=None): + return QuantityUnitPair( + quantity=quantity, unit=unit, vertpositionindex=verticalpositionindex + ) + +def harmonic_values(iscorrection: bool): + function = "harmonic-correction" if iscorrection else "harmonic" + return dict( + name=f"boundary_{function}", + function=function, + quantityunitpair=[ + quantityunitpair("harmonic component", "minutes"), + quantityunitpair("waterlevelbnd amplitude", "m"), + quantityunitpair("waterlevelbnd phase", "deg"), + ], + datablock=[ + ["0", "1.23", "2.34"], + ["60", "3.45", "4.56"], + ], + ) + +def qhtable_values(): + return dict( + name="boundary_qhtable", + function="qhtable", + quantityunitpair=[ + quantityunitpair("qhbnd discharge", "m3/s"), + quantityunitpair("qhbnd waterlevel", "m"), + ], + datablock=[ + ["1.23", "2.34"], + ["3.45", "4.56"], + ], + ) + + +def constant_values(): + return dict( + name="boundary_constant", + function="constant", + offset="1.23", + factor="2.34", + timeinterpolation=TimeInterpolation.linear, + quantityunitpair=[ + quantityunitpair("waterlevelbnd", "m"), + ], + datablock=[ + ["3.45"], + ], + ) + +def astronomic_values(iscorrection: bool, quantityunitpair): + function = "astronomic-correction" if iscorrection else "astronomic" + return dict( + name=f"boundary_{function}", + function=function, + quantityunitpair=[ + quantityunitpair("astronomic component", "-"), + quantityunitpair("waterlevelbnd amplitude", "m"), + quantityunitpair("waterlevelbnd phase", "deg"), + ], + datablock=[ + ["A0", "1.23", "2.34"], + ["M4", "3.45", "4.56"], + ["N2", "5.67", "6.78"], + ], + ) + + +class TestForcingBase: + @pytest.mark.parametrize( + "input,expected", + [ + ("TimeSeries", "timeseries"), + ("haRmoniC", "harmonic"), + ("ASTRONOMIC", "astronomic"), + ("harmonic-Correction", "harmonic-correction"), + ("Astronomic-Correction", "astronomic-correction"), + ("t3D", "t3d"), + ("QHtable", "qhtable"), + ("Constant", "constant"), + ("DOESNOTEXIST", "DOESNOTEXIST"), + ("doesnotexist", "doesnotexist"), + ], + ) + def test_parses_function_case_insensitive(self, input, expected): + forcing = ForcingBase(function=input, name="somename", quantityunitpair=[]) + + assert forcing.function == expected + + @pytest.mark.parametrize( + "missing_field", + [ + "quantity", + "unit", + ], + ) + def test_create_forcingbase_missing_field_raises_correct_error( + self, missing_field: str + ): + values = dict( + name="Boundary2", + function="constant", + quantity="dischargebnd", + unit="m³/s", + datablock=[["100"]], + ) + + del values[missing_field] + + with pytest.raises(ValidationError) as error: + ForcingBase(**values) + + expected_message = f"{missing_field} is not provided" + assert expected_message in str(error.value) + + @pytest.mark.parametrize( + "quantities,units", + [ + (["time", "dischargebnd"], "m³/s"), + ( + ["time", "dischargebnd", "extra"], + ["minutes since 2021-01-01 00:00:00", "m³/s"], + ), + ], + ) + def test_create_forcingbase_mismatch_number_of_quantities_units_raises_correct_error( + self, quantities, units + ): + values = dict( + name="Boundary2", + function="constant", + quantity=quantities, + unit=units, + datablock=[["100"]], + ) + + with pytest.raises(ValidationError) as error: + ForcingBase(**values) + + expected_message = "Number of quantities should be equal to number of units" + assert expected_message in str(error.value) + + +class TestForcingModel: + """ + Wrapper class to test the logic of the ForcingModel class in hydrolib.core.dflowfm.bc.models.py. + """ + + def test_forcing_model(self): + filepath = ( + test_data_dir + / "input/e02/f101_1D-boundaries/c01_steady-state-flow/BoundaryConditions.bc" + ) + m = ForcingModel(filepath) + assert len(m.forcing) == 13 + assert isinstance(m.forcing[-1], TimeSeries) + + def test_read_bc_missing_field_raises_correct_error(self): + bc_file = "missing_field.bc" + identifier = "Boundary2" + + filepath = invalid_test_data_dir / bc_file + + with pytest.raises(ValidationError) as error: + ForcingModel(filepath) + + expected_message1 = f"{bc_file} -> forcing -> 1 -> {identifier}" + expected_message2 = "quantity is not provided" + assert expected_message1 in str(error.value) + assert expected_message2 in str(error.value) + + def test_save_forcing_model(self, time_series_values, t3d_values): + bc_file = Path(test_output_dir / TEST_BC_FILE) + reference_file = Path(test_reference_dir / "bc" / TEST_BC_FILE) + forcingmodel = ForcingModel() + forcingmodel.filepath = bc_file + + timeseries = TimeSeries(**time_series_values) + harmonic = Harmonic(**harmonic_values(False)) + harmoniccorrection = HarmonicCorrection(**harmonic_values(True)) + t3d = T3D(**t3d_values) + qhtable = QHTable(**qhtable_values()) + constant = Constant(**constant_values()) + + forcingmodel.forcing.append(timeseries) + forcingmodel.forcing.append(harmonic) + forcingmodel.forcing.append(harmoniccorrection) + forcingmodel.forcing.append(t3d) + forcingmodel.forcing.append(qhtable) + forcingmodel.forcing.append(constant) + + forcingmodel.serializer_config.float_format = ".3f" + forcingmodel.serializer_config.float_format_datablock = ".4f" + forcingmodel.save() + + assert bc_file.is_file() == True + assert_files_equal(bc_file, reference_file, skip_lines=[0, 3]) + + @pytest.mark.parametrize("cls", [Astronomic, AstronomicCorrection]) + def test_astronomic_values_with_strings_in_datablock_are_parsed_correctly( + self, cls + ): + try: + is_correction = cls == AstronomicCorrection + _ = cls(**astronomic_values(is_correction, quantityunitpair)) + except ValidationError: + pytest.fail( + f"No validation error should be raised when creating an {cls.__name__}" + ) + + def test_representation_is_correct(self): + forcing = ForcingBase( + name="some_name", + function="some_function", + quantityunitpair=[ + QuantityUnitPair(quantity="some_quantity", unit="some_unit") + ], + datablock=[[1.2, 2.3]], + ) + + str_representation_as_single = str(forcing) + str_representation_in_list = str([forcing]) + + # datablock should be omitted when a `ForcingBase` is represented from within a list + expected_result = "comments=Comments() datablock={0} name='some_name' function='some_function' quantityunitpair=[QuantityUnitPair(quantity='some_quantity', unit='some_unit', vertpositionindex=None)]" + assert str_representation_as_single == expected_result.format("[[1.2, 2.3]]") + assert str_representation_in_list == "[{0}]".format( + expected_result.format("''") + ) + + def test_forcing_model_correct_default_serializer_config(self): + model = ForcingModel() + + assert model.serializer_config.section_indent == 0 + assert model.serializer_config.property_indent == 0 + assert model.serializer_config.datablock_indent == 0 + assert model.serializer_config.float_format == "" + assert model.serializer_config.datablock_spacing == 2 + assert model.serializer_config.comment_delimiter == "#" + assert model.serializer_config.skip_empty_properties == True + + def test_forcing_model_with_datablock_that_has_nan_values_should_raise_error(self): + datablock = np.random.uniform(low=-40, high=130.3, size=(4, 2)) * np.nan + datablock_list = datablock.tolist() + + with pytest.raises(ValidationError) as error: + TimeSeries( + name="east2_0001", + quantityunitpair=[ + QuantityUnitPair( + quantity="time", unit="seconds since 2022-01-01 00:00:00 +00:00" + ), + QuantityUnitPair(quantity="waterlevel", unit="m"), + ], + timeInterpolation=TimeInterpolation.linear, + datablock=datablock_list, + ) + + expected_message = "NaN is not supported in datablocks." + assert expected_message in str(error.value) \ No newline at end of file