test_WriteApi.py

# coding: utf-8

from __future__ import absolute_import

import datetime
import os
import sys
import unittest
from collections import namedtuple
from datetime import timedelta
from multiprocessing.pool import ApplyResult

import httpretty
import pytest

import influxdb_client
from influxdb_client import Point, WritePrecision, InfluxDBClient
from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS, PointSettings
from influxdb_client.rest import ApiException
from tests.base_test import BaseTest


class SynchronousWriteTest(BaseTest):

    def setUp(self) -> None:
        super().setUp()

        os.environ['data_center'] = "LA"

        self.id_tag = "132-987-655"
        self.customer_tag = "California Miner"
        self.data_center_key = "data_center"

        self.write_client = self.client.write_api(write_options=SYNCHRONOUS,
                                                  point_settings=PointSettings(**{"id": self.id_tag,
                                                                                  "customer": self.customer_tag,
                                                                                  self.data_center_key:
                                                                                      '${env.data_center}'}))

    def tearDown(self) -> None:
        self.write_client.close()
        super().tearDown()

    def test_write_line_protocol(self):
        bucket = self.create_test_bucket()

        record = "h2o_feet,location=coyote_creek level\\ water_level=1 1"
        self.write_client.write(bucket.name, self.org, record)

        result = self.query_api.query(f"from(bucket:\"{bucket.name}\") |> range(start: 1970-01-01T00:00:00.000000001Z)",
                                      self.org)

        self.assertEqual(len(result), 1)
        self.assertEqual(result[0].records[0].get_measurement(), "h2o_feet")
        self.assertEqual(result[0].records[0].get_value(), 1.0)
        self.assertEqual(result[0].records[0].get_field(), "level water_level")
        self.assertEqual(result[0].records[0].get_time(),
                         datetime.datetime(1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))

        self.delete_test_bucket(bucket)

    def test_write_precision(self):
        bucket = self.create_test_bucket()

        self.write_client.write(org="my-org", bucket=bucket.name, record="air,location=Python humidity=99",
                                write_precision=WritePrecision.MS)

        result = self.query_api.query(
            "from(bucket:\"" + bucket.name + "\") |> range(start: 1970-01-01T00:00:00.000000001Z) |> last()", self.org)

        self.assertEqual(len(result), 1)

        self.delete_test_bucket(bucket)

    def test_write_records_list(self):
        bucket = self.create_test_bucket()

        _record1 = "h2o_feet,location=coyote_creek level\\ water_level=1 1"
        _record2 = "h2o_feet,location=coyote_creek level\\ water_level=2 2"

        record_list = [_record1, _record2]

        self.write_client.write(bucket.name, self.org, record_list)

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'

        flux_result = self.client.query_api().query(query)

        self.assertEqual(1, len(flux_result))

        records = flux_result[0].records

        self.assertEqual(2, len(records))

        self.assertEqual("h2o_feet", records[0].get_measurement())
        self.assertEqual(1, records[0].get_value())
        self.assertEqual("level water_level", records[0].get_field())

        self.assertEqual("h2o_feet", records[1].get_measurement())
        self.assertEqual(2, records[1].get_value())
        self.assertEqual("level water_level", records[1].get_field())
        self.delete_test_bucket(bucket)

    def test_write_points_unicode(self):
        bucket = self.create_test_bucket()

        measurement = "h2o_feet_ěščřĚŠČŘ"
        field_name = "field_ěščř"
        utf8_val = "Přerov 🍺"
        tag = "tag_ěščř"
        tag_value = "tag_value_ěščř"

        p = Point(measurement)
        p.field(field_name, utf8_val)
        p.tag(tag, tag_value)
        record_list = [p]

        self.write_client.write(bucket.name, self.org, record_list)

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'
        flux_result = self.client.query_api().query(query)
        self.assertEqual(1, len(flux_result))
        rec = flux_result[0].records[0]

        self.assertEqual(self.id_tag, rec["id"])
        self.assertEqual(self.customer_tag, rec["customer"])
        self.assertEqual("LA", rec[self.data_center_key])

        self.assertEqual(measurement, rec.get_measurement())
        self.assertEqual(utf8_val, rec.get_value())
        self.assertEqual(field_name, rec.get_field())

    def test_write_using_default_tags(self):
        bucket = self.create_test_bucket()

        measurement = "h2o_feet"
        field_name = "water_level"
        val = "1.0"
        val2 = "2.0"
        tag = "location"
        tag_value = "creek level"

        p = Point(measurement)
        p.field(field_name, val)
        p.tag(tag, tag_value)
        p.time(1)

        p2 = Point(measurement)
        p2.field(field_name, val2)
        p2.tag(tag, tag_value)
        p2.time(2)

        record_list = [p, p2]

        self.write_client.write(bucket.name, self.org, record_list)

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'
        flux_result = self.client.query_api().query(query)
        self.assertEqual(1, len(flux_result))
        rec = flux_result[0].records[0]
        rec2 = flux_result[0].records[1]

        self.assertEqual(self.id_tag, rec["id"])
        self.assertEqual(self.customer_tag, rec["customer"])
        self.assertEqual("LA", rec[self.data_center_key])

        self.assertEqual(self.id_tag, rec2["id"])
        self.assertEqual(self.customer_tag, rec2["customer"])
        self.assertEqual("LA", rec2[self.data_center_key])

        self.delete_test_bucket(bucket)

    def test_write_result(self):
        _bucket = self.create_test_bucket()

        _record = "h2o_feet,location=coyote_creek level\\ water_level=1 1"
        result = self.write_client.write(_bucket.name, self.org, _record)

        # The success response is 204 - No Content
        self.assertEqual(None, result)

    def test_write_error(self):
        _bucket = self.create_test_bucket()

        _record = "h2o_feet,location=coyote_creek level\\"

        with self.assertRaises(ApiException) as cm:
            self.write_client.write(_bucket.name, self.org, _record)
        exception = cm.exception

        self.assertEqual(400, exception.status)
        self.assertEqual("Bad Request", exception.reason)

    def test_write_dictionary(self):
        _bucket = self.create_test_bucket()
        _point = {"measurement": "h2o_feet", "tags": {"location": "coyote_creek"},
                  "time": "2009-11-10T23:00:00Z", "fields": {"water_level": 1.0}}

        self.write_client.write(_bucket.name, self.org, _point)

        result = self.query_api.query(
            "from(bucket:\"" + _bucket.name + "\") |> range(start: 1970-01-01T00:00:00.000000001Z) |> last()", self.org)

        self.assertEqual(len(result), 1)
        self.assertEqual(result[0].records[0].get_measurement(), "h2o_feet")
        self.assertEqual(result[0].records[0].get_value(), 1.0)
        self.assertEqual(result[0].records[0].values.get("location"), "coyote_creek")
        self.assertEqual(result[0].records[0].get_field(), "water_level")
        self.assertEqual(result[0].records[0].get_time(),
                         datetime.datetime(2009, 11, 10, 23, 0, tzinfo=datetime.timezone.utc))

        self.delete_test_bucket(_bucket)

    def test_write_bytes(self):
        _bucket = self.create_test_bucket()
        _bytes = "h2o_feet,location=coyote_creek level\\ water_level=1 1".encode("utf-8")

        self.write_client.write(_bucket.name, self.org, _bytes)

        result = self.query_api.query(
            f"from(bucket:\"{_bucket.name}\") |> range(start: 1970-01-01T00:00:00.000000001Z)", self.org)

        self.assertEqual(len(result), 1)
        self.assertEqual(result[0].records[0].get_measurement(), "h2o_feet")
        self.assertEqual(result[0].records[0].get_value(), 1.0)
        self.assertEqual(result[0].records[0].values.get("location"), "coyote_creek")
        self.assertEqual(result[0].records[0].get_field(), "level water_level")
        self.assertEqual(result[0].records[0].get_time(),
                         datetime.datetime(1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))

        self.delete_test_bucket(_bucket)

    def test_write_tuple(self):
        bucket = self.create_test_bucket()

        _record1 = "h2o_feet,location=coyote_creek level\\ water_level=1 1"
        _record2 = "h2o_feet,location=coyote_creek level\\ water_level=2 2"
        _bytes = "h2o_feet,location=coyote_creek level\\ water_level=3 3".encode("utf-8")

        p = (Point("h2o_feet").tag("location", "coyote_creek").field("level water_level", 4.0).time(4))

        tuple = (_record1, _record2, _bytes, (p, ))

        self.write_client = self.client.write_api(write_options=SYNCHRONOUS)
        self.write_client.write(bucket.name, self.org, tuple)

        query = f'from(bucket:"{bucket.name}") |> range(start: 1970-01-01T00:00:00.000000001Z)'

        flux_result = self.client.query_api().query(query)

        self.assertEqual(1, len(flux_result))

        records = flux_result[0].records

        self.assertEqual(4, len(records))

        self.assertEqual("h2o_feet", records[0].get_measurement())
        self.assertEqual(1, records[0].get_value())
        self.assertEqual("level water_level", records[0].get_field())

        self.assertEqual("h2o_feet", records[1].get_measurement())
        self.assertEqual(2, records[1].get_value())
        self.assertEqual("level water_level", records[1].get_field())

        self.assertEqual("h2o_feet", records[2].get_measurement())
        self.assertEqual(3, records[2].get_value())
        self.assertEqual("level water_level", records[2].get_field())

        self.assertEqual("h2o_feet", records[3].get_measurement())
        self.assertEqual(4, records[3].get_value())
        self.assertEqual("level water_level", records[3].get_field())
        self.delete_test_bucket(bucket)

    def test_write_data_frame(self):
        from influxdb_client.extras import pd

        bucket = self.create_test_bucket()

        now = pd.Timestamp('1970-01-01 00:00+00:00')
        data_frame = pd.DataFrame(data=[["coyote_creek", 1], ["coyote_creek", 2]],
                                  index=[now + timedelta(hours=1), now + timedelta(hours=2)],
                                  columns=["location", "water water_level"])

        self.write_client.write(bucket.name, record=data_frame, data_frame_measurement_name='h2o_feet',
                                data_frame_tag_columns=['location'])

        result = self.query_api.query(
            f"from(bucket:\"{bucket.name}\") |> range(start: 1970-01-01T00:00:00.000000001Z)", self.org)

        self.assertEqual(1, len(result))
        self.assertEqual(2, len(result[0].records))

        self.assertEqual(result[0].records[0].get_measurement(), "h2o_feet")
        self.assertEqual(result[0].records[0].get_value(), 1.0)
        self.assertEqual(result[0].records[0].values.get("location"), "coyote_creek")
        self.assertEqual(result[0].records[0].get_field(), "water water_level")
        self.assertEqual(result[0].records[0].get_time(),
                         datetime.datetime(1970, 1, 1, 1, 0, tzinfo=datetime.timezone.utc))

        self.assertEqual(result[0].records[1].get_measurement(), "h2o_feet")
        self.assertEqual(result[0].records[1].get_value(), 2.0)
        self.assertEqual(result[0].records[1].values.get("location"), "coyote_creek")
        self.assertEqual(result[0].records[1].get_field(), "water water_level")
        self.assertEqual(result[0].records[1].get_time(),
                         datetime.datetime(1970, 1, 1, 2, 0, tzinfo=datetime.timezone.utc))

        self.delete_test_bucket(bucket)

    def test_write_data_frame_without_measurement_name(self):
        from influxdb_client.extras import pd

        bucket = self.create_test_bucket()

        now = pd.Timestamp('1970-01-01 00:00+00:00')
        data_frame = pd.DataFrame(data=[["coyote_creek", 1.0], ["coyote_creek", 2.0]],
                                  index=[now + timedelta(hours=1), now + timedelta(hours=2)],
                                  columns=["location", "water_level"])

        with self.assertRaises(TypeError) as cm:
            self.write_client.write(bucket.name, record=data_frame)
        exception = cm.exception

        self.assertEqual('"data_frame_measurement_name" is a Required Argument', exception.__str__())

    def test_use_default_org(self):
        bucket = self.create_test_bucket()

        record = "h2o_feet,location=coyote_creek level\\ water_level=1 1"
        self.write_client.write(bucket.name, record=record)

        result = self.query_api.query(
            f"from(bucket:\"{bucket.name}\") |> range(start: 1970-01-01T00:00:00.000000001Z)", self.org)

        self.assertEqual(len(result), 1)
        self.assertEqual(result[0].records[0].get_measurement(), "h2o_feet")
        self.assertEqual(result[0].records[0].get_value(), 1.0)
        self.assertEqual(result[0].records[0].get_field(), "level water_level")
        self.assertEqual(result[0].records[0].get_time(),
                         datetime.datetime(1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))

    def test_write_empty_data(self):
        bucket = self.create_test_bucket()

        result = self.query_api.query(
            "from(bucket:\"" + bucket.name + "\") |> range(start: 1970-01-01T00:00:00.000000001Z) |> last()", self.org)

        self.assertEqual(len(result), 0)

    def test_write_point_different_precision(self):
        bucket = self.create_test_bucket()

        point1 = Point('test_precision') \
            .field('power', 10) \
            .tag('powerFlow', 'low') \
            .time(datetime.datetime(2020, 4, 20, 6, 30, tzinfo=datetime.timezone.utc), WritePrecision.S)

        point2 = Point('test_precision') \
            .field('power', 20) \
            .tag('powerFlow', 'high') \
            .time(datetime.datetime(2020, 4, 20, 5, 30, tzinfo=datetime.timezone.utc), WritePrecision.MS)

        writer = self.client.write_api(write_options=SYNCHRONOUS)
        writer.write(bucket.name, self.org, [point1, point2])

        result = self.query_api.query(
            f"from(bucket:\"{bucket.name}\") |> range(start: 1970-01-01T00:00:00.000000001Z) |> last() "
            "|> sort(columns: [\"_time\"], desc: false)", self.org)

        self.assertEqual(len(result), 2)
        self.assertEqual(len(result[0].records), 1)
        self.assertEqual(len(result[1].records), 1)
        self.assertEqual(result[0].records[0].get_time(),
                         datetime.datetime(2020, 4, 20, 5, 30, tzinfo=datetime.timezone.utc))
        self.assertEqual(result[1].records[0].get_time(),
                         datetime.datetime(2020, 4, 20, 6, 30, tzinfo=datetime.timezone.utc))

    def test_write_point_with_default_tags(self):
        bucket = self.create_test_bucket()

        point = Point("h2o_feet")\
            .field("water_level", 1)\
            .tag("location", "creek level")

        self.write_client.write(bucket.name, self.org, point)

        flux_result = self.client.query_api().query(f'from(bucket:"{bucket.name}") |> range(start: 1970-01-01T00:00:00.000000001Z)')
        self.assertEqual(1, len(flux_result))

        record = flux_result[0].records[0]

        self.assertEqual(self.id_tag, record["id"])
        self.assertEqual(self.customer_tag, record["customer"])
        self.assertEqual("LA", record[self.data_center_key])

    def test_write_list_of_list_point_with_default_tags(self):
        bucket = self.create_test_bucket()

        point = Point("h2o_feet")\
            .field("water_level", 1)\
            .tag("location", "creek level")

        self.write_client.write(bucket.name, self.org, [[point]])

        flux_result = self.client.query_api().query(f'from(bucket:"{bucket.name}") |> range(start: 1970-01-01T00:00:00.000000001Z)')
        self.assertEqual(1, len(flux_result))

        record = flux_result[0].records[0]

        self.assertEqual(self.id_tag, record["id"])
        self.assertEqual(self.customer_tag, record["customer"])
        self.assertEqual("LA", record[self.data_center_key])

    def test_check_write_permission_by_empty_data(self):
        client = InfluxDBClient(url="http://localhost:8086", token="my-token-wrong", org="my-org")
        write_api = client.write_api(write_options=SYNCHRONOUS)

        with self.assertRaises(ApiException) as cm:
            write_api.write("my-bucket", self.org, b'')
        exception = cm.exception

        self.assertEqual(401, exception.status)
        self.assertEqual("Unauthorized", exception.reason)

        client.close()

    def test_write_query_data_nanoseconds(self):

        from influxdb_client.client.util.date_utils_pandas import PandasDateTimeHelper
        import influxdb_client.client.util.date_utils as date_utils

        date_utils.date_helper = PandasDateTimeHelper()

        bucket = self.create_test_bucket()

        point = Point("h2o_feet") \
            .field("water_level", 155) \
            .tag("location", "creek level")\
            .time('1996-02-25T21:20:00.001001231Z')

        self.write_client.write(bucket.name, self.org, [point])

        flux_result = self.client.query_api().query(
            f'from(bucket:"{bucket.name}") |> range(start: 1970-01-01T00:00:00.000000001Z)')
        self.assertEqual(1, len(flux_result))

        record = flux_result[0].records[0]

        self.assertEqual(self.id_tag, record["id"])
        self.assertEqual(record["_value"], 155)
        self.assertEqual(record["location"], "creek level")
        self.assertEqual(record["_time"].year, 1996)
        self.assertEqual(record["_time"].month, 2)
        self.assertEqual(record["_time"].day, 25)
        self.assertEqual(record["_time"].hour, 21)
        self.assertEqual(record["_time"].minute, 20)
        self.assertEqual(record["_time"].second, 00)
        self.assertEqual(record["_time"].microsecond, 1001)
        self.assertEqual(record["_time"].nanosecond, 231)

        date_utils.date_helper = None


