def METHOD_NAME():
    def _execute(scheduler_address, instance):
        with execute_job(
            reconstructable(dask_engine_job),
            run_config={
                "execution": {"config": {"cluster": {"existing": {"address": scheduler_address}}}},
            },
            instance=instance,
        ) as result:
            return result
    async def _run_test():
        with instance_for_test() as instance:
            async with Scheduler() as scheduler:
                async with Worker(scheduler.address) as _:
                    result = await asyncio.get_event_loop().run_in_executor(
                        None, _execute, scheduler.address, instance
                    )
                    assert result.success
                    assert result.output_for_node("simple") == 1
    asyncio.get_event_loop().run_until_complete(_run_test())