code: def test_task_group_expand_kwargs_with_upstream(
      dag_maker, session, caplog
  ):
      with dag_maker() as dag:
  
          @dag.task
          def t1():
              return [{"a": 1}, {"a": 2}]
  
          @task_group("tg1")
          def tg1(a, b):
              @dag.task()
              def t2():
                  return [a, b]
  
              t2()
  
          tg1.expand_kwargs(t1())
  
      dr = dag_maker.create_dagrun()
      dr.task_instance_scheduling_decisions()
      assert "Cannot expand" not in caplog.text
      assert (
          "missing upstream values: ['expand_kwargs() argument']"
          not in caplog.text
      )
  
file_path: airflow/tests/decorators/test_task_group.py
imports: from __future__ import annotations
  from datetime import timedelta
  import pendulum
  import pytest
  from airflow.decorators import dag, task_group
  from airflow.models.expandinput import DictOfListsExpandInput, ListOfDictsExpandInput, MappedArgument
  from airflow.operators.empty import EmptyOperator
  from airflow.utils.task_group import MappedTaskGroup
label_desc: Use image node_id 9 for calling a global function with example usage: test_task_group_expand_kwargs_with_upstream(dag_maker, session, caplog) and returns: list, list