async def METHOD_NAME(self, flow, agent):
    """
    This test sets the "sum" task to Failed before running the flow.
    It then sets it to Scheduled and reruns from that point. This tests whether "sum"
    properly loads its upstream mapped states.
    """
    flow_run_id = await api.runs.create_flow_run(flow_id=flow.server_id)
    # ----------------------------------------------------------
    # first run - start with numbers 1
    # set sum to paused so it won't run
    await api.states.set_task_run_state(
        task_run_id=await api.runs.get_or_create_task_run(
            flow_run_id=flow_run_id, task_id=flow.sum.id
        ),
        state=Failed(),
    )
    await agent.run_scheduled(flow_id=flow.server_id)
    # wait for states to be written to the database
    await asyncio.sleep(1.0)
    fr = await models.FlowRun.where(id=flow_run_id).first(
        {
            "serialized_state": True,
            "task_runs": {
                "task": {"slug"},
                "serialized_state": True,
                "map_index": True,
            },
        },
    )
    task_states = {
        (tr.task.slug, tr.map_index): state_schema.load(tr.serialized_state)
        for tr in fr.task_runs
    }
    assert fr.serialized_state["type"] == "Failed"
    assert len(task_states) == 6
    # numbers task
    assert task_states[(flow.numbers.slug, -1)].is_successful()
    # add parent task
    assert task_states[(flow.add.slug, -1)].is_mapped()
    # add children tasks
    for i in range(3):
        assert task_states[(flow.add.slug, i)].is_successful()
    # sum task
    assert task_states[(flow.sum.slug, -1)].is_failed()
    # ----------------------------------------------------------
    # second run - make sum scheduled and set flow to running
    await api.states.set_task_run_state(
        task_run_id=await api.runs.get_or_create_task_run(
            flow_run_id=flow_run_id, task_id=flow.sum.id
        ),
        state=Scheduled(),
    )
    await api.states.set_flow_run_state(flow_run_id, state=Running())
    from prefect.utilities.debug import raise_on_exception
    with raise_on_exception():
        await agent.run_scheduled(flow_id=flow.server_id)
        # wait for states to be written to the database
        await asyncio.sleep(1.0)
    fr = await models.FlowRun.where(id=flow_run_id).first(
        {
            "serialized_state": True,
            "task_runs": {
                "task": {"slug"},
                "serialized_state": True,
                "map_index": True,
            },
        },
    )
    task_states = {
        (tr.task.slug, tr.map_index): state_schema.load(tr.serialized_state)
        for tr in fr.task_runs
    }
    assert fr.serialized_state["type"] == "Success"
    # numbers task
    assert task_states[(flow.numbers.slug, -1)].is_successful()
    # add parent task
    assert task_states[(flow.add.slug, -1)].is_mapped()
    # add children tasks
    for i in range(3):
        assert task_states[(flow.add.slug, i)].is_successful()
    # sum task
    assert task_states[(flow.sum.slug, -1)].is_successful()
    assert prefect.RESULTS[(flow.sum.slug, -1)] == 9