class_docstr: Tests keda.
class_docstr_tok: ["Tests","keda","."]
docstr: Verify default keda db connection.
docstr_tok: ["Verify","default","keda","db","connection","."]
code: def test_default_keda_db_connection(self):
      """Verify default keda db connection."""
      import base64
  
      docs = render_chart(
          values={
              "workers": {"keda": {"enabled": True}},
              "executor": "CeleryExecutor",
          },
          show_only=[
              "templates/workers/worker-deployment.yaml",
              "templates/workers/worker-kedaautoscaler.yaml",
              "templates/secrets/metadata-connection-secret.yaml",
          ],
      )
      worker_deployment = docs[0]
      keda_autoscaler = docs[1]
      metadata_connection_secret = docs[2]
  
      worker_container_env_vars = jmespath.search(
          "spec.template.spec.containers[?name=='worker'].env[].name",
          worker_deployment,
      )
      assert "AIRFLOW_CONN_AIRFLOW_DB" in worker_container_env_vars
      assert "KEDA_DB_CONN" not in worker_container_env_vars
  
      secret_data = jmespath.search("data", metadata_connection_secret)
      assert "connection" in secret_data.keys()
      assert (
          "@release-name-postgresql"
          in base64.b64decode(secret_data["connection"]).decode()
      )
      assert "kedaConnection" not in secret_data.keys()
  
      autoscaler_connection_env_var = jmespath.search(
          "spec.triggers[0].metadata.connectionFromEnv", keda_autoscaler
      )
      assert autoscaler_connection_env_var == "AIRFLOW_CONN_AIRFLOW_DB"
  
file_path: airflow/helm_tests/other/test_keda.py
imports: from __future__ import annotations
  import jmespath
  import pytest
  from tests.charts.helm_template_generator import render_chart
label_desc: Use image node_id 8 for calling the TestKeda obj's underlying member method code with example usage: obj.test_default_keda_db_connection() without return types