Break test using parametrize

Author: enekomartinmartinez

pysd/py_backend/output.py

@@ -362,6 +362,9 @@ def __create_ds_vars(self, model, capture_elements, time_dim=True):
             var = self.ds.createVariable(key, "f8", dims, **kwargs)
             # adding metadata for each var from the model.doc
             for col in model.doc.columns:
+                if col in ["Subscripts", "Limits"]:
+                    # pass those that cannot be saved as attributes
+                    continue
                 var.setncattr(
                     col,
                     model.doc.loc[model.doc["Py Name"] == key, col].values[0]

tests/pytest_pysd/pytest_output.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from warnings import simplefilter, catch_warnings
+import shutil
 
 import pytest
 import numpy as np
@@ -14,29 +14,40 @@
 import pysd
 
 
-_root = Path(__file__).parent.parent
-
-test_model_look = _root.joinpath(
+test_model_look = Path(
     "test-models/tests/get_lookups_subscripted_args/"
-    + "test_get_lookups_subscripted_args.mdl")
-test_model_constants = _root.joinpath(
+    "test_get_lookups_subscripted_args.mdl"
+)
+test_model_constants = Path(
     "test-models/tests/get_constants_subranges/"
     "test_get_constants_subranges.mdl"
 )
-test_model_numeric_coords = _root.joinpath(
+test_model_numeric_coords = Path(
     "test-models/tests/subscript_1d_arrays/"
     "test_subscript_1d_arrays.mdl"
 )
-test_variable_step = _root.joinpath(
+test_variable_step = Path(
     "test-models/tests/control_vars/"
     "test_control_vars.mdl"
 )
-test_partial_definitions = _root.joinpath(
+test_partial_definitions = Path(
     "test-models/tests/partial_range_definitions/"
     "test_partial_range_definitions.mdl"
 )
 
 
+@pytest.fixture
+@pytest.mark.filterwarnings("ignore")
+def model(_root, tmp_path, model_path):
+    """
+    Copy model to the tmp_path and translate it
+    """
+
+    target = tmp_path / model_path.parent
+    shutil.copytree(_root / model_path.parent, target)
+    return pysd.read_vensim(target / model_path.name)
+
+
 class TestOutput():
 
     def test_output_handler_interface(self):
@@ -124,151 +135,167 @@ class EmptyHandler(OutputHandlerInterface):
             EmptyHandler.add_run_elements(
                 EmptyHandler, "model", "capture")
 
-    def test_output_nc(self, shared_tmpdir):
-        model = pysd.read_vensim(test_model_look)
-        model.progress = False
-
-        out_file = shared_tmpdir.joinpath("results.nc")
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            model.run(output_file=out_file)
+    @pytest.mark.parametrize(
+        "model_path,dims,values",
+        [
+            (
+                test_model_look,
+                {
+                    "Rows": 2,
+                    "Dim": 2,
+                    "time": 61
+                },
+                {
+                    "lookup_1d_time": (("time",), None),
+                    "d2d": (("time", "Rows", "Dim"), None),
+                    "initial_time": (tuple(), 0),
+                    "final_time": (tuple(), 30),
+                    "saveper": (tuple(), 0.5),
+                    "time_step": (tuple(), 0.5)
+                }
+
+            ),
+            (
+                test_model_constants,
+                {
+                    "dim1": 5,
+                    "dim1a": 2,
+                    "dim1c": 3,
+                    'time': 2
+                },
+                {
+                    "constant": (
+                        ("dim1",),
+                        np.array([0.,  0.,  1., 15., 50.])
+                    )
+                }
+            ),
+            (
+                test_model_numeric_coords,
+                {
+                    "One Dimensional Subscript": 3,
+                    'time': 101
+                },
+                {
+                    "rate_a": (
+                        ("One Dimensional Subscript",),
+                        np.array([0.01, 0.02, 0.03])),
+                    "stock_a": (
+                        ("time", "One Dimensional Subscript"),
+                        np.array([
+                            np.arange(0, 1.0001, 0.01),
+                            np.arange(0, 2.0001, 0.02),
+                            np.arange(0, 3.0001, 0.03)],
+                            dtype=float).transpose()),
+                    "time": (("time",), np.arange(0.0, 101.0, 1.0))
+                }
+            ),
+            (
+                test_variable_step,
+                {
+                    "time": 25
+                },
+                {
+                    "final_time": (
+                        ("time",),
+                        np.array([
+                            10., 10., 10., 10., 10., 10.,
+                            50., 50., 50., 50., 50., 50.,
+                            50., 50., 50., 50., 50., 50.,
+                            50., 50., 50., 50., 50., 50., 50.
+                        ])),
+                    "initial_time": (
+                        ("time",),
+                        np.array([
+                            0., 0., 0., 0.2, 0.2, 0.2,
+                            0.2, 0.2, 0.2, 0.2, 0.2, 0.2,
+                            0.2, 0.2, 0.2, 0.2, 0.2, 0.2,
+                            0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2
+                        ])),
+                    "time_step": (
+                        ("time",),
+                        np.array([
+                            1., 1., 1., 1., 0.5, 0.5,
+                            0.5, 0.5, 0.5, 0.5, 0.5, 0.5,
+                            0.5, 0.5, 0.5, 0.5, 0.5, 0.5,
+                            0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5
+                        ])),
+                    "saveper": (
+                        ("time",),
+                        np.array([
+                            1., 1., 1., 1., 0.5, 0.5,
+                            0.5, 0.5, 0.5, 0.5, 0.5, 0.5,
+                            0.5, 0.5, 0.5, 0.5, 5., 5.,
+                            5., 5., 5., 5., 5., 5., 5.]))
+                }
+            ),
+            (
+                test_partial_definitions,
+                {
+                    "my range": 5,
+                    "time": 11
+                },
+                {
+                    "partial_data": (("time", "my range"), None),
+                    "partial_constants": (("my range",), None)
+                }
+            )
+        ],
+        ids=["lookups", "constants", "numeric_coords", "variable_step",
+             "partial_definitions"]
+    )
+    @pytest.mark.filterwarnings("ignore")
+    def test_output_nc(self, tmp_path, model, dims, values):
+
+        out_file = tmp_path.joinpath("results.nc")
+
+        model.run(output_file=out_file)
 
         with nc.Dataset(out_file, "r") as ds:
             assert ds.ncattrs() == [
                 'description', 'model_file', 'timestep', 'initial_time',
                 'final_time']
-            assert list(ds.dimensions.keys()) == ["Rows", "Dim", "time"]
+            assert list(ds.dimensions) == list(dims)
             # dimensions are stored as variables
-            assert ds["Rows"].size == 2
-            assert "Rows" in ds.variables.keys()
-            assert "time" in ds.variables.keys()
-            # scalars do not have the time dimension
-            assert ds["initial_time"].size == 1
-            # cache step variables have the "time" dimension
-            assert ds["lookup_1d_time"].dimensions == ("time",)
-
-            assert ds["d2d"].dimensions == ("time", "Rows", "Dim")
-
-            with catch_warnings(record=True) as w:
-                simplefilter("always")
-                assert ds["d2d"].Comment == "Missing"
-                assert ds["d2d"].Units == "Missing"
-
-        # test cache run variables with dimensions
-        model2 = pysd.read_vensim(test_model_constants)
-        model2.progress = False
-
-        out_file2 = shared_tmpdir.joinpath("results2.nc")
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            model2.run(output_file=out_file2)
-
-        with nc.Dataset(out_file2, "r") as ds:
-            assert ds["time_step"].size == 1
-            assert "constant" in list(ds.variables.keys())
-            assert ds["constant"].dimensions == ("dim1",)
-
-            with catch_warnings(record=True) as w:
-                simplefilter("always")
-                assert ds["dim1"][:].data.dtype == "S1"
-
-        # dimension with numeric coords
-        model3 = pysd.read_vensim(test_model_numeric_coords)
-        model3.progress = False
-
-        out_file3 = shared_tmpdir.joinpath("results3.nc")
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            model3.run(output_file=out_file3)
-
-        # using xarray instead of netCDF4 to load the dataset
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            ds = xr.open_dataset(out_file3, engine="netcdf4")
-
-        assert "time" in ds.dims
-        assert ds["rate_a"].dims == ("One Dimensional Subscript",)
-        assert ds["stock_a"].dims == ("time", "One Dimensional Subscript")
-
-        # coordinates get dtype=object when their length is different
-        assert ds["One Dimensional Subscript"].data.dtype == "O"
-
-        # check data
-        assert np.array_equal(
-                ds["time"].data, np.arange(0.0, 101.0, 1.0))
-        assert np.allclose(
-                ds["stock_a"][0, :].data, np.array([0.0, 0.0, 0.0]))
-        assert np.allclose(
-                ds["stock_a"][-1, :].data, np.array([1.0, 2.0, 3.0]))
-        assert ds["rate_a"].shape == (3,)
-
-        # variable attributes
-        assert list(model.doc.columns) == list(ds["stock_a"].attrs.keys())
-
-        # global attributes
-        assert list(ds.attrs.keys()) == [
-            'description', 'model_file', 'timestep', 'initial_time',
-            'final_time']
-
-        model4 = pysd.read_vensim(test_variable_step)
-        model4.progress = False
-
-        out_file4 = shared_tmpdir.joinpath("results4.nc")
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            model4.run(output_file=out_file4)
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            ds = xr.open_dataset(out_file4, engine="netcdf4")
-
-        # global attributes for variable timestep
-        assert ds.attrs["timestep"] == "Variable"
-        assert ds.attrs["final_time"] == "Variable"
-
-        # saveper final_time and time_step have time dimension
-        assert ds["saveper"].dims == ("time",)
-        assert ds["time_step"].dims == ("time",)
-        assert ds["final_time"].dims == ("time",)
-
-        assert np.unique(ds["time_step"]).size == 2
-
-    def test_output_nc2(self, shared_tmpdir):
-        # dimension with numeric coords
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            model5 = pysd.read_vensim(test_partial_definitions)
-        model5.progress = False
-
-        out_file5 = shared_tmpdir.joinpath("results5.nc")
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            model5.run(output_file=out_file5)
-
-        # using xarray instead of netCDF4 to load the dataset
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            ds = xr.open_dataset(out_file5, engine="netcdf4")
-
-        print(ds)
-
-    @pytest.mark.parametrize("fmt,sep", [("csv", ","), ("tab", "\t")])
-    def test_output_csv(self, fmt, sep, capsys, shared_tmpdir):
-        model = pysd.read_vensim(test_model_look)
-        model.progress = False
-
-        out_file = shared_tmpdir.joinpath("results." + fmt)
-
-        with catch_warnings(record=True) as w:
-            simplefilter("always")
-            model.run(output_file=out_file)
+            for dim, n in dims.items():
+                # check dimension size
+                assert ds[dim].size == n
+                assert dim in ds.variables.keys()
+                # check dimension type
+                if dim != "time":
+                    assert ds[dim].dtype in ["S1", str]
+                else:
+                    assert ds[dim].dtype == float
+
+            for var, (dim, val) in values.items():
+                # check variable dimensions
+                assert ds[var].dimensions == dim
+                if val is not None:
+                    # check variable values if given
+                    assert np.all(np.isclose(ds[var][:].data, val))
+
+            # Check variable attributes
+            doc = model.doc
+            doc.set_index("Py Name", drop=False, inplace=True)
+            doc.drop(columns=["Subscripts", "Limits"], inplace=True)
+
+            for var in doc["Py Name"]:
+                if doc.loc[var, "Type"] == "Lookup":
+                    continue
+                for key in doc.columns:
+                    assert getattr(ds[var], key) == (doc.loc[var, key]
+                                                     or "Missing")
+
+    @pytest.mark.parametrize(
+        "model_path,fmt,sep",
+        [
+            (test_model_look, "csv", ","),
+            (test_model_look, "tab", "\t")])
+    @pytest.mark.filterwarnings("ignore")
+    def test_output_csv(self, fmt, sep, capsys, model, tmp_path):
+        out_file = tmp_path.joinpath("results." + fmt)
+
+        model.run(output_file=out_file)
 
         captured = capsys.readouterr()  # capture stdout
         assert f"Data saved in '{out_file}'" in captured.out
@@ -282,10 +309,10 @@ def test_output_csv(self, fmt, sep, capsys, shared_tmpdir):
         assert "lookup 3d time[B;Row1]" in df.columns or \
             "lookup 3d time[B,Row1]" in df.columns
 
-    def test_dataset_handler_step_setter(self, shared_tmpdir):
-        model = pysd.read_vensim(test_model_look)
+    @pytest.mark.parametrize("model_path", [test_model_look])
+    def test_dataset_handler_step_setter(self, tmp_path, model):
         capture_elements = set()
-        results = shared_tmpdir.joinpath("results.nc")
+        results = tmp_path.joinpath("results.nc")
         output = ModelOutput(model, capture_elements, results)
 
         # Dataset handler step cannot be modified from the outside