Hello, I'm having a difficult time getting my test...
# ask-community
s
Hello, I'm having a difficult time getting my testing_pipeline to work. I've been following this documentation: https://docs.dagster.io/tutorial/advanced-tutorial/scheduling https://docs.dagster.io/concepts/testing The pipeline is supposed to run using @daily_schedule and returns a config that looks like:
Copy code
@daily_schedule(
    pipeline_name="my_pipeline",
    start_date=datetime(2021, 4, 1),
    execution_time=time(hour=8, minute=15),
    execution_timezone="UTC",
)
def my_schedule(date):
    return {
        "solids": {
            "my_solid": {
                "config": {
                    "date": date.strftime("%Y-%m-%d")
                }
            }
        }
    }
My test is:
Copy code
def test_pipeline_pipeline():
    date = datetime(2021, 5, 1)
    res = execute_pipeline(
        my_pipeline,
        mode="unit_test",
        run_config={
            "solids": {
                "my_solid": {
                        "config": {
                            "date": date.strftime("%Y-%m-%d")
                    }
                }
            }
        }
    )
    assert isinstance(res, PipelineExecutionResult)
    assert res.success

    assert res.output_for_solid("validate_data")
My latest error is:
dagster.check.ParameterCheckError: Param "msg" is not a str. Got AttributeError("'SolidExecutionContext' object has no attribute 'date'") which is type <class 'AttributeError'>
At this point I'm very confused and not quite sure how the config works between the scheduling and the testing. Any help is greatly appreciated.
p
Hi Sarah! Do you mind including the solid definitions that make up your pipeline
my_pipeline
?
s
Copy code
@solid(
    required_resource_keys={'my_api'},
    input_defs=[InputDefinition("start", Nothing)]
)
def my_solid(context) -> dagster_pd.DataFrame:
    try:
        end_date = (datetime.strptime(context.date, date_format) + timedelta(days=1)).strftime(date_format)
        return context.resources.my_api.fetch_network(context.date, end_date)
    except Exception as e:
        context.log.error(e)

@solid(required_resource_keys={'redacted_db'})
def write_df_to_table(context, df: dagster_pd.DataFrame) -> dagster_pd.DataFrame:
    context.resources.redacted_db.write_pandas(db_table, df)
    return df

@solid(required_resource_keys={'redacted_db'})
def delete_daily_data(context, df: dagster_pd.DataFrame) -> dagster_pd.DataFrame:
    query = f'DELETE FROM {db_table} WHERE date::DATE = \'{context.date}\';'
    <http://context.log.info|context.log.info>(f'Running the following query against database: \n{query}')
    context.resources.redacted_db.query(query)
    return df

@solid(required_resource_keys={'redacted_db'})
def validate_data(context, df: dagster_pd.DataFrame) -> bool:
    expected_count = 1000
    threshold = .02

    query = f'SELECT count(*) from REDACTED_TABLENAME where date::DATE = \'{context.date}\''
    is_valid = context.resources.redacted_db.validate(df, query, threshold=threshold, min_amt=expected_count)

    return is_valid
I've changed some names of things for safety and whatnot
p
I think
context.date
should probably be
context.solid_config['date']
. The error message could definitely be better here
s
I'll give it a shot!
That worked! Thanks!!
🎉 1