I’m trying to write a unit test which mocks a method which cancels workflows.
Here is the example taken from (python/ray/workflow/tests/test_metadata.py
):
def test_running_and_canceled_workflow(workflow_start_regular, tmp_path):
workflow_id = "simple"
flag = tmp_path / "flag"
@workflow.step
def simple():
flag.touch()
time.sleep(1000)
return 0
simple.step().run_async(workflow_id)
# Wait until step runs to make sure pre-run metadata is written
while not flag.exists():
time.sleep(1)
workflow_metadata = workflow.get_metadata(workflow_id)
assert workflow_metadata["status"] == "RUNNING"
assert "start_time" in workflow_metadata["stats"]
assert "end_time" not in workflow_metadata["stats"]
workflow.cancel(workflow_id)
workflow_metadata = workflow.get_metadata(workflow_id)
assert workflow_metadata["status"] == "CANCELED"
Is there any way to test workflow.cancel
with mocking the state of a workflow? I’d like to skip writing workflow.step
and simple.step().run_async(workflow_id)
explicitly.