JP-3930: Make sure input is not modified inside Steps

changes/9725.general.rst

@@ -0,0 +1 @@
+Ensure that all steps do not modify input datamodels.

jwst/ami/tests/test_ami_analyze.py

@@ -1,8 +1,5 @@
 """Unit tests for ami_analyze module and step."""
 
-import math
-
-import numpy as np
 import pytest
 import stpipe
 from astropy.io import fits
@@ -46,16 +43,9 @@ def test_ami_analyze_step_no_affine(example_model):
     AmiAnalyzeStep.call(example_model, affine2d=None)
 
 
-def create_throughput(nelem):
-    """Create a symmetric dummy throughput function that has values near
-    0 on the wings and near 1 at the center.
-    """
-    ctr = int(nelem / 2.0)
-
-    lower_half = [2.0 / (1.0 + math.e ** (-5.0 * i / ctr)) - 1.0 for i in range(ctr)]
-
-    throughput = np.zeros(nelem, dtype=np.float32)
-    throughput[:ctr] = lower_half
-    throughput[ctr:] = lower_half[::-1]  # mirror image for upper half
-
-    return throughput
+def test_output_is_not_input(example_model):
+    results = AmiAnalyzeStep.call(example_model)
+    assert example_model.meta.cal_step.ami_analyze is None
+    for result in results:
+        assert result is not example_model
+        assert result.meta.cal_step.ami_analyze == "COMPLETE"

jwst/ami/tests/test_ami_average.py

@@ -12,4 +12,9 @@ def test_ami_average_deprecated(example_model):
         image_list.append(ImageModel(example_model.data[i, :, :]))
 
     with pytest.deprecated_call():
-        AmiAverageStep.call(image_list)
+        result = AmiAverageStep.call(image_list)
+        assert result is not image_list
+        assert result.meta.cal_step.ami_average == "COMPLETE"
+        for model in image_list:
+            assert result is not model
+            assert model.meta.cal_step.ami_average is None

jwst/ami/tests/test_ami_normalize.py

@@ -176,3 +176,12 @@ def test_ami_normalize(oi_data, ref_data):
     assert np.allclose(result.t3["T3PHI"], RAW_PHI - REF_PHI)
     assert np.allclose(result.q4["Q4AMP"], np.log(RAW_AMP / REF_AMP))
     assert np.allclose(result.q4["Q4PHI"], RAW_PHI)
+
+
+def test_output_is_not_input(oi_data, ref_data):
+    result = AmiNormalizeStep.call(oi_data, ref_data)
+    assert result is not oi_data
+    assert result is not ref_data
+    assert oi_data.meta.cal_step.ami_normalize is None
+    assert ref_data.meta.cal_step.ami_normalize is None
+    assert result.meta.cal_step.ami_normalize == "COMPLETE"

jwst/assign_wcs/tests/test_assign_wcs_step.py

@@ -0,0 +1,46 @@
+"""Test the AssignWCSStep."""
+
+import numpy as np
+from stdatamodels.jwst import datamodels
+
+from jwst.assign_wcs import AssignWcsStep
+from jwst.assign_wcs.tests.test_miri import create_hdul as create_miri
+from jwst.assign_wcs.tests.test_nircam import create_hdul as create_nircam
+from jwst.assign_wcs.tests.test_niriss import create_hdul as create_niriss
+
+
+def test_assign_wcs_step_miri_ifu():
+    hdul = create_miri(detector="MIRIFULONG", channel="34", band="MEDIUM")
+    model = datamodels.CubeModel(hdul)
+    model.data = np.zeros((3, 40, 50))
+    result = AssignWcsStep.call(model)
+    assert result is not model
+    assert result.meta.cal_step.assign_wcs == "COMPLETE"
+    assert model.meta.cal_step.assign_wcs is None
+
+
+def test_assign_wcs_step_nis_wfss():
+    hdul = create_niriss(filtername="GR150R", pupil="F200W", exptype="NIS_WFSS")
+    model = datamodels.ImageModel(hdul)
+    result = AssignWcsStep.call(model)
+    assert result is not model
+    assert result.meta.cal_step.assign_wcs == "COMPLETE"
+    assert model.meta.cal_step.assign_wcs is None
+
+
+def test_assign_wcs_step_nrc_wfss():
+    hdul = create_nircam(exptype="NRC_WFSS", filtername="F444W", pupil="GRISMR")
+    model = datamodels.ImageModel(hdul)
+    result = AssignWcsStep.call(model)
+    assert result is not model
+    assert result.meta.cal_step.assign_wcs == "COMPLETE"
+    assert model.meta.cal_step.assign_wcs is None
+
+
+def test_unsupported_input(caplog):
+    model = datamodels.SlitModel()
+    result = AssignWcsStep.call(model)
+    assert result is not model
+    assert "type is not supported" in caplog.text
+    assert result.meta.cal_step.assign_wcs == "SKIPPED"
+    assert model.meta.cal_step.assign_wcs is None

jwst/badpix_selfcal/badpix_selfcal_step.py

