Add changes

Signed-off-by: Divya Madala <divyaasm@amazon.com>

src/validation_workflow/rpm/validation_rpm.py

@@ -80,10 +80,12 @@ def validate_metadata(self, product_type: str) -> None:
         # The context the meta data should be based on type OpenSearch or OpenSearchDashBoards
         if product_type == "opensearch":
             ref_map['Summary'] = "An open source distributed and RESTful search engine"
-            ref_map['Description'] = "OpenSearch makes it easy to ingest, search, visualize, and analyze your data\nFor more information, see: https://opensearch.org/"
+            ref_map[
+                'Description'] = "OpenSearch makes it easy to ingest, search, visualize, and analyze your data\nFor more information, see: https://opensearch.org/"
         else:
             ref_map['Summary'] = "Open source visualization dashboards for OpenSearch"
-            ref_map['Description'] = "OpenSearch Dashboards is the visualization tool for data in OpenSearch\nFor more information, see: https://opensearch.org/"
+            ref_map[
+                'Description'] = "OpenSearch Dashboards is the visualization tool for data in OpenSearch\nFor more information, see: https://opensearch.org/"
 
         meta_map = {}
         for line in stdout.split('\n'):
@@ -102,25 +104,27 @@ def validate_metadata(self, product_type: str) -> None:
                 elif value == 'arm64':
                     assert meta_map.get(key) == 'aarch64'
             else:
-                if meta_map.get(key) == value:
-                    logging.info(value)
+                assert meta_map.get(key) == value
             logging.info(f"Meta data for {key} is validated")
 
         logging.info(f"Validation for {product_type} meta data of RPM distribution completed.")
 
     def validate_signature(self) -> None:
         (_, stdout, _) = execute(f'rpm -K -v {os.path.join(self.tmp_dir.path, self.filename)}', ".")
         logging.info(stdout)
-        key_list = ["Header V4 RSA/SHA512 Signature, key ID 9310d3fc", "Header SHA256 digest", "Header SHA1 digest", "Payload SHA256 digest", "V4 RSA/SHA512 Signature, key ID 9310d3fc", "MD5 digest"]
+        key_list = ["Header V4 RSA/SHA512 Signature, key ID 9310d3fc", "Header SHA256 digest", "Header SHA1 digest",
+                    "Payload SHA256 digest", "V4 RSA/SHA512 Signature, key ID 9310d3fc", "MD5 digest"]
         present_key = []
         for line in stdout.rstrip('\n').split('\n'):
             key = line.split(':')[0].strip()
             if key != os.path.join(self.tmp_dir.path, self.filename):
-                if "OK" == line.split(':')[1].strip():
-                    logging.info(f"{key} is validated as: {line}")
-                    present_key.append(key)
+                assert "OK" == line.split(':')[1].strip()
+                logging.info(f"{key} is validated as: {line}")
+                present_key.append(key)
         logging.info("Validation of all key digests starts: ")
         for digest in key_list:
             if digest in present_key:
                 logging.info(f'Key digest "{digest}" is validated to be present.')
+            else:
+                raise ValueError(f'Key digest "{digest}" is not found')
         logging.info("Validation for signature of RPM distribution completed.")

tests/tests_validation_workflow/test_validation_rpm.py

@@ -215,7 +215,8 @@ def test_cleanup(self, mock_temporary_directory: Mock, mock_validation_args: Moc
     @patch('validation_workflow.rpm.validation_rpm.ValidationArgs')
     @patch('system.temporary_directory.TemporaryDirectory')
     def test_validate_metadata(self, mock_temporary_directory: Mock, mock_validation_args: Mock, mock_logging_info: Mock, mock_execute: Mock) -> None:
-        mock_execute.return_value = (None, 'Name: opensearch\nVersion: 1.2.3\nArchitecture: x86_64\nDescription: This is a test application\n "OpenSearch makes it easy to ingest, search, visualize, and analyze your data\nFor more information, see: https://opensearch.org/', None)
+        mock_execute.return_value = (None, 'Name: opensearch\nVersion: 1.2.3\nArchitecture: x86_64\nDescription: This is a test application\n'
+                                           ' "OpenSearch makes it easy to ingest, search, visualize, and analyze your data\nFor more information, see: https://opensearch.org/', None)
 
         validate_rpm = ValidateRpm(mock_validation_args.return_value, mock_temporary_directory.return_value)
         mock_temporary_directory.return_value.path = "/tmp/trytytyuit/"
@@ -232,7 +233,7 @@ def test_validate_metadata(self, mock_temporary_directory: Mock, mock_validation
         mock_logging_info.assert_any_call('Validation for opensearch meta data of RPM distribution completed.')
 
         mock_execute.assert_called_once_with(
-            f'rpm -qip /tmp/trytytyuit/example.rpm', '.'
+            'rpm -qip /tmp/trytytyuit/example.rpm', '.'
         )
 
     @patch('validation_workflow.rpm.validation_rpm.execute')
@@ -253,6 +254,5 @@ def test_validate_signature(self, mock_temporary_directory: Mock, mock_validatio
         mock_logging_info.assert_any_call('Validation of all key digests starts: ')
         mock_logging_info.assert_any_call('Validation for signature of RPM distribution completed.')
         mock_execute.assert_called_once_with(
-            f'rpm -K -v /tmp/trytytyuit/example.rpm', '.'
+            'rpm -K -v /tmp/trytytyuit/example.rpm', '.'
         )
-