1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92
|
import pytest
from azure.kusto.data.data_format import IngestionMappingKind, DataFormat
from azure.kusto.ingest import IngestionProperties, ColumnMapping, TransformationMethod
from azure.kusto.ingest.exceptions import KustoDuplicateMappingError, KustoMissingMappingError, KustoMappingError
def test_duplicate_reference_and_column_mappings_raises():
"""Tests invalid ingestion properties."""
with pytest.raises(KustoDuplicateMappingError):
IngestionProperties(
database="database", table="table", column_mappings=[ColumnMapping("test", "int")], ingestion_mapping_reference="ingestionMappingReference"
)
def test_mapping_kind_without_mapping_raises():
with pytest.raises(KustoMissingMappingError):
IngestionProperties(database="database", table="table", ingestion_mapping_kind=IngestionMappingKind.CSV)
def test_mapping_kind_data_format_mismatch():
with pytest.raises(KustoMappingError):
IngestionProperties(
database="database",
table="table",
ingestion_mapping_reference="ingestionMappingReference",
data_format=DataFormat.JSON,
ingestion_mapping_kind=IngestionMappingKind.CSV,
)
def test_mapping_kind_data_format_invalid_no_name():
with pytest.raises(KustoMappingError):
IngestionProperties(
database="database",
table="table",
column_mappings=[ColumnMapping("", "int")],
data_format=DataFormat.JSON,
ingestion_mapping_kind=IngestionMappingKind.JSON,
)
def test_mapping_kind_data_format_invalid_no_path():
with pytest.raises(KustoMappingError):
IngestionProperties(
database="database",
table="table",
column_mappings=[ColumnMapping("test", "int")],
data_format=DataFormat.JSON,
ingestion_mapping_kind=IngestionMappingKind.JSON,
)
def test_mapping_kind_data_format_with_path():
IngestionProperties(
database="database",
table="table",
column_mappings=[ColumnMapping("test", "int", "path")],
data_format=DataFormat.JSON,
ingestion_mapping_kind=IngestionMappingKind.JSON,
)
def test_mapping_kind_data_format_with_transform():
IngestionProperties(
database="database",
table="table",
column_mappings=[ColumnMapping("test", "int", transform=TransformationMethod.SOURCE_LINE_NUMBER)],
data_format=DataFormat.JSON,
ingestion_mapping_kind=IngestionMappingKind.JSON,
)
def test_mapping_kind_data_format_with_no_properties():
with pytest.raises(KustoMappingError):
IngestionProperties(
database="database",
table="table",
column_mappings=[ColumnMapping("test", "int")],
data_format=DataFormat.AVRO,
ingestion_mapping_kind=IngestionMappingKind.AVRO,
)
def test_with_constant_value():
IngestionProperties(
database="database",
table="table",
column_mappings=[ColumnMapping("test", "int", const_value="1")],
data_format=DataFormat.PARQUET,
ingestion_mapping_kind=IngestionMappingKind.PARQUET,
)
|