Code cleanup for ER3

* Updated unittests, finished ER3 cleanup

Co-authored-by: Jim Hofman <[email protected]>

src/opera/pge/base_pge.py

@@ -24,8 +24,8 @@
 
 import os
 from datetime import datetime
+from functools import lru_cache
 from os.path import abspath, basename, exists, join, splitext
-from functools import cache
 
 from yamale import YamaleError
 
@@ -36,10 +36,10 @@
 from opera.util.logger import PgeLogger
 from opera.util.logger import default_log_file_name
 from opera.util.metfile import MetFile
-from opera.util.run_utils import get_checksum
-from opera.util.run_utils import get_extension
 from opera.util.run_utils import create_qa_command_line
 from opera.util.run_utils import create_sas_command_line
+from opera.util.run_utils import get_checksum
+from opera.util.run_utils import get_extension
 from opera.util.run_utils import time_and_execute
 from opera.util.time import get_catalog_metadata_datetime_str
 from opera.util.time import get_time_for_filename
@@ -274,13 +274,12 @@ def _checksum_output_products(self):
 
         return checksums
 
-    @cache
+    @lru_cache
     def _create_catalog_metadata(self):
         """
         Returns the catalog metadata as a MetFile instance. Once generated, the
         catalog metadata is cached for the life of the PGE instance.
         """