@@ -233,10 +233,10 @@ def _parse_inputs(input_data, selfcal_list, bkg_list):
                     "Input data contains multiple science exposures. "
                     "This is not supported in calwebb_spec2 steps."
                 )
-            input_sci = sci_models[0]
+            input_sci = sci_models[0].copy()
 
         elif isinstance(input_dm, dm.IFUImageModel) or isinstance(input_dm, dm.ImageModel):
-            input_sci = input_dm
+            input_sci = input_dm.copy()
 
         else:
             raise TypeError(

jwst/badpix_selfcal/tests/test_badpix_selfcal.py

@@ -80,11 +80,11 @@ def create_reference_files(datamodel):
 @pytest.fixture(scope="module")
 def background():
     """
-    Create background IFUImageModel for testing. This is a mockup of the expected
-    background data in a .rate file.
+    Create background IFUImageModel for testing.
+
+    This is a mockup of the expected background data in a .rate file.
     Three components: random noise, low-order variability, and outliers
     """
-
     # random noise
     rng = np.random.default_rng(seed=77)
     shp = (1024, 1032)
@@ -120,7 +120,9 @@ def background():
 
 @pytest.fixture(scope="module")
 def sci(background):
-    """Create science IFUImageMoel for testing.
+    """
+    Create science IFUImageModel for testing.
+
     Same data as background with different noise.
     """
     hdul = create_hdul(detector="MIRIFULONG", channel="34", band="LONG")
@@ -134,8 +136,11 @@ def sci(background):
 
 @pytest.fixture(scope="module")
 def asn(tmp_cwd_module, background, sci):
-    """Create association for testing. Needs at least
-    two background images to properly test the step."""
+    """
+    Create association for testing.
+
+    Needs at least two background images to properly test the step.
+    """
 
     sci_path = tmp_cwd_module / "sci.fits"
     sci.save(sci_path)
@@ -167,7 +172,10 @@ def asn(tmp_cwd_module, background, sci):
 
 
 def test_input_parsing(asn, sci, background):
-    """Test that the input parsing function correctly identifies the science,
+    """
+    Test the input parsing function.
+
+    It should correctly identify the science,
     background, and selfcal exposures in the association file
     given association vs imagemodel inputs, and selfcal_list vs not
 
@@ -229,11 +237,20 @@ def test_input_parsing(asn, sci, background):
     assert len(selfcal_list) == 4
 
 
+def test_bad_input():
+    input_data = dm.SlitModel()
+    with pytest.raises(TypeError, match="Cannot continue"):
+        _parse_inputs(input_data, [], [])
+
+
+def test_bad_input_in_container():
+    input_data = dm.ModelContainer([dm.ImageModel(), dm.ImageModel()])
+    with pytest.raises(ValueError, match="multiple science exposures"):
+        _parse_inputs(input_data, [], [])
+
+
 def test_background_flagger_mrs(background):
-    """
-    Ensure the right number of outliers are found, and that
-    true outliers are among those.
-    """
+    """Ensure the right number of true outliers are found."""
 
     bg = background.data
 
@@ -252,6 +269,8 @@ def test_background_flagger_mrs(background):
 
 def test_apply_flags(background):
     """
+    Test apply_flags.
+
     Ensure that flagged pixels are set to NaN in the data and err arrays,
     and that the DQ flag is set to 1.
     """
@@ -276,10 +295,7 @@ def test_apply_flags(background):
 
 @pytest.mark.parametrize("dset", ["sci", "asn"])
 def test_badpix_selfcal_step(request, dset):
-    """Test the badpix_selfcal step. This is a functional test that checks
-    that the step runs without error. The output will be checked by the
-    regtest.
-    """
+    """Smoke test for the badpix_selfcal step."""
     input_data = request.getfixturevalue(dset)
     result = BadpixSelfcalStep.call(input_data, skip=False, force_single=True)
 
@@ -294,19 +310,29 @@ def test_badpix_selfcal_step(request, dset):
         assert isinstance(result[0], dm.IFUImageModel)
         assert len(result[1]) == 2
 
+    # Make sure input is not modified
+    assert result[0] is not input_data
+    if dset == "sci":
+        assert input_data.meta.cal_step.badpix_selfcal is None
+
 
 def test_expected_fail_sci(sci):
     """Test that the step fails as expected when given only a science exposure
     and force_single is False.
     """
     result = BadpixSelfcalStep.call(sci, skip=False, force_single=False)
     assert result[0].meta.cal_step.badpix_selfcal == "SKIPPED"
+    assert result[0] is not sci
+    assert sci.meta.cal_step.badpix_selfcal is None
 
 
 @pytest.fixture(scope="module")
 def vertical_striping():
-    """2-D array with vertical stripes and some outliers, used to test that
-    the step is flagging in the correct (along-dispersion) direction
+    """
+    Make an array with vertical stripes.
+
+    Array is 2-D with vertical stripes and some outliers, used to test that
+    the step is flagging in the correct (along-dispersion) direction.
     """
     shp = (102, 96)
     stripes = np.zeros(shp)
@@ -342,3 +368,17 @@ def test_dispersion_direction(vertical_striping, dispaxis):
 
     elif dispaxis is None:
         assert len(flagged_indices[0]) == 1
+
+
+def test_save_bkg(asn, tmp_path):
+    """Test background saving."""
+    BadpixSelfcalStep.call(
+        asn, output_dir=str(tmp_path), output_file="test", save_flagged_bkg=True, save_results=True
+    )
+    expected = [
+        "test_badpix_selfcal_bkg_0.fits",
+        "test_badpix_selfcal_bkg_1.fits",
+        "test_badpixselfcalstep.fits",
+    ]
+    for filename in expected:
+        assert (tmp_path / filename).exists()

jwst/barshadow/tests/test_barshadow_step.py

@@ -52,6 +52,10 @@ def test_barshadow_step(nirspec_mos_model):
     result = BarShadowStep.call(model)
     assert result.meta.cal_step.barshadow == "COMPLETE"
 
+    # make sure input is not modified
+    assert result is not model
+    assert model.meta.cal_step.barshadow is None
+
     # check all slits for appropriate correction
     for slit in result.slits:
         assert slit.barshadow_corrected is True
@@ -131,6 +135,11 @@ def test_barshadow_no_reffile(monkeypatch, nirspec_mos_model):
     assert result.meta.cal_step.barshadow == "SKIPPED"
     assert result.slits[0].barshadow.size == 0
     assert result.slits[0].barshadow_corrected is None
+
+    # make sure input is not modified
+    assert result is not model
+    assert model.meta.cal_step.barshadow is None
+
     result.close()
 
 

jwst/charge_migration/charge_migration_step.py

@@ -37,14 +37,14 @@ def process(self, step_input):
         """
         # Open the input data model
         with datamodels.RampModel(step_input) as input_model:
+            # Work on a copy
+            result = input_model.copy()
+
             if input_model.data.shape[1] < 3:  # skip step if only 1 or 2 groups/integration
                 log.info("Too few groups per integration; skipping charge_migration")
 
-                input_model.meta.cal_step.charge_migration = "SKIPPED"
-                return input_model
-
-            # Work on a copy
-            result = input_model.copy()
+                result.meta.cal_step.charge_migration = "SKIPPED"
+                return result
 
             # Retrieve the parameter value(s)
             signal_threshold = self.signal_threshold

jwst/charge_migration/tests/test_charge_migration.py

@@ -220,6 +220,22 @@ def test_nearest_neighbor_3():
     npt.assert_array_equal(out_model.groupdq, gdq_check)
 
 
+def test_output_is_not_input():
+    """Test that input is not modified."""
+    ngroups, nints, nrows, ncols = 4, 1, 1, 1
+    ramp_model, pixdq, groupdq, err = create_mod_arrays(ngroups, nints, nrows, ncols)
+
+    result = ChargeMigrationStep.call(ramp_model)
+
+    # Successful completion
+    status = result.meta.cal_step.charge_migration
+    npt.assert_string_equal(status, "COMPLETE")
+
+    # Output is not input
+    assert result is not ramp_model
+    assert ramp_model.meta.cal_step.charge_migration is None
+
+
 def test_too_few_groups():
     """
     Test that processing for datasets having too few (<3) groups per integration
@@ -232,10 +248,13 @@ def test_too_few_groups():
     sig_thresh = 100.0
 
     result = ChargeMigrationStep.call(ramp_model, skip=False, signal_threshold=sig_thresh)
-    status = result.meta.cal_step.charge_migration
 
+    status = result.meta.cal_step.charge_migration
     npt.assert_string_equal(status, "SKIPPED")
 
+    assert result is not ramp_model
+    assert ramp_model.meta.cal_step.charge_migration is None
+
 
 def create_mod_arrays(ngroups, nints, nrows, ncols):
     """

jwst/clean_flicker_noise/tests/test_clean_flicker_noise_step.py

@@ -51,6 +51,9 @@ def test_run_in_pipeline(skip):
     else:
         assert cleaned.meta.cal_step.clean_flicker_noise == "COMPLETE"
 
+    # Either way, input model is not modified
+    assert input_model.meta.cal_step.clean_flicker_noise is None
+
     input_model.close()
     cleaned.close()
 
@@ -185,3 +188,18 @@ def test_autoparam_failed(caplog, monkeypatch):
     assert "Auto parameter setting failed" in caplog.text
     assert "Using input parameters as provided" in caplog.text
     assert "apply_flat_field: True" not in caplog.text
+
+
+def test_output_is_not_input():
+    input_model = make_small_ramp_model()
+    cleaned = CleanFlickerNoiseStep.call(input_model)
+
+    # successful completion
+    assert cleaned.meta.cal_step.clean_flicker_noise == "COMPLETE"
+
+    # make sure input is not modified
+    assert cleaned is not input_model
+    assert input_model.meta.cal_step.clean_flicker_noise is None
+
+    input_model.close()
+    cleaned.close()