# -*- coding: utf-8 -*-
"""Unit tests for misc module."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

from datetime import timedelta

import json
import unittest
import warnings
import requests_mock

from influxdb.tests import skipIfPYpy, using_pypy
from nose.tools import raises

from .client_test import _mocked_session

if not using_pypy:
    import pandas as pd
    from pandas.util.testing import assert_frame_equal
    from influxdb import DataFrameClient


@skipIfPYpy
class TestDataFrameClient(unittest.TestCase):
    """Set up a test DataFrameClient object."""

    def setUp(self):
        """Instantiate a TestDataFrameClient object."""
        # By default, raise exceptions on warnings
        warnings.simplefilter('error', FutureWarning)

    def test_write_points_from_dataframe(self):
        """Test write points from df in TestDataFrameClient object."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[["1", 1, 1.0], ["2", 2, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["column_one", "column_two",
                                          "column_three"])
        expected = (
            b"foo column_one=\"1\",column_two=1i,column_three=1.0 0\n"
            b"foo column_one=\"2\",column_two=2i,column_three=2.0 "
            b"3600000000000\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')

            cli.write_points(dataframe, 'foo')
            self.assertEqual(m.last_request.body, expected)

            cli.write_points(dataframe, 'foo', tags=None)
            self.assertEqual(m.last_request.body, expected)

    def test_dataframe_write_points_with_whitespace_measurement(self):
        """write_points should escape white space in measurements."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[["1", 1, 1.0], ["2", 2, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["column_one", "column_two",
                                          "column_three"])
        expected = (
            b"meas\\ with\\ space "
            b"column_one=\"1\",column_two=1i,column_three=1.0 0\n"
            b"meas\\ with\\ space "
            b"column_one=\"2\",column_two=2i,column_three=2.0 "
            b"3600000000000\n"
        )
        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)
            cli = DataFrameClient(database='db')
            cli.write_points(dataframe, 'meas with space')
            self.assertEqual(m.last_request.body, expected)

    def test_dataframe_write_points_with_whitespace_in_column_names(self):
        """write_points should escape white space in column names."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[["1", 1, 1.0], ["2", 2, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["column one", "column two",
                                          "column three"])
        expected = (
            b"foo column\\ one=\"1\",column\\ two=1i,column\\ three=1.0 0\n"
            b"foo column\\ one=\"2\",column\\ two=2i,column\\ three=2.0 "
            b"3600000000000\n"
        )
        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)
            cli = DataFrameClient(database='db')
            cli.write_points(dataframe, 'foo')
            self.assertEqual(m.last_request.body, expected)

    def test_write_points_from_dataframe_with_none(self):
        """Test write points from df in TestDataFrameClient object."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[["1", None, 1.0], ["2", 2.0, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["column_one", "column_two",
                                          "column_three"])
        expected = (
            b"foo column_one=\"1\",column_three=1.0 0\n"
            b"foo column_one=\"2\",column_two=2.0,column_three=2.0 "
            b"3600000000000\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')

            cli.write_points(dataframe, 'foo')
            self.assertEqual(m.last_request.body, expected)

            cli.write_points(dataframe, 'foo', tags=None)
            self.assertEqual(m.last_request.body, expected)

    def test_write_points_from_dataframe_with_line_of_none(self):
        """Test write points from df in TestDataFrameClient object."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[[None, None, None], ["2", 2.0, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["column_one", "column_two",
                                          "column_three"])
        expected = (
            b"foo column_one=\"2\",column_two=2.0,column_three=2.0 "
            b"3600000000000\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')

            cli.write_points(dataframe, 'foo')
            self.assertEqual(m.last_request.body, expected)

            cli.write_points(dataframe, 'foo', tags=None)
            self.assertEqual(m.last_request.body, expected)

    def test_write_points_from_dataframe_with_all_none(self):
        """Test write points from df in TestDataFrameClient object."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[[None, None, None], [None, None, None]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["column_one", "column_two",
                                          "column_three"])
        expected = (
            b"\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')

            cli.write_points(dataframe, 'foo')
            self.assertEqual(m.last_request.body, expected)

            cli.write_points(dataframe, 'foo', tags=None)
            self.assertEqual(m.last_request.body, expected)

    def test_write_points_from_dataframe_in_batches(self):
        """Test write points in batch from df in TestDataFrameClient object."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[["1", 1, 1.0], ["2", 2, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["column_one", "column_two",
                                          "column_three"])
        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')
            self.assertTrue(cli.write_points(dataframe, "foo", batch_size=1))

    def test_write_points_from_dataframe_with_tag_columns(self):
        """Test write points from df w/tag in TestDataFrameClient object."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[['blue', 1, "1", 1, 1.0],
                                       ['red', 0, "2", 2, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["tag_one", "tag_two", "column_one",
                                          "column_two", "column_three"])
        expected = (
            b"foo,tag_one=blue,tag_two=1 "
            b"column_one=\"1\",column_two=1i,column_three=1.0 "
            b"0\n"
            b"foo,tag_one=red,tag_two=0 "
            b"column_one=\"2\",column_two=2i,column_three=2.0 "
            b"3600000000000\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')

            cli.write_points(dataframe, 'foo',
                             tag_columns=['tag_one', 'tag_two'])
            self.assertEqual(m.last_request.body, expected)

            cli.write_points(dataframe, 'foo',
                             tag_columns=['tag_one', 'tag_two'], tags=None)
            self.assertEqual(m.last_request.body, expected)

    def test_write_points_from_dataframe_with_tag_cols_and_global_tags(self):
        """Test write points from df w/tag + cols in TestDataFrameClient."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[['blue', 1, "1", 1, 1.0],
                                       ['red', 0, "2", 2, 2.0]],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["tag_one", "tag_two", "column_one",
                                          "column_two", "column_three"])
        expected = (
            b"foo,global_tag=value,tag_one=blue,tag_two=1 "
            b"column_one=\"1\",column_two=1i,column_three=1.0 "
            b"0\n"
            b"foo,global_tag=value,tag_one=red,tag_two=0 "
            b"column_one=\"2\",column_two=2i,column_three=2.0 "
            b"3600000000000\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')

            cli.write_points(dataframe, 'foo',
                             tag_columns=['tag_one', 'tag_two'],
                             tags={'global_tag': 'value'})
            self.assertEqual(m.last_request.body, expected)

    def test_write_points_from_dataframe_with_tag_cols_and_defaults(self):
        """Test default write points from df w/tag in TestDataFrameClient."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(data=[['blue', 1, "1", 1, 1.0, 'hot'],
                                       ['red', 0, "2", 2, 2.0, 'cold']],
                                 index=[now, now + timedelta(hours=1)],
                                 columns=["tag_one", "tag_two", "column_one",
                                          "column_two", "column_three",
                                          "tag_three"])
        expected_tags_and_fields = (
            b"foo,tag_one=blue "
            b"column_one=\"1\",column_two=1i "
            b"0\n"
            b"foo,tag_one=red "
            b"column_one=\"2\",column_two=2i "
            b"3600000000000\n"
        )

        expected_tags_no_fields = (
            b"foo,tag_one=blue,tag_two=1 "
            b"column_one=\"1\",column_two=1i,column_three=1.0,"
            b"tag_three=\"hot\" 0\n"
            b"foo,tag_one=red,tag_two=0 "
            b"column_one=\"2\",column_two=2i,column_three=2.0,"
            b"tag_three=\"cold\" 3600000000000\n"
        )

        expected_fields_no_tags = (
            b"foo,tag_one=blue,tag_three=hot,tag_two=1 "
            b"column_one=\"1\",column_two=1i,column_three=1.0 "
            b"0\n"
            b"foo,tag_one=red,tag_three=cold,tag_two=0 "
            b"column_one=\"2\",column_two=2i,column_three=2.0 "
            b"3600000000000\n"
        )

        expected_no_tags_no_fields = (
            b"foo "
            b"tag_one=\"blue\",tag_two=1i,column_one=\"1\","
            b"column_two=1i,column_three=1.0,tag_three=\"hot\" "
            b"0\n"
            b"foo "
            b"tag_one=\"red\",tag_two=0i,column_one=\"2\","
            b"column_two=2i,column_three=2.0,tag_three=\"cold\" "
            b"3600000000000\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')

            cli.write_points(dataframe, 'foo',
                             field_columns=['column_one', 'column_two'],
                             tag_columns=['tag_one'])
            self.assertEqual(m.last_request.body, expected_tags_and_fields)

            cli.write_points(dataframe, 'foo',
                             tag_columns=['tag_one', 'tag_two'])
            self.assertEqual(m.last_request.body, expected_tags_no_fields)

            cli.write_points(dataframe, 'foo',
                             field_columns=['column_one', 'column_two',
                                            'column_three'])
            self.assertEqual(m.last_request.body, expected_fields_no_tags)

            cli.write_points(dataframe, 'foo')
            self.assertEqual(m.last_request.body, expected_no_tags_no_fields)

    def test_write_points_from_dataframe_with_tag_escaped(self):
        """Test write points from df w/escaped tag in TestDataFrameClient."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        dataframe = pd.DataFrame(
            data=[
                ['blue orange', "1", 1, 'hot=cold'],  # space, equal
                ['red,green', "2", 2, r'cold\fire'],  # comma, backslash
                ['some', "2", 2, ''],                 # skip empty
                ['some', "2", 2, None],               # skip None
                ['', "2", 2, None],                   # all tags empty
            ],
            index=pd.period_range(now, freq='H', periods=5),
            columns=["tag_one", "column_one", "column_two", "tag_three"]
        )

        expected_escaped_tags = (
            b"foo,tag_one=blue\\ orange,tag_three=hot\\=cold "
            b"column_one=\"1\",column_two=1i "
            b"0\n"
            b"foo,tag_one=red\\,green,tag_three=cold\\\\fire "
            b"column_one=\"2\",column_two=2i "
            b"3600000000000\n"
            b"foo,tag_one=some "
            b"column_one=\"2\",column_two=2i "
            b"7200000000000\n"
            b"foo,tag_one=some "
            b"column_one=\"2\",column_two=2i "
            b"10800000000000\n"
            b"foo "
            b"column_one=\"2\",column_two=2i "
            b"14400000000000\n"
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)
            cli = DataFrameClient(database='db')
            cli.write_points(dataframe, 'foo',
                             field_columns=['column_one', 'column_two'],
                             tag_columns=['tag_one', 'tag_three'])
            self.assertEqual(m.last_request.body, expected_escaped_tags)

    def test_write_points_from_dataframe_with_numeric_column_names(self):
        """Test write points from df with numeric cols."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        # df with numeric column names
        dataframe = pd.DataFrame(data=[["1", 1, 1.0], ["2", 2, 2.0]],
                                 index=[now, now + timedelta(hours=1)])

        expected = (
            b'foo,hello=there 0=\"1\",1=1i,2=1.0 0\n'
            b'foo,hello=there 0=\"2\",1=2i,2=2.0 3600000000000\n'
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')
            cli.write_points(dataframe, "foo", {"hello": "there"})

            self.assertEqual(m.last_request.body, expected)

    def test_write_points_from_dataframe_with_numeric_precision(self):
        """Test write points from df with numeric precision."""
        now = pd.Timestamp('1970-01-01 00:00+00:00')
        # df with numeric column names
        dataframe = pd.DataFrame(data=[["1", 1, 1.1111111111111],
                                       ["2", 2, 2.2222222222222]],
                                 index=[now, now + timedelta(hours=1)])

        expected_default_precision = (
            b'foo,hello=there 0=\"1\",1=1i,2=1.11111111111 0\n'
            b'foo,hello=there 0=\"2\",1=2i,2=2.22222222222 3600000000000\n'
        )

        expected_specified_precision = (
            b'foo,hello=there 0=\"1\",1=1i,2=1.1111 0\n'
            b'foo,hello=there 0=\"2\",1=2i,2=2.2222 3600000000000\n'
        )

        expected_full_precision = (
            b'foo,hello=there 0=\"1\",1=1i,2=1.1111111111111 0\n'
            b'foo,hello=there 0=\"2\",1=2i,2=2.2222222222222 3600000000000\n'
        )

        with requests_mock.Mocker() as m:
            m.register_uri(requests_mock.POST,
                           "http://localhost:8086/write",
                           status_code=204)

            cli = DataFrameClient(database='db')
            cli.write_points(dataframe, "foo", {"hello": "there"})

            self.assertEqual(m.last_request.body, expected_default_precision)

            cli = DataFrameClient(database='db')
            cli.write_points(dataframe, "foo", {"hello": "there"},
                             numeric_precision=4)

            self.assertEqual(m.last_request.body, expected_specified_precision)

            cli = DataFrameClient(database='db')
            cli.write_points(dataframe, "foo", {"hello": "there"},
                             numeric_precision='full')

            self.assertEqual(m.last_request.body, expected_full_precision)

    def test_write_points_from_dataframe_with_period_index(self):
        """Test write points from df with period index."""
        dataframe = pd.DataFrame(data=[["1", 1, 1.0], ["2", 2, 2.0]],