class WriteApiTestMock(BaseTest):

    def setUp(self) -> None:
        httpretty.enable()
        httpretty.reset()

        conf = influxdb_client.configuration.Configuration()
        conf.host = "http://localhost"
        conf.debug = False

        self.influxdb_client = InfluxDBClient(url=conf.host, token="my-token")

    def tearDown(self) -> None:
        self.influxdb_client.close()
        httpretty.disable()

    def test_writes_synchronous_without_retry(self):
        httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=503)

        self.write_client = self.influxdb_client.write_api(write_options=SYNCHRONOUS)
        with self.assertRaises(ApiException) as cm:
            self.write_client.write("my-bucket", "my-org", "h2o_feet,location=coyote_creek water_level=1 1")
        exception = cm.exception

        self.assertEqual("Service Unavailable", exception.reason)
        self.assertEqual(1, len(httpretty.httpretty.latest_requests))

    def test_writes_asynchronous_without_retry(self):
        httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=503)

        self.write_client = self.influxdb_client.write_api(write_options=ASYNCHRONOUS)
        with self.assertRaises(ApiException) as cm:
            self.write_client.write("my-bucket", "my-org", "h2o_feet,location=coyote_creek water_level=1 1").get()
        exception = cm.exception

        self.assertEqual("Service Unavailable", exception.reason)
        self.assertEqual(1, len(httpretty.httpretty.latest_requests))

    def test_writes_default_tags_dict_without_tag(self):
        httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=204)

        point_settings = PointSettings(**{"id": "132-987-655", "customer": "California Miner"})
        self.write_client = self.influxdb_client.write_api(write_options=SYNCHRONOUS,
                                                           point_settings=point_settings)

        self.write_client.write("my-bucket", "my-org", {"measurement": "h2o", "fields": {"level": 1.0}, "time": 1})

        requests = httpretty.httpretty.latest_requests
        self.assertEqual(1, len(requests))
        self.assertEqual("h2o,customer=California\\ Miner,id=132-987-655 level=1 1", requests[0].parsed_body)

    def test_redirect(self):
        from urllib3 import Retry
        Retry.DEFAULT_REMOVE_HEADERS_ON_REDIRECT = frozenset()
        Retry.DEFAULT.remove_headers_on_redirect = Retry.DEFAULT_REMOVE_HEADERS_ON_REDIRECT
        self.influxdb_client.close()

        self.influxdb_client = InfluxDBClient(url="http://localhost", token="my-token", org="my-org")

        httpretty.register_uri(httpretty.POST, uri="http://localhost2/api/v2/write", status=204)
        httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=301,
                               adding_headers={'Location': 'http://localhost2/api/v2/write'})

        self.write_client = self.influxdb_client.write_api(write_options=SYNCHRONOUS)

        self.write_client.write("my-bucket", "my-org", {"measurement": "h2o", "fields": {"level": 1.0}, "time": 1})

        requests = httpretty.httpretty.latest_requests
        self.assertEqual(2, len(requests))
        self.assertEqual('Token my-token', requests[0].headers['Authorization'])
        self.assertEqual('Token my-token', requests[1].headers['Authorization'])

        from urllib3 import Retry
        Retry.DEFAULT.remove_headers_on_redirect = Retry.DEFAULT_REMOVE_HEADERS_ON_REDIRECT

    def test_named_tuple(self):
        httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=204)

        self.write_client = self.influxdb_client.write_api(write_options=SYNCHRONOUS)

        Factory = namedtuple('Factory', ['measurement', 'position', 'customers'])
        factory = Factory(measurement='factory', position="central europe", customers=123456)

        self.write_client.write("my-bucket", "my-org", factory,
                                record_measurement_key="measurement",
                                record_tag_keys=["position"],
                                record_field_keys=["customers"])

        requests = httpretty.httpretty.latest_requests
        self.assertEqual(1, len(requests))
        self.assertEqual("factory,position=central\\ europe customers=123456i", requests[0].parsed_body)

    @pytest.mark.skipif(sys.version_info < (3, 8), reason="requires python3.8 or higher")
    def test_data_class(self):
        httpretty.register_uri(httpretty.POST, uri="http://localhost/api/v2/write", status=204)

        self.write_client = self.influxdb_client.write_api(write_options=SYNCHRONOUS)

        from dataclasses import dataclass

        @dataclass
        class Car:
            engine: str
            type: str
            speed: float

        car = Car('12V-BT', 'sport-cars', 125.25)
        self.write_client.write("my-bucket", "my-org",
                                record=car,
                                record_measurement_name="performance",
                                record_tag_keys=["engine", "type"],
                                record_field_keys=["speed"])

        requests = httpretty.httpretty.latest_requests
        self.assertEqual(1, len(requests))
        self.assertEqual("performance,engine=12V-BT,type=sport-cars speed=125.25", requests[0].parsed_body)


