Remove the fallback operator for async task (astronomer#1538)

Since we decided to close issue astronomer#1273, the fallback for the async operator class is no longer needed. Therefore, this PR removes the fallback.

cosmos/operators/_asynchronous/base.py

@@ -18,7 +18,7 @@ def _create_async_operator_class(profile_type: str, dbt_class: str) -> Any:
 
     The function constructs a class path string for an asynchronous operator, based on the provided `profile_type` and
     `dbt_class`. It attempts to import the corresponding class dynamically and return it. If the class cannot be found,
-    it falls back to returning the `DbtRunLocalOperator` class.
+    it raises an error.
 
     :param profile_type: The dbt profile type
     :param dbt_class: The dbt class name. Example DbtRun, DbtTest.
@@ -30,9 +30,8 @@ def _create_async_operator_class(profile_type: str, dbt_class: str) -> Any:
         module = importlib.import_module(module_path)
         operator_class = getattr(module, class_name)
         return operator_class
-    except (ModuleNotFoundError, AttributeError):
-        log.info("Error in loading class: %s. falling back to DbtRunLocalOperator", class_path)
-        return DbtRunLocalOperator
+    except (ModuleNotFoundError, AttributeError) as e:
+        raise ImportError(f"Error in loading class: {class_path}. Unable to find the specified operator class.") from e
 
 
 class DbtRunAirflowAsyncFactoryOperator(DbtRunLocalOperator):  # type: ignore[misc]

tests/operators/_asynchronous/test_base.py

@@ -8,25 +8,31 @@
 from cosmos.operators._asynchronous import TeardownAsyncOperator
 from cosmos.operators._asynchronous.base import DbtRunAirflowAsyncFactoryOperator, _create_async_operator_class
 from cosmos.operators._asynchronous.bigquery import DbtRunAirflowAsyncBigqueryOperator
+from cosmos.operators._asynchronous.databricks import DbtRunAirflowAsyncDatabricksOperator
 from cosmos.operators.local import DbtRunLocalOperator
 
 
 @pytest.mark.parametrize(
     "profile_type, dbt_class, expected_operator_class",
     [
         ("bigquery", "DbtRun", DbtRunAirflowAsyncBigqueryOperator),
-        ("snowflake", "DbtRun", DbtRunLocalOperator),
-        ("bigquery", "DbtTest", DbtRunLocalOperator),
+        ("databricks", "DbtRun", DbtRunAirflowAsyncDatabricksOperator),
     ],
 )
-def test_create_async_operator_class_success(profile_type, dbt_class, expected_operator_class):
+def test_create_async_operator_class(profile_type, dbt_class, expected_operator_class):
     """Test the successful loading of the async operator class."""
 
     operator_class = _create_async_operator_class(profile_type, dbt_class)
 
     assert operator_class == expected_operator_class
 
 
+def test_create_async_operator_class_unsupported():
+
+    with pytest.raises(ImportError, match="Error in loading class"):
+        _create_async_operator_class("test_profile", "DbtRun")
+
+
 @pytest.fixture
 def profile_config_mock():
     """Fixture to create a mock ProfileConfig."""
@@ -46,13 +52,6 @@ def test_create_async_operator_class_valid():
         assert result == mock_class
 
 
-def test_create_async_operator_class_fallback():
-    """Test _create_async_operator_class falls back to DbtRunLocalOperator when import fails."""
-    with patch("cosmos.operators._asynchronous.base.importlib.import_module", side_effect=ModuleNotFoundError):
-        result = _create_async_operator_class("bigquery", "DbtRun")
-        assert result == DbtRunLocalOperator
-
-
 class MockAsyncOperator(DbtRunLocalOperator):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)