add ssl_verify_cert support to DruidDbApiHook.get_conn

providers/apache/druid/docs/operators.rst

@@ -54,7 +54,7 @@ the connection metadata is structured as follows:
      - Druid broker port (default: 8082)
    * - Extra: JSON
      - Additional connection configuration, such as:
-       ``{"endpoint": "/druid/v2/sql/", "method": "POST"}``
+       ``{"endpoint": "/druid/v2/sql/", "method": "POST", "ssl_verify_cert": false}``
 
 An example usage of the SQLExecuteQueryOperator to connect to Apache Druid is as follows:
 

providers/apache/druid/src/airflow/providers/apache/druid/hooks/druid.py

@@ -227,6 +227,7 @@ def get_conn(self) -> connect:
             user=conn.login,
             password=conn.password,
             context=self.context,
+            ssl_verify_cert=conn.extra_dejson.get("ssl_verify_cert", True),
         )
         self.log.info("Get the connection to druid broker on %s using user %s", conn.host, conn.login)
         return druid_broker_conn

providers/apache/druid/tests/unit/apache/druid/hooks/test_druid.py

@@ -426,6 +426,35 @@ def test_get_conn_with_context(
             user="test_login",
             password="test_password",
             context=passed_context,
+            ssl_verify_cert=True,
+        )
+
+    @patch("airflow.providers.apache.druid.hooks.druid.DruidDbApiHook.get_connection")
+    @patch("airflow.providers.apache.druid.hooks.druid.connect")
+    def test_get_conn_respects_ssl_verify_cert(self, mock_connect, mock_get_connection):
+        get_conn_value = MagicMock()
+        get_conn_value.host = "test_host"
+        get_conn_value.conn_type = "https"
+        get_conn_value.login = "test_login"
+        get_conn_value.password = "test_password"
+        get_conn_value.port = 10000
+        get_conn_value.extra_dejson = {
+            "endpoint": "/test/endpoint",
+            "schema": "https",
+            "ssl_verify_cert": False,
+        }
+        mock_get_connection.return_value = get_conn_value
+        hook = DruidDbApiHook()
+        hook.get_conn()
+        mock_connect.assert_called_with(
+            host="test_host",
+            port=10000,
+            path="/test/endpoint",
+            scheme="https",
+            user="test_login",
+            password="test_password",
+            context={},
+            ssl_verify_cert=False,
         )
 
     def test_get_uri(self):