Dolor Oculus
09/18/2020, 9:33 PMwith Flow() as flow:
e = Extract()
t = Transform(e)
l = Load(t)
state = flow.run()
syntax? I'm getting key not found results on asserting state.result[e], and wondering if you have to do it in the way given in the testing-flows link above to unit test the flows.Dolor Oculus
09/18/2020, 9:34 PMDolor Oculus
09/18/2020, 9:35 PMnicholas
Dolor Oculus
09/19/2020, 12:55 AMdef test_simple_flow():
@task
def extract():
return 42
@task
def transform(x: int) -> int:
return int(x / 2 + 2)
@task
def load(x: int) -> str:
return str(x)
with Flow(__name__, environment=LocalEnvironment()) as flow:
e = extract()
t = transform(e)
l = load(t)
state = flow.run()
assert state.is_successful()
assert state.result[e].is_successful()
assert state.result[t].is_successful()
assert state.result[l].is_successful()
assert state.result[e].result == 42
assert state.result[t].result == 23
assert state.result[l].result == "23"
nicholas