# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License
import asyncio
import io
import os
import pathlib
import platform
import pytest
import random
import sys
import time
import uuid
from azure.identity import DefaultAzureCredential
from typing import Optional, ClassVar, Callable

from azure.kusto.data.env_utils import get_env, prepare_app_key_auth
from azure.kusto.data import KustoClient, KustoConnectionStringBuilder
from azure.kusto.data.aio import KustoClient as AsyncKustoClient
from azure.kusto.data.data_format import DataFormat, IngestionMappingKind
from azure.kusto.data.exceptions import KustoServiceError
from azure.kusto.ingest import (
    QueuedIngestClient,
    KustoStreamingIngestClient,
    IngestionProperties,
    ColumnMapping,
    ValidationPolicy,
    ValidationOptions,
    ValidationImplications,
    ReportLevel,
    ReportMethod,
    FileDescriptor,
    BlobDescriptor,
    StreamDescriptor,
    ManagedStreamingIngestClient,
)
from azure.kusto.ingest._resource_manager import _ResourceUri


@pytest.fixture(params=["ManagedStreaming", "NormalClient"])
def is_managed_streaming(request):
    return request.param == "ManagedStreaming"


@pytest.fixture(scope="module")
def event_loop_policy(request):
    if platform.system() == "Windows":
        return asyncio.WindowsSelectorEventLoopPolicy()
    return asyncio.DefaultEventLoopPolicy()


