Hey friends! Curious to get people's opinions on my testing strategy. My workflow is all filebased processing, so I was thinking of importing my task into say test_task1.py, running it against an input with a known output, and then writing a second task that compares the actual output to the known output, and combining both into a workflow. I could then aggregate all these task-test-workflows as subworkflows into one large testing suite parent workflow. What do ya reckon? Am I over-engineering the socks off this?

can you not test your worklows locally?

Hey Jay, thanks for chiming in. Most of them I can, yes, I guess I'm looking to setup something a bit more formal/declarative so I'm not just eyeballing it / doing a lot of repeated clicking.

oh i mean just use pytest

I see.. so you basically execute your whole workflow with pytest locally and then run a bunch of asserts between expected and actual outputs? How well does pytest play with Flyte native objects like FlyteFiles? I don't suppose your test suite is open source so I could take a gander? 😅

you might need to mock few things but overall pretty easy to setup. However our test image has access to cloud so we use Flytefile as it is

Oh interesting... great suggestion on looking into flytekit tests, I was barking up te wrong tree in the flytesnacks repo. Is this vaguely the approach you're suggesting?

def test_lp():
    @task
    def t1(a: int) -> typing.NamedTuple("OutputsBC", t1_int_output=int, c=str):
        a = a + 2
        return a, "world-" + str(a)

    @workflow
    def wf(a: int) -> (str, str):
        x, y = t1(a=a)
        u, v = t1(a=x)
        return y, v

    lp = launch_plan.LaunchPlan.get_or_create(wf, "get_or_create1")
    lp2 = launch_plan.LaunchPlan.get_or_create(wf, "get_or_create1")
    assert lp.name == "get_or_create1"
    assert lp is lp2

    default_lp = launch_plan.LaunchPlan.get_or_create(wf)
    default_lp2 = launch_plan.LaunchPlan.get_or_create(wf)
    assert default_lp is default_lp2

    with pytest.raises(ValueError):
        launch_plan.LaunchPlan.get_or_create(wf, default_inputs={"a": 3})

    lp_with_defaults = launch_plan.LaunchPlan.create("get_or_create2", wf, default_inputs={"a": 3})
    assert lp_with_defaults.parameters.parameters["a"].default.scalar.primitive.integer == 3

Hmm that’s a little bit different we just have a pre determined launch plans so we just import the workflow or launch plan and compare the outputs

Gotcha - so your setup might be closer to:

import my_task
def test_my_task():
    out_ = my_task(in_)
    assert out_ == expected_out

And since you can run everything locally you just run

pytest test .

(or wtv) and it handles everything. I think that would work for most of my tasks, however one relies on a daemon running in another pod, so that might have to be e2e by definition. It's true what you're saying though, there's never 1 right way to test things, always a balancing act. Thanks for sharing your insights, you've set me on a more productive path I think! Lots to think about...