Page MenuHomePhabricator
Differential D4761 Diff 23634 python_modules/dagster/dagster_tests/api_tests/test_api_snapshot_partition.py

Changeset View

Standalone View

View Options

python_modules/dagster/dagster_tests/api_tests/test_api_snapshot_partition.py

Show All 11 Lines
) )
from dagster.core.host_representation import ( from dagster.core.host_representation import (
ExternalPartitionConfigData, ExternalPartitionConfigData,
ExternalPartitionExecutionErrorData, ExternalPartitionExecutionErrorData,
ExternalPartitionNamesData, ExternalPartitionNamesData,
ExternalPartitionSetExecutionParamData, ExternalPartitionSetExecutionParamData,
ExternalPartitionTagsData, ExternalPartitionTagsData,
) )
from dagster.core.test_utils import instance_for_test
from .utils import get_bar_grpc_repo_handle, get_bar_repo_handle from .utils import get_bar_grpc_repo_handle, get_bar_repo_handle
def test_external_partition_names(): def test_external_partition_names():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
data = sync_get_external_partition_names(repository_handle, "baz_partitions") with instance_for_test() as instance:
data = sync_get_external_partition_names(instance, repository_handle, "baz_partitions")
assert isinstance(data, ExternalPartitionNamesData) assert isinstance(data, ExternalPartitionNamesData)
assert data.partition_names == list(string.ascii_lowercase) assert data.partition_names == list(string.ascii_lowercase)
def test_external_partition_names_error(): def test_external_partition_names_error():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
error = sync_get_external_partition_names(repository_handle, "error_partitions") with instance_for_test() as instance:
error = sync_get_external_partition_names(instance, repository_handle, "error_partitions")
assert isinstance(error, ExternalPartitionExecutionErrorData) assert isinstance(error, ExternalPartitionExecutionErrorData)
assert "womp womp" in error.error.to_string() assert "womp womp" in error.error.to_string()
def test_external_partition_names_grpc(): def test_external_partition_names_grpc():
with get_bar_grpc_repo_handle() as repository_handle: with get_bar_grpc_repo_handle() as repository_handle:
with instance_for_test() as instance:
data = sync_get_external_partition_names_grpc( data = sync_get_external_partition_names_grpc(
repository_handle.repository_location_handle.client, repository_handle, "baz_partitions" repository_handle.repository_location_handle.client,
instance,
repository_handle,
"baz_partitions",
) )
assert isinstance(data, ExternalPartitionNamesData) assert isinstance(data, ExternalPartitionNamesData)
assert data.partition_names == list(string.ascii_lowercase) assert data.partition_names == list(string.ascii_lowercase)
def test_external_partition_names_error_grpc(): def test_external_partition_names_error_grpc():
with get_bar_grpc_repo_handle() as repository_handle: with get_bar_grpc_repo_handle() as repository_handle:
with instance_for_test() as instance:
error = sync_get_external_partition_names_grpc( error = sync_get_external_partition_names_grpc(
repository_handle.repository_location_handle.client, repository_handle.repository_location_handle.client,
instance,
repository_handle, repository_handle,
"error_partitions", "error_partitions",
) )
assert isinstance(error, ExternalPartitionExecutionErrorData) assert isinstance(error, ExternalPartitionExecutionErrorData)
assert "womp womp" in error.error.to_string() assert "womp womp" in error.error.to_string()
def test_external_partitions_config(): def test_external_partitions_config():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
data = sync_get_external_partition_config(repository_handle, "baz_partitions", "c") with instance_for_test() as instance:
data = sync_get_external_partition_config(
instance, repository_handle, "baz_partitions", "c"
)
assert isinstance(data, ExternalPartitionConfigData) assert isinstance(data, ExternalPartitionConfigData)
assert data.run_config assert data.run_config
assert data.run_config["solids"]["do_input"]["inputs"]["x"]["value"] == "c" assert data.run_config["solids"]["do_input"]["inputs"]["x"]["value"] == "c"
def test_external_partitions_config_error(): def test_external_partitions_config_error():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
error = sync_get_external_partition_config(repository_handle, "error_partition_config", "c") with instance_for_test() as instance:
error = sync_get_external_partition_config(
instance, repository_handle, "error_partition_config", "c"
)
assert isinstance(error, ExternalPartitionExecutionErrorData) assert isinstance(error, ExternalPartitionExecutionErrorData)
def test_external_partitions_config_grpc(): def test_external_partitions_config_grpc():
with get_bar_grpc_repo_handle() as repository_handle: with get_bar_grpc_repo_handle() as repository_handle:
with instance_for_test() as instance:
data = sync_get_external_partition_config_grpc( data = sync_get_external_partition_config_grpc(
repository_handle.repository_location_handle.client, repository_handle.repository_location_handle.client,
instance,
repository_handle, repository_handle,
"baz_partitions", "baz_partitions",
"c", "c",
) )
assert isinstance(data, ExternalPartitionConfigData) assert isinstance(data, ExternalPartitionConfigData)
assert data.run_config assert data.run_config
assert data.run_config["solids"]["do_input"]["inputs"]["x"]["value"] == "c" assert data.run_config["solids"]["do_input"]["inputs"]["x"]["value"] == "c"
def test_external_partitions_config_error_grpc(): def test_external_partitions_config_error_grpc():
with get_bar_grpc_repo_handle() as repository_handle: with get_bar_grpc_repo_handle() as repository_handle:
with instance_for_test() as instance:
error = sync_get_external_partition_config_grpc( error = sync_get_external_partition_config_grpc(
repository_handle.repository_location_handle.client, repository_handle.repository_location_handle.client,
instance,
repository_handle, repository_handle,
"error_partition_config", "error_partition_config",
"c", "c",
) )
assert isinstance(error, ExternalPartitionExecutionErrorData) assert isinstance(error, ExternalPartitionExecutionErrorData)
def test_external_partitions_tags(): def test_external_partitions_tags():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
data = sync_get_external_partition_tags(repository_handle, "baz_partitions", "c") with instance_for_test() as instance:
data = sync_get_external_partition_tags(instance, repository_handle, "baz_partitions", "c")
assert isinstance(data, ExternalPartitionTagsData) assert isinstance(data, ExternalPartitionTagsData)
assert data.tags assert data.tags
assert data.tags["foo"] == "bar" assert data.tags["foo"] == "bar"
def test_external_partitions_tags_error(): def test_external_partitions_tags_error():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
error = sync_get_external_partition_tags(repository_handle, "error_partition_tags", "c") with instance_for_test() as instance:
error = sync_get_external_partition_tags(
instance, repository_handle, "error_partition_tags", "c"
)
assert isinstance(error, ExternalPartitionExecutionErrorData) assert isinstance(error, ExternalPartitionExecutionErrorData)
def test_external_partitions_tags_grpc(): def test_external_partitions_tags_grpc():
with get_bar_grpc_repo_handle() as repository_handle: with get_bar_grpc_repo_handle() as repository_handle:
with instance_for_test() as instance:
data = sync_get_external_partition_tags_grpc( data = sync_get_external_partition_tags_grpc(
repository_handle.repository_location_handle.client, repository_handle.repository_location_handle.client,
instance,
repository_handle, repository_handle,
"baz_partitions", "baz_partitions",
"c", "c",
) )
assert isinstance(data, ExternalPartitionTagsData) assert isinstance(data, ExternalPartitionTagsData)
assert data.tags assert data.tags
assert data.tags["foo"] == "bar" assert data.tags["foo"] == "bar"
def test_external_partitions_tags_error_grpc(): def test_external_partitions_tags_error_grpc():
with get_bar_grpc_repo_handle() as repository_handle: with get_bar_grpc_repo_handle() as repository_handle:
with instance_for_test() as instance:
error = sync_get_external_partition_tags_grpc( error = sync_get_external_partition_tags_grpc(
repository_handle.repository_location_handle.client, repository_handle.repository_location_handle.client,
instance,
repository_handle, repository_handle,
"error_partition_tags", "error_partition_tags",
"c", "c",
) )
assert isinstance(error, ExternalPartitionExecutionErrorData) assert isinstance(error, ExternalPartitionExecutionErrorData)
def test_external_partition_set_execution_params(): def test_external_partition_set_execution_params():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
with instance_for_test() as instance:
data = sync_get_external_partition_set_execution_param_data( data = sync_get_external_partition_set_execution_param_data(
repository_handle, "baz_partitions", ["a", "b", "c"] instance, repository_handle, "baz_partitions", ["a", "b", "c"]
) )
assert isinstance(data, ExternalPartitionSetExecutionParamData) assert isinstance(data, ExternalPartitionSetExecutionParamData)
assert len(data.partition_data) == 3 assert len(data.partition_data) == 3
def test_external_partition_set_execution_params_grpc(): def test_external_partition_set_execution_params_grpc():
with get_bar_grpc_repo_handle() as repository_handle: with get_bar_grpc_repo_handle() as repository_handle:
with instance_for_test() as instance:
data = sync_get_external_partition_set_execution_param_data_grpc( data = sync_get_external_partition_set_execution_param_data_grpc(
repository_handle.repository_location_handle.client, repository_handle.repository_location_handle.client,
instance,
repository_handle, repository_handle,
"baz_partitions", "baz_partitions",
["a", "b", "c"], ["a", "b", "c"],
) )
assert isinstance(data, ExternalPartitionSetExecutionParamData) assert isinstance(data, ExternalPartitionSetExecutionParamData)
assert len(data.partition_data) == 3 assert len(data.partition_data) == 3
def test_external_partition_set_execution_params_error(): def test_external_partition_set_execution_params_error():
repository_handle = get_bar_repo_handle() repository_handle = get_bar_repo_handle()
with instance_for_test() as instance:
error = sync_get_external_partition_set_execution_param_data( error = sync_get_external_partition_set_execution_param_data(
repository_handle, "error_partitions", ["a", "b", "c"] instance, repository_handle, "error_partitions", ["a", "b", "c"]
) )
assert isinstance(error, ExternalPartitionExecutionErrorData) assert isinstance(error, ExternalPartitionExecutionErrorData)
assert "womp womp" in error.error.to_string() assert "womp womp" in error.error.to_string()