Pass parameters to parent class in SnowflakeSensorAsync (#1519)

* Pass `parameters` to parent class in SnowflakeSensorAsync

Currently the SnowflakeSensorAsync does not propagate `parameters` to its parent class, SqlSensor. Because of this, `parameters` are not passed for execution and can cause task failures when doing so.

astronomer/providers/snowflake/sensors/snowflake.py

@@ -46,7 +46,7 @@ def __init__(
         *,
         snowflake_conn_id: str,
         sql: str,
-        parameters: str | None = None,
+        parameters: dict[str, Any] | None = None,
         success: str | None = None,
         failure: str | None = None,
         fail_on_empty: bool = False,
@@ -63,6 +63,7 @@ def __init__(
         super().__init__(
             conn_id=snowflake_conn_id,
             sql=sql,
+            parameters=parameters,
             success=success,
             failure=failure,
             fail_on_empty=fail_on_empty,

tests/snowflake/sensors/test_snowflake.py

@@ -103,3 +103,25 @@ def test_soft_fail_enable(self, context):
         )
         with pytest.raises(AirflowSkipException):
             sensor.execute(context)
+
+    @mock.patch("airflow.providers.common.sql.sensors.sql.SqlSensor.__init__")
+    def test_call_with_parameters(self, mock_sql_sensor):
+        """Ensure SQL parameters are passed to the SqlSensor parent class."""
+        SnowflakeSensorAsync(
+            sql="SELECT * FROM %(src)s;",
+            task_id="snowflake_sensor",
+            snowflake_conn_id=CONN_ID,
+            parameters={"src": "my_table"},
+        )
+
+        mock_sql_sensor.assert_called_once_with(
+            conn_id=CONN_ID,
+            task_id="snowflake_sensor",
+            sql="SELECT * FROM %(src)s;",
+            parameters={"src": "my_table"},
+            success=None,
+            failure=None,
+            fail_on_empty=False,
+            hook_params=None,
+            default_args={},
+        )