Replace pytest-arraydiff with regression data style

carsus/conftest.py

@@ -67,6 +67,8 @@ def pytest_configure(config):
 from sqlalchemy.orm import Session
 from carsus import init_db
 
+from carsus.tests.fixtures.regression_data import regression_data
+
 DATA_DIR_PATH = Path(__file__).parent / "tests" / "data"
 
 
@@ -80,6 +82,17 @@ def pytest_addoption(parser):
     parser.addoption(
         "--refdata", dest="refdata", default=None, help="carsus-refdata folder location"
     )
+    parser.addoption(
+        "--carsus-regression-data",
+        default=None,
+        help="Path to the Carsus regression data directory",
+    )
+    parser.addoption(
+        "--generate-regression-data",
+        action="store_true",
+        default=False,
+        help="generate regression data instead of testing",
+    )
 
 
 def pytest_collection_modifyitems(config, items):

carsus/io/tests/test_chianti.py

@@ -1,9 +1,11 @@
 import pytest
+import pandas as pd
+from pandas import testing as pdt
 
 from numpy.testing import assert_almost_equal
 from carsus.io.chianti_ import ChiantiIonReader, ChiantiIngester, ChiantiReader
 from carsus.model import Level, Ion, Line, ECollision
-
+from carsus.tests.fixtures.regression_data import RegressionData
 
 
 @pytest.fixture
@@ -16,29 +18,32 @@ def ch_ingester(memory_session):
 class TestChiantiIonReader:
     @pytest.fixture(scope="class", params=["ne_2", "n_5"])
     def ch_ion_reader(self, request):
-        return ChiantiIonReader(request.param)
-
-    @pytest.mark.array_compare(file_format="pd_hdf")
-    def test_chianti_bound_levels(self, ch_ion_reader):
-        bound_levels = ch_ion_reader.bound_levels
-        return bound_levels
-
-    @pytest.mark.array_compare(file_format="pd_hdf")
-    def test_chianti_bound_lines(self, ch_ion_reader):
-        bound_lines = ch_ion_reader.bound_lines
-        return bound_lines
-
-    @pytest.mark.array_compare(file_format="pd_hdf")
-    def test_chianti_reader_read_levels(self, ch_ion_reader):
-        return ch_ion_reader.levels
-
-    @pytest.mark.array_compare(file_format="pd_hdf")
-    def test_chianti_reader_read_lines(self, ch_ion_reader):
-        return ch_ion_reader.lines
-
-    @pytest.mark.array_compare(file_format="pd_hdf")
-    def test_chianti_reader_read_collisions(self, ch_ion_reader):
-        return ch_ion_reader.collisions
+        yield ChiantiIonReader(request.param)
+
+    def test_chianti_bound_levels(self, regression_data, ch_ion_reader):
+        actual = ch_ion_reader.bound_levels
+        expected = regression_data.sync_hdf_store(actual)
+        pdt.assert_equal(actual, expected)
+
+    def test_chianti_bound_lines(self, regression_data, ch_ion_reader):
+        actual = ch_ion_reader.bound_lines
+        expected = regression_data.sync_hdf_store(actual)
+        pdt.assert_equal(actual, expected)
+
+    def test_chianti_reader_read_levels(self, regression_data, ch_ion_reader):
+        actual = ch_ion_reader.levels
+        expected = regression_data.sync_hdf_store(actual)
+        pdt.assert_equal(actual, expected)
+
+    def test_chianti_reader_read_lines(self, regression_data, ch_ion_reader):
+        actual = ch_ion_reader.lines
+        expected = regression_data.sync_hdf_store(actual)
+        pdt.assert_equal(actual, expected)
+
+    def test_chianti_reader_read_collisions(self, regression_data, ch_ion_reader):
+        actual = ch_ion_reader.collisions
+        expected = regression_data.sync_hdf_store(actual)
+        pdt.assert_equal(actual, expected)
 
 
 

carsus/tests/fixtures/regression_data.py

