Hiya prefect folks! I have the following composition test for a flow that totally works. This composition test literally perfectly describes the flow in question: ``````

Hi @chelseatroy, could you move the code to the thread to clear up the main channel a bit when you get the chance? I think with is because the Flow object creates copies of the task. Let me test this a bit and get back to you

from prefect.core.edge import Edge

from src.flows.prereview_engagement_feature_store_flow import *

print(flow.tasks)
assert get_last_executed_value in flow.tasks
assert update_last_executed_value in flow.tasks
assert extract_from_snowflake in flow.tasks
assert dataframe_to_feature_group in flow.tasks
assert len(flow.tasks) == 4

...

However it fails at the very first assertion. I have printed flow.tasks and I see as follows.

{<Task: get_last_executed_value>, <Task: extract_from_snowflake>, <Task: update_last_executed_value>, <Task: dataframe_to_feature_group>}

So if you have a flow:

from prefect import Flow, task

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

@task
def bcd(x):
    return x

with Flow("..") as flow:
    a = abc(1)
    b = bcd(a)

So @Kevin Kho, are you saying to ad that line, or replace some line with that line?

Sorry,

testing.py

is the name of the file of my first file. So I am importing the Flow object and tasks from it

Hm. @Kevin Kho I believe that that is what I am doing here:

from src.flows.prereview_engagement_feature_store_flow import *

The assert on my example is on

a

, the copy of the task, not

abc()

, the task definition

Ah, I see

Yep!