diff --git a/src/opera/pge/base_pge.py b/src/opera/pge/base_pge.py index 1357e7a9..e04d8327 100644 --- a/src/opera/pge/base_pge.py +++ b/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): diff --git a/src/opera/pge/dswx_pge.py b/src/opera/pge/dswx_pge.py index 63b9c8f8..7634fab8 100644 --- a/src/opera/pge/dswx_pge.py +++ b/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 diff --git a/src/opera/test/data/test_base_pge_config.yaml b/src/opera/test/data/test_base_pge_config.yaml index f3e6013b..7735c935 100644 --- a/src/opera/test/data/test_base_pge_config.yaml +++ b/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: diff --git a/src/opera/test/data/test_sas_qa_bad_extension_config.yaml b/src/opera/test/data/test_sas_qa_bad_extension_config.yaml new file mode 100644 index 00000000..d307f64f --- /dev/null +++ b/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 diff --git a/src/opera/test/pge/test_base_pge.py b/src/opera/test/pge/test_base_pge.py index 49f80395..f9b62423 100644 --- a/src/opera/test/pge/test_base_pge.py +++ b/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() diff --git a/src/opera/test/pge/test_dswx_pge.py b/src/opera/test/pge/test_dswx_pge.py index cc646b78..d6fa41d6 100644 --- a/src/opera/test/pge/test_dswx_pge.py +++ b/src/opera/test/pge/test_dswx_pge.py @@ -36,7 +36,7 @@ import opera.util.img_utils from opera.pge import DSWxExecutor, RunConfig from opera.util import PgeLogger -from opera.util.img_utils import MockGdal, get_geotiff_metadata +from opera.util.img_utils import MockGdal class DSWxPgeTestCase(unittest.TestCase): @@ -83,15 +83,7 @@ def tearDown(self) -> None: self.input_file.close() self.working_dir.cleanup() - def get_geotiff_metadata_patch(self): - """ - Patch for img_utils.get_geotiff_metadata. Needed because test_dswx_pge_execution - will not produce a legitimate output DSWx-HLS file to obtain metadata - from when assigning output filenames. - """ - return MockGdal.MockGdalDataset().GetMetadata() - - @patch.object(opera.util.img_utils, "get_geotiff_metadata", get_geotiff_metadata_patch) + @patch.object(opera.util.img_utils, "gdal", MockGdal) def test_dswx_pge_execution(self): """ Test execution of the DSWxExecutor class and its associated mixins using @@ -307,7 +299,7 @@ def test_dswx_pge_output_validation(self): if os.path.exists(test_runconfig_path): os.unlink(test_runconfig_path) - @patch.object(opera.util.img_utils, "get_geotiff_metadata", get_geotiff_metadata_patch) + @patch.object(opera.util.img_utils, "gdal", MockGdal) def test_geotiff_filename(self): """Test _geotiff_filename() method""" runconfig_path = join(self.data_dir, 'test_dswx_hls_config.yaml') @@ -319,7 +311,7 @@ def test_geotiff_filename(self): image_files = glob.glob(join(pge.runconfig.output_product_path, "*.tif")) for i in range(len(image_files)): file_name = pge._geotiff_filename(image_files[i]) - md = self.get_geotiff_metadata_patch() + md = MockGdal.MockGdalDataset().GetMetadata() file_name_regex = rf"{pge.PROJECT}_{pge.LEVEL}_{md['PRODUCT_TYPE']}_{md['PRODUCT_SOURCE']}_" \ rf"{md['SPACECRAFT_NAME']}_{md['HLS_DATASET'].split('.')[2]}_\d{{8}}T\d{{6}}_" \ rf"{'.'.join(md['HLS_DATASET'].split('.')[-2:])}_\d{{3}}.tif?" diff --git a/src/opera/test/util/test_render_jinja2.py b/src/opera/test/util/test_render_jinja2.py index 0f5b12e8..4969d441 100644 --- a/src/opera/test/util/test_render_jinja2.py +++ b/src/opera/test/util/test_render_jinja2.py @@ -15,14 +15,14 @@ # """ -===================== +================= test_render_jinja2.py -===================== +================= Unit tests for the util/render_jinja2.py module. - """ import os +import shutil import tempfile import unittest from os.path import abspath, join @@ -35,6 +35,7 @@ class RenderJinja2TestCase(unittest.TestCase): """Base test class using unittest""" + starting_dir = None working_dir = None test_dir = None @@ -56,33 +57,46 @@ def setUpClass(cls) -> None: @classmethod def tearDownClass(cls) -> None: - """At completion re-establish starting directory""" + """ + At completion re-establish starting directory + ------- + """ os.chdir(cls.starting_dir) def setUp(self) -> None: - """Use the temporary directory as the working directory""" + """ + Use the temporary directory as the working directory + ------- + """ self.working_dir = tempfile.TemporaryDirectory( prefix="test_met_file_", suffix='_temp', dir=os.curdir ) os.chdir(self.working_dir.name) def tearDown(self) -> None: - """Return to starting directory""" + """ + Return to starting directory + ------- + """ os.chdir(self.test_dir) self.working_dir.cleanup() def get_data(self): + """ + Returns a "movie" dictionary as the test data + ------- + + """ data = { "movies": [ { "title": 'Terminator', - "description": 'A soldier is sent back in time to protect ' - 'an important woman from a killing android.' + "description": 'A soldier is sent back in time to protect an important ' + 'woman from a killing android.' }, { "title": 'The Sandlot', - "description": 'Boys have a magical summer of baseball and ' - 'discovery.' + "description": 'Boys have a magical summer of baseball and discovery.' }, { "title": 'The Lion King', @@ -93,69 +107,70 @@ def get_data(self): return data def remove_key(self, movie_dict, key): - """Remove all passed 'key' from dictionary""" + """ + Remove all passed 'key' from dictionary + Returns + ------- + movie_dict : dict + The modified dictionary + """ for entry in movie_dict['movies']: entry.pop(key) def testRenderJinja2(self): """ Use a simple template to test the basic functionality of the module. - Test the 'dunder functions defined in the LoggingUndefined class. - Test that the proper substitutions are made in the rendered html file. + Test the 'dunder functions defined in the LoggingUndefined class + Test that the proper substitutions are made in the rendered html file + ------- """ - self.logger.write('info', "opera_pge", 0, 'Dummy entry in log') template_file = join(self.data_dir, 'render_jinja_test_template.html') data = self.get_data() - # run with a logger - rendered_template = render_jinja2(template_file, data, self.logger) - + rendered_text = render_jinja2(template_file, data, self.logger) # Verify the titles were properly added to the html file - self.assertIn('Terminator', rendered_template) - self.assertIn('The Sandlot', rendered_template) - self.assertIn('The Lion King', rendered_template) - + self.assertIn('Terminator', rendered_text) + self.assertIn('The Sandlot', rendered_text) + self.assertIn('The Lion King', rendered_text) # Verify the descriptions were properly added into the html file - self.assertIn('A soldier is sent back in time to protect an important woman from a killing android.', rendered_template) - self.assertIn('Boys have a magical summer of baseball and discovery.', rendered_template) - self.assertIn('A young lion prince is born in Africa.', rendered_template) + self.assertIn('A soldier is sent back in time to protect an important woman from a killing android.', + rendered_text) + self.assertIn('Boys have a magical summer of baseball and discovery.', rendered_text) + self.assertIn('A young lion prince is born in Africa.', rendered_text) # run without a logger - rendered_template = render_jinja2(template_file, data) - + rendered_text = render_jinja2(template_file, data) # Verify the titles were properly added to the html file - self.assertIn('Terminator', rendered_template) - self.assertIn('The Sandlot', rendered_template) - self.assertIn('The Lion King', rendered_template) - + self.assertIn('Terminator', rendered_text) + self.assertIn('The Sandlot', rendered_text) + self.assertIn('The Lion King', rendered_text) # Verify the descriptions were properly added into the html file - self.assertIn('A soldier is sent back in time to protect an important woman from a killing android.', rendered_template) - self.assertIn('Boys have a magical summer of baseball and discovery.', rendered_template) - self.assertIn('A young lion prince is born in Africa.', rendered_template) + self.assertIn('A soldier is sent back in time to protect an important woman from a killing android.', + rendered_text) + self.assertIn('Boys have a magical summer of baseball and discovery.', rendered_text) + self.assertIn('A young lion prince is born in Africa.', rendered_text) + # Move a template with another name into the temp directory. + shutil.copy(template_file, join(os.getcwd(), 'render_jinja_test_template_2.html')) + # Verify that os.getcmd() is used to find the new template. + render_jinja2('render_jinja_test_template_2.html', data) # Remove the title fields and verify the '!Not Found! is returned new_data = self.get_data() self.remove_key(new_data, 'title') - rendered_template = render_jinja2(template_file, new_data, self.logger) - - self.assertIn('!Not found!', rendered_template) - + rendered_text = render_jinja2(template_file, new_data, self.logger) + self.assertIn('!Not found!', rendered_text) # Verify the log has been updated. stream = self.logger.get_stream_object() self.assertIn('Missing/undefined ISO metadata template variable:', stream.getvalue()) - # Run again without a logger, default behavior of jinja2.Undefined should - # be to provide empty string for missing placeholders + # Run again without a logger and expect a KeyError new_data = self.get_data() self.remove_key(new_data, 'title') - - rendered_template = render_jinja2(template_file, new_data) - self.assertNotIn('Terminator', rendered_template) - self.assertNotIn('The Sandlot', rendered_template) - self.assertNotIn('The Lion King', rendered_template) + render_jinja2(template_file, new_data) + self.assertRaises(KeyError) diff --git a/src/opera/test/util/test_run_utils.py b/src/opera/test/util/test_run_utils.py index 2fa46619..90f31ad3 100644 --- a/src/opera/test/util/test_run_utils.py +++ b/src/opera/test/util/test_run_utils.py @@ -27,6 +27,7 @@ import tempfile import unittest from os.path import abspath +from unittest.mock import patch from pkg_resources import resource_filename @@ -161,6 +162,30 @@ def test_create_qa_command_line(self): for option in options: self.assertIn(option, command_line) + def _access_mock(self, executable_path='./', mode=os.F_OK): + """Mock os.access with os.F_OK returning True and os.X_OK returning False""" + if mode == os.F_OK: + return True + if mode == os.X_OK: + return False + + @patch.object(os, 'access', _access_mock) + def test_create_sas_command_line_exception(self): + """Test exception handling for create_sas_command_line() function""" + cmd = '' + runconfig_path = '' + options = ['Hello from test_create_sas_command_line function.', '--'] + with self.assertRaises(OSError): + create_sas_command_line(cmd, runconfig_path, options) + + @patch.object(os, 'access', _access_mock) + def test_qa_command_line_exception(self): + """Test exception handling for create_qa_command_line() function""" + cmd = '' + options = ['Hello from test_create_qa_command_line function.'] + with self.assertRaises(OSError): + create_qa_command_line(cmd, options) + def test_time_and_execute(self): """Tests for run_utils.time_and_execute()""" program_path = 'echo' diff --git a/src/opera/util/img_utils.py b/src/opera/util/img_utils.py index 004c51d9..6f89a6bc 100644 --- a/src/opera/util/img_utils.py +++ b/src/opera/util/img_utils.py @@ -24,7 +24,7 @@ from collections import namedtuple from datetime import datetime -from functools import cache +from functools import lru_cache from os.path import exists @@ -85,7 +85,7 @@ def Open(filename): gdal = MockGdal # pragma: no cover -@cache +@lru_cache def get_geotiff_metadata(filename): """ Returns the set of metadata fields associated to the provided GeoTIFF diff --git a/src/opera/util/render_jinja2.py b/src/opera/util/render_jinja2.py index 72545e74..4d229128 100644 --- a/src/opera/util/render_jinja2.py +++ b/src/opera/util/render_jinja2.py @@ -22,13 +22,14 @@ Adapted by: Jim Hofman """ import os + import jinja2 -from opera.util.logger import PgeLogger + from opera.util.error_codes import ErrorCode +from opera.util.logger import PgeLogger def _make_undefined_handler_class(logger: PgeLogger): - """ Factory function, returns a child class of the jinja2.Undefined class for use when rendering templates. @@ -67,18 +68,18 @@ def _log_message(undef): class LoggingUndefined(jinja2.Undefined): """Override the default behavior which can raise an exception""" - def _fail_with_undefined_error(self, *args, **kwargs): + def _fail_with_undefined_error(self, *args, **kwargs): # pragma no cover _log_message(self) def __str__(self): _log_message(self) return "!Not found!" - def __iter__(self): + def __iter__(self): # pragma no cover _log_message(self) return super().__iter__() - def __bool__(self): + def __bool__(self): # pragma no cover _log_message(self) return super().__bool__() @@ -89,8 +90,7 @@ def __getattr__(self, name): return LoggingUndefined -def render_jinja2(template_filename: str, input_data: dict, - logger: PgeLogger = None): +def render_jinja2(template_filename: str, input_data: dict, logger: PgeLogger = None): """ Renders from a jinja2 template using the specified input data. Writes the rendered output to the specified output file. diff --git a/src/opera/util/usage_metrics.py b/src/opera/util/usage_metrics.py index 919e3f3c..75fd1d79 100644 --- a/src/opera/util/usage_metrics.py +++ b/src/opera/util/usage_metrics.py @@ -105,7 +105,7 @@ def get_self_peak_vmm_kb(): if platform != "linux" or not os.path.exists(status_file): raise EnvironmentError - with open(status_file, "r", encoding='utf-8') as infile: + with open(status_file, "r", encoding='utf-8') as infile: # pragma no cover for line in infile.readlines(): if line.startswith("VmPeak:"): vm_peak_str = line.replace("VmPeak:", "").replace("kB", "")