class TestE2E:
    """A class to define mappings to deft table."""

    input_folder_path: ClassVar[str]
    engine_cs: ClassVar[Optional[str]]
    dm_cs: ClassVar[Optional[str]]
    test_db: ClassVar[Optional[str]]
    client: ClassVar[KustoClient]
    test_table: ClassVar[str]
    current_count: ClassVar[int]
    ingest_client: ClassVar[QueuedIngestClient]
    streaming_ingest_client: ClassVar[KustoStreamingIngestClient]
    managed_streaming_ingest_client: ClassVar[ManagedStreamingIngestClient]
    csv_file_path: ClassVar[str]
    tsv_file_path: ClassVar[str]
    zipped_csv_file_path: ClassVar[str]
    json_file_path: ClassVar[str]
    zipped_json_file_path: ClassVar[str]
    cred: ClassVar[Callable[[], DefaultAzureCredential]]
    async_cred: ClassVar[Callable[[], DefaultAzureCredential]]
    dm_kcsb: KustoConnectionStringBuilder

    CHUNK_SIZE = 1024

    @staticmethod
    def get_test_table_csv_mappings():
        """A method to define csv mappings to test table."""
        mappings = list()
        mappings.append(ColumnMapping(column_name="rownumber", column_type="int", ordinal=0))
        mappings.append(ColumnMapping(column_name="rowguid", column_type="string", ordinal=1))
        mappings.append(ColumnMapping(column_name="xdouble", column_type="real", ordinal=2))
        mappings.append(ColumnMapping(column_name="xfloat", column_type="real", ordinal=3))
        mappings.append(ColumnMapping(column_name="xbool", column_type="bool", ordinal=4))
        mappings.append(ColumnMapping(column_name="xint16", column_type="int", ordinal=5))
        mappings.append(ColumnMapping(column_name="xint32", column_type="int", ordinal=6))
        mappings.append(ColumnMapping(column_name="xint64", column_type="long", ordinal=7))
        mappings.append(ColumnMapping(column_name="xuint8", column_type="long", ordinal=8))
        mappings.append(ColumnMapping(column_name="xuint16", column_type="long", ordinal=9))
        mappings.append(ColumnMapping(column_name="xuint32", column_type="long", ordinal=10))
        mappings.append(ColumnMapping(column_name="xuint64", column_type="long", ordinal=11))
        mappings.append(ColumnMapping(column_name="xdate", column_type="datetime", ordinal=12))
        mappings.append(ColumnMapping(column_name="xsmalltext", column_type="string", ordinal=13))
        mappings.append(ColumnMapping(column_name="xtext", column_type="string", ordinal=14))
        mappings.append(ColumnMapping(column_name="xnumberAsText", column_type="string", ordinal=15))
        mappings.append(ColumnMapping(column_name="xtime", column_type="timespan", ordinal=16))
        mappings.append(ColumnMapping(column_name="xtextWithNulls", column_type="string", ordinal=17))
        mappings.append(ColumnMapping(column_name="xdynamicWithNulls", column_type="dynamic", ordinal=18))
        return mappings

    @staticmethod
    def table_json_mappings():
        """A method to define json mappings to test table."""
        mappings = list()
        mappings.append(ColumnMapping(column_name="rownumber", path="$.rownumber", column_type="int"))
        mappings.append(ColumnMapping(column_name="rowguid", path="$.rowguid", column_type="string"))
        mappings.append(ColumnMapping(column_name="xdouble", path="$.xdouble", column_type="real"))
        mappings.append(ColumnMapping(column_name="xfloat", path="$.xfloat", column_type="real"))
        mappings.append(ColumnMapping(column_name="xbool", path="$.xbool", column_type="bool"))
        mappings.append(ColumnMapping(column_name="xint16", path="$.xint16", column_type="int"))
        mappings.append(ColumnMapping(column_name="xint32", path="$.xint32", column_type="int"))
        mappings.append(ColumnMapping(column_name="xint64", path="$.xint64", column_type="long"))
        mappings.append(ColumnMapping(column_name="xuint8", path="$.xuint8", column_type="long"))
        mappings.append(ColumnMapping(column_name="xuint16", path="$.xuint16", column_type="long"))
        mappings.append(ColumnMapping(column_name="xuint32", path="$.xuint32", column_type="long"))
        mappings.append(ColumnMapping(column_name="xuint64", path="$.xuint64", column_type="long"))
        mappings.append(ColumnMapping(column_name="xdate", path="$.xdate", column_type="datetime"))
        mappings.append(ColumnMapping(column_name="xsmalltext", path="$.xsmalltext", column_type="string"))
        mappings.append(ColumnMapping(column_name="xtext", path="$.xtext", column_type="string"))
        mappings.append(ColumnMapping(column_name="xnumberAsText", path="$.xnumberAsText", column_type="string"))
        mappings.append(ColumnMapping(column_name="xtime", path="$.xtime", column_type="timespan"))
        mappings.append(ColumnMapping(column_name="xtextWithNulls", path="$.xtextWithNulls", column_type="string"))
        mappings.append(ColumnMapping(column_name="xdynamicWithNulls", path="$.xdynamicWithNulls", column_type="dynamic"))
        return mappings

    @staticmethod
    def table_json_mapping_reference():
        """A method to get json mappings reference to test table."""
        return """'['
                    '    { "column" : "rownumber", "datatype" : "int", "Properties":{"Path":"$.rownumber"}},'
                    '    { "column" : "rowguid", "datatype" : "string", "Properties":{"Path":"$.rowguid"}},'
                    '    { "column" : "xdouble", "datatype" : "real", "Properties":{"Path":"$.xdouble"}},'
                    '    { "column" : "xfloat", "datatype" : "real", "Properties":{"Path":"$.xfloat"}},'
                    '    { "column" : "xbool", "datatype" : "bool", "Properties":{"Path":"$.xbool"}},'
                    '    { "column" : "xint16", "datatype" : "int", "Properties":{"Path":"$.xint16"}},'
                    '    { "column" : "xint32", "datatype" : "int", "Properties":{"Path":"$.xint32"}},'
                    '    { "column" : "xint64", "datatype" : "long", "Properties":{"Path":"$.xint64"}},'
                    '    { "column" : "xuint8", "datatype" : "long", "Properties":{"Path":"$.xuint8"}},'
                    '    { "column" : "xuint16", "datatype" : "long", "Properties":{"Path":"$.xuint16"}},'
                    '    { "column" : "xuint32", "datatype" : "long", "Properties":{"Path":"$.xuint32"}},'
                    '    { "column" : "xuint64", "datatype" : "long", "Properties":{"Path":"$.xuint64"}},'
                    '    { "column" : "xdate", "datatype" : "datetime", "Properties":{"Path":"$.xdate"}},'
                    '    { "column" : "xsmalltext", "datatype" : "string", "Properties":{"Path":"$.xsmalltext"}},'
                    '    { "column" : "xtext", "datatype" : "string", "Properties":{"Path":"$.xtext"}},'
                    '    { "column" : "xnumberAsText", "datatype" : "string", "Properties":{"Path":"$.rowguid"}},'
                    '    { "column" : "xtime", "datatype" : "timespan", "Properties":{"Path":"$.xtime"}},'
                    '    { "column" : "xtextWithNulls", "datatype" : "string", "Properties":{"Path":"$.xtextWithNulls"}},'
                    '    { "column" : "xdynamicWithNulls", "datatype" : "dynamic", "Properties":{"Path":"$.xdynamicWithNulls"}},'
                    ']'"""

    @staticmethod
    def get_file_path() -> str:
        current_dir = os.getcwd()
        path_parts = ["azure-kusto-ingest", "tests", "input"]
        missing_path_parts = []
        for path_part in path_parts:
            if path_part not in current_dir:
                missing_path_parts.append(path_part)
        return os.path.join(current_dir, *missing_path_parts)

    @classmethod
    def engine_kcsb_from_env(cls, is_async=False) -> KustoConnectionStringBuilder:
        return KustoConnectionStringBuilder.with_azure_token_credential(
            cls.engine_cs,
            credential=cls.cred() if not is_async else cls.async_cred(),
        )

    @classmethod
    def dm_kcsb_from_env(cls) -> KustoConnectionStringBuilder:
        return KustoConnectionStringBuilder.with_azure_token_credential(cls.dm_cs, credential=cls.cred())

    @classmethod
    def setup_class(cls):
        # DM CS can be composed from engine CS
        cls.engine_cs = get_env("ENGINE_CONNECTION_STRING")
        cls.dm_cs = get_env("DM_CONNECTION_STRING", default=cls.engine_cs.replace("//", "//ingest-"))

        # Called to set the env variables for the default azure credentials
        prepare_app_key_auth(optional=True)

        cls.test_db = get_env("TEST_DATABASE")
        cls.test_blob = get_env("TEST_BLOB", optional=True)

        cls.cred = lambda: DefaultAzureCredential(exclude_interactive_browser_credential=False)
        # Async credentials don't support interactive browser authentication for now, so until they do, we'll use the sync default credential for async tests
        cls.async_cred = lambda: DefaultAzureCredential(exclude_interactive_browser_credential=False)

        # Init clients
        python_version = "_".join([str(v) for v in sys.version_info[:3]])
        cls.test_table = "python_test_{0}_{1}_{2}".format(python_version, str(int(time.time())), random.randint(1, 100000))

        cls.client = KustoClient(cls.engine_kcsb_from_env())
        cls.ingest_client = QueuedIngestClient(cls.dm_kcsb_from_env())
        cls.streaming_ingest_client = KustoStreamingIngestClient(cls.engine_kcsb_from_env())
        cls.managed_streaming_ingest_client = ManagedStreamingIngestClient(cls.engine_kcsb_from_env(), cls.dm_kcsb_from_env())

        cls.input_folder_path = cls.get_file_path()

        cls.csv_file_path = os.path.join(cls.input_folder_path, "dataset.csv")
        cls.tsv_file_path = os.path.join(cls.input_folder_path, "dataset.tsv")
        cls.zipped_csv_file_path = os.path.join(cls.input_folder_path, "dataset.csv.gz")
        cls.json_file_path = os.path.join(cls.input_folder_path, "dataset.json")
        cls.zipped_json_file_path = os.path.join(cls.input_folder_path, "dataset.jsonz.gz")

        cls.current_count = 0

        cls.client.execute(
            cls.test_db,
            f".create table {cls.test_table} (rownumber: int, rowguid: string, xdouble: real, xfloat: real, xbool: bool, xint16: int, xint32: int, xint64: long, xuint8: long, xuint16: long, xuint32: long, xuint64: long, xdate: datetime, xsmalltext: string, xtext: string, xnumberAsText: string, xtime: timespan, xtextWithNulls: string, xdynamicWithNulls: dynamic)",
        )
        cls.client.execute(cls.test_db, f".create table {cls.test_table} ingestion json mapping 'JsonMapping' {cls.table_json_mapping_reference()}")

        cls.client.execute(cls.test_db, f".alter table {cls.test_table} policy streamingingestion enable ")

        # Clear the cache to guarantee that subsequent streaming ingestion requests incorporate database and table schema changes
        # See https://docs.microsoft.com/azure/data-explorer/kusto/management/data-ingestion/clear-schema-cache-command
        cls.client.execute(cls.test_db, ".clear database cache streamingingestion schema")

    @classmethod
    def teardown_class(cls):
        cls.client.execute(cls.test_db, ".drop table {} ifexists".format(cls.test_table))
        cls.client.close()
        cls.ingest_client.close()
        cls.streaming_ingest_client.close()
        cls.managed_streaming_ingest_client.close()

    @classmethod
    async def get_async_client(cls) -> AsyncKustoClient:
        return AsyncKustoClient(cls.engine_kcsb_from_env(is_async=True))

    # assertions
    @classmethod
    async def assert_rows_added(cls, expected: int, timeout=60):
        actual = 0
        while timeout > 0:
            time.sleep(1)
            timeout -= 1

            try:
                command = "{} | count".format(cls.test_table)
                response = cls.client.execute(cls.test_db, command)
                async_client = await cls.get_async_client()
                response_from_async = await async_client.execute(cls.test_db, command)
            except KustoServiceError:
                continue

            if response is not None:
                row = response.primary_results[0][0]
                row_async = response_from_async.primary_results[0][0]
                actual = int(row["Count"]) - cls.current_count
                # this is done to allow for data to arrive properly
                if actual >= expected:
                    assert row_async == row, "Mismatch answers between async('{0}') and sync('{1}') clients".format(row_async, row)
                    break

        cls.current_count += actual
        assert actual == expected, "Row count expected = {0}, while actual row count = {1}".format(expected, actual)

    @pytest.mark.asyncio
    async def test_csv_ingest_existing_table(self, is_managed_streaming):
        csv_ingest_props = IngestionProperties(
            self.test_db,
            self.test_table,
            data_format=DataFormat.CSV,
            column_mappings=self.get_test_table_csv_mappings(),
            report_level=ReportLevel.FailuresAndSuccesses,
            flush_immediately=True,
        )

        client = self.streaming_ingest_client if is_managed_streaming else self.ingest_client

        for f in [self.csv_file_path, self.zipped_csv_file_path]:
            client.ingest_from_file(f, csv_ingest_props)

        await self.assert_rows_added(20)

    @pytest.mark.asyncio
    async def test_csv_ingest_ignore_first_record(self, is_managed_streaming):
        csv_ingest_props = IngestionProperties(
            self.test_db,
            self.test_table,
            data_format=DataFormat.CSV,
            column_mappings=self.get_test_table_csv_mappings(),
            report_level=ReportLevel.FailuresAndSuccesses,
            flush_immediately=True,
            ignore_first_record=True,
        )

        for f in [self.csv_file_path, self.zipped_csv_file_path]:
            self.ingest_client.ingest_from_file(f, csv_ingest_props)

        await self.assert_rows_added(18)

    @pytest.mark.asyncio
    async def test_json_ingest_existing_table(self):
        json_ingestion_props = IngestionProperties(
            self.test_db,
            self.test_table,
            flush_immediately=True,
            data_format=DataFormat.JSON,
            column_mappings=self.table_json_mappings(),
            report_level=ReportLevel.FailuresAndSuccesses,
        )

        for f in [self.json_file_path, self.zipped_json_file_path]:
            self.ingest_client.ingest_from_file(f, json_ingestion_props)

        await self.assert_rows_added(4)

    @pytest.mark.asyncio
    async def test_json_ingest_existing_table_no_mapping(self):
        json_ingestion_props = IngestionProperties(
            self.test_db,
            self.test_table,
            flush_immediately=True,
            data_format=DataFormat.JSON,
            report_level=ReportLevel.FailuresAndSuccesses,
        )

        for f in [self.json_file_path, self.zipped_json_file_path]:
            self.ingest_client.ingest_from_file(f, json_ingestion_props)

        await self.assert_rows_added(4)

    @pytest.mark.asyncio
    async def test_ingest_complicated_props(self):
        validation_policy = ValidationPolicy(
            validation_options=ValidationOptions.ValidateCsvInputConstantColumns, validation_implications=ValidationImplications.Fail
        )
        json_ingestion_props = IngestionProperties(
            self.test_db,
            self.test_table,
            data_format=DataFormat.JSON,
            column_mappings=self.table_json_mappings(),
            additional_tags=["a", "b"],
            ingest_if_not_exists=["aaaa", "bbbb"],
            ingest_by_tags=["ingestByTag"],
            drop_by_tags=["drop", "drop-by"],
            flush_immediately=False,
            report_level=ReportLevel.FailuresAndSuccesses,
            report_method=ReportMethod.Queue,
            validation_policy=validation_policy,
        )

        file_paths = [self.json_file_path, self.zipped_json_file_path]
        fds = [FileDescriptor(fp, 0, uuid.uuid4()) for fp in file_paths]

        for fd in fds:
            self.ingest_client.ingest_from_file(fd, json_ingestion_props)

        await self.assert_rows_added(4)

    @pytest.mark.asyncio
    async def test_ingest_from_stream(self, is_managed_streaming):
        validation_policy = ValidationPolicy(
            validation_options=ValidationOptions.ValidateCsvInputConstantColumns, validation_implications=ValidationImplications.Fail
        )
        json_ingestion_props = IngestionProperties(
            self.test_db,
            self.test_table,
            data_format=DataFormat.JSON,
            column_mappings=self.table_json_mappings(),
            additional_tags=["a", "b"],
            ingest_if_not_exists=["aaaa", "bbbb"],
            ingest_by_tags=["ingestByTag"],
            drop_by_tags=["drop", "drop-by"],
            flush_immediately=False,
            report_level=ReportLevel.FailuresAndSuccesses,
            report_method=ReportMethod.Queue,
            validation_policy=validation_policy,
        )
        text = io.StringIO(pathlib.Path(self.json_file_path).read_text())
        zipped = io.BytesIO(pathlib.Path(self.zipped_json_file_path).read_bytes())

        client = self.managed_streaming_ingest_client if is_managed_streaming else self.ingest_client

        client.ingest_from_stream(text, json_ingestion_props)
        client.ingest_from_stream(StreamDescriptor(zipped, is_compressed=True), json_ingestion_props)

        await self.assert_rows_added(4)

    @pytest.mark.asyncio
    async def test_json_ingestion_ingest_by_tag(self):
        json_ingestion_props = IngestionProperties(
            self.test_db,
            self.test_table,
            data_format=DataFormat.JSON,
            column_mappings=self.table_json_mappings(),
            ingest_if_not_exists=["ingestByTag"],
            report_level=ReportLevel.FailuresAndSuccesses,
            drop_by_tags=["drop", "drop-by"],
            flush_immediately=True,
        )

        for f in [self.json_file_path, self.zipped_json_file_path]:
            self.ingest_client.ingest_from_file(f, json_ingestion_props)

        await self.assert_rows_added(0)

    @pytest.mark.asyncio
    async def test_tsv_ingestion_csv_mapping(self):
        tsv_ingestion_props = IngestionProperties(
            self.test_db,
            self.test_table,
            flush_immediately=True,
            data_format=DataFormat.TSV,
            column_mappings=self.get_test_table_csv_mappings(),
            report_level=ReportLevel.FailuresAndSuccesses,
        )

        self.ingest_client.ingest_from_file(self.tsv_file_path, tsv_ingestion_props)

        await self.assert_rows_added(10)

    @pytest.mark.asyncio
    async def test_ingest_blob(self):
        if not self.test_blob:
            pytest.skip("Provide blob SAS uri with 'dataset.csv'")

        csv_ingest_props = IngestionProperties(
            self.test_db,
            self.test_table,
            data_format=DataFormat.CSV,
            column_mappings=self.get_test_table_csv_mappings(),
            report_level=ReportLevel.FailuresAndSuccesses,
            flush_immediately=True,
        )

        blob_len = 1578
        self.ingest_client.ingest_from_blob(BlobDescriptor(self.test_blob, blob_len), csv_ingest_props)

        await self.assert_rows_added(10)

        # Don't provide size hint
        self.ingest_client.ingest_from_blob(BlobDescriptor(self.test_blob, size=None), csv_ingest_props)

        await self.assert_rows_added(10)

    @pytest.mark.asyncio
    async def test_streaming_ingest_from_opened_file(self, is_managed_streaming):
        ingestion_properties = IngestionProperties(database=self.test_db, table=self.test_table, data_format=DataFormat.CSV)

        client = self.managed_streaming_ingest_client if is_managed_streaming else self.streaming_ingest_client
        with open(self.csv_file_path, "r") as stream:
            client.ingest_from_stream(stream, ingestion_properties=ingestion_properties)

        await self.assert_rows_added(10, timeout=120)

    @pytest.mark.asyncio
    async def test_streaming_ingest_from_csv_file(self):
        ingestion_properties = IngestionProperties(database=self.test_db, table=self.test_table, flush_immediately=True, data_format=DataFormat.CSV)

        for f in [self.csv_file_path, self.zipped_csv_file_path]:
            self.streaming_ingest_client.ingest_from_file(f, ingestion_properties=ingestion_properties)

        await self.assert_rows_added(20, timeout=120)

    @pytest.mark.asyncio
    async def test_streaming_ingest_from_json_file(self):
        ingestion_properties = IngestionProperties(
            database=self.test_db,
            table=self.test_table,
            flush_immediately=True,
            data_format=DataFormat.JSON,
            ingestion_mapping_reference="JsonMapping",
            ingestion_mapping_kind=IngestionMappingKind.JSON,
        )

        for f in [self.json_file_path, self.zipped_json_file_path]:
            self.streaming_ingest_client.ingest_from_file(f, ingestion_properties=ingestion_properties)

        await self.assert_rows_added(4, timeout=120)

    @pytest.mark.asyncio
    async def test_streaming_ingest_from_csv_io_streams(self):
        ingestion_properties = IngestionProperties(database=self.test_db, table=self.test_table, data_format=DataFormat.CSV)
        byte_sequence = b'0,00000000-0000-0000-0001-020304050607,0,0,0,0,0,0,0,0,0,0,2014-01-01T01:01:01.0000000Z,Zero,"Zero",0,00:00:00,,null'
        bytes_stream = io.BytesIO(byte_sequence)
        self.streaming_ingest_client.ingest_from_stream(bytes_stream, ingestion_properties=ingestion_properties)

        str_sequence = '0,00000000-0000-0000-0001-020304050607,0,0,0,0,0,0,0,0,0,0,2014-01-01T01:01:01.0000000Z,Zero,"Zero",0,00:00:00,,null'
        str_stream = io.StringIO(str_sequence)
        self.streaming_ingest_client.ingest_from_stream(str_stream, ingestion_properties=ingestion_properties)

        await self.assert_rows_added(2, timeout=120)

    @pytest.mark.asyncio
    async def test_streaming_ingest_from_json_io_streams(self):
        ingestion_properties = IngestionProperties(
            database=self.test_db,
            table=self.test_table,
            data_format=DataFormat.JSON,
            flush_immediately=True,
            ingestion_mapping_reference="JsonMapping",
            ingestion_mapping_kind=IngestionMappingKind.JSON,
        )

        byte_sequence = b'{"rownumber": 0, "rowguid": "00000000-0000-0000-0001-020304050607", "xdouble": 0.0, "xfloat": 0.0, "xbool": 0, "xint16": 0, "xint32": 0, "xint64": 0, "xunit8": 0, "xuint16": 0, "xunit32": 0, "xunit64": 0, "xdate": "2014-01-01T01:01:01Z", "xsmalltext": "Zero", "xtext": "Zero", "xnumberAsText": "0", "xtime": "00:00:00", "xtextWithNulls": null, "xdynamicWithNulls": ""}'
        bytes_stream = io.BytesIO(byte_sequence)
        self.streaming_ingest_client.ingest_from_stream(bytes_stream, ingestion_properties=ingestion_properties)

        str_sequence = '{"rownumber": 0, "rowguid": "00000000-0000-0000-0001-020304050607", "xdouble": 0.0, "xfloat": 0.0, "xbool": 0, "xint16": 0, "xint32": 0, "xint64": 0, "xunit8": 0, "xuint16": 0, "xunit32": 0, "xunit64": 0, "xdate": "2014-01-01T01:01:01Z", "xsmalltext": "Zero", "xtext": "Zero", "xnumberAsText": "0", "xtime": "00:00:00", "xtextWithNulls": null, "xdynamicWithNulls": ""}'
        str_stream = io.StringIO(str_sequence)
        self.streaming_ingest_client.ingest_from_stream(str_stream, ingestion_properties=ingestion_properties)

        await self.assert_rows_added(2, timeout=120)

    @pytest.mark.asyncio
    async def test_streaming_ingest_from_dataframe(self):
        from pandas import DataFrame

        fields = [
            "rownumber",
            "rowguid",
            "xdouble",
            "xfloat",
            "xbool",
            "xint16",
            "xint32",
            "xint64",
            "xunit8",
            "xuint16",
            "xunit32",
            "xunit64",
            "xdate",
            "xsmalltext",
            "xtext",
            "xnumberAsText",
            "xtime",
            "xtextWithNulls",
            "xdynamicWithNulls",
        ]

        guid = uuid.uuid4()

        dynamic_value = ["me@dummy.com", "you@dummy.com", "them@dummy.com"]
        rows = [[0, str(guid), 0.0, 0.0, 0, 0, 0, 0, 0, 0, 0, 0, "2014-01-01T01:01:01Z", "Zero", "Zero", "0", "00:00:00", None, dynamic_value]]
        df = DataFrame(data=rows, columns=fields)
        ingestion_properties = IngestionProperties(database=self.test_db, table=self.test_table, flush_immediately=True)
        self.ingest_client.ingest_from_dataframe(df, ingestion_properties)

        await self.assert_rows_added(1, timeout=120)

        a = self.client.execute(self.test_db, f"{self.test_table} | where rowguid == '{guid}'")
        assert a.primary_results[0].rows[0]["xdynamicWithNulls"] == dynamic_value

    @pytest.mark.asyncio
    async def test_streaming_ingest_from_blob(self, is_managed_streaming):
        ingestion_properties = IngestionProperties(
            database=self.test_db,
            table=self.test_table,
            data_format=DataFormat.JSON,
            ingestion_mapping_reference="JsonMapping",
            ingestion_mapping_kind=IngestionMappingKind.JSON,
        )
        export_containers_list = self.ingest_client._resource_manager._kusto_client.execute("NetDefaultDB", ".show export containers")
        containers = [_ResourceUri(s["StorageRoot"]) for s in export_containers_list.primary_results[0]]

        for c in containers:
            self.ingest_client._resource_manager._ranked_storage_account_set.add_storage_account(c.storage_account_name)

        with FileDescriptor(self.json_file_path).open(False) as stream:
            blob_descriptor = self.ingest_client.upload_blob(
                containers,
                FileDescriptor(self.json_file_path),
                ingestion_properties.database,
                ingestion_properties.table,
                stream,
                None,
                10 * 60,
                3,
            )
            if is_managed_streaming:
                self.managed_streaming_ingest_client.ingest_from_blob(blob_descriptor, ingestion_properties)
            else:
                self.streaming_ingest_client.ingest_from_blob(blob_descriptor, ingestion_properties)

        await self.assert_rows_added(2, timeout=120)