Hi Could you please help me to figure out how to test pipeli

Question

Hi Could you please help me to figure out how to test pipeline if it uses some external resources Are there proper ways to so Or the only way is to set up an environment DB etc and write special test mode for the pipeline I tried to use execute pipeline and solid selection But I cannot figure out how to pass inputs in solids in the middle code in the thread

Andrey Alekseev · Answer

``` pipeline def my pipeline df = load data from dwh alias load target data don t need to run df = data preparation df result = ml stuff df ``` df simple pandas dataframe I have simple specified dagster pandas type and data preparation accepts it ```PandasDataFrame = create dagster pandas dataframe type name= PandasDataFrame ``` load data from dwh uses custom sqlalchemy code so i don t have any resources Tried to do this ```def test pipeline run dataframe run config = solids divide data over column inputs df dataframe config solid selection = data preparation ml stuff execute pipeline my pipeline run config=run config solid selection=solid selection ``` An it results in

Andrey Alekseev · Answer

``` miniconda3 envs etna lib python3 8 site packages dagster core execution api py 298 in execute pipeline return logged execute pipeline miniconda3 envs etna lib python3 8 site packages dagster core telemetry py 89 in wrap result = f args kwargs miniconda3 envs etna lib python3 8 site packages dagster core execution api py 342 in logged execute pipeline pipeline run = instance create run for pipeline miniconda3 envs etna lib python3 8 site packages dagster core instance init py 518 in create run for pipeline execution plan = create execution plan miniconda3 envs etna lib python3 8 site packages dagster core execution api py 565 in create execution plan environment config = EnvironmentConfig build pipeline def run config mode=mode miniconda3 envs etna lib python3 8 site packages dagster core system config objects py 96 in build config evr = process config environment type run config miniconda3 envs etna lib python3 8 site packages dagster config validate py 333 in process config validate evr = validate config config type config dict miniconda3 envs etna lib python3 8 site packages dagster config validate py 58 in validate config return validate config from snap miniconda3 envs etna lib python3 8 site packages dagster config validate py 68 in validate config from snap return validate config miniconda3 envs etna lib python3 8 site packages dagster config validate py 103 in validate config return validate shape config context config value miniconda3 envs etna lib python3 8 site packages dagster config validate py 261 in validate shape config return validate shape config context config value check for extra incoming fields=True miniconda3 envs etna lib python3 8 site packages dagster config validate py 234 in validate shape config field evr = validate config context for field snap field snap config value name miniconda3 envs etna lib python3 8 site packages dagster config validate py 103 in validate config return validate shape config context config value miniconda3 envs etna lib python3 8 site packages dagster config validate py 261 in validate shape config return validate shape config context config value check for extra incoming fields=True miniconda3 envs etna lib python3 8 site packages dagster config validate py 234 in validate shape config field evr = validate config context for field snap field snap config value name miniconda3 envs etna lib python3 8 site packages dagster config validate py 103 in validate config return validate shape config context config value miniconda3 envs etna lib python3 8 site packages dagster config validate py 261 in validate shape config return validate shape config context config value check for extra incoming fields=True miniconda3 envs etna lib python3 8 site packages dagster config validate py 234 in validate shape config field evr = validate config context for field snap field snap config value name miniconda3 envs etna lib python3 8 site packages dagster config validate py 103 in validate config return validate shape config context config value miniconda3 envs etna lib python3 8 site packages dagster config validate py 261 in validate shape config return validate shape config context config value check for extra incoming fields=True miniconda3 envs etna lib python3 8 site packages dagster config validate py 234 in validate shape config field evr = validate config context for field snap field snap config value name miniconda3 envs etna lib python3 8 site packages dagster config validate py 101 in validate config return validate selector config context config value miniconda3 envs etna lib python3 8 site packages dagster config validate py 155 in validate selector config if config value == miniconda3 envs etna lib python3 8 site packages pandas core ops init py 833 in f other = align method FRAME self other axis=None miniconda3 envs etna lib python3 8 site packages pandas core ops init py 676 in align method FRAME right = to series right right = def to series right msg = Unable to coerce to Series length must be req len given given len if axis is not None and left get axis name axis == index if len left index = len right raise ValueError msg format req len=len left index given len=len right right = left constructor sliced right index=left index else if len left columns = len right gt raise ValueError msg format req len=len left columns given len=len right E ValueError Unable to coerce to Series length must be 3 given 0```

Andrey Alekseev · Answer

Same error when I change ```def data preparation context df PandasDataFrame ``` to ```def data preparation context df Any ```

Leor · Answer

hmmm that seems like an issue with your dataframe not dagster itself

Leor · Answer

can you confirm that the code works outside the dagster abstraction

Andrey Alekseev · Answer

Yep I have test for single solid and it works Here is the full code ```import pandas as pd import pytest from dagster import execute solid from src solids features data preparation import data preparation pytest fixture def create dataframe df = pd DataFrame from dict date pd date range start= 2020 01 01 end= 2020 02 01 df value = 42 df segment = Moscow return df pytest mark smoke def test solid run create dataframe input values = df create dataframe run config = solids data preparation config target column value divide by segment date column date res = execute solid data preparation input values=input values run config=run config assert res success``` Also it works as full pipeline when it loads data from DB and sends it to data preparation solid

Leor · Answer

aight

Andrey Alekseev · Answer

Maybe it is probably better just to use modes and resources and test the the whole pipeline

Leor · Answer

Yes probably

sandy · Answer

hey < Andrey Alekseev> this is a very sensible thing to want to do the issue you re running into is because it s not possible to pass pandas dataframes through dagster config i e dagster doesn t know what to do with ``` inputs df dataframe ``` we would like to make this cleaner to express but I believe the best way to make this happen right now would be with a custom dagster type loader ```from dagster pandas data frame import dataframe loader dagster type loader config schema=Selector mock def mock df loader context config if list config items 0 0 == mock df = pd DataFrame from dict date pd date range start= 2020 01 01 end= 2020 02 01 df value = 42 df segment = Moscow return df PandasDataFrame = create dagster pandas dataframe type name= PandasDataFrame loader=mock df loader ``` then you would use the following as run config in your test ``` run config = solids divide data over column inputs df mock config ```

Andrey Alekseev · Answer

Thanks Will try that