Differential D8684 Diff 40813 examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_partitions.py

Changeset View

Standalone View

View Options

examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_partitions.py

	import pytest			import pytest
	from dagster import build_schedule_context, execute_pipeline, validate_run_config			from dagster import (
				build_schedule_context,
				execute_pipeline,
				validate_run_config,
				)
	from docs_snippets.concepts.partitions_schedules_sensors.partition_definition import (			from docs_snippets.concepts.partitions_schedules_sensors.partition_definition import (
	date_partition_set,			date_partition_set,
	my_schedule,			my_schedule,
	test_my_partition_set,			test_my_partition_set,
	weekday_partition_set,			weekday_partition_set,
	)			)
	from docs_snippets.concepts.partitions_schedules_sensors.pipeline import my_data_pipeline			from docs_snippets.concepts.partitions_schedules_sensors.pipeline import (
				my_data_pipeline,
				)


	def test_pipeline():			def test_pipeline():
	result = execute_pipeline(			result = execute_pipeline(
	my_data_pipeline,			my_data_pipeline,
	{"solids": {"process_data_for_date": {"config": {"date": "2018-05-01"}}}},			{
				"solids": {
				"process_data_for_date": {"config": {"date": "2018-05-01"}}
				}
				},
	)			)
	assert result.success			assert result.success


	@pytest.mark.parametrize("partition_set", [date_partition_set, weekday_partition_set])			@pytest.mark.parametrize(
				"partition_set", [date_partition_set, weekday_partition_set]
				)
	def test_pipeline_with_partition_set(partition_set):			def test_pipeline_with_partition_set(partition_set):
	for partition in partition_set.partition_fn():			for partition in partition_set.partition_fn():
	run_config = partition_set.run_config_for_partition(partition)			run_config = partition_set.run_config_for_partition(partition)
	result = execute_pipeline(			result = execute_pipeline(
	my_data_pipeline,			my_data_pipeline,
	run_config=run_config,			run_config=run_config,
	)			)
	assert result.success			assert result.success
	Show All 10 Lines