Abhas P

09/29/2021, 12:46 AM
Hi, I am trying to test the flow compositions as pointed in this article, I declare my tasks as a decorator in the flow file(instead of a Class kinda definition as pointed in the link) like :
Copy code

def load():
I try to test the flow composition like this :
Copy code
from flows.sample_flow import load
from flows import sample_flow as flow_file

def test_flow_composition():
    load_task = load
    flow_tasks = flow_file.flow.tasks

    assert flow_file.flow.terminal_tasks() == set([load_task])

    # assert load_task in flow_tasks    - same things happens here , load_task is clearlya n element of the flow_tasks set but it asserts false
I get this error, which is little weird (I suspect object signatures here :
Copy code
>       assert flow_file.flow.terminal_tasks() == set([load_task])
E       assert {<Task: load>} == {<Task: load>}
E         Extra items in the left set:
E         <Task: load>
E         Extra items in the right set:
E         <Task: load>
E         Full diff:
E           {<Task: load>}
Can you help me with this ?

Kevin Kho

09/29/2021, 3:44 PM
I’ve been trying this myself but I can’t figure it out. Will need to task the team
I got an answer on this
So I have a file called
that does:
Copy code
from prefect import Flow, task
from prefect.client.secrets import Secret

def load():
    return 1

with Flow("name") as flow:
    x = load()
And then my test script:
Copy code
import secret_test
from prefect import task

def test_flow_composition():
    assert set([secret_test.x]) == secret_test.flow.terminal_tasks()
    assert secret_test.x in secret_test.flow.tasks

Basically using a task in the flow block creates a new copy because tasks can be used repeatedly so you need to reference that variable