class AsynchronousWriteTest(BaseTest):

    def setUp(self) -> None:
        super().setUp()

        os.environ['data_center'] = "LA"

        self.id_tag = "132-987-655"
        self.customer_tag = "California Miner"
        self.data_center_key = "data_center"

        self.write_client = self.client.write_api(write_options=ASYNCHRONOUS,
                                                  point_settings=PointSettings(**{"id": self.id_tag,
                                                                                  "customer": self.customer_tag,
                                                                                  self.data_center_key:
                                                                                      '${env.data_center}'}))

    def tearDown(self) -> None:
        self.write_client.close()
        super().tearDown()

    def test_write_result(self):
        _bucket = self.create_test_bucket()

        _record = "h2o_feet,location=coyote_creek level\\ water_level=1 1"
        result = self.write_client.write(_bucket.name, self.org, _record)

        self.assertEqual(ApplyResult, type(result))
        self.assertEqual(None, result.get())
        self.delete_test_bucket(_bucket)

    def test_write_dictionaries(self):
        bucket = self.create_test_bucket()

        _point1 = {"measurement": "h2o_feet", "tags": {"location": "coyote_creek"},
                   "time": "2009-11-10T22:00:00Z", "fields": {"water_level": 1.0}}
        _point2 = {"measurement": "h2o_feet", "tags": {"location": "coyote_creek"},
                   "time": "2009-11-10T23:00:00Z", "fields": {"water_level": 2.0}}

        _point_list = [_point1, _point2]

        async_result = self.write_client.write(bucket.name, self.org, _point_list)
        async_result.get()

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'

        flux_result = self.client.query_api().query(query)

        self.assertEqual(1, len(flux_result))

        records = flux_result[0].records

        self.assertEqual(2, len(records))

        self.assertEqual("h2o_feet", records[0].get_measurement())
        self.assertEqual(1, records[0].get_value())
        self.assertEqual("water_level", records[0].get_field())
        self.assertEqual("coyote_creek", records[0].values.get('location'))
        self.assertEqual(records[0].get_time(),
                         datetime.datetime(2009, 11, 10, 22, 0, tzinfo=datetime.timezone.utc))

        self.assertEqual("h2o_feet", records[1].get_measurement())
        self.assertEqual(2, records[1].get_value())
        self.assertEqual("water_level", records[1].get_field())
        self.assertEqual("coyote_creek", records[1].values.get('location'))
        self.assertEqual(records[1].get_time(),
                         datetime.datetime(2009, 11, 10, 23, 0, tzinfo=datetime.timezone.utc))

        self.delete_test_bucket(bucket)

    def test_use_default_tags_with_dictionaries(self):
        bucket = self.create_test_bucket()

        _point1 = {"measurement": "h2o_feet", "tags": {"location": "coyote_creek"},
                   "time": "2009-11-10T22:00:00Z", "fields": {"water_level": 1.0}}
        _point2 = {"measurement": "h2o_feet", "tags": {"location": "coyote_creek"},
                   "time": "2009-11-10T23:00:00Z", "fields": {"water_level": 2.0}}

        _point_list = [_point1, _point2]

        async_result = self.write_client.write(bucket.name, self.org, _point_list)
        async_result.get()

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'

        flux_result = self.client.query_api().query(query)

        self.assertEqual(1, len(flux_result))

        records = flux_result[0].records

        self.assertEqual(2, len(records))

        rec = records[0]
        rec2 = records[1]

        self.assertEqual(self.id_tag, rec["id"])
        self.assertEqual(self.customer_tag, rec["customer"])
        self.assertEqual("LA", rec[self.data_center_key])

        self.assertEqual(self.id_tag, rec2["id"])
        self.assertEqual(self.customer_tag, rec2["customer"])
        self.assertEqual("LA", rec2[self.data_center_key])

        self.delete_test_bucket(bucket)

    def test_use_default_tags_with_data_frame(self):
        from influxdb_client.extras import pd

        bucket = self.create_test_bucket()

        now = pd.Timestamp('1970-01-01 00:00+00:00')
        data_frame = pd.DataFrame(data=[["coyote_creek", 1.0], ["coyote_creek", 2.0]],
                                  index=[now + timedelta(hours=1), now + timedelta(hours=2)],
                                  columns=["location", "water_level"])

        async_result = self.write_client.write(bucket.name, record=data_frame, data_frame_measurement_name='h2o_feet',
                                               data_frame_tag_columns=['location'])
        async_result.get()

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'

        flux_result = self.client.query_api().query(query)

        self.assertEqual(1, len(flux_result))

        records = flux_result[0].records

        self.assertEqual(2, len(records))

        rec = records[0]
        rec2 = records[1]

        self.assertEqual(self.id_tag, rec["id"])
        self.assertEqual(self.customer_tag, rec["customer"])
        self.assertEqual("LA", rec[self.data_center_key])

        self.assertEqual(self.id_tag, rec2["id"])
        self.assertEqual(self.customer_tag, rec2["customer"])
        self.assertEqual("LA", rec2[self.data_center_key])

        self.delete_test_bucket(bucket)

    def test_write_bytes(self):
        bucket = self.create_test_bucket()

        _bytes1 = "h2o_feet,location=coyote_creek level\\ water_level=1 1".encode("utf-8")
        _bytes2 = "h2o_feet,location=coyote_creek level\\ water_level=2 2".encode("utf-8")

        _bytes_list = [_bytes1, _bytes2]

        async_result = self.write_client.write(bucket.name, self.org, _bytes_list, write_precision=WritePrecision.S)
        async_result.get()

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'

        flux_result = self.client.query_api().query(query)

        self.assertEqual(1, len(flux_result))

        records = flux_result[0].records

        self.assertEqual(2, len(records))

        self.assertEqual("h2o_feet", records[0].get_measurement())
        self.assertEqual(1, records[0].get_value())
        self.assertEqual("level water_level", records[0].get_field())
        self.assertEqual("coyote_creek", records[0].values.get('location'))
        self.assertEqual(records[0].get_time(),
                         datetime.datetime(1970, 1, 1, 0, 0, 1, tzinfo=datetime.timezone.utc))

        self.assertEqual("h2o_feet", records[1].get_measurement())
        self.assertEqual(2, records[1].get_value())
        self.assertEqual("level water_level", records[1].get_field())
        self.assertEqual("coyote_creek", records[1].values.get('location'))
        self.assertEqual(records[1].get_time(),
                         datetime.datetime(1970, 1, 1, 0, 0, 2, tzinfo=datetime.timezone.utc))

        self.delete_test_bucket(bucket)

    def test_write_point_different_precision(self):
        bucket = self.create_test_bucket()

        _point1 = Point("h2o_feet").tag("location", "coyote_creek").field("level water_level", 5.0).time(5,
                                                                                                         WritePrecision.S)
        _point2 = Point("h2o_feet").tag("location", "coyote_creek").field("level water_level", 6.0).time(6,
                                                                                                         WritePrecision.US)

        _point_list = [_point1, _point2]

        async_results = self.write_client.write(bucket.name, self.org, _point_list)
        self.assertEqual(2, len(async_results))
        for async_result in async_results:
            async_result.get()

        query = f'from(bucket:"{bucket.name}") |> range(start: 1970-01-01T00:00:00.000000001Z) '\
                '|> sort(columns: [\"_time\"], desc: false)'

        flux_result = self.client.query_api().query(query)

        self.assertEqual(1, len(flux_result))

        records = flux_result[0].records

        self.assertEqual(2, len(records))
        self.assertEqual(records[0].get_time(),
                         datetime.datetime(1970, 1, 1, 0, 0, 0, 6, tzinfo=datetime.timezone.utc))
        self.assertEqual(records[1].get_time(),
                         datetime.datetime(1970, 1, 1, 0, 0, 5, tzinfo=datetime.timezone.utc))



