uncomment and use get_asn

romancal/regtest/regtestdata.py

@@ -25,6 +25,8 @@
 from deepdiff.operator import BaseOperator
 from gwcs.wcstools import grid_from_bounding_box
 
+from romancal.associations.load_asn import load_asn
+
 # from romancal.lib.suffix import replace_suffix
 from romancal.stpipe import RomanStep
 
@@ -225,51 +227,51 @@
 
         return self.truth
 
-    # def get_asn(self, path=None, docopy=True, get_members=True):
-    #     """Copy association and association members from Artifactory remote
-    #     resource to the CWD/truth.
-    #
-    #     Updates self.input and self.input_remote upon completion
-    #
-    #     Parameters
-    #     ----------
-    #     path: str
-    #         The remote path
-    #
-    #     docopy : bool
-    #         Switch to control whether or not to copy a file
-    #         into the test output directory when running the test.
-    #         If you wish to open the file directly from remote
-    #         location or just to set path to source, set this to `False`.
-    #         Default: `True`
-    #
-    #     get_members: bool
-    #         If an association is the input, retrieve the members.
-    #         Otherwise, do not.
-    #     """
-    #     if path is None:
-    #         path = self.input_remote
-    #     else:
-    #         self.input_remote = path
-    #     if docopy is None:
-    #         docopy = self.docopy
-    #
-    #     # Get the association JSON file
-    #     self.input = get_bigdata(self._inputs_root, self._env, path,
-    #         docopy=docopy)
-    #     with open(self.input) as fp:
-    #         asn = load_asn(fp)
-    #         self.asn = asn
-    #
-    #     # Get each member in the association as well
-    #     if get_members:
-    #         for product in asn['products']:
-    #             for member in product['members']:
-    #                 fullpath = os.path.join(
-    #                     os.path.dirname(self.input_remote),
-    #                     member['expname'])
-    #                 get_bigdata(self._inputs_root, self._env, fullpath,
-    #                             docopy=self.docopy)
+    def get_asn(self, path=None, docopy=True, get_members=True):
+        """Copy association and association members from Artifactory remote
+        resource to the CWD/truth.
+
+        Updates self.input and self.input_remote upon completion
+
+        Parameters
+        ----------
+        path: str
+            The remote path
+
+        docopy : bool
+            Switch to control whether or not to copy a file
+            into the test output directory when running the test.
+            If you wish to open the file directly from remote
+            location or just to set path to source, set this to `False`.
+            Default: `True`
+
+        get_members: bool
+            If an association is the input, retrieve the members.
+            Otherwise, do not.
+        """
+        if path is None:
+            path = self.input_remote
+        else:
+            self.input_remote = path
+        if docopy is None:
+            docopy = self.docopy
+
+        # Get the association JSON file
+        self.input = get_bigdata(self._inputs_root, self._env, path, docopy=docopy)
+        with open(self.input) as fp:
+            asn = load_asn(fp)
+            self.asn = asn
+
+        # Get each member in the association as well
+        if get_members:
+            for product in asn["products"]:
+                for member in product["members"]:
+                    fullpath = os.path.join(
+                        os.path.dirname(self.input_remote), member["expname"]
+                    )
+                    get_bigdata(
+                        self._inputs_root, self._env, fullpath, docopy=self.docopy
+                    )
 
     def to_asdf(self, path):
         tree = eval(str(self))

romancal/regtest/test_mos_pipeline.py

@@ -23,19 +23,7 @@ def passfail(bool_expr):
 @metrics_logger("DMS356", "DMS374")
 def test_level3_mos_pipeline(rtdata, ignore_asdf_paths):
     """Tests for level 3 processing requirements DMS356"""
-
-    cal_files = [
-        "WFI/image/r0000101001001001001_01101_0001_WFI01_cal.asdf",
-        "WFI/image/r0000101001001001001_01101_0002_WFI01_cal.asdf",
-        "WFI/image/r0000101001001001001_01101_0003_WFI01_cal.asdf",
-    ]
-
-    for cal_file in cal_files:
-        rtdata.get_data(cal_file)
-
-    input_asn = "L3_regtest_asn.json"
-    rtdata.get_data(f"WFI/image/{input_asn}")
-    rtdata.input = input_asn
+    rtdata.get_asn("WFI/image/L3_regtest_asn.json")
 
     # Test Pipeline
     output = "r0099101001001001001_F158_visit_i2d.asdf"
@@ -135,19 +123,7 @@ def test_level3_mos_pipeline(rtdata, ignore_asdf_paths):
 @metrics_logger("DMS373")
 def test_hlp_mosaic_pipeline(rtdata, ignore_asdf_paths):
     """Tests for level 3 mosaic requirements DMS373"""
-
-    cal_files = [
-        "WFI/image/r0000101001001001001_01101_0001_WFI01_cal.asdf",
-        "WFI/image/r0000101001001001001_01101_0002_WFI01_cal.asdf",
-        "WFI/image/r0000101001001001001_01101_0003_WFI01_cal.asdf",
-    ]
-
-    for cal_file in cal_files:
-        rtdata.get_data(cal_file)
-
-    input_asn = "L3_mosaic_asn.json"
-    rtdata.get_data(f"WFI/image/{input_asn}")
-    rtdata.input = input_asn
+    rtdata.get_asn("WFI/image/L3_mosaic_asn.json")
 
     # Test Pipeline
     output = "r0099101001001001001_r274dp63x31y81_prompt_F158_i2d.asdf"

romancal/regtest/test_resample.py

@@ -12,18 +12,11 @@
 @metrics_logger("DMS342", "DMS343", "DMS344", "DMS345")
 @pytest.mark.bigdata
 def test_resample_single_file(rtdata, ignore_asdf_paths):
-    input_data = [
-        "r0000101001001001001_01101_0001_WFI01_cal.asdf",
-        "r0000101001001001001_01101_0002_WFI01_cal.asdf",
-    ]
     output_data = "mosaic_resamplestep.asdf"
 
-    [rtdata.get_data(f"WFI/image/{data}") for data in input_data]
-    asnfn = "mosaic_asn.json"
-    rtdata.get_data(f"WFI/image/{asnfn}")
+    rtdata.get_asn("WFI/image/mosaic_asn.json")
     rtdata.get_truth(f"truth/WFI/image/{output_data}")
 
-    rtdata.input = asnfn
     rtdata.output = output_data
 
     # instantiate ResampleStep (for running and log access)