def METHOD_NAME(
    empty_data_context: AbstractDataContext,
    valid_file_path: pathlib.Path,
    test_df_pandas: pd.DataFrame,
    spark_session,
    spark_df_from_pandas_df,
):
    my_config_variables = {"pipeline_filename": __file__}
    empty_data_context.config_variables.update(my_config_variables)
    spark_df = spark_df_from_pandas_df(spark_session, test_df_pandas)
    spark_datasource = empty_data_context.sources.add_spark("my_spark_datasource")
    batch_metadata = {
        "no_curly_pipeline_filename": "$pipeline_filename",
        "curly_pipeline_filename": "${pipeline_filename}",
        "pipeline_step": "transform_3",
    }
    dataframe_asset = spark_datasource.add_dataframe_asset(
        name="my_dataframe_asset",
        batch_metadata=batch_metadata,
    )
    assert dataframe_asset.batch_metadata == batch_metadata
    batch_list = dataframe_asset.get_batch_list_from_batch_request(
        dataframe_asset.build_batch_request(dataframe=spark_df)
    )
    assert len(batch_list) == 1
    substituted_batch_metadata = copy.deepcopy(batch_metadata)
    substituted_batch_metadata.update(
        {
            "no_curly_pipeline_filename": __file__,
            "curly_pipeline_filename": __file__,
        }
    )
    assert batch_list[0].metadata == substituted_batch_metadata