Split long regtests into reusable fixtures and smaller named tests

changes/1426.general.rst

@@ -0,0 +1 @@
+Break up long regression tests to avoid needing to okify results twice.

romancal/regtest/test_catalog.py

@@ -3,58 +3,28 @@
 import asdf
 import pytest
 
-from romancal.source_catalog.source_catalog_step import SourceCatalogStep
 from romancal.stpipe import RomanStep
 
+# mark all tests in this module
+pytestmark = [pytest.mark.bigdata, pytest.mark.soctests]
 
-def passfail(bool_expr):
-    """set pass fail"""
-    if bool_expr:
-        return "Pass"
-    return "Fail"
 
+@pytest.fixture(
+    scope="module",
+    params=[
+        "r0099101001001001001_F158_visit_i2d.asdf",
+        "r0000101001001001001_01101_0001_WFI01_cal.asdf",
+    ],
+    ids=["L3", "L2"],
+)
+def run_source_catalog(rtdata_module, request):
+    rtdata = rtdata_module
 
-def check_catalog_fields(model, log, modeltype):
-    fields = model.dtype.names
-    has_pos = ("ra_centroid" in fields) and ("dec_centroid" in fields)
-    log.info(
-        f"DMS374 MSG: {modeltype} Catalog contains sky coordinates of sources? :"
-        + passfail(has_pos)
-    )
+    inputfn = request.param
 
-    has_flux = "aper_total_flux" in fields
-    log.info(f"DMS375 MSG: {modeltype} Catalog contains fluxes? :" + passfail(has_flux))
+    outputfn = inputfn.rsplit("_", 1)[0] + "_cat.asdf"
+    rtdata.output = outputfn
 
-    has_type = "is_extended" in fields
-    log.info(
-        f"DMS376 MSG: {modeltype} Catalog contains source classification? :"
-        + passfail(has_type)
-    )
-
-    has_flux_err = "aper_total_flux_err" in fields
-    log.info(
-        f"DMS386 MSG: {modeltype} Catalog contains flux uncertainties? :"
-        + passfail(has_flux_err)
-    )
-
-    has_flags = "flags" in fields
-    log.info(
-        f"DMS387 MSG: {modeltype} Catalog contains DQ flags? :" + passfail(has_flags)
-    )
-
-    return has_pos & has_flux & has_type & has_flux_err & has_flags
-
-
-@pytest.mark.bigdata
-@pytest.mark.soctests
-def test_catalog_l3(rtdata, ignore_asdf_paths):
-    # DMS374: positions on ICRF
-    # DMS375: fluxes
-    # DMS376: type of source
-    # DMS386: flux uncertainties
-    # DMS387: DQ flags
-    inputfn = "r0099101001001001001_F158_visit_i2d.asdf"
-    outputfn = inputfn.replace("_i2d", "_cat")
     rtdata.get_data(f"WFI/image/{inputfn}")
     rtdata.input = inputfn
     rtdata.get_truth(f"truth/WFI/image/{outputfn}")
@@ -64,36 +34,30 @@ def test_catalog_l3(rtdata, ignore_asdf_paths):
         rtdata.input,
     ]
     RomanStep.from_cmdline(args)
-    catalogfp = asdf.open(outputfn)
-    catalog = catalogfp["roman"]["source_catalog"]
-    step = SourceCatalogStep()
-    assert check_catalog_fields(catalog, step.log, "L3")
+    return rtdata_module
 
-    # no compare_asdf on the catalogs
 
+@pytest.fixture(scope="module")
+def catalog(run_source_catalog):
+    with asdf.open(run_source_catalog.output) as af:
+        yield af["roman"]["source_catalog"]
 
-@pytest.mark.bigdata
-@pytest.mark.soctests
-def test_catalog_l2(rtdata, ignore_asdf_paths):
-    # DMS374: positions on ICRF
-    # DMS375: fluxes
-    # DMS376: type of source
-    # DMS386: flux uncertainties
-    # DMS387: DQ flags
-    inputfn = "r0000101001001001001_01101_0001_WFI01_cal.asdf"
-    outputfn = inputfn.replace("cal", "cat")
-    rtdata.get_data(f"WFI/image/{inputfn}")
-    rtdata.input = inputfn
-    rtdata.get_truth(f"truth/WFI/image/{outputfn}")
 
-    args = [
-        "romancal.step.SourceCatalogStep",
-        rtdata.input,
-    ]
-    RomanStep.from_cmdline(args)
-    catalogfp = asdf.open(outputfn)
-    catalog = catalogfp["roman"]["source_catalog"]
-    step = SourceCatalogStep()
+@pytest.fixture(scope="module")
+def fields(catalog):
+    return catalog.dtype.names
+
 
