Differential D8684 Diff 40807 examples/docs_snippets/docs_snippets_tests/concepts_tests/resources_tests/test_modes_resources.py

Changeset View

Standalone View

View Options

examples/docs_snippets/docs_snippets_tests/concepts_tests/resources_tests/test_modes_resources.py

	from dagster import build_init_resource_context, build_solid_context, execute_pipeline			from dagster import (
				build_init_resource_context,
				build_solid_context,
				execute_pipeline,
				)
	from docs_snippets.concepts.modes_resources.modes_resources import (			from docs_snippets.concepts.modes_resources.modes_resources import (
	cereal_fetcher,			cereal_fetcher,
	db_resource,			db_resource,
	emit_foo,			emit_foo,
	foo_resource,			foo_resource,
	pipeline_with_mode,			pipeline_with_mode,
	solid_requires_resources,			solid_requires_resources,
	test_cm_resource,			test_cm_resource,
	test_my_resource,			test_my_resource,
	test_my_resource_with_context,			test_my_resource_with_context,
	)			)


	def test_cereal_fetcher():			def test_cereal_fetcher():
	assert cereal_fetcher(None)			assert cereal_fetcher(None)


	def test_database_resource():			def test_database_resource():
	class BasicDatabase:			class BasicDatabase:
	def execute_query(self, query):			def execute_query(self, query):
	pass			pass

	solid_requires_resources(build_solid_context(resources={"database": BasicDatabase()}))			solid_requires_resources(
				build_solid_context(resources={"database": BasicDatabase()})
				)


	def test_resource_testing_examples():			def test_resource_testing_examples():
	test_my_resource()			test_my_resource()
	test_my_resource_with_context()			test_my_resource_with_context()
	test_cm_resource()			test_cm_resource()


	def test_pipeline_with_mode_example():			def test_pipeline_with_mode_example():
	result = execute_pipeline(pipeline_with_mode, mode="ab_mode")			result = execute_pipeline(pipeline_with_mode, mode="ab_mode")
	assert result.success			assert result.success

	result = execute_pipeline(pipeline_with_mode, mode="c_mode")			result = execute_pipeline(pipeline_with_mode, mode="c_mode")
	assert result.success			assert result.success


	def test_resource_dependencies_example():			def test_resource_dependencies_example():
	assert emit_foo(build_init_resource_context(resources={"foo": foo_resource})) == "foo"			assert (
				emit_foo(build_init_resource_context(resources={"foo": foo_resource}))
				== "foo"
				)


	def test_resource_config_example():			def test_resource_config_example():
	dbconn = db_resource(build_init_resource_context(config={"connection": "foo"}))			dbconn = db_resource(
				build_init_resource_context(config={"connection": "foo"})
				)
	assert dbconn.connection == "foo"			assert dbconn.connection == "foo"