class_docstr: Tests keda.
class_docstr_tok: ["Tests","keda","."]
docstr: Verify keda sql query uses configured concurrency.
docstr_tok: ["Verify","keda","sql","query","uses","configured","concurrency","."]
code: def test_keda_concurrency(self, executor, concurrency):
      """Verify keda sql query uses configured concurrency."""
      docs = render_chart(
          values={
              "workers": {
                  "keda": {"enabled": True},
                  "persistence": {"enabled": False},
              },
              "executor": executor,
              "config": {"celery": {"worker_concurrency": concurrency}},
          },
          show_only=["templates/workers/worker-kedaautoscaler.yaml"],
      )
      expected_query = self.build_query(
          executor=executor, concurrency=concurrency
      )
      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 5 for calling the TestKeda obj's underlying member method code with example usage: obj.test_keda_concurrency(executor, concurrency) without return types