test hooks without requiring jwst

tests/test_hooks.py

@@ -1,32 +1,66 @@
+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):
         self.log.info("Shoveling...")
         return input_data
 
 
-class CancelNoiseStep(Step):
+class CancelNoiseStep(FakeStep):
     class_alias = "cancelnoise"
 
     def process(self, input_data):
         self.log.info("De-noising...")
         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,31 +186,26 @@ 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}",
             ]
         }
     }
     MyPipeline.call(model, steps=steps)
 
     # 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
+    )