Changeset View
Changeset View
Standalone View
Standalone View
docs/content/concepts/testing.mdx
| Show First 20 Lines • Show All 88 Lines • ▼ Show 20 Lines | |||||
| @solid | @solid | ||||
| def do_something(context, data): | def do_something(context, data): | ||||
| output = process(data) | output = process(data) | ||||
| return output | return output | ||||
| @pipeline( | @pipeline( | ||||
| mode_defs=[ | mode_defs=[ | ||||
| ModeDefinition(name="unit_test", resource_defs={"api": ResourceDefinition.mock_resource()}), | ModeDefinition( | ||||
| name="unit_test", | |||||
| resource_defs={"api": ResourceDefinition.mock_resource()}, | |||||
| ), | |||||
| ModeDefinition(name="prod", resource_defs={"api": api_client}), | ModeDefinition(name="prod", resource_defs={"api": api_client}), | ||||
| ] | ] | ||||
| ) | ) | ||||
| def download_pipeline(): | def download_pipeline(): | ||||
| do_something(get_data()) | do_something(get_data()) | ||||
| ``` | ``` | ||||
| In this example, the business logic (i.e., pipelines and solids) remains the same in two different environments (i.e., resources), while the `api` resource gets mocked in the test mode using a helper method [`mock_resource`](/\_apidocs/modes-resources#dagster.ResourceDefinition.mock_resource) from the <PyObject object="ResourceDefinition" /> class. | In this example, the business logic (i.e., pipelines and solids) remains the same in two different environments (i.e., resources), while the `api` resource gets mocked in the test mode using a helper method [`mock_resource`](/\_apidocs/modes-resources#dagster.ResourceDefinition.mock_resource) from the <PyObject object="ResourceDefinition" /> class. | ||||
| ▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines | |||||
| ```python file=/concepts/solids_pipelines/unit_tests.py startafter=start_test_resource_def_marker endbefore=end_test_resource_def_marker | ```python file=/concepts/solids_pipelines/unit_tests.py startafter=start_test_resource_def_marker endbefore=end_test_resource_def_marker | ||||
| @resource(config_schema={"my_str": str}) | @resource(config_schema={"my_str": str}) | ||||
| def my_foo_resource(context): | def my_foo_resource(context): | ||||
| return context.resource_config["my_str"] | return context.resource_config["my_str"] | ||||
| def test_solid_resource_def(): | def test_solid_resource_def(): | ||||
| context = build_solid_context(resources={"foo": my_foo_resource.configured({"my_str": "bar"})}) | context = build_solid_context( | ||||
| resources={"foo": my_foo_resource.configured({"my_str": "bar"})} | |||||
| ) | |||||
| assert solid_requires_foo(context) == "found bar" | assert solid_requires_foo(context) == "found bar" | ||||
| ``` | ``` | ||||
| ### Testing pipeline execution with configs | ### Testing pipeline execution with configs | ||||
| Sometimes, you may want to test with different configuration. You can execute pipeline with a specified run config via the `run_config`: | Sometimes, you may want to test with different configuration. You can execute pipeline with a specified run config via the `run_config`: | ||||
| ```python file=/concepts/solids_pipelines/unit_tests.py startafter=start_test_pipeline_with_config endbefore=end_test_pipeline_with_config | ```python file=/concepts/solids_pipelines/unit_tests.py startafter=start_test_pipeline_with_config endbefore=end_test_pipeline_with_config | ||||
| def test_pipeline_with_config(): | def test_pipeline_with_config(): | ||||
| result = execute_pipeline( | result = execute_pipeline( | ||||
| do_math, | do_math, | ||||
| run_config={ | run_config={ | ||||
| "solids": {"add_one": {"inputs": {"num": 2}}, "add_two": {"inputs": {"num": 3}}} | "solids": { | ||||
| "add_one": {"inputs": {"num": 2}}, | |||||
| "add_two": {"inputs": {"num": 3}}, | |||||
| } | |||||
| }, | }, | ||||
| ) | ) | ||||
| assert result.success | assert result.success | ||||
| assert result.output_for_solid("add_one") == 3 | assert result.output_for_solid("add_one") == 3 | ||||
| assert result.output_for_solid("add_two") == 5 | assert result.output_for_solid("add_two") == 5 | ||||
| assert result.output_for_solid("subtract") == -2 | assert result.output_for_solid("subtract") == -2 | ||||
| Show All 11 Lines | def test_subset_execution(): | ||||
| ) | ) | ||||
| assert result.success | assert result.success | ||||
| assert result.output_for_solid("add_one") == 2 | assert result.output_for_solid("add_one") == 2 | ||||
| assert result.output_for_solid("add_two") == 3 | assert result.output_for_solid("add_two") == 3 | ||||
| # solid_result_list returns List[SolidExecutionResult] | # solid_result_list returns List[SolidExecutionResult] | ||||
| # this checks to see that only two were executed | # this checks to see that only two were executed | ||||
| assert {solid_result.solid.name for solid_result in result.solid_result_list} == { | assert { | ||||
| solid_result.solid.name for solid_result in result.solid_result_list | |||||
| } == { | |||||
| "add_one", | "add_one", | ||||
| "add_two", | "add_two", | ||||
| } | } | ||||
| ``` | ``` | ||||
| ### Testing event stream | ### Testing event stream | ||||
| The event stream is the most generic way that a solid communicates what happened during its computation. Solids communicate events for starting, input/output type checking, and user-provided events such as expectations, materializations, and outputs. | The event stream is the most generic way that a solid communicates what happened during its computation. Solids communicate events for starting, input/output type checking, and user-provided events such as expectations, materializations, and outputs. | ||||
| ```python file=/concepts/solids_pipelines/unit_tests.py startafter=start_test_event_stream endbefore=end_test_event_stream | ```python file=/concepts/solids_pipelines/unit_tests.py startafter=start_test_event_stream endbefore=end_test_event_stream | ||||
| def test_event_stream(): | def test_event_stream(): | ||||
| pipeline_result = execute_pipeline( | pipeline_result = execute_pipeline( | ||||
| emit_events_pipeline, {"solids": {"emit_events_solid": {"inputs": {"input_num": 1}}}} | emit_events_pipeline, | ||||
| {"solids": {"emit_events_solid": {"inputs": {"input_num": 1}}}}, | |||||
| ) | ) | ||||
| assert pipeline_result.success | assert pipeline_result.success | ||||
| solid_result = pipeline_result.result_for_solid("emit_events_solid") | solid_result = pipeline_result.result_for_solid("emit_events_solid") | ||||
| assert isinstance(solid_result, SolidExecutionResult) | assert isinstance(solid_result, SolidExecutionResult) | ||||
| # when one has multiple outputs, you need to specify output name | # when one has multiple outputs, you need to specify output name | ||||
| Show All 16 Lines | ( | ||||
| expectation_event, | expectation_event, | ||||
| materialization_event, | materialization_event, | ||||
| _num_output_event, | _num_output_event, | ||||
| _num_handled_output_operation, | _num_handled_output_operation, | ||||
| _success, | _success, | ||||
| ) = solid_result.step_events | ) = solid_result.step_events | ||||
| # apologies for verboseness here! we can do better. | # apologies for verboseness here! we can do better. | ||||
| expectation_result = expectation_event.event_specific_data.expectation_result | expectation_result = ( | ||||
| expectation_event.event_specific_data.expectation_result | |||||
| ) | |||||
| assert isinstance(expectation_result, ExpectationResult) | assert isinstance(expectation_result, ExpectationResult) | ||||
| assert expectation_result.success | assert expectation_result.success | ||||
| assert expectation_result.label == "positive" | assert expectation_result.label == "positive" | ||||
| materialization = materialization_event.event_specific_data.materialization | materialization = ( | ||||
| materialization_event.event_specific_data.materialization | |||||
| ) | |||||
| assert isinstance(materialization, AssetMaterialization) | assert isinstance(materialization, AssetMaterialization) | ||||
| assert materialization.label == "persisted_string" | assert materialization.label == "persisted_string" | ||||
| ``` | ``` | ||||