Differential D4920 Diff 24623 python_modules/dagster/dagster_tests/cli_tests/command_tests/test_launch_command.py

Changeset View

Standalone View

python_modules/dagster/dagster_tests/cli_tests/command_tests/test_launch_command.py

from __future__ import print_function		from __future__ import print_function

import re		import re

import pytest		import pytest
from click.testing import CliRunner		from click.testing import CliRunner

from dagster.cli.pipeline import execute_launch_command, pipeline_launch_command		from dagster.cli.pipeline import execute_launch_command, pipeline_launch_command
		from dagster.core.errors import DagsterUserCodeProcessError
from dagster.core.test_utils import new_cwd		from dagster.core.test_utils import new_cwd
from dagster.utils import file_relative_path		from dagster.utils import file_relative_path

from .test_cli_commands import (		from .test_cli_commands import (
default_cli_test_instance,		default_cli_test_instance,
grpc_server_bar_cli_args,		grpc_server_bar_cli_args,
launch_command_contexts,		launch_command_contexts,
		non_existant_python_origin_target_args,
python_bar_cli_args,		python_bar_cli_args,
valid_external_pipeline_target_cli_args_with_preset,		valid_external_pipeline_target_cli_args_with_preset,
)		)


def run_launch(kwargs, instance, expected_count=None):		def run_launch(kwargs, instance, expected_count=None):
run = execute_launch_command(instance, kwargs)		run = execute_launch_command(instance, kwargs)
assert run		assert run
Show All 20 Lines	with gen_pipeline_args as (cli_args, uses_legacy_repository_yaml_format, instance):
"You are using the legacy repository yaml format. Please update your file "		"You are using the legacy repository yaml format. Please update your file "
),		),
):		):
run_launch(cli_args, instance, expected_count=1)		run_launch(cli_args, instance, expected_count=1)
else:		else:
run_launch(cli_args, instance, expected_count=1)		run_launch(cli_args, instance, expected_count=1)


		def test_launch_non_existant_file():
		with default_cli_test_instance() as instance:
		kwargs = non_existant_python_origin_target_args()

		with pytest.raises(DagsterUserCodeProcessError):
		run_launch(kwargs, instance)


@pytest.mark.parametrize("pipeline_cli_args", valid_external_pipeline_target_cli_args_with_preset())		@pytest.mark.parametrize("pipeline_cli_args", valid_external_pipeline_target_cli_args_with_preset())
def test_launch_pipeline_cli(pipeline_cli_args):		def test_launch_pipeline_cli(pipeline_cli_args):
with default_cli_test_instance() as instance:		with default_cli_test_instance() as instance:
cli_args, uses_legacy_repository_yaml_format = pipeline_cli_args		cli_args, uses_legacy_repository_yaml_format = pipeline_cli_args
if uses_legacy_repository_yaml_format:		if uses_legacy_repository_yaml_format:
with pytest.warns(		with pytest.warns(
UserWarning,		UserWarning,
match=re.escape(		match=re.escape(
▲ Show 20 Lines • Show All 110 Lines • Show Last 20 Lines