File: test_ingestion_properties.py

package info (click to toggle)
azure-kusto-python 5.0.5-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 6,704 kB
  • sloc: python: 10,633; sh: 13; makefile: 3
file content (92 lines) | stat: -rw-r--r-- 3,215 bytes parent folder | download | duplicates (3)
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,
    )