class_docstr: Tests keda.
class_docstr_tok: ["Tests","keda","."]
docstr: Verify keda sql query ignores kubernetes queue when CKE is used.
  
  Sometimes a user might want to use a different queue name for k8s executor tasks,
  and we also verify here that we use the configured queue name in that case.
docstr_tok: ["Verify","keda","sql","query","ignores","kubernetes","queue","when","CKE","is","used",".","Sometimes","a","user","might","want","to","use","a","different","queue","name","for","k8s","executor","tasks",",","and","we","also","verify","here","that","we","use","the","configured","queue","name","in","that","case","."]
code: def test_keda_query_kubernetes_queue(
      self, executor, queue, should_filter
  ):
      """
      Verify keda sql query ignores kubernetes queue when CKE is used.
  
      Sometimes a user might want to use a different queue name for k8s executor tasks,
      and we also verify here that we use the configured queue name in that case.
      """
      values = {
          "workers": {
              "keda": {"enabled": True},
              "persistence": {"enabled": False},
          },
          "executor": executor,
      }
      if queue:
          values.update(
              {
                  "config": {
                      "celery_kubernetes_executor": {
                          "kubernetes_queue": queue
                      }
                  }
              }
          )
      docs = render_chart(
          values=values,
          show_only=["templates/workers/worker-kedaautoscaler.yaml"],
      )
      expected_query = self.build_query(executor=executor, queue=queue)
      assert (
          jmespath.search("spec.triggers[0].metadata.query", docs[0])
          == expected_query
      )
  
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 6 for calling the TestKeda obj's underlying member method code with example usage: obj.test_keda_query_kubernetes_queue(executor, queue, should_filter) without return types