chelseatroy

    chelseatroy

    8 months ago
    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: ``````
    Kevin Kho

    Kevin Kho

    8 months ago
    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
    chelseatroy

    chelseatroy

    8 months ago
    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>}
    When I print out get_last_executed_value I get:
    <Task: get_last_executed_value>
    Kevin Kho

    Kevin Kho

    8 months ago
    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)
    The test would be:
    from testing import *
    
    assert a in flow
    Because a copy is create with the Functional API
    Thanks for moving it!
    chelseatroy

    chelseatroy

    8 months ago
    So @Kevin Kho, are you saying to ad that line, or replace some line with that line?
    And where does
    testing
    come from? Because when I do that line it is telling me that's not a thing 😞
    Kevin Kho

    Kevin Kho

    8 months ago
    Sorry,
    testing.py
    is the name of the file of my first file. So I am importing the Flow object and tasks from it
    I am saying you need to assign your task copies to a variable, and then assert on that variable because it’s a copy
    chelseatroy

    chelseatroy

    8 months ago
    Hm. @Kevin Kho I believe that that is what I am doing here:
    from src.flows.prereview_engagement_feature_store_flow import *
    Kevin Kho

    Kevin Kho

    8 months ago
    The assert on my example is on
    a
    , the copy of the task, not
    abc()
    , the task definition
    chelseatroy

    chelseatroy

    8 months ago
    Ah, I see
    okay, I'll try that. Thank you!
    Kevin Kho

    Kevin Kho

    8 months ago
    Yep!