Hi guys, has anyone done testing for mapped flow? I'm trying to write a test to verify if a task is available in the flow but I can't seem to access the tasks that are inside the mapped flow

Hi Scarlett, would you be able to show us the mapped flow? Or a minimal reproducible example of your flow?

from prefect import Flow, task, case, apply_map
from prefect.tasks.control_flow import merge

@task
def inc(x):
    return x + 1

@task
def negate(x):
    return -x

@task
def is_even(x):
    return x % 2 == 0

def inc_or_negate(x):
    cond = is_even(x)
    # If x is even, increment it
    with case(cond, True):
        res1 = inc(x)
    # If x is odd, negate it
    with case(cond, False):
        res2 = negate(x)
    return merge(res1, res2)

with Flow("apply-map example") as flow:
    result = apply_map(inc_or_negate, range(4))

So if you have a simple flow like in the example above which I copied from the docs for Mapping, when I tried (below codes), it doesn't seem to work. How do I access the tasks that are part of the mapped flow?

assert cond in flow.tasks

Just double checking that you are using prefect 1.0 correct?

Apply map and mapping in general works a bit differently since it dynamically spins up new child task runs at runtime but those don't exist in the static DAG structure. I think you could do: assert result in flow.tasks But then, it will dynamically generate mapped tasks

Yes, so I'm trying to make sure the right tasks are added to inc_or_negate(). I don't care about the child tasks but just the fact that cond, res1 and res2 are there and the dependencies between each task are correct.

This is what Prefect tests already do as part of the tests available in the Prefect repository - you can check mapping tests there. But I'm not sure how helpful this would be in your use case - this way you are testing Prefect core functionality while unit tests of your flows should rather test your business logic I encourage you to examine tests in our repo and then decide what makes sense to test

Let me try to be a bit clearer. I'm not testing the apply_map() function. What I'm trying to do, let's say I have the following tasks:

def full_flow(table):
    e = extract(...)
    x = export(...)
    t = transform(...)
    l = load(...)
with Flow("example") as flow:
    list_of_tables = get_tables(...)
    run = apply_map(full_flow, table=list_of_tables)

What I want to test in this flow is that I expect the task e and x to happen before t can happen. I understand that we can set upstream_task(), but I want to make sure the flow composition is as we expect especially when we have more complicated flow than this one. I'm having problems accessing the e, x, t, l tasks in flow.tasks. I can see when I look into flow.tasks, I can see all of the tasks there including

list_of_tables

. However, if I try evaluate the following statements, only the first one return true

list_of_tables in flow.tasks
e in flow.tasks
x in flow.tasks
t in flow.tasks
l in flow.tasks

I think the issue is I can't access e, x, t, or l because they are part of a mapped flow even though I can see them in flow.tasks.

I believe that your test is replicating Prefect tests, i.e. testing Prefect system rather than your dataflow logic

Ok, thank you for your clarification