Improve Airflow parser functionality

conftest.py

@@ -22,6 +22,8 @@
 
 pytest_plugins = ["jupyter_server.pytest_plugin"]
 
+TEST_CATALOG_NAME = 'new_test_catalog'
+
 KFP_COMPONENT_CACHE_INSTANCE = {
     "display_name": "KFP Example Components",
     "metadata": {
@@ -40,6 +42,7 @@
     "schema_name": "elyra-airflow-examples-catalog"
 }
 
+
 @pytest.fixture
 def component_cache_instance(request):
     """Creates an instance of a component cache and removes after test."""
@@ -68,3 +71,22 @@ def component_cache_instance(request):
     # Test was not parametrized, so component instance is not needed
     except AttributeError:
         yield None
+
+
+@pytest.fixture
+def metadata_manager_with_teardown():
+    """
+    This fixture provides a MetadataManager instance for certain tests that modify the component
+    catalog. This ensures the catalog instance is removed even when the test fails
+    """
+    metadata_manager = MetadataManager(schemaspace=ComponentCatalogs.COMPONENT_CATALOGS_SCHEMASPACE_ID)
+
+    # Run test with provided metadata manager
+    yield metadata_manager
+
+    # Remove test catalog
+    try:
+        if metadata_manager.get(TEST_CATALOG_NAME):
+            metadata_manager.remove(TEST_CATALOG_NAME)
+    except Exception:
+        pass

elyra/pipeline/catalog_connector.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 #
 from abc import abstractmethod
+from copy import deepcopy
 import hashlib
 from http import HTTPStatus
 import os
@@ -240,8 +241,12 @@ def read_component_definitions(self, catalog_instance: Metadata) -> Dict[str, Di
             # the catalog entry hash for each entry in the catalog
             keys_to_hash = self.get_hash_keys()
 
+            # Add display_name attribute to the metadata dictionary
+            catalog_metadata = deepcopy(catalog_instance.metadata)
+            catalog_metadata['display_name'] = catalog_instance.display_name
+
             # Add catalog entry data dictionaries to the thread queue
-            for entry_data in self.get_catalog_entries(catalog_instance.metadata):
+            for entry_data in self.get_catalog_entries(catalog_metadata):
                 catalog_entry_q.put_nowait(entry_data)
 
         except NotImplementedError as e:
@@ -268,7 +273,7 @@ def read_with_thread():
                     self.log.debug(f"Attempting read of definition for catalog entry with identifying information: "
                                    f"{str(catalog_entry_data)}...")
                     definition = self.read_catalog_entry(catalog_entry_data=catalog_entry_data,
-                                                         catalog_metadata=catalog_instance.metadata)
+                                                         catalog_metadata=catalog_metadata)
 
                     # Ignore this entry if no definition content is returned
                     if not definition:

elyra/pipeline/component_catalog.py

@@ -108,7 +108,7 @@ def check_outstanding_threads(self) -> bool:
                 # Report successful join for threads that have previously logged a
                 # cache update duration warning
                 if thread.last_warn_time != thread.task_start_time:
-                    self.log.info(f"Cache update for catalog '{thread.name}' has"
+                    self.log.info(f"Cache update for catalog '{thread.name}' has "
                                   f"completed after {cumulative_run_time} seconds")
 
         return outstanding_threads

elyra/tests/pipeline/kfp/test_component_parser_kfp.py

@@ -17,12 +17,11 @@
 from types import SimpleNamespace
 
 from conftest import KFP_COMPONENT_CACHE_INSTANCE
+from conftest import TEST_CATALOG_NAME
 import jupyter_core.paths
 import pytest
 
-from elyra.metadata.manager import MetadataManager
 from elyra.metadata.metadata import Metadata
-from elyra.metadata.schemaspaces import ComponentCatalogs
 from elyra.pipeline.catalog_connector import FilesystemComponentCatalogConnector
 from elyra.pipeline.catalog_connector import UrlComponentCatalogConnector
 from elyra.pipeline.component_catalog import ComponentCache
@@ -47,7 +46,7 @@ def test_component_catalog_can_load_components_from_registries(component_cache_i
     assert len(components) > 0
 
 
-def test_modify_component_catalogs():
+def test_modify_component_catalogs(metadata_manager_with_teardown):
     # Initialize a ComponentCache instance and wait for all worker threads to compete
     component_catalog = ComponentCache.instance()
     component_catalog.wait_for_all_cache_updates()
@@ -58,8 +57,6 @@ def test_modify_component_catalogs():
     # Components must be sorted by id for the equality comparison with later component lists
     initial_components = sorted(initial_components, key=lambda component: component.id)
 
-    metadata_manager = MetadataManager(schemaspace=ComponentCatalogs.COMPONENT_CATALOGS_SCHEMASPACE_ID)
-
     # Create new registry instance with a single URL-based component
     paths = [_get_resource_path('kfp_test_operator.yaml')]
 
@@ -70,17 +67,11 @@ def test_modify_component_catalogs():
         "paths": paths
     }
     registry_instance = Metadata(schema_name="local-file-catalog",
-                                 name="new_test_registry",
+                                 name=TEST_CATALOG_NAME,
                                  display_name="New Test Registry",
                                  metadata=instance_metadata)
 
-    try:
-        if metadata_manager.get("new_test_registry"):
-            metadata_manager.remove("new_test_registry")
-    except Exception:
-        pass
-
-    metadata_manager.create("new_test_registry", registry_instance)
+    metadata_manager_with_teardown.create(TEST_CATALOG_NAME, registry_instance)
 
     # Wait for update to complete
     component_catalog.wait_for_all_cache_updates()
@@ -96,7 +87,7 @@ def test_modify_component_catalogs():
 
     # Modify the test registry to add an additional path to
     paths.append(_get_resource_path('kfp_test_operator_no_inputs.yaml'))
-    metadata_manager.update("new_test_registry", registry_instance)
+    metadata_manager_with_teardown.update(TEST_CATALOG_NAME, registry_instance)
 
     # Wait for update to complete
     component_catalog.wait_for_all_cache_updates()
@@ -111,7 +102,7 @@ def test_modify_component_catalogs():
     assert 'Test Operator No Inputs' in modified_component_names
 
     # Delete the test registry
-    metadata_manager.remove("new_test_registry")
+    metadata_manager_with_teardown.remove(TEST_CATALOG_NAME)
 
     # Wait for update to complete
     component_catalog.wait_for_all_cache_updates()
@@ -131,16 +122,14 @@ def test_modify_component_catalogs():
     assert initial_palette == post_delete_palette
 
 
-def test_directory_based_component_catalog():
+def test_directory_based_component_catalog(metadata_manager_with_teardown):
     # Initialize a ComponentCache instance and wait for all worker threads to compete
     component_catalog = ComponentCache.instance()
     component_catalog.wait_for_all_cache_updates()
 
     # Get initial set of components from the current active registries
     initial_components = component_catalog.get_all_components(RUNTIME_PROCESSOR)
 
-    metadata_manager = MetadataManager(schemaspace=ComponentCatalogs.COMPONENT_CATALOGS_SCHEMASPACE_ID)
-
     # Create new directory-based registry instance with components in ../../test/resources/components
     registry_path = _get_resource_path('')
     instance_metadata = {
@@ -150,17 +139,11 @@ def test_directory_based_component_catalog():
         "paths": [registry_path]
     }
     registry_instance = Metadata(schema_name="local-directory-catalog",
-                                 name="new_test_registry",
+                                 name=TEST_CATALOG_NAME,
                                  display_name="New Test Registry",
                                  metadata=instance_metadata)
 
-    try:
-        if metadata_manager.get("new_test_registry"):
-            metadata_manager.remove("new_test_registry")
-    except Exception:
-        pass
-
-    metadata_manager.create("new_test_registry", registry_instance)
+    metadata_manager_with_teardown.create(TEST_CATALOG_NAME, registry_instance)
 
     # Wait for update to complete
     component_catalog.wait_for_all_cache_updates()
@@ -175,9 +158,6 @@ def test_directory_based_component_catalog():
     assert 'Test Operator' in added_component_names
     assert 'Test Operator No Inputs' in added_component_names
 
-    # Remove the test instance
-    metadata_manager.remove("new_test_registry")
-
 
 def test_parse_kfp_component_file():
     # Define the appropriate reader for a filesystem-type component definition