Adding more tests

Author: erikvansebille

parcels/particleset.py

@@ -729,7 +729,9 @@ def execute(
         if output_file:
             output_file.metadata["parcels_kernels"] = self._kernel.name
 
-        if np.isnat(dt) or dt is None:
+        if (dt is not None) and (not isinstance(dt, np.timedelta64)):
+            raise TypeError("dt must be a np.timedelta64 object")
+        if dt is None or np.isnat(dt):
             dt = np.timedelta64(1, "s")
         self._data["dt"][:] = dt
         sign_dt = np.sign(dt).astype(int)

tests/v4/test_kernel.py

@@ -0,0 +1,75 @@
+import numpy as np
+import pytest
+
+from parcels import (
+    AdvectionRK4,
+    Field,
+    FieldSet,
+    ParticleSet,
+    xgcm,
+)
+from parcels._datasets.structured.generic import datasets as datasets_structured
+from parcels.xgrid import XGrid
+from tests.common_kernels import MoveEast, MoveNorth
+
+
+@pytest.fixture
+def fieldset() -> FieldSet:
+    ds = datasets_structured["ds_2d_left"]
+    grid = XGrid(xgcm.Grid(ds))
+    U = Field("U", ds["U (A grid)"], grid, mesh_type="flat")
+    V = Field("V", ds["V (A grid)"], grid, mesh_type="flat")
+    return FieldSet([U, V])
+
+
+def test_multi_kernel_reuse_varnames(fieldset):
+    pset = ParticleSet(fieldset, lon=[0.5], lat=[0.5])
+
+    # Testing for merging of two Kernels with the same variable declared
+    def MoveEast1(particle, fieldset, time):  # pragma: no cover
+        add_lon = 0.2
+        particle_dlon += add_lon  # noqa
+
+    def MoveEast2(particle, fieldset, time):  # pragma: no cover
+        particle_dlon += add_lon  # noqa
+
+    pset.execute([MoveEast1, MoveEast2], runtime=np.timedelta64(2, "s"))
+    assert np.allclose(pset.lon, [0.9], atol=1e-5)  # should be 0.5 + 0.2 + 0.2 = 0.9
+
+
+def test_combined_kernel_from_list(fieldset):
+    """
+    Test pset.Kernel(List[function])
+
+    Tests that a Kernel can be created from a list functions, or a list of
+    mixed functions and kernel objects.
+    """
+    pset = ParticleSet(fieldset, lon=[0.5], lat=[0.5])
+    kernels_single = pset.Kernel([AdvectionRK4])
+    kernels_functions = pset.Kernel([AdvectionRK4, MoveEast, MoveNorth])
+
+    # Check if the kernels were combined correctly
+    assert kernels_single.funcname == "AdvectionRK4"
+    assert kernels_functions.funcname == "AdvectionRK4MoveEastMoveNorth"
+
+
+def test_combined_kernel_from_list_error_checking(fieldset):
+    """
+    Test pset.Kernel(List[function])
+
+    Tests that various error cases raise appropriate messages.
+    """
+    pset = ParticleSet(fieldset, lon=[0.5], lat=[0.5])
+
+    # Test that list has to be non-empty
+    with pytest.raises(ValueError):
+        pset.Kernel([])
+
+    # Test that list has to be all functions
+    with pytest.raises(ValueError):
+        pset.Kernel([AdvectionRK4, "something else"])
+
+    # Can't mix kernel objects and functions in list
+    with pytest.raises(ValueError):
+        kernels_mixed = pset.Kernel([pset.Kernel(AdvectionRK4), MoveEast, MoveNorth])
+        assert kernels_mixed.funcname == "AdvectionRK4MoveEastMoveNorth"

tests/v4/test_particleset_execute.py

@@ -81,6 +81,36 @@ def test_execution_endtime(fieldset, starttime, endtime, dt):
     assert abs(pset.time_nextloop - endtime) < np.timedelta64(1, "ms")
 
 
+@pytest.mark.parametrize(
+    "starttime, runtime, dt",
+    [(0, 10, 1), (0, 10, 3), (2, 16, 3), (20, 10, -1), (20, 0, -2), (5, 15, None)],
+)
+def test_execution_runtime(fieldset, starttime, runtime, dt):
+    starttime = fieldset.time_interval.left + np.timedelta64(starttime, "s")
+    runtime = np.timedelta64(runtime, "s")
+    sign_dt = 1 if dt is None else np.sign(dt)
+    dt = np.timedelta64(dt, "s")
+    pset = ParticleSet(fieldset, time=starttime, lon=0, lat=0)
+    pset.execute(DoNothing, runtime=runtime, dt=dt)
+    assert abs(pset.time_nextloop - starttime - runtime * sign_dt) < np.timedelta64(1, "ms")
+
+
+def test_execution_fail_python_exception(fieldset, npart=10):
+    pset = ParticleSet(fieldset, lon=np.linspace(0, 1, npart), lat=np.linspace(1, 0, npart))
+
+    def PythonFail(particle, fieldset, time):  # pragma: no cover
+        if particle.time >= fieldset.time_interval.left + np.timedelta64(10, "s"):
+            raise RuntimeError("Enough is enough!")
+        else:
+            pass
+
+    with pytest.raises(RuntimeError):
+        pset.execute(PythonFail, runtime=np.timedelta64(20, "s"), dt=np.timedelta64(2, "s"))
+    assert len(pset) == npart
+    assert pset.time[0] == fieldset.time_interval.left + np.timedelta64(10, "s")
+    assert all([time == fieldset.time_interval.left + np.timedelta64(8, "s") for time in pset.time[1:]])
+
+
 @pytest.mark.parametrize("verbose_progress", [True, False])
 def test_uxstommelgyre_pset_execute(verbose_progress):
     ds = datasets_unstructured["stommel_gyre_delaunay"]