code: def test_pyspark_decorator_with_connection(
      self, spark_mock, conf_mock, dag_maker
  ):
      config = FakeConfig()
  
      builder = mock.MagicMock()
      spark_mock.builder.config.return_value = builder
      builder.getOrCreate.return_value = builder
      builder.sparkContext.return_value = builder
  
      conf_mock.return_value = config
  
      @task.pyspark(
          conn_id="pyspark_local",
          config_kwargs={"spark.executor.memory": "2g"},
      )
      def f(spark, sc):
          import random
  
          assert spark is not None
          assert sc is not None
          return [random.random() for _ in range(100)]
  
      with dag_maker():
          ret = f()
  
      dr = dag_maker.create_dagrun()
      ret.operator.run(
          start_date=dr.execution_date, end_date=dr.execution_date
      )
      ti = dr.get_task_instances()[0]
      assert len(ti.xcom_pull()) == 100
      assert config.get("spark.master") == "spark://none"
      assert config.get("spark.executor.memory") == "2g"
      assert config.get("spark.remote") is None
      assert config.get("spark.app.name")
  
      spark_mock.builder.config.assert_called_once_with(
          conf=conf_mock()
      )
  
file_path: airflow/tests/providers/apache/spark/decorators/test_pyspark.py
imports: from __future__ import annotations
  from typing import Any
  from unittest import mock
  import pytest
  from airflow.decorators import task
  from airflow.models import Connection
  from airflow.utils import db, timezone
label_desc: Use image node_id 2 for calling the TestPysparkDecorator obj's underlying member method code with example usage: obj.test_pyspark_decorator_with_connection(spark_mock, conf_mock, dag_maker) and returns: unknown