acoustic-carpenter-78188
11/02/2023, 9:28 PMimport pytest
from mock import patch as _system_patch
from flytekit.core.task import task
from flytekit.core.testing import patch as flyte_patch
from flytekit.core.workflow import ImperativeWorkflow, workflow
@task
def t1(a: str) -> str:
return a + " world"
wb = ImperativeWorkflow(name="my.workflow")
wb.add_workflow_input("in1", str)
node = wb.add_entity(t1, a=wb.inputs["in1"])
wb.add_workflow_output("from_n0t1", node.outputs["o0"])
@_system_patch("flytekit.core.workflow.ImperativeWorkflow.execute")
def test_return_none_errors(mock_execute):
mock_execute.return_value = None
with pytest.raises(Exception):
wb(in1="hello")
@flyte_patch(wb)
def test_imperative_patching(mock_wb):
mock_wb.return_value = "hi"
@workflow
def my_functional_wf(a: str) -> str:
x = wb(in1=a)
return x
assert my_functional_wf(a="hello") == "hi"
It is unclear what is happening. Somehow, the flyte_patch
patching logic is interfering with the normal mock.patch
. The mere presence of the second test, basically makes it seem like the mock.patch
isn't there on the first test. Basically ImperativeWorkflow.execute
is not patched and returns what it normally would return, and hence the exception isn't raised.
If you copy paste the wb
workflow and use a new variable for it for the second test, then things pass. This will take some digging into the mock library to understand how they're interfering with each other.
flyteorg/flyteacoustic-carpenter-78188
11/02/2023, 9:28 PM