code: def test_xcom_zip(dag_maker, session, fillvalue, expected_results):
      results = set()
      with dag_maker(session=session) as dag:
  
          @dag.task
          def push_letters():
              return ["a", "b", "c"]
  
          @dag.task
          def push_numbers():
              return [1, 2, 3, 4]
  
          @dag.task
          def pull(value):
              results.add(value)
  
          pull.expand(
              value=push_letters().zip(
                  push_numbers(), fillvalue=fillvalue
              )
          )
  
      dr = dag_maker.create_dagrun()
  
      # Run "push_letters" and "push_numbers".
      decision = dr.task_instance_scheduling_decisions(session=session)
      assert sorted(ti.task_id for ti in decision.schedulable_tis) == [
          "push_letters",
          "push_numbers",
      ]
      for ti in decision.schedulable_tis:
          ti.run(session=session)
      session.commit()
  
      # Run "pull".
      decision = dr.task_instance_scheduling_decisions(session=session)
      assert sorted(ti.task_id for ti in decision.schedulable_tis) == [
          "pull"
      ] * len(expected_results)
      for ti in decision.schedulable_tis:
          ti.run(session=session)
  
      assert results == expected_results
  
imports: from __future__ import annotations
  import pytest
  from airflow.models.xcom_arg import XComArg
  from airflow.operators.bash import BashOperator
  from airflow.operators.python import PythonOperator
  from airflow.utils.types import NOTSET
  from tests.test_utils.config import conf_vars
  from tests.test_utils.db import clear_db_dags, clear_db_runs
label_desc: Use image node_id 4 for calling a global function with example usage: test_xcom_zip(dag_maker, session, fillvalue, expected_results) and returns: list, list