def METHOD_NAME(
    databricks_client_factory,
    mocker: MockerFixture,
    databricks_resource_key: Optional[str],
) -> None:
    mock_submit_run = mocker.patch("databricks.sdk.JobsAPI.submit")
    mock_get_run = mocker.patch("databricks.sdk.JobsAPI.get_run")
    databricks_job_configuration = {
        "new_cluster": {
            "spark_version": "2.1.0-db3-scala2.11",
            "num_workers": 2,
        },
        "notebook_task": {
            "notebook_path": "/Users/dagster@example.com/PrepareData",
        },
    }
    mock_submit_run_response = mocker.Mock()
    mock_submit_run_response.bind.return_value = {"run_id": 1}
    mock_submit_run.return_value = mock_submit_run_response
    mock_get_run.side_effect = _mock_get_run_response()
    if databricks_resource_key is not None:
        METHOD_NAME = create_databricks_submit_run_op(
            databricks_job_configuration=databricks_job_configuration,
            poll_interval_seconds=0.01,
            databricks_resource_key=databricks_resource_key,
        )
    else:
        METHOD_NAME = create_databricks_submit_run_op(
            databricks_job_configuration=databricks_job_configuration,
            poll_interval_seconds=0.01,
        )
    databricks_resource_name = (
        databricks_resource_key if databricks_resource_key is not None else "databricks"
    )
    @job(
        resource_defs={
            databricks_resource_name: databricks_client_factory(
                host="https://abc123.cloud.databricks.com/", token="token"
            )
        }
    )
    def test_databricks_job() -> None:
        METHOD_NAME()
    result = test_databricks_job.execute_in_process()
    assert result.success
    mock_submit_run.assert_called_once_with(
        tasks=[jobs.SubmitTask.from_dict(databricks_job_configuration)],
    )
    assert mock_get_run.call_count == 4