-
         catalog_metadata = {
             'PGE_Name': self.runconfig.pge_name,
             'PGE_Version': opera.__version__,
@@ -377,7 +376,7 @@ def _geotiff_filename(self, inter_filename):
             The file name to assign to GeoTIFF product(s) created by this PGE.
 
         """
-        base_filename = splitext(inter_filename)[0]
+        base_filename = splitext(basename(inter_filename))[0]
         return self._core_filename() + f"_{base_filename}.tif"
 
     def _catalog_metadata_filename(self):
@@ -490,7 +489,7 @@ def _assign_filename(self, input_filepath, output_dir):
         try:
             os.rename(input_filepath, final_filepath)
         except OSError as err:
-            msg = f"Failed to rename output file {input_filepath}, reason: {str(err)}"
+            msg = f"Failed to rename output file {basename(input_filepath)}, reason: {str(err)}"
             self.logger.critical(self.name, ErrorCode.FILE_MOVE_FAILED, msg)
 
     def _stage_output_files(self):

src/opera/pge/dswx_pge.py

@@ -27,8 +27,8 @@
 from os.path import abspath, exists, isdir, join
 
 from opera.util.error_codes import ErrorCode
-from opera.util.img_utils import get_geotiff_metadata
 from opera.util.img_utils import get_geotiff_hls_dataset
+from opera.util.img_utils import get_geotiff_metadata
 from opera.util.img_utils import get_geotiff_spacecraft_name
 from opera.util.img_utils import get_hls_filename_fields
 from opera.util.render_jinja2 import render_jinja2

src/opera/test/data/test_base_pge_config.yaml

@@ -26,7 +26,8 @@ RunConfig:
         ProductIdentifier: EXAMPLE
         ProgramPath: echo
         ProgramOptions:
-          - hello world
+          - 'hello world > base_pge_test/outputs/dswx_hls.tif;'
+          - '/bin/echo hello world'
         ErrorCodeBase: 100000
         SchemaPath: sample_sas_schema.yaml
         IsoTemplatePath: sample_iso_template.xml.jinja2
@@ -39,6 +40,7 @@ RunConfig:
 
       DebugLevelGroup:
         DebugSwitch: False
+        ExecuteViaShell: True
 
     SAS:
       input_subset:

src/opera/test/data/test_sas_qa_bad_extension_config.yaml

@@ -0,0 +1,104 @@
+RunConfig:
+  Name: OPERA-BASE-PGE-TEST-CONFIG
+
+  Groups:
+
+    PGE:
+      PGENameGroup:
+        PGEName: BASE_PGE_TEST
+
+      InputFilesGroup:
+        InputFilePaths:
+          - input/input_file01.h5
+          - input/input_file02.h5
+
+      DynamicAncillaryFilesGroup:
+        AncillaryFileMap:
+          DEMFile: input/input_dem.vrt
+
+      ProductPathGroup:
+        ProductCounter: 005
+        OutputProductPath: base_pge_test/outputs/
+        ScratchPath: base_pge_test/scratch/
+        SASOutputFile: output_file.abc
+
+      PrimaryExecutable:
+        ProductIdentifier: EXAMPLE
+        ProgramPath: echo
+        ProgramOptions:
+          - 'hello world > base_pge_test/outputs/output_file.abc'
+        ErrorCodeBase: 100000
+        SchemaPath: sample_sas_schema.yaml
+        IsoTemplatePath: sample_iso_template.xml.jinja2
+
+      QAExecutable:
+        Enabled: True
+        ProgramPath: echo
+        ProgramOptions:
+         -  hello from qa executable
+
+      DebugLevelGroup:
+        DebugSwitch: False
+        ExecuteViaShell: True
+
+    SAS:
+      input_subset:
+        list_of_frequencies:
+          A:
+          B:
+        fullcovariance: False
+
+      dem_download:
+        source:
+        top_left:
+          x:
+          y:
+        bottom_right:
+          x:
+          y:
+
+      pre_process:
+        azimuth_looks: 1
+        range_looks: 1
+
+      rtc:
+        output_type: gamma0
+        algorithm_type: area_projection
+        input_terrain_radiometry: sigma0
+        rtc_min_value_db: -30
+
+      geocode:
+        algorithm_type: area_projection
+        memory_mode: auto
+        geogrid_upsampling: 1
+        save_nlooks: True
+        save_rtc: True
+        abs_rad_cal: 1
+        outputEPSG:
+        output_posting:
+          A:
+            x_posting:
+            y_posting:
+          B:
+            x_posting:
+            y_posting:
+        x_snap: 100
+        y_snap: 100
+        top_left:
+          y_abs:
+          x_abs:
+        bottom_right:
+          y_abs:
+          x_abs:
+
+      noise_correction:
+        apply_correction: False
+        correction_type: None
+
+      worker:
+        internet_access: False
+        gpu_enabled: False
+
+      QA:
+        validate: False
+        quality: False

src/opera/test/pge/test_base_pge.py

@@ -28,11 +28,16 @@
 from io import StringIO
 from os.path import abspath, join
 from pathlib import Path
+from unittest.mock import patch
 
 from pkg_resources import resource_filename
 
+import yaml
+
+import opera
 from opera.pge import PgeExecutor, RunConfig
 from opera.util import PgeLogger
+from opera.util import run_utils
 
 
 class BasePgeTestCase(unittest.TestCase):
@@ -47,7 +52,6 @@ def setUpClass(cls) -> None:
         cls.starting_dir = abspath(os.curdir)
         cls.test_dir = resource_filename(__name__, "")
         cls.data_dir = join(cls.test_dir, os.pardir, "data")
-
         os.chdir(cls.test_dir)
 
     @classmethod
@@ -57,7 +61,6 @@ def tearDownClass(cls) -> None:
 
     def setUp(self) -> None:
         """Use the temporary directory as the working directory"""
-
         self.working_dir = tempfile.TemporaryDirectory(
             prefix="test_base_pge_", suffix='temp', dir=os.curdir
         )
@@ -275,6 +278,164 @@ def test_geotiff_filename(self):
         file_name_regex = rf'{pge.PROJECT}_{pge.LEVEL}_BasePge_\d{{8}}T\d{{6}}_\d{{3}}_{name}{{1,2}}?'
         self.assertEqual(re.match(file_name_regex, file_name).group(), file_name)
 
+    def _makedirs_mock(self, mode=511, exist_ok=False):
+        """Mock function for os.makedirs that always raises OSError"""
+        raise OSError("Mock OSError from os.makedirs")
+
+    @patch.object(os, 'makedirs', _makedirs_mock)
+    def test_setup_directories_exception(self):
+        """Test IOError exception in _setup_directories()"""
+        runconfig_path = join(self.data_dir, 'test_sas_qa_config.yaml')
+        pge = PgeExecutor(pge_name='BasePgeFilenameTest', runconfig_path=runconfig_path)
+        pge._initialize_logger()
+        pge._load_runconfig()
+
+        with self.assertRaises(RuntimeError):
+            pge._setup_directories()
+
+        expected_log_file = pge.logger.get_file_name()
+        self.assertTrue(os.path.exists(expected_log_file))
+
+        # Open the log file, and check that we got expected error message
+        with open(expected_log_file, 'r', encoding='utf-8') as infile:
+            log_contents = infile.read()
+
+        self.assertIn(
+            'Could not create one or more working directories. reason: \n'
+            'Mock OSError from os.makedirs',
+            log_contents
+        )
+
+    def create_sas_command_line_mock(self, sas_program_path='', sas_runconfig_filepath='',
+                                     sas_program_options=None):
+        """Mock run_util.create_qa_command_line()"""
+        raise OSError("Mock OSError from run_utils.create_sas_command_line")
+
+    @patch.object(opera.pge.base_pge, 'create_sas_command_line', create_sas_command_line_mock)
+    def test_run_sas_executable_exception(self):
+        """Test IOError exception in _run_sas_executable()"""
+        runconfig_path = join(self.data_dir, 'test_sas_qa_config.yaml')
+        pge = PgeExecutor(pge_name='PgeQATest', runconfig_path=runconfig_path)
+
+        with self.assertRaises(RuntimeError):
+            pge.run()
+
+        expected_log_file = pge.logger.get_file_name()
+        self.assertTrue(os.path.exists(expected_log_file))
+
+        # Open the log file, and check that we got expected error message
+        with open(expected_log_file, 'r', encoding='utf-8') as infile:
+            log_contents = infile.read()
+
+        self.assertIn(
+            'Failed to create SAS command line, reason: '
+            'Mock OSError from run_utils.create_sas_command_line',
+            log_contents
+        )
+
+    def create_qa_command_line_mock(self, qa_program_path='./', qa_program_options=None):
+        """Mock function for run_utils.create_qa_command_line that always raises OSError"""
+        raise OSError("Mock OSError from run_utils.create_qa_command_line")
+
+    @patch.object(opera.pge.base_pge, 'create_qa_command_line', create_qa_command_line_mock)
+    def test_run_sas_qa_executable_exception(self):
+        """Test OSError in _run_sas_qa_executable()"""
+        runconfig_path = join(self.data_dir, 'test_sas_qa_config.yaml')
+        pge = PgeExecutor(pge_name='PgeQATest', runconfig_path=runconfig_path)
+        pge.run_preprocessor()
+        pge.run_sas_executable()
+
+        with self.assertRaises(RuntimeError):
+            pge._run_sas_qa_executable()
+
+        expected_log_file = pge.logger.get_file_name()
+        self.assertTrue(os.path.exists(expected_log_file))
+
+        # Open the log file, and check that we got expected error message
+        with open(expected_log_file, 'r', encoding='utf-8') as infile:
+            log_contents = infile.read()
+
+        self.assertIn(
+            'Failed to create QA command line, reason: '
+            'Mock OSError from run_utils.create_qa_command_line',
+            log_contents
+        )
+
+    def test_assign_filename_with_unknown_extension(self):
+        """
+        Test _assign_filename against an output file with an unknown extension.
+        File renaming should be skipped.
+        """
+        runconfig_path = join(self.data_dir, 'test_sas_qa_bad_extension_config.yaml')
+        pge = PgeExecutor(pge_name='PgeQATest', runconfig_path=runconfig_path)
+        pge.run_preprocessor()
+        pge.run_sas_executable()
+        pge._stage_output_files()
+
+        expected_log_file = pge.logger.get_file_name()
+        self.assertTrue(os.path.exists(expected_log_file))
+
+        # Open the log file, and check that we got expected error message
+        with open(expected_log_file, 'r', encoding='utf-8') as infile:
+            log_contents = infile.read()
+
+        self.assertIn(
+            'No rename function configured for file "output_file.abc", skipping assignment',
+            log_contents
+        )
+
+    def _os_rename_mock(self, input_filepath='./', final_filepath='./'):
+        """Mock function for os.rename that always raises OSError"""
+        raise OSError("Mock OSError from os.rename")
+
+    @patch.object(os, 'rename', _os_rename_mock)
+    def test_assign_filename_exception(self):
+        """Test IOError exception in _assign_filename()"""
+        runconfig_path = join(self.data_dir, 'test_base_pge_config.yaml')
+        pge = PgeExecutor(pge_name='PgeQATest', runconfig_path=runconfig_path)
+
+        with self.assertRaises(RuntimeError):
+            pge.run()
+
+        expected_log_file = pge.logger.get_file_name()
+        self.assertTrue(os.path.exists(expected_log_file))
+
+        # Open the log file, and check that we got expected error message
+        with open(expected_log_file, 'r', encoding='utf-8') as infile:
+            log_contents = infile.read()
+
+        self.assertIn(
+            "Failed to rename output file dswx_hls.tif, reason: "
+            "Mock OSError from os.rename",
+            log_contents
+        )
+
+    def safe_dump_mock(self, sas_config='./', outfile='./', sort_keys=False):
+        """Mock function for yaml.safe_dump() that always raises OSError"""
+        raise OSError("Mock OSError from yaml.safe_dump")
+
+    @patch.object(yaml, 'safe_dump', safe_dump_mock)
+    def test_isolate_sas_runconfig_exception(self):
+        """Test IOError exception in _isolate_sas_runconfig()"""
+        runconfig_path = join(self.data_dir, 'test_sas_qa_config.yaml')
+        pge = PgeExecutor(pge_name='PgeQATest', runconfig_path=runconfig_path)
+
+        with self.assertRaises(RuntimeError):
+            pge.run()
+
+        expected_log_file = pge.logger.get_file_name()
+        self.assertTrue(os.path.exists(expected_log_file))
+
+        # Open the log file, and check that we got expected error message
+        with open(expected_log_file, 'r', encoding='utf-8') as infile:
+            log_contents = infile.read()
+
+        self.assertIn(
+            'Failed to create SAS config file base_pge_test/scratch/test_sas_qa_config_sas.yaml, '
+            'reason: Mock OSError from yaml.safe_dump',
+            log_contents
+        )
+
 
 if __name__ == "__main__":
     unittest.main()