[Fix]: Removing race condition within DiagManagerMonitor testing

.github/workflows/unit_tests.yaml

@@ -5,8 +5,8 @@ on:
   pull_request: # ... a PR is opened / updated
   merge_group: # ... the PR is added to the merge queue
   push:
-    branches:
-      - main # ... when merging into the main branch
+    # branches:
+    #   - main # ... when merging into the main branch
 
 # cancel running jobs if theres a newer push
 concurrency:
@@ -19,7 +19,18 @@ jobs:
     strategy:
       matrix:
         python-version: ['3.11', '3.12', '3.13']
-    name: Python ${{ matrix.python-version }}
+        extra: ['test,zarr']
+        marker-serial: ['not parallel and not gpu and not pyfms']
+        marker-parallel: ['parallel and not gpu and not pyfms']
+        include:
+          # add pyfms tests for 3.12
+          - extra: 'test,pyfms'
+            marker-serial: 'pyfms and not parallel'
+            marker-parallel: 'pyfms and parallel'
+            python-version: '3.12'
+      # don't cancel other jobs if one fails
+      fail-fast: false
+    name: Python ${{ matrix.python-version }}${{ contains(matrix.extra, 'pyfms') && ' (pyFMS)' || '' }}
     steps:
       - name: Checkout repository
         uses: actions/checkout@v6
@@ -31,17 +42,20 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
 
-      - name: Install mpi (MPICH flavor)
-        run: pip3 install mpich
+      - name: Install dependencies
+        run: |
+          sudo apt-get update --fix-missing
+          sudo apt-get install libopenmpi-dev netcdf-bin libnetcdf-dev libnetcdff-dev nco libyaml-dev diffutils
 
       - name: Install Python packages
-        run: pip3 install .[test,zarr]
+        run: |
+          pip3 install -v .[${{matrix.extra}}]
 
       - name: Run serial cpu tests
-        run: coverage run --rcfile=pyproject.toml -m pytest -m "not parallel and not gpu" tests
+        run: coverage run --rcfile=pyproject.toml -m pytest -s -m "${{matrix.marker-serial}}" tests
 
       - name: Run parallel cpu tests
-        run: mpiexec -np 6 coverage run --rcfile=pyproject.toml -m mpi4py -m pytest -m "parallel and not gpu" tests
+        run: mpiexec -np 6 --oversubscribe coverage run --rcfile=pyproject.toml -m mpi4py -m pytest -s -m "${{matrix.marker-parallel}}" tests
 
       - name: Output code coverage
         run: |

docs/docstrings/monitor/diag_manager_monitor.md

@@ -0,0 +1,3 @@
+# diag_manager_monitor
+
+::: monitor.diag_manager_monitor

mkdocs.yml

@@ -75,6 +75,7 @@ nav:
           - "subtile_grid_sizer": docstrings/initialization/subtile_grid_sizer.md
       - monitor:
           - "convert": docstrings/monitor/convert.md
+          - "diag_manager_monitor": docstrings/monitor/diag_manager_monitor.md
           - "netcdf_monitor": docstrings/monitor/netcdf_monitor.md
           - "protocol": docstrings/monitor/protocol.md
           - "zarr_monitor": docstrings/monitor/zarr_monitor.md

ndsl/optional_imports.py

@@ -17,6 +17,11 @@ def __call__(self, *args: Any, **kwargs: dict) -> None:
 except ModuleNotFoundError as err:
     zarr = RaiseWhenAccessed(err)
 
+try:
+    import pyfms
+except ModuleNotFoundError as err:
+    pyfms = RaiseWhenAccessed(err)
+
 try:
     import cupy
 except ImportError:

pyproject.toml

@@ -124,6 +124,8 @@ module = [
 markers = [
   # tests running on a cpu (e.g. with cupy)
   "gpu",
+  # tests relying on the optional pyFMS dependency
+  "pyfms",
   # tests relying on at least two MPI processes
   "parallel",
   # tests relying on the optional zarr dependency

tests/test_dm_monitor_cubed.py → tests/monitor/test_dm_monitor_cubed.py

@@ -21,9 +21,7 @@
 )
 from ndsl.config import Backend
 from ndsl.initialization import SubtileGridSizer
-
-
-pyfms = pytest.importorskip("pyfms")
+from ndsl.optional_imports import pyfms
 
 
 # init fms mpi and set up a simple domain
@@ -52,7 +50,7 @@ def fms_mpp_init():
     return domain_id
 
 
-def _create_input(reduction: str = "none"):
+def _create_input(reduction: str = "none") -> None:
     diag_config = {
         "title": "ndsl_diag_manager_test",
         "base_date": "1 1 1 0 0 0",
@@ -89,9 +87,9 @@ def _create_input(reduction: str = "none"):
 
 
 # Simple test, uses a lat/lon grid and (1, npes) layout
+@pytest.mark.pyfms
 @pytest.mark.parallel
-def test_dm_monitor():
-
+def test_dm_monitor() -> None:
     npes = MPIComm()._comm.Get_size()
     if npes % 6 != 0:
         raise RuntimeError("this test requires npes to be a multiple of 6 to run")
@@ -203,7 +201,7 @@ def test_dm_monitor():
     pe = MPIComm()._comm.Get_rank() + 1
     filename = "diag_manager_cubed_sphere.tile" + str(pe) + ".nc"
     assert Path(filename).exists()
-    ds = xr.open_mfdataset(filename, decode_times=True)
+    ds = xr.open_dataset(filename, decode_times=True)
     assert "var1" in ds
     np.testing.assert_array_equal(ds["var1"].shape, (ntimesteps, ny, nx))
     assert "var2" in ds

tests/test_dm_monitor_single.py → tests/monitor/test_dm_monitor_single.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 
 import cftime
+import dask
 import numpy as np
 import pytest
 import xarray as xr
@@ -22,12 +23,10 @@
 )
 from ndsl.config import Backend
 from ndsl.initialization import SubtileGridSizer
+from ndsl.optional_imports import pyfms
 
 
-pyfms = pytest.importorskip("pyfms")
-
-
-def _create_input(reduction: str = "none"):
+def _create_input() -> None:
     diag_config = {
         "title": "ndsl_diag_manager_test",
         "base_date": "2 1 1 1 1 1",
@@ -63,10 +62,10 @@ def _create_input(reduction: str = "none"):
         f.write(text_content)
 
 
-def test_dm_monitor_single_tile():
+@pytest.mark.pyfms
+def test_dm_monitor_single_tile() -> None:
     # mpi info
     npes = MPIComm()._comm.Get_size()
-    pe = MPIComm()._comm.Get_rank()
     # tile parameters for quantities/domains
     nx = 8
     ny = 8
@@ -220,7 +219,9 @@ def test_dm_monitor_single_tile():
 
     # check output!
     assert Path("diag_manager_single_tile.nc").exists()
-    ds = xr.open_mfdataset("diag_manager_single_tile.nc", decode_times=True)
+    with dask.config.set(scheduler="synchronous"):
+        ds = xr.open_dataset("diag_manager_single_tile.nc", decode_times=True)
+    print("Dataset opened through xarray")
     assert "var_2d" in ds
     np.testing.assert_array_equal(ds["var_2d"].shape, (ntimesteps, nx, ny))
     assert ds["var_2d"].dims == ("time", "y", "x")