Differential D8501 Diff 40123 python_modules/libraries/dagster-airflow/dagster_airflow_tests/test_compile.py

Changeset View

Standalone View

python_modules/libraries/dagster-airflow/dagster_airflow_tests/test_compile.py

	from dagster.core.definitions.pipeline_base import InMemoryPipeline			from dagster.core.definitions.pipeline_base import InMemoryPipeline
	from dagster.core.execution.plan.plan import ExecutionPlan			from dagster.core.execution.plan.plan import ExecutionPlan
	from dagster.core.system_config.objects import ResolvedRunConfig			from dagster.core.system_config.objects import ResolvedRunConfig
	from dagster_airflow.compile import coalesce_execution_steps			from dagster_airflow.compile import coalesce_execution_steps
	from dagster_test.toys.composition import composition			from dagster_test.toys.composition import composition_job


	def test_compile():			def test_compile():
	resolved_run_config = ResolvedRunConfig.build(			resolved_run_config = ResolvedRunConfig.build(
	composition,			composition_job,
	{"solids": {"add_four": {"inputs": {"num": {"value": 1}}}}},			{"solids": {"add_four": {"inputs": {"num": {"value": 1}}}}},
	)			)

	plan = ExecutionPlan.build(InMemoryPipeline(composition), resolved_run_config)			plan = ExecutionPlan.build(InMemoryPipeline(composition_job), resolved_run_config)

	res = coalesce_execution_steps(plan)			res = coalesce_execution_steps(plan)
	assert set(res.keys()) == {			assert set(res.keys()) == {
	"add_four.add",			"add_four.add",
	"div_four.div_two",			"div_four.div_two",
	"div_four.div_two_2",			"div_four.div_two_2",
	"add_four.emit_two.emit_one_2",			"add_four.emit_two.emit_one_2",
	"add_four.emit_two_2.add",			"add_four.emit_two_2.add",
	"int_to_float",			"int_to_float",
	"add_four.emit_two_2.emit_one_2",			"add_four.emit_two_2.emit_one_2",
	"add_four.emit_two.add",			"add_four.emit_two.add",
	"add_four.emit_two_2.emit_one",			"add_four.emit_two_2.emit_one",
	"add_four.emit_two.emit_one",			"add_four.emit_two.emit_one",
	}			}