-    assert check_catalog_fields(catalog, step.log, "L2")
-    # no compare_asdf on the catalogs
+@pytest.mark.parametrize(
+    "field",
+    (
+        "ra_centroid",  # DMS374 positions on ICRF
+        "dec_centroid",  # DMS374 positions on ICRF
+        "aper_total_flux",  # DMS375 fluxes
+        "is_extended",  # DMS376 type of source
+        "aper_total_flux_err",  # DMS386 flux uncertainties
+        "flags",  # DMS387 dq_flags
+    ),
+)
+def test_has_field(fields, field):
+    assert field in fields

romancal/regtest/test_mos_pipeline.py

@@ -9,18 +9,13 @@
 
 from .regtestdata import compare_asdf
 
+# mark all tests in this module
+pytestmark = [pytest.mark.bigdata, pytest.mark.soctests]
 
-def passfail(bool_expr):
-    """set pass fail"""
-    if bool_expr:
-        return "Pass"
-    return "Fail"
 
-
-@pytest.mark.bigdata
-@pytest.mark.soctests
-def test_level3_mos_pipeline(rtdata, ignore_asdf_paths):
-    """Tests for level 3 processing requirements DMS356"""
+@pytest.fixture(scope="module")
+def run_mos(rtdata_module):
+    rtdata = rtdata_module
     rtdata.get_asn("WFI/image/L3_regtest_asn.json")
 
     # Test Pipeline
@@ -32,123 +27,89 @@ def test_level3_mos_pipeline(rtdata, ignore_asdf_paths):
     ]
     MosaicPipeline.from_cmdline(args)
     rtdata.get_truth(f"truth/WFI/image/{output}")
-    diff = compare_asdf(rtdata.output, rtdata.truth, **ignore_asdf_paths)
-    assert diff.identical, diff.report()
+    return rtdata
+
+
+@pytest.fixture(scope="module")
+def output_filename(run_mos):
+    return run_mos.output
+
+
+@pytest.fixture(scope="module")
+def output_model(output_filename):
+    with rdm.open(output_filename) as model:
+        yield model
+
+
+@pytest.fixture(scope="module")
+def truth_filename(run_mos):
+    return run_mos.truth
 
-    # Generate thumbnail image
-    input_file = "r0099101001001001001_F158_visit_i2d.asdf"
-    thumbnail_file = "r0099101001001001001_F158_visit_thumb.png"
 
-    preview_cmd = f"stpreview to {input_file} {thumbnail_file} 256 256 roman"
+@pytest.fixture(scope="module")
+def thumbnail_filename(output_filename):
+    thumbnail_filename = output_filename.rsplit("_", 1)[0] + "_thumb.png"
+    preview_cmd = f"stpreview to {output_filename} {thumbnail_filename} 256 256 roman"
     os.system(preview_cmd)  # nosec
+    return thumbnail_filename
 
-    # Generate preview image
-    input_file = "r0099101001001001001_F158_visit_i2d.asdf"
-    preview_file = "r0099101001001001001_F158_visit_preview.png"
-    preview_cmd = f"stpreview to {input_file} {preview_file} 1080 1080 roman"
+
+@pytest.fixture(scope="module")
+def preview_filename(output_filename):
+    preview_filename = output_filename.rsplit("_", 1)[0] + "_preview.png"
+    preview_cmd = f"stpreview to {output_filename} {preview_filename} 1080 1080 roman"
     os.system(preview_cmd)  # nosec
+    return preview_filename
 
