async def METHOD_NAME(self, flow, agent):
    flow_run_id = await api.runs.create_flow_run(flow_id=flow.server_id)
    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 len(task_states) == 6
    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