diff --git a/tests/test_hooks.py b/tests/test_hooks.py index 4d7d40a6..3eb96375 100644 --- a/tests/test_hooks.py +++ b/tests/test_hooks.py @@ -1,11 +1,44 @@ +import os + +import asdf import pytest -pytest.importorskip("jwst") +from stpipe.pipeline import Pipeline +from stpipe.step import Step + + +class FakeDataModel: + def __init__(self, data_id=None): + self.data_id = data_id + + @property + def crds_observatory(self): + return "jwst" + + def get_crds_parameters(self): + return {} -from jwst.stpipe import Pipeline, Step # noqa: E402 + 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') + """ -class ShovelPixelsStep(Step): + @classmethod + def _datamodels_open(cls, init, **kwargs): + return init + + +class ShovelPixelsStep(FakeStep): class_alias = "shovelpixels" def process(self, input_data): @@ -13,7 +46,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 +54,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 +72,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 +96,20 @@ 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 +@pytest.fixture(autouse=True, scope="function") +def disable_crds(monkeypatch): + monkeypatch.setitem(os.environ, "STPIPE_DISABLE_CRDS_STEPPARS", "True") + yield + + def test_hook_as_step_class(caplog): """Test an imported Step subclass can be a hook""" - datamodels = pytest.importorskip("stdatamodels.jwst.datamodels") - model = datamodels.ImageModel((10, 10)) + model = FakeDataModel() steps = { "cancelnoise": { @@ -79,8 +126,7 @@ def test_hook_as_step_class(caplog): def test_hook_as_step_instance(caplog): """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": { @@ -96,8 +142,7 @@ def test_hook_as_step_instance(caplog): def test_hook_as_string_of_importable_step_class(caplog): """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": { @@ -113,8 +158,7 @@ def test_hook_as_string_of_importable_step_class(caplog): def test_hook_as_string_of_step_instance(caplog): """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": { @@ -130,8 +174,8 @@ def test_hook_as_string_of_step_instance(caplog): def test_hook_as_string_of_importable_function(caplog): """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 +186,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): """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 +205,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 + )