@@ -0,0 +1,171 @@
+import os
+import re
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import pytest
+
+def convert_to_snake_case(s):
+    s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", s)
+    return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower()
+
+
+class RegressionData:
+    def __init__(self, request) -> None:
+        self.request = request
+        if request.config.getoption("--carsus-regression-data") is None:
+            pytest.skip("--carsus-regression-data was not specified")
+        regression_data_path = Path(
+            request.config.getoption("--carsus-regression-data")
+        )
+        self.regression_data_path = Path(
+            os.path.expandvars(regression_data_path.expanduser())
+        )
+        self.enable_generate_reference = request.config.getoption(
+            "--generate-regression-data"
+        )
+        self.fname = f"{self.fname_prefix}.UNKNOWN_FORMAT"
+
+    @property
+    def module_name(self):
+        return self.request.node.module.__name__
+
+    @property
+    def test_name(self):
+        return self.request.node.name
+
+    @property
+    def fname_prefix(self):
+        double_under = re.compile(r"[:\[\]{}]")
+        no_space = re.compile(r'[,"\']')  # quotes and commas
+
+        name = double_under.sub("__", self.test_name)
+        name = no_space.sub("", name)
+        return name
+
+    @property
+    def relative_regression_data_dir(self):
+        relative_data_dir = Path(self.module_name.replace(".", "/"))
+        if self.request.cls is not None:
+            relative_data_dir /= convert_to_snake_case(
+                self.request.cls.__name__
+            )
+        return relative_data_dir
+
+    @property
+    def absolute_regression_data_dir(self):
+        return self.regression_data_path / self.relative_regression_data_dir
+
+    @property
+    def fpath(self):
+        return self.absolute_regression_data_dir / self.fname
+
+    def sync_dataframe(self, data, key="data"):
+        """
+        Synchronizes the dataframe with the regression data.
+
+        Parameters
+        ----------
+        data : DataFrame
+            The dataframe to be synchronized.
+        key : str, optional
+            The key to use for storing the dataframe in the regression data file. Defaults to "data".
+
+        Returns
+        -------
+        DataFrame or None
+            The synchronized dataframe if `enable_generate_reference` is `False`, otherwise `None`.
+        """
+        self.fname = f"{self.fname_prefix}.h5"
+        if self.enable_generate_reference:
+            self.fpath.parent.mkdir(parents=True, exist_ok=True)
+            data.to_hdf(
+                self.fpath,
+                key=key,
+            )
+            pytest.skip("Skipping test to generate reference data")
+        else:
+            return pd.read_hdf(self.fpath, key=key)
+
+    def sync_ndarray(self, data):
+        """
+        Synchronizes the ndarray with the regression data.
+
+        Parameters
+        ----------
+        data : ndarray
+            The ndarray to be synchronized.
+
+        Returns
+        -------
+        ndarray or None
+            The synchronized ndarray if `enable_generate_reference` is `False`, otherwise `None`.
+        """
+        self.fname = f"{self.fname_prefix}.npy"
+        if self.enable_generate_reference:
+            self.fpath.parent.mkdir(parents=True, exist_ok=True)
+            self.fpath.parent.mkdir(parents=True, exist_ok=True)
+            np.save(self.fpath, data)
+            pytest.skip("Skipping test to generate reference data")
+        else:
+            return np.load(self.fpath)
+
+    def sync_str(self, data):
+        """
+        Synchronizes the string with the regression data.
+
+        Parameters
+        ----------
+        data : str
+            The string to be synchronized.
+
+        Returns
+        -------
+        str or None
+            The synchronized string if `enable_generate_reference` is `False`, otherwise `None`.
+        """
+        self.fname = f"{self.fname_prefix}.txt"
+        if self.enable_generate_reference:
+            self.fpath.parent.mkdir(parents=True, exist_ok=True)
+            with self.fpath.open("w") as fh:
+                fh.write(data)
+            pytest.skip(
+                f"Skipping test to generate regression_data {self.fpath} data"
+            )
+        else:
+            with self.fpath.open("r") as fh:
+                return fh.read()
+
+    def sync_hdf_store(self, module, update_fname=True):
+        """
+        Synchronizes the HDF store with the regression data.
+
+        Parameters
+        ----------
+        module : object
+            The module to be synchronized.
+        update_fname : bool, optional
+            Whether to update the file name. Defaults to True.
+
+        Returns
+        -------
+        HDFStore or None
+            The synchronized HDF store if `enable_generate_reference` is `False`, otherwise `None`.
+        """
+        if update_fname:
+            self.fname = f"{self.fname_prefix}.h5"
+        if self.enable_generate_reference:
+            self.fpath.parent.mkdir(parents=True, exist_ok=True)
+            with pd.HDFStore(self.fpath, mode="w") as store:
+                module.to_hdf(store, overwrite=True)
+            pytest.skip(
+                f"Skipping test to generate regression data: {self.fpath}"
+            )
+        else:
+            return pd.HDFStore(self.fpath, mode="r")
+
+
+@pytest.fixture(scope="function")
+def regression_data(request):
+    return RegressionData(request)