Changeset View
Changeset View
Standalone View
Standalone View
python_modules/dagster-graphql/dagster_graphql_tests/test_cli.py
| import json | import json | ||||
| import os | import os | ||||
| import time | import time | ||||
| from contextlib import contextmanager | from contextlib import contextmanager | ||||
| from click.testing import CliRunner | from click.testing import CliRunner | ||||
| from dagster_graphql.cli import ui | from dagster_graphql.cli import ui | ||||
| from dagster import ( | from dagster import seven | ||||
| InputDefinition, | |||||
| Int, | |||||
| OutputDefinition, | |||||
| ScheduleDefinition, | |||||
| lambda_solid, | |||||
| pipeline, | |||||
| repository, | |||||
| seven, | |||||
| solid, | |||||
| ) | |||||
| from dagster.core.storage.pipeline_run import PipelineRunStatus | from dagster.core.storage.pipeline_run import PipelineRunStatus | ||||
| from dagster.core.test_utils import instance_for_test_tempdir | from dagster.core.test_utils import instance_for_test_tempdir | ||||
| from dagster.utils import file_relative_path | from dagster.utils import file_relative_path | ||||
| @contextmanager | @contextmanager | ||||
| def dagster_cli_runner(): | def dagster_cli_runner(): | ||||
| with seven.TemporaryDirectory() as dagster_home_temp: | with seven.TemporaryDirectory() as dagster_home_temp: | ||||
| with instance_for_test_tempdir( | with instance_for_test_tempdir( | ||||
| dagster_home_temp, | dagster_home_temp, | ||||
| overrides={ | overrides={ | ||||
| "run_launcher": { | "run_launcher": { | ||||
| "module": "dagster.core.launcher.sync_in_memory_run_launcher", | "module": "dagster.core.launcher.sync_in_memory_run_launcher", | ||||
| "class": "SyncInMemoryRunLauncher", | "class": "SyncInMemoryRunLauncher", | ||||
| } | } | ||||
| }, | }, | ||||
| ): | ): | ||||
| yield CliRunner(env={"DAGSTER_HOME": dagster_home_temp}) | yield CliRunner(env={"DAGSTER_HOME": dagster_home_temp}) | ||||
| @lambda_solid(input_defs=[InputDefinition("num", Int)], output_def=OutputDefinition(Int)) | |||||
| def add_one(num): | |||||
| return num + 1 | |||||
| @lambda_solid(input_defs=[InputDefinition("num", Int)], output_def=OutputDefinition(Int)) | |||||
| def mult_two(num): | |||||
| return num * 2 | |||||
| @pipeline | |||||
| def math(): | |||||
| mult_two(add_one()) | |||||
| @solid(config_schema={"gimme": str}) | |||||
| def needs_config(context): | |||||
| return context.solid_config["gimme"] | |||||
| @lambda_solid | |||||
| def no_config(): | |||||
| return "ok" | |||||
| @pipeline | |||||
| def subset_test(): | |||||
| no_config() | |||||
| needs_config() | |||||
| def define_schedules(): | |||||
| math_hourly_schedule = ScheduleDefinition( | |||||
| name="math_hourly_schedule", | |||||
| cron_schedule="0 0 * * *", | |||||
| pipeline_name="math", | |||||
| run_config={"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | |||||
| ) | |||||
| return [math_hourly_schedule] | |||||
| @repository | |||||
| def test(): | |||||
| return [math, subset_test] + define_schedules() | |||||
| def test_basic_introspection(): | def test_basic_introspection(): | ||||
| query = "{ __schema { types { name } } }" | query = "{ __schema { types { name } } }" | ||||
| workspace_path = file_relative_path(__file__, "./cli_test_workspace.yaml") | workspace_path = file_relative_path(__file__, "./cli_test_workspace.yaml") | ||||
| with dagster_cli_runner() as runner: | with dagster_cli_runner() as runner: | ||||
| result = runner.invoke(ui, ["-w", workspace_path, "-t", query]) | result = runner.invoke(ui, ["-w", workspace_path, "-t", query]) | ||||
| assert result.exit_code == 0 | assert result.exit_code == 0 | ||||
| Show All 18 Lines | |||||
| def test_basic_variables(): | def test_basic_variables(): | ||||
| query = """ | query = """ | ||||
| query FooBar($pipelineName: String! $repositoryName: String! $repositoryLocationName: String!){ | query FooBar($pipelineName: String! $repositoryName: String! $repositoryLocationName: String!){ | ||||
| pipelineOrError(params:{pipelineName: $pipelineName repositoryName: $repositoryName repositoryLocationName: $repositoryLocationName}) | pipelineOrError(params:{pipelineName: $pipelineName repositoryName: $repositoryName repositoryLocationName: $repositoryLocationName}) | ||||
| { ... on Pipeline { name } } | { ... on Pipeline { name } } | ||||
| } | } | ||||
| """ | """ | ||||
| variables = '{"pipelineName": "math", "repositoryName": "test", "repositoryLocationName": "<<in_process>>"}' | variables = '{"pipelineName": "math", "repositoryName": "test", "repositoryLocationName": "test_cli_location"}' | ||||
| workspace_path = file_relative_path(__file__, "./cli_test_workspace.yaml") | workspace_path = file_relative_path(__file__, "./cli_test_workspace.yaml") | ||||
| with dagster_cli_runner() as runner: | with dagster_cli_runner() as runner: | ||||
| result = runner.invoke(ui, ["-w", workspace_path, "-v", variables, "-t", query]) | result = runner.invoke(ui, ["-w", workspace_path, "-v", variables, "-t", query]) | ||||
| assert result.exit_code == 0 | assert result.exit_code == 0 | ||||
| result_data = json.loads(result.output) | result_data = json.loads(result.output) | ||||
| Show All 26 Lines | |||||
| """ | """ | ||||
| def test_start_execution_text(): | def test_start_execution_text(): | ||||
| variables = seven.json.dumps( | variables = seven.json.dumps( | ||||
| { | { | ||||
| "executionParams": { | "executionParams": { | ||||
| "selector": { | "selector": { | ||||
| "repositoryLocationName": "<<in_process>>", | "repositoryLocationName": "test_cli_location", | ||||
| "repositoryName": "test", | "repositoryName": "test", | ||||
| "pipelineName": "math", | "pipelineName": "math", | ||||
| }, | }, | ||||
| "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | ||||
| "mode": "default", | "mode": "default", | ||||
| } | } | ||||
| } | } | ||||
| ) | ) | ||||
| Show All 18 Lines | |||||
| def test_start_execution_file(): | def test_start_execution_file(): | ||||
| variables = seven.json.dumps( | variables = seven.json.dumps( | ||||
| { | { | ||||
| "executionParams": { | "executionParams": { | ||||
| "selector": { | "selector": { | ||||
| "pipelineName": "math", | "pipelineName": "math", | ||||
| "repositoryLocationName": "<<in_process>>", | "repositoryLocationName": "test_cli_location", | ||||
| "repositoryName": "test", | "repositoryName": "test", | ||||
| }, | }, | ||||
| "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | ||||
| "mode": "default", | "mode": "default", | ||||
| } | } | ||||
| } | } | ||||
| ) | ) | ||||
| Show All 23 Lines | def test_start_execution_save_output(): | ||||
| """ | """ | ||||
| Test that the --output flag saves the GraphQL response to the specified file | Test that the --output flag saves the GraphQL response to the specified file | ||||
| """ | """ | ||||
| variables = seven.json.dumps( | variables = seven.json.dumps( | ||||
| { | { | ||||
| "executionParams": { | "executionParams": { | ||||
| "selector": { | "selector": { | ||||
| "repositoryLocationName": "<<in_process>>", | "repositoryLocationName": "test_cli_location", | ||||
| "repositoryName": "test", | "repositoryName": "test", | ||||
| "pipelineName": "math", | "pipelineName": "math", | ||||
| }, | }, | ||||
| "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | ||||
| "mode": "default", | "mode": "default", | ||||
| } | } | ||||
| } | } | ||||
| ) | ) | ||||
| Show All 30 Lines | with dagster_cli_runner() as runner: | ||||
| ) | ) | ||||
| def test_start_execution_predefined(): | def test_start_execution_predefined(): | ||||
| variables = seven.json.dumps( | variables = seven.json.dumps( | ||||
| { | { | ||||
| "executionParams": { | "executionParams": { | ||||
| "selector": { | "selector": { | ||||
| "repositoryLocationName": "<<in_process>>", | "repositoryLocationName": "test_cli_location", | ||||
| "repositoryName": "test", | "repositoryName": "test", | ||||
| "pipelineName": "math", | "pipelineName": "math", | ||||
| }, | }, | ||||
| "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | ||||
| "mode": "default", | "mode": "default", | ||||
| } | } | ||||
| } | } | ||||
| ) | ) | ||||
| Show All 14 Lines | with dagster_cli_runner() as runner: | ||||
| ) | ) | ||||
| def test_logs_in_start_execution_predefined(): | def test_logs_in_start_execution_predefined(): | ||||
| variables = seven.json.dumps( | variables = seven.json.dumps( | ||||
| { | { | ||||
| "executionParams": { | "executionParams": { | ||||
| "selector": { | "selector": { | ||||
| "repositoryLocationName": "<<in_process>>", | "repositoryLocationName": "test_cli_location", | ||||
| "repositoryName": "test", | "repositoryName": "test", | ||||
| "pipelineName": "math", | "pipelineName": "math", | ||||
| }, | }, | ||||
| "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | "runConfigData": {"solids": {"add_one": {"inputs": {"num": {"value": 123}}}}}, | ||||
| "mode": "default", | "mode": "default", | ||||
| } | } | ||||
| } | } | ||||
| ) | ) | ||||
| Show All 38 Lines | |||||