-    # expected catalog and segmentation files
-    catalog_file = "r0099101001001001001_F158_visit_cat.asdf"
-    segm_file = "r0099101001001001001_F158_visit_segm.asdf"
-
-    # Perform DMS tests
-    # Initial prep
-    model = rdm.open(rtdata.output)
-    pipeline = MosaicPipeline()
-
-    # DMS356 result is an ImageModel
-    pipeline.log.info(
-        "DMS356 MSG: Testing that result is a Level 3 mosaic model......."
-        + passfail(isinstance(model, rdm.datamodels.MosaicModel))
-    )
-
-    # DMS356 Test that skymatch step is complete
-    pipeline.log.info(
-        "Status of the step:             skymatch    "
-        + str(model.meta.cal_step.skymatch)
-    )
-    # DMS356 Test that the thumbnail image exists
-    pipeline.log.info(
-        "Status of the step:             thumbnail image    "
-        + passfail(os.path.isfile(thumbnail_file))
-    )
-    # DMS356 Test that the preview image exists
-    pipeline.log.info(
-        "Status of the step:             preview image    "
-        + passfail(os.path.isfile(preview_file))
-    )
-    # DMS374 Test that the output catalog exists
-    pipeline.log.info(
-        "Check that the catalog file exists   " + passfail(os.path.isfile(catalog_file))
-    )
-    # DMS374 Test that the segmentation file exists
-    pipeline.log.info(
-        "Check that the degmentation file exists   "
-        + passfail(os.path.isfile(segm_file))
-    )
-    pipeline.log.info(
-        "DMS400 MSG: Testing completion of skymatch in the Level 3  output......."
-        + passfail(model.meta.cal_step.skymatch == "COMPLETE")
-    )
-    assert model.meta.cal_step.skymatch == "COMPLETE"
-    pipeline.log.info(
-        "Status of the step:             skymatch    "
-        + str(model.meta.cal_step.skymatch)
-    )
-    pipeline.log.info(
-        "DMS400 MSG: SkyMatchStep added meta.background? :"
-        f'  {hasattr(model.meta.individual_image_meta, "background")}'
-    )
-    assert hasattr(model.meta.individual_image_meta, "background")
-
-    pipeline.log.info(
-        "DMS400 MSG: SkyMatchStep populated meta.background.level? :"
-        f"  {any(model.meta.individual_image_meta.background['level'] != 0)}"
-    )
-    assert any(model.meta.individual_image_meta.background["level"] != 0)
-    pipeline.log.info(
-        "DMS86 MSG: Testing completion of outlier detection in the Level 3 image output......."
-        + passfail(model.meta.cal_step.outlier_detection == "COMPLETE")
-    )
-    assert model.meta.cal_step.outlier_detection == "COMPLETE"
-    pipeline.log.info(
-        "Status of the step:             outlier_detection    "
-        + str(model.meta.cal_step.outlier_detection)
-    )
-    assert model.meta.cal_step.resample == "COMPLETE"
-    pipeline.log.info(
-        "Status of the step:             resample          "
-        + str(model.meta.cal_step.resample)
-    )
-    pipeline.log.info(
-        "DMS86 MSG: Testing completion of resample in the Level 3 image output......."
-        + passfail(model.meta.cal_step.resample == "COMPLETE")
-    )
-
-
-@pytest.mark.bigdata
-@pytest.mark.soctests
-def test_hlp_mosaic_pipeline(rtdata, ignore_asdf_paths):
-    """Tests for level 3 mosaic requirements DMS373"""
-    rtdata.get_asn("WFI/image/L3_mosaic_asn.json")
 
-    # Test Pipeline
-    output = "r0099101001001001001_r274dp63x31y81_prompt_F158_i2d.asdf"
-    rtdata.output = output
-    args = [
-        "roman_mos",
-        rtdata.input,
-    ]
-    MosaicPipeline.from_cmdline(args)
-    rtdata.get_truth(f"truth/WFI/image/{output}")
-    pipeline = MosaicPipeline()
-    diff = compare_asdf(rtdata.output, rtdata.truth, **ignore_asdf_paths)
+def test_output_matches_truth(output_filename, truth_filename, ignore_asdf_paths):
+    # DMS356
+    diff = compare_asdf(output_filename, truth_filename, **ignore_asdf_paths)
     assert diff.identical, diff.report()
 
-    model = rdm.open(rtdata.output, lazy_load=False)
 
-    pipeline.log.info(
-        "DMS373 MSG: Testing the creation of a Level 3 mosaic image resampled to a skycell"
-        + passfail(model.meta.cal_step.resample == "COMPLETE")
-    )
+def test_thumbnail_exists(thumbnail_filename):
+    # DMS356
+    assert os.path.isfile(thumbnail_filename)
+
+
+def test_preview_exists(preview_filename):
+    # DMS356
+    assert os.path.isfile(preview_filename)
+
+
+@pytest.mark.parametrize("suffix", ("cat", "segm"))
+def test_file_exists(output_filename, suffix):
+    # DMS374 for catalog and segm
+    expected_filename = output_filename.rsplit("_", 1)[0] + f"_{suffix}.asdf"
+    assert os.path.isfile(expected_filename)
+
+
+def test_output_is_mosaic(output_model):
+    # DMS356
+    assert isinstance(output_model, rdm.datamodels.MosaicModel)
+
+
+@pytest.mark.parametrize(
+    "step_name",
+    (
+        "skymatch",
+        "outlier_detection",
+        "resample",
+    ),
+)
+def test_steps_ran(output_model, step_name):
+    # DMS356
+    # DMS400 for skymatch
+    # DMS86 for outlier_detection and resample
+    assert getattr(output_model.meta.cal_step, step_name) == "COMPLETE"
+
+
+def test_added_background(output_model):
+    # DMS400
+    assert hasattr(output_model.meta.individual_image_meta, "background")
+
+
+def test_added_background_level(output_model):
+    # DMS400
+    assert any(output_model.meta.individual_image_meta.background["level"] != 0)