return {"key1": "value1", "key2": "value2"}
        f.override(task_id="my_task_id", do_xcom_push=True)("arg1", "arg2", kwarg1="kwarg1")
    dr = dag_maker.create_dagrun()
    (ti,) = dr.task_instances
    mock_hook.return_value.get_xcom_sidecar_container_image.return_value = XCOM_IMAGE
    dag.get_task("my_task_id").execute(context=ti.get_template_context(session=session))
    mock_hook.assert_called_once_with(
        conn_id=None,
        in_cluster=False,
        cluster_context="default",
        config_file="/tmp/fake_file",
    )
    assert mock_create_pod.call_count == 1
    assert mock_hook.return_value.get_xcom_sidecar_container_image.call_count == 1
    containers = mock_create_pod.call_args[1]["pod"].spec.containers
    
    assert len(containers) == 2
    assert containers[0].command[0] == "bash"
    assert len(containers[0].args) == 0
    assert containers[0].env[0].name == "__PYTHON_SCRIPT"
    assert containers[0].env[0].value
    assert containers[0].env[1].name == "__PYTHON_INPUT"
    assert containers[0].env[1].value
    
    decoded_input = pickle.loads(base64.b64decode(containers[0].env[1].value))
    assert decoded_input == {"args": ("arg1", "arg2"), "kwargs": {"kwarg1": "kwarg1"}}
    
    assert containers[1].image == XCOM_IMAGE
    assert containers[1].volume_mounts[0].mount_path == "/airflow/xcom"