如何使用 pytest 测试 kfp 组件答案

【问题标题】：How to test kfp components with pytest如何使用 pytest 测试 kfp 组件
【发布时间】：2022-10-19 11:56:00
【问题描述】：

我正在尝试使用 pytest 对来自 kfp.v2.ds1（在管道上工作）的 kubeflow 组件进行本地测试，但在输入/输出参数和固定装置方面遇到了困难。

下面是一个代码示例来说明这个问题：

首先，我创建了一个夹具来模拟数据集。这个夹具也是一个 kubeflow 组件。

# ./fixtures/

    @pytest.fixture
    @component()
    def sample_df(dataset: Output[Dataset]):
         df = pd.DataFrame(
             {
                 'name': ['Ana', 'Maria', 'Josh'],
                 'age': [15, 19, 22],
             }
         )
         dataset.path += '.csv'
         df.to_csv(dataset.path, index=False)
         return

让我们假设组件的年龄翻倍。

# ./src/
    @component()
    def double_ages(df_input: Input[Dataset], df_output: Output[Dataset]):
         df = pd.read_csv(df_input.path)
         
         double_df = df.copy()
         double_df['age'] = double_df['age']*2

         df_output.path += '.csv'
         double_df.to_csv(df_output.path, index=False)

然后，测试：

#./tests/

@pytest.mark.usefixtures("sample_df")
def test_double_ages(sample_df):

    expected_df = pd.DataFrame(
        {
            'name': ['Ana', 'Maria', 'Josh'],
            'age': [30, 38, 44],
        }
    )

    df_component = double_ages(sample_df)    # This is where I call the component, sample_df is an Input[Dataset]
    df_output = df_component.outputs['df_output']
    df = pd.read_csv(df_output.path)
    
    assert df['age'].tolist() == expected_df['age'].tolist()

但这就是问题发生的时候。应该作为输出传递的 Output[Dataset] 不是，因此组件无法正常使用它，然后我会在assert df['age'].tolist() == expected_df['age'].tolist() 上收到以下错误：

AttributeError：“TaskOutputArgument”对象没有属性“路径”

显然，该对象的类型是TaskOutputArgument，而不是Dataset。

有谁知道如何解决这个问题？或者如何正确使用 pytest 和 kfp 组件？我在互联网上搜索了很多，但找不到关于它的线索。

【问题讨论】：

标签： python pytest kubeflow kubeflow-pipelines kfp

【解决方案1】：

我花了一些时间对此进行调查，我的结论是单个组件不应该由 kfp 的设计进行单元测试。这意味着您必须依靠对每个组件的逻辑进行单元测试，将该逻辑的每一部分包装在一个组件中，然后测试 kfp 管道的端到端功能。

我同意，如果有一种方法可以轻松地模拟输入和输出，那将是非常酷的，但我挖得很深，目前这似乎不是预期的用途（或简单的 hack）。

【讨论】：