QQ how can I assert the state of a dependent flow is successful? I'm running the flow using StartFlowRun

Hey @Arun Giridharan, use

wait=True

in your

StartFlowRun

Hi @Kevin Kho What I have currently is a setup like this.

flow_run = StartFlowRun(flow_name="process", project_name="sandbox", wait=true)


with prefect.Flow("download") as flow_download:
    fp = "plip"
    check = check_file(fp)
    download = download_file(check)
    run = flow_run(run_name=f"process_{fp}", parameters={'fp': fp})

For my test I would like to

assert

that

process

actually passed. I can verify

download

passed using the

state.is_sucessful()

method. Is there a way to do this for

process

?

wait=True

will raise the state so if it fails, that task should fail also. Are you checking for success in a state handler?

def test_flow_results(monkeypatch):

flow = create_flow()

state = flow.run(

parameters={

...

},

}

)

# Test running the flow does not produce any obvious errors

assert state.is_successful()

Yeah it’s hard cuz under the hood, that task is doing an API call. What I saw one user do is subclass

StartFlowRun

and if it’s running with an env variable, just

raise success

, otherwise actually run the task. So you can do something like that to test it. Similarly, I think you have to Mock the task itself to test it so that the API call to start a flow run is not triggered. With your Mock, you can

raise Success

to raise the state.