Configurator Unit Tests

tests/test_configurators/conftest.py

@@ -0,0 +1,59 @@
+import keras
+import pytest
+from bayesflow.experimental.configurators import Configurator
+
+
+@pytest.fixture(params=[2, 3])
+def batch_size(request):
+    return request.param
+
+
+@pytest.fixture(params=[2, 3])
+def set_size(request):
+    return request.param
+
+
+@pytest.fixture(params=[2,3])
+def num_features(request):
+    return request.param
+
+
+@pytest.fixture(params=[True, False])
+def random_data(request, batch_size, set_size, num_features):
+    data = {
+        "var1": keras.random.normal((batch_size, set_size, num_features)),
+        "var2": keras.random.normal((batch_size, set_size, num_features)),
+        "var3": keras.random.normal((batch_size, set_size, num_features)),
+        "summary_inputs": keras.random.normal((batch_size, set_size, num_features)),
+        "summary_conditions": keras.random.normal((batch_size, set_size, num_features))
+    }
+    if request.param:
+        data["summary_outputs"] = keras.random.normal((batch_size, set_size, num_features))
+    return data
+
+
+@pytest.fixture(params=[True, False])
+def test_params(request):
+    args = {
+        "inference_variables": ["var1"],
+        "inference_conditions": ["var2", "var3"],
+        "summary_variables": ["var1"],
+        "summary_conditions": ["var2"]
+    }
+    if request.param:
+        args["inference_conditions"].append("summary_outputs")
+    return args
+
+
+@pytest.fixture(params=[True, False])
+def configurator(request, test_params):
+    if request.param:
+        return Configurator(
+            inference_variables=test_params["inference_variables"]
+        )
+    return Configurator(
+        inference_variables=test_params["inference_variables"],
+        inference_conditions=test_params["inference_conditions"],
+        summary_variables=test_params["summary_variables"],
+        summary_conditions=test_params["summary_conditions"]
+    )

tests/test_configurators/test_configurators.py

@@ -0,0 +1,48 @@
+from keras import ops
+import pytest
+
+
+def test_inference_vars_filter(random_data, configurator):
+    # Tests for correct output shape when querying inference variables
+    filtered_data = configurator.configure_inference_variables(random_data)
+    expected = ops.concatenate([random_data[v] for v in configurator.inference_variables], axis=-1)
+    assert filtered_data.shape == expected.shape
+
+
+def test_inferences_conds_filter(random_data, configurator):
+    # Tests for correct output shape when querying inference conditions w.r.t. summary_outputs
+    if not configurator.inference_conditions:
+        if "summary_outputs" in random_data:
+            assert configurator.configure_inference_conditions(random_data).shape == random_data["summary_outputs"].shape
+        else:
+            assert configurator.configure_inference_conditions(random_data) is None
+    elif not "summary_outputs" in random_data and "summary_outputs" in configurator.inference_conditions:
+        with pytest.raises(KeyError):
+            filtered_data = configurator.configure_inference_conditions(random_data)
+    else:
+        filtered_data = configurator.configure_inference_conditions(random_data)
+        tensors = [random_data[v] for v in configurator.inference_conditions]
+        if "summary_outputs" in random_data and not "summary_outputs" in configurator.inference_conditions:
+            tensors.append(random_data["summary_outputs"])
+        expected = ops.concatenate(tensors, axis=-1)
+        assert filtered_data.shape == expected.shape
+
+
+def test_summary_vars_filter(random_data, configurator):
+    # Tests for correct output shape when querying summary variables
+    if not configurator.summary_variables:
+        assert configurator.configure_summary_variables(random_data) is None
+    else:
+        filtered_data = configurator.configure_summary_variables(random_data)
+        expected = ops.concatenate([random_data[v] for v in configurator.summary_variables], axis=-1)
+        assert filtered_data.shape == expected.shape
+
+
+def test_summary_conds_filter(random_data, configurator):
+    # Tests for correct output shape when querying summary conditions
+    if not configurator.summary_conditions:
+        assert configurator.configure_summary_conditions(random_data) is None
+    else:
+        filtered_data = configurator.configure_summary_conditions(random_data)
+        expected = ops.concatenate([random_data[v] for v in configurator.summary_conditions], axis=-1)
+        assert filtered_data.shape == expected.shape