(graphql-context-test-suite-13) Move one test over to hijacking start…

…. Found a bug

Summary:
This diff is a good demonstration of why both duplicated code paths are bad
and why APIs like create_run are way too low level and should not exist. One should
be forced to have a higher level API where you must pass an ExternalPipeline rather
than force duplicated logic at every callsite.

Depends on D3083

Test Plan: BK

Reviewers: alangenfeld, max

Reviewed By: alangenfeld

Differential Revision: https://dagster.phacility.com/D3087

python_modules/dagster-graphql/dagster_graphql/implementation/execution/launch_execution.py

@@ -10,6 +10,7 @@
 from dagster.core.events import EngineEventData
 from dagster.core.storage.pipeline_run import PipelineRunStatus
 from dagster.core.utils import make_new_run_id
+from dagster.utils import merge_dicts
 from dagster.utils.error import SerializableErrorInfo, serializable_error_info_from_exc_info
 
 from ..external import (
@@ -88,7 +89,7 @@ def do_launch(graphene_info, execution_params, is_reexecuted=False):
         environment_dict=execution_params.environment_dict,
         mode=execution_params.mode,
         step_keys_to_execute=step_keys_to_execute,
-        tags=execution_params.execution_metadata.tags,
+        tags=merge_dicts(external_pipeline.tags, execution_params.execution_metadata.tags),
         root_run_id=execution_params.execution_metadata.root_run_id,
         parent_run_id=execution_params.execution_metadata.parent_run_id,
         status=PipelineRunStatus.NOT_STARTED,

python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_execute_pipeline.py

@@ -79,27 +79,30 @@ def test_basic_start_pipeline_execution_with_preset(self, graphql_context):
         assert uuid.UUID(result.data['startPipelineExecution']['run']['runId'])
         assert result.data['startPipelineExecution']['run']['pipeline']['name'] == 'csv_hello_world'
 
+    def test_basic_start_pipeline_execution_with_pipeline_def_tags(self, graphql_context):
+        result = execute_dagster_graphql(
+            graphql_context,
+            START_PIPELINE_EXECUTION_QUERY,
+            variables={
+                'executionParams': {
+                    'selector': {'name': 'hello_world_with_tags'},
+                    'mode': 'default',
+                },
+            },
+        )
 
-def test_basic_start_pipeline_execution_with_pipeline_def_tags(graphql_context):
-    result = execute_dagster_graphql(
-        graphql_context,
-        START_PIPELINE_EXECUTION_QUERY,
-        variables={
-            'executionParams': {'selector': {'name': 'hello_world_with_tags'}, 'mode': 'default',},
-        },
-    )
-
-    assert not result.errors
-    assert result.data['startPipelineExecution']['run']['tags'] == [
-        {'key': 'tag_key', 'value': 'tag_value'}
-    ]
+        assert not result.errors
+        assert result.data['startPipelineExecution']['run']['tags'] == [
+            {'key': 'tag_key', 'value': 'tag_value'}
+        ]
 
-    # just test existence
-    assert result.data['startPipelineExecution']['__typename'] == 'StartPipelineRunSuccess'
-    assert uuid.UUID(result.data['startPipelineExecution']['run']['runId'])
-    assert (
-        result.data['startPipelineExecution']['run']['pipeline']['name'] == 'hello_world_with_tags'
-    )
+        # just test existence
+        assert result.data['startPipelineExecution']['__typename'] == 'StartPipelineRunSuccess'
+        assert uuid.UUID(result.data['startPipelineExecution']['run']['runId'])
+        assert (
+            result.data['startPipelineExecution']['run']['pipeline']['name']
+            == 'hello_world_with_tags'
+        )
 
 
 def test_basic_start_pipeline_execution_with_non_existent_preset(graphql_context):