class PointSettingTest(BaseTest):

    def setUp(self) -> None:
        super().setUp()
        self.id_tag = "132-987-655"
        self.customer_tag = "California Miner"

    def tearDown(self) -> None:
        self.write_client.close()
        super().tearDown()

    def test_point_settings(self):
        self.write_client = self.client.write_api(write_options=SYNCHRONOUS,
                                                  point_settings=PointSettings(**{"id": self.id_tag,
                                                                                  "customer": self.customer_tag}))

        default_tags = self.write_client._point_settings.defaultTags

        self.assertEqual(self.id_tag, default_tags.get("id"))
        self.assertEqual(self.customer_tag, default_tags.get("customer"))

    def test_point_settings_with_add(self):
        os.environ['data_center'] = "LA"

        point_settings = PointSettings()
        point_settings.add_default_tag("id", self.id_tag)
        point_settings.add_default_tag("customer", self.customer_tag)
        point_settings.add_default_tag("data_center", "${env.data_center}")

        self.write_client = self.client.write_api(write_options=SYNCHRONOUS, point_settings=point_settings)

        default_tags = self.write_client._point_settings.defaultTags

        self.assertEqual(self.id_tag, default_tags.get("id"))
        self.assertEqual(self.customer_tag, default_tags.get("customer"))
        self.assertEqual("LA", default_tags.get("data_center"))


