Pryce
07/29/2023, 10:51 PMJay Ganbat
07/29/2023, 11:04 PMPryce
07/29/2023, 11:06 PMJay Ganbat
07/29/2023, 11:08 PMPryce
07/29/2023, 11:12 PMJay Ganbat
07/29/2023, 11:14 PMPryce
07/29/2023, 11:22 PMdef 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
Jay Ganbat
07/29/2023, 11:25 PMPryce
07/29/2023, 11:28 PMimport 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...