Abhas P

    11 months ago
    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 :
    def load():
    I try to test the flow composition like this :
    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 :
    >       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

    Kevin Kho

    11 months ago
    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:
    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:
    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