class DefaultTagsConfiguration(BaseTest):

    def setUp(self) -> None:
        super().setUp()

        os.environ['data_center'] = "LA"

        self.id_tag = "132-987-655"
        self.customer_tag = "California Miner"
        self.data_center_key = "data_center"

        os.environ['INFLUXDB_V2_TOKEN'] = "my-token"
        os.environ['INFLUXDB_V2_TIMEOUT'] = "6000"
        os.environ['INFLUXDB_V2_ORG'] = "my-org"

        os.environ['INFLUXDB_V2_TAG_ID'] = self.id_tag
        os.environ['INFLUXDB_V2_TAG_CUSTOMER'] = self.customer_tag
        os.environ['INFLUXDB_V2_TAG_DATA_CENTER'] = "${env.data_center}"

    def tearDown(self) -> None:
        self.write_client.close()
        super().tearDown()

    def test_connection_option_from_conf_file(self):
        self.client.close()
        self.client = InfluxDBClient.from_config_file(self._path_to_config(), self.debug)

        self.assertEqual("http://localhost:8086", self.client.url)
        self._check_connection_settings()

    def test_connection_option_from_env(self):
        self.client.close()
        self.client = InfluxDBClient.from_env_properties(self.debug)

        self.assertEqual("http://localhost:8086", self.client.url)
        self._check_connection_settings()

    def _check_connection_settings(self):
        self.write_client = self.client.write_api(write_options=SYNCHRONOUS)

        self.assertEqual("my-org", self.client.org)
        self.assertEqual("my-token", self.client.token)
        self.assertEqual(6000, self.client.api_client.configuration.timeout)

    def test_default_tags_from_conf_file(self):
        self.client.close()
        self.client = InfluxDBClient.from_config_file(self._path_to_config(), self.debug)

        self._write_point()

    def test_default_tags_from_env(self):
        self.client.close()
        self.client = InfluxDBClient.from_env_properties(self.debug)

        self._write_point()

    def _write_point(self):
        self.write_client = self.client.write_api(write_options=SYNCHRONOUS)

        bucket = self.create_test_bucket()

        measurement = "h2o_feet"
        field_name = "water_level"
        val = "1.0"
        tag = "location"
        tag_value = "creek level"

        p = Point(measurement)
        p.field(field_name, val)
        p.tag(tag, tag_value)

        record_list = [p]

        self.write_client.write(bucket.name, self.org, record_list)

        query = 'from(bucket:"' + bucket.name + '") |> range(start: 1970-01-01T00:00:00.000000001Z)'
        flux_result = self.client.query_api().query(query)
        self.assertEqual(1, len(flux_result))
        rec = flux_result[0].records[0]

        self.assertEqual(self.id_tag, rec["id"])
        self.assertEqual(self.customer_tag, rec["customer"])
        self.assertEqual("LA", rec[self.data_center_key])

        self.delete_test_bucket(bucket)

    @staticmethod
    def _path_to_config():
        return os.path.dirname(os.path.realpath(__file__)) + "/config.ini"


if __name__ == '__main__':
    unittest.main()