code: def test_execute(self):
      # Given / When
      conn_id = "spark_default"
      operator = SparkSubmitOperator(
          task_id="spark_submit_job",
          spark_binary="sparky",
          dag=self.dag,
          **self._config,
      )
  
      # Then expected results
      expected_dict = {
          "conf": {"parquet.compression": "SNAPPY"},
          "files": "hive-site.xml",
          "py_files": "sample_library.py",
          "archives": "sample_archive.zip#SAMPLE",
          "driver_class_path": "parquet.jar",
          "jars": "parquet.jar",
          "packages": "com.databricks:spark-avro_2.11:3.2.0",
          "exclude_packages": "org.bad.dependency:1.0.0",
          "repositories": "http://myrepo.org",
          "total_executor_cores": 4,
          "executor_cores": 4,
          "executor_memory": "22g",
          "keytab": "privileged_user.keytab",
          "principal": "user/spark@airflow.org",
          "proxy_user": "sample_user",
          "name": "{{ task_instance.task_id }}",
          "num_executors": 10,
          "status_poll_interval": 30,
          "verbose": True,
          "application": "test_application.py",
          "driver_memory": "3g",
          "java_class": "com.foo.bar.AppMain",
          "application_args": [
              "-f",
              "foo",
              "--bar",
              "bar",
              "--start",
              "{{ macros.ds_add(ds, -1)}}",
              "--end",
              "{{ ds }}",
              "--with-spaces",
              "args should keep embedded spaces",
          ],
          "spark_binary": "sparky",
          "queue": "yarn_dev_queue2",
          "deploy_mode": "client2",
          "use_krb5ccache": True,
          "properties_file": "conf/spark-custom.conf",
      }
  
      assert conn_id == operator._conn_id
      assert expected_dict["application"] == operator._application
      assert expected_dict["conf"] == operator._conf
      assert expected_dict["files"] == operator._files
      assert expected_dict["py_files"] == operator._py_files
      assert expected_dict["archives"] == operator._archives
      assert (
          expected_dict["driver_class_path"]
          == operator._driver_class_path
      )
      assert expected_dict["jars"] == operator._jars
      assert expected_dict["packages"] == operator._packages
      assert (
          expected_dict["exclude_packages"]
          == operator._exclude_packages
      )
      assert expected_dict["repositories"] == operator._repositories
      assert (
          expected_dict["total_executor_cores"]
          == operator._total_executor_cores
      )
      assert expected_dict["executor_cores"] == operator._executor_cores
      assert (
          expected_dict["executor_memory"] == operator._executor_memory
      )
      assert expected_dict["keytab"] == operator._keytab
      assert expected_dict["principal"] == operator._principal
      assert expected_dict["proxy_user"] == operator._proxy_user
      assert expected_dict["name"] == operator._name
      assert expected_dict["num_executors"] == operator._num_executors
      assert (
          expected_dict["status_poll_interval"]
          == operator._status_poll_interval
      )
      assert expected_dict["verbose"] == operator._verbose
      assert expected_dict["java_class"] == operator._java_class
      assert expected_dict["driver_memory"] == operator._driver_memory
      assert (
          expected_dict["application_args"]
          == operator._application_args
      )
      assert expected_dict["spark_binary"] == operator._spark_binary
      assert expected_dict["queue"] == operator._queue
      assert expected_dict["deploy_mode"] == operator._deploy_mode
      assert (
          expected_dict["properties_file"] == operator._properties_file
      )
      assert expected_dict["use_krb5ccache"] == operator._use_krb5ccache
  
file_path: airflow/tests/providers/apache/spark/operators/test_spark_submit.py
imports: from __future__ import annotations
  from datetime import timedelta
  import pytest
  from airflow.models import DagRun, TaskInstance
  from airflow.models.dag import DAG
  from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
  from airflow.utils import timezone
label_desc: Use image node_id 2 for calling the TestSparkSubmitOperator obj's underlying member method code with example usage: obj.test_execute() without return types