Differential D7406 Diff 35370 python_modules/libraries/dagstermill/dagstermill_tests/test_solids.py

Changeset View

Standalone View

python_modules/libraries/dagstermill/dagstermill_tests/test_solids.py

	Show First 20 Lines • Show All 60 Lines • ▼ Show 20 Lines
	@contextmanager			@contextmanager
	def exec_for_test(fn_name, env=None, raise_on_error=True, **kwargs):			def exec_for_test(fn_name, env=None, raise_on_error=True, **kwargs):
	result = None			result = None
	recon_pipeline = ReconstructablePipeline.for_module("dagstermill.examples.repository", fn_name)			recon_pipeline = ReconstructablePipeline.for_module("dagstermill.examples.repository", fn_name)

	with instance_for_test() as instance:			with instance_for_test() as instance:
	try:			try:
	result = execute_pipeline(			result = execute_pipeline(
	recon_pipeline, env, instance=instance, raise_on_error=raise_on_error, **kwargs			recon_pipeline,
				env,
				instance=instance,
				raise_on_error=raise_on_error,
				**kwargs,
	)			)
	yield result			yield result
	finally:			finally:
	if result:			if result:
	cleanup_result_notebook(result)			cleanup_result_notebook(result)


	@pytest.mark.notebook_test			@pytest.mark.notebook_test
	▲ Show 20 Lines • Show All 351 Lines • ▼ Show 20 Lines


	def test_custom_description():			def test_custom_description():
	test_description = "custom description"			test_description = "custom description"
	test_solid = define_dagstermill_solid(			test_solid = define_dagstermill_solid(
	BACKING_NB_NAME, BACKING_NB_PATH, description=test_description			BACKING_NB_NAME, BACKING_NB_PATH, description=test_description
	)			)
	assert test_solid.description == test_description			assert test_solid.description == test_description


				@pytest.mark.notebook_test
				def test_retries(capsys):
				with exec_for_test("retries_pipeline", raise_on_error=False) as result:

				assert result.result_for_solid("yield_retry").retry_attempts == 1

				# the raise_retry solid should trigger a warning to use yield_event
				warn_found = False
				captured = capsys.readouterr()
				for line in captured.err.split("\n"):
				if "Use dagstermill.yield_event with RetryRequested or Failure" in line:
				warn_found = True

				assert warn_found


				@pytest.mark.notebook_test
				def test_failure(capsys):
				with exec_for_test("failure_pipeline", raise_on_error=False) as result:
				assert (
				result.result_for_solid("yield_failure").failure_data.user_failure_data.description
				== "bad bad notebook"
				)

				# the raise_failure solid should trigger a warning to use yield_event
				warn_found = False
				captured = capsys.readouterr()
				for line in captured.err.split("\n"):
				if "Use dagstermill.yield_event with RetryRequested or Failure" in line:
				warn_found = True

				assert warn_found