如何进入州长核心地方任务的最终状态

Question

我构建了一个流，如果kwarg为空，它会隐式跳过运行给定任务。

我在任务函数中使用这样的东西来跳过逻辑：

if kwargs.get('processors', Hierarchy()).__len__() == 0:
raise signals.SKIP('skipping task',
result=Prediction())    

我想构建一些单元测试，以确保跳过所述任务的最终状态。在任务级别获取状态的最简单方法是什么？

我可以从文档中看到如何获得流，但不能获得任务。

更新

为了补充Chris的回答，我使用了他提出的第一个选项。由于我的流是在测试之外定义的，所以我创建了一个简单的函数来获取一组跳过的任务。在测试中，将其与本应跳过的任务列表进行了比较：

def get_skipped_tasks(flow_state):
return set(key.name for key, value in flow_state.result.items() if value.is_skipped())

Answer 1

为了完整性，我将在这里包含一些方法；例如，我将使用以下基本流程：

from prefect import task, Flow
from prefect.engine.signals import SKIP
import random

@task
def random_number():
return random.randint(0, 100)
@task
def is_even(num):
if num % 2:
raise SKIP("odd number")
return True
with Flow("dummy") as flow:
even_task = is_even(random_number)

运行整个流程

交互运行时，您可以始终运行整个流，并从父流运行状态访问单个任务状态；请注意，当您"；呼叫"；一个任务(例如is_even(random_number)(创建了一个副本，因此您需要正确跟踪这些副本。

flow_state = flow.run()
assert flow_state.result[even_task].is_skipped() # for example

用模拟数据运行一段流

当以交互方式运行时，您还可以传递任务字典->声明跑步者将予以尊重；这些状态可以选择性地提供数据：

from prefect.engine.state import Success
mocked_state = Success(result=2)

flow_state = flow.run(task_states={random_number: mocked_state})
assert not flow_state.result[even_task].is_skipped()

使用TaskRunner

最后，如果您想单独对此任务运行基于状态的测试，可以使用TaskRunner。这会使变得更加复杂，因为您必须使用Edges重新创建上游依赖关系。

from prefect.engine.task_runner import TaskRunner
from prefect.edge import Edge
runner = TaskRunner(task=even_task)
edge = Edge(key="num", upstream_task=random_number, downstream_task=even_task)

task_state = runner.run(upstream_states={edge: mocked_state})
assert not task_state.is_skipped()