Add default conn name to asana provider operators

providers/asana/src/airflow/providers/asana/operators/asana_tasks.py

@@ -53,9 +53,9 @@ class AsanaCreateTaskOperator(BaseOperator):
     def __init__(
         self,
         *,
-        conn_id: str,
         name: str,
         task_parameters: dict | None = None,
+        conn_id: str = "asana_default",
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -92,9 +92,9 @@ class AsanaUpdateTaskOperator(BaseOperator):
     def __init__(
         self,
         *,
-        conn_id: str,
         asana_task_gid: str,
         task_parameters: dict,
+        conn_id: str = "asana_default",
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -124,8 +124,8 @@ class AsanaDeleteTaskOperator(BaseOperator):
     def __init__(
         self,
         *,
-        conn_id: str,
         asana_task_gid: str,
+        conn_id: str = "asana_default",
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -160,8 +160,8 @@ class AsanaFindTaskOperator(BaseOperator):
     def __init__(
         self,
         *,
-        conn_id: str,
         search_parameters: dict | None = None,
+        conn_id: str = "asana_default",
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)

providers/asana/tests/unit/asana/operators/test_asana_tasks.py

@@ -37,6 +37,27 @@ class TestAsanaTaskOperators:
     @pytest.fixture(autouse=True)
     def setup_connections(self, create_connection_without_db):
         create_connection_without_db(Connection(conn_id="asana_test", conn_type="asana", password="test"))
+        create_connection_without_db(Connection(conn_id="asana_default", conn_type="asana", password="test"))
+
+    @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
+    def test_asana_create_task_operator_with_default_conn(self, mock_asana_hook):
+        """
+        Tests that the AsanaCreateTaskOperator uses the default connection.
+        """
+
+        mock_hook_instance = MagicMock()
+        mock_asana_hook.return_value = mock_hook_instance
+        mock_hook_instance.create_task.return_value = {"gid": "1"}
+
+        create_task = AsanaCreateTaskOperator(
+            task_id="create_task",
+            name="test",
+            task_parameters={"workspace": "1"},
+        )
+        result = create_task.execute({})
+        assert create_task.conn_id == "asana_default"
+        mock_hook_instance.create_task.assert_called_once_with("test", {"workspace": "1"})
+        assert result == "1"
 
     @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
     def test_asana_create_task_operator(self, mock_asana_hook):
@@ -58,6 +79,26 @@ def test_asana_create_task_operator(self, mock_asana_hook):
         mock_hook_instance.create_task.assert_called_once_with("test", {"workspace": "1"})
         assert result == "1"
 
+    @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
+    def test_asana_find_task_operator_with_default_conn(self, mock_asana_hook):
+        """
+        Tests that the AsanaFindTaskOperator uses the default connection.
+        """
+
+        mock_hook_instance = MagicMock()
+        mock_asana_hook.return_value = mock_hook_instance
+        mock_hook_instance.find_task.return_value = {"gid": "1"}
+
+        find_task = AsanaFindTaskOperator(
+            task_id="find_task",
+            search_parameters={"project": "test"},
+        )
+        assert find_task.conn_id == "asana_default"
+
+        result = find_task.execute({})
+        mock_hook_instance.find_task.assert_called_once_with({"project": "test"})
+        assert result == {"gid": "1"}
+
     @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
     def test_asana_find_task_operator(self, mock_asana_hook):
         """
@@ -77,6 +118,23 @@ def test_asana_find_task_operator(self, mock_asana_hook):
         mock_hook_instance.find_task.assert_called_once_with({"project": "test"})
         assert result == {"gid": "1"}
 
+    @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
+    def test_asana_update_task_operator_default_conn(self, mock_asana_hook):
+        """
+        Tests that the AsanaUpdateTaskOperator uses the default connection.
+        """
+
+        mock_hook_instance = MagicMock()
+        mock_asana_hook.return_value = mock_hook_instance
+        update_task = AsanaUpdateTaskOperator(
+            task_id="update_task",
+            asana_task_gid="test",
+            task_parameters={"completed": True},
+        )
+        assert update_task.conn_id == "asana_default"
+        update_task.execute({})
+        mock_hook_instance.update_task.assert_called_once_with("test", {"completed": True})
+
     @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
     def test_asana_update_task_operator(self, mock_asana_hook):
         """
@@ -93,6 +151,23 @@ def test_asana_update_task_operator(self, mock_asana_hook):
         update_task.execute({})
         mock_hook_instance.update_task.assert_called_once_with("test", {"completed": True})
 
+    @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
+    def test_asana_delete_task_operator_with_default_conn(self, mock_asana_hook):
+        """
+        Tests that the AsanaDeleteTaskOperator uses the default connection.
+        """
+
+        mock_hook_instance = MagicMock()
+        mock_asana_hook.return_value = mock_hook_instance
+
+        delete_task = AsanaDeleteTaskOperator(
+            task_id="delete_task",
+            asana_task_gid="test",
+        )
+        delete_task.execute({})
+        assert delete_task.conn_id == "asana_default"
+        mock_hook_instance.delete_task.assert_called_once_with("test")
+
     @patch("airflow.providers.asana.operators.asana_tasks.AsanaHook")
     def test_asana_delete_task_operator(self, mock_asana_hook):
         """