diff --git a/tests/conftest.py b/tests/conftest.py index 01a870ba..fe3cde76 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,3 +13,12 @@ def tmp_cwd(tmp_path): yield tmp_path finally: os.chdir(old_dir) + + +@pytest.fixture() +def disable_crds_steppars(monkeypatch): + """ + Disable crds steppars (by setting the environment variable) + """ + monkeypatch.setitem(os.environ, "STPIPE_DISABLE_CRDS_STEPPARS", "True") + yield diff --git a/tests/test_hooks.py b/tests/test_hooks.py index 4d7d40a6..15dc5623 100644 --- a/tests/test_hooks.py +++ b/tests/test_hooks.py @@ -1,11 +1,41 @@ -import pytest +import asdf -pytest.importorskip("jwst") +from stpipe.pipeline import Pipeline +from stpipe.step import Step -from jwst.stpipe import Pipeline, Step # noqa: E402 +class FakeDataModel: + def __init__(self, data_id=None): + self.data_id = data_id -class ShovelPixelsStep(Step): + @property + def crds_observatory(self): + return "jwst" + + def get_crds_parameters(self): + return {} + + def save(self, filename): + asdf.AsdfFile({"data_id": self.data_id}).write_to(filename) + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + +class FakeStep(Step): + spec = """ + output_ext = string(default='asdf') + """ + + @classmethod + def _datamodels_open(cls, init, **kwargs): + return init + + +class ShovelPixelsStep(FakeStep): class_alias = "shovelpixels" def process(self, input_data): @@ -13,7 +43,7 @@ def process(self, input_data): return input_data -class CancelNoiseStep(Step): +class CancelNoiseStep(FakeStep): class_alias = "cancelnoise" def process(self, input_data): @@ -21,12 +51,13 @@ def process(self, input_data): return input_data -class HookStep(Step): +class HookStep(FakeStep): class_alias = "myhook" spec = """ param1 = string(default="bar") param2 = float(default=1) + output_ext = string(default='asdf') """ def process(self, input_data): @@ -38,11 +69,19 @@ def process(self, input_data): class MyPipeline(Pipeline): class_alias = "mypipeline" + spec = """ + output_ext = string(default='asdf') + """ + step_defs = { # noqa: RUF012 "shovelpixels": ShovelPixelsStep, "cancelnoise": CancelNoiseStep, } + @classmethod + def _datamodels_open(cls, init, **kwargs): + return init + def process(self, input_data): result = self.shovelpixels(input_data) result = self.cancelnoise(result) @@ -54,15 +93,14 @@ def hook_function(input_data): import logging log = logging.getLogger(__name__) - log.info("Running hook_function on data array of size %s", input_data.shape) + log.info("Running hook_function on data %s", input_data.data_id) return input_data -def test_hook_as_step_class(caplog): +def test_hook_as_step_class(caplog, disable_crds_steppars): """Test an imported Step subclass can be a hook""" - datamodels = pytest.importorskip("stdatamodels.jwst.datamodels") - model = datamodels.ImageModel((10, 10)) + model = FakeDataModel() steps = { "cancelnoise": { @@ -77,10 +115,9 @@ def test_hook_as_step_class(caplog): assert "cancelnoise.myhook" in caplog.text -def test_hook_as_step_instance(caplog): +def test_hook_as_step_instance(caplog, disable_crds_steppars): """Test an imported Step subclass instance with parameters can be a hook""" - datamodels = pytest.importorskip("stdatamodels.jwst.datamodels") - model = datamodels.ImageModel((10, 10)) + model = FakeDataModel() steps = { "shovelpixels": { @@ -94,10 +131,9 @@ def test_hook_as_step_instance(caplog): assert "Running HookStep with foo and 3" in caplog.text -def test_hook_as_string_of_importable_step_class(caplog): +def test_hook_as_string_of_importable_step_class(caplog, disable_crds_steppars): """Test a string of a fully-qualified path to Step subclass can be a hook""" - datamodels = pytest.importorskip("stdatamodels.jwst.datamodels") - model = datamodels.ImageModel((10, 10)) + model = FakeDataModel() steps = { "shovelpixels": { @@ -111,10 +147,9 @@ def test_hook_as_string_of_importable_step_class(caplog): assert "Running HookStep" in caplog.text -def test_hook_as_string_of_step_instance(caplog): +def test_hook_as_string_of_step_instance(caplog, disable_crds_steppars): """Test a string of a fully-qualified Step instance w/params""" - datamodels = pytest.importorskip("stdatamodels.jwst.datamodels") - model = datamodels.ImageModel((10, 10)) + model = FakeDataModel() steps = { "shovelpixels": { @@ -128,10 +163,10 @@ def test_hook_as_string_of_step_instance(caplog): assert "Running HookStep with foo and 2" in caplog.text -def test_hook_as_string_of_importable_function(caplog): +def test_hook_as_string_of_importable_function(caplog, disable_crds_steppars): """Test a string of a fully-qualified function path can be a hook""" - datamodels = pytest.importorskip("stdatamodels.jwst.datamodels") - model = datamodels.ImageModel((10, 10)) + data_id = 42 + model = FakeDataModel(data_id) steps = { "shovelpixels": { @@ -142,23 +177,18 @@ def test_hook_as_string_of_importable_function(caplog): } MyPipeline.call(model, steps=steps) - assert "Running hook_function on data array of size (10, 10)" in caplog.text + assert f"Running hook_function on data {data_id}" in caplog.text -def test_hook_as_systemcall(caplog, tmp_cwd): +def test_hook_as_systemcall(caplog, tmp_cwd, disable_crds_steppars): """Test a string of a terminal command""" - datamodels = pytest.importorskip("stdatamodels.jwst.datamodels") - model = datamodels.ImageModel((10, 10)) - filename = "test_hook_as_subprocess.fits" - path = tmp_cwd / filename - model.save(path) + model = FakeDataModel() - # Run post_hooks of "fitsinfo" and "fitsheader" CLI scripts from astropy + # Run post_hook CLI scripts steps = { "shovelpixels": { "post_hooks": [ - "fitsinfo {0}", - "fitsheader {0}", + "asdftool info {0}", ] } } @@ -166,7 +196,7 @@ def test_hook_as_systemcall(caplog, tmp_cwd): # Logs from fitsinfo assert "SystemCall instance created" in caplog.text - assert "Spawning 'fitsinfo stpipe.MyPipeline.shovelpixels.post_hook0" in caplog.text - - # logs from fitsheader - assert "DATAMODL= 'ImageModel'" in caplog.text + assert ( + "Spawning 'asdftool info stpipe.MyPipeline.shovelpixels.post_hook0" + in caplog.text + )