Hello, can anyone help me? thanks :smile:, I'm trying to create unit/integration tests to check my f...

Raimundo Pereira De Souza Neto

03/29/2022, 5:35 PM

Hello, can anyone help me? thanks 😄, I'm trying to create unit/integration tests to check my flows and tasks. I'm using Prefect Orion

prefect = "2.0a13"

, I did read this link, but I'm using async flows. More infos in the thread.

Raimundo Pereira De Souza Neto

03/29/2022, 5:36 PM

My async flow:

Copy code

@flow(name="AutomaticLostFlow")
async def automatic_lost_flow() -> None:
    await automatic_lost_task()

My async task:

Copy code

@task(name="AutomaticLostTask")
async def automatic_lost_task():
    logger = get_run_logger()
    r = await lost.automatic_lost()
    <http://logger.info|logger.info>(f"{r} onboardings gone to lost stage")

lost.automatic_lost()

is an async query in the database

Raimundo Pereira De Souza Neto

03/29/2022, 5:38 PM

I would like to know how to test my flow (

automatic_lost_flow()

)

Anna Geller

03/29/2022, 5:38 PM

the docs you shared are for Prefect 1.0 though. For Orion, check out https://orion-docs.prefect.io/

Anna Geller

03/29/2022, 5:39 PM

perhaps this can help? https://github.com/PrefectHQ/prefect/blob/orion/tests/test_flows.py

Raimundo Pereira De Souza Neto

03/29/2022, 5:42 PM

Maybe this helps yes. Thanks a lot, I will read now.

😄 1

Open in Slack

Previous Next

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.