# -*- coding: utf-8 -*-
"""
unit tests for checking the good/expected interaction between :

+ the python client.. (obviously)
+ and a *_real_* server instance running.

This basically duplicates what's in client_test.py
 but without mocking around every call.

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

from functools import partial
import os
import time
import unittest
import warnings

# By default, raise exceptions on warnings
warnings.simplefilter('error', FutureWarning)

from influxdb import InfluxDBClient
from influxdb.exceptions import InfluxDBClientError

from influxdb.tests import skipIfPYpy, using_pypy, skipServerTests
from influxdb.tests.server_tests.base import ManyTestCasesWithServerMixin
from influxdb.tests.server_tests.base import SingleTestCaseWithServerMixin

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


THIS_DIR = os.path.abspath(os.path.dirname(__file__))


def point(serie_name, timestamp=None, tags=None, **fields):
    res = {'measurement': serie_name}
    if timestamp:
        res['time'] = timestamp
    if tags:
        res['tags'] = tags
    res['fields'] = fields
    return res


dummy_point = [  # some dummy points
    {
        "measurement": "cpu_load_short",
        "tags": {
            "host": "server01",
            "region": "us-west"
        },
        "time": "2009-11-10T23:00:00Z",
        "fields": {
            "value": 0.64
        }
    }
]

dummy_points = [  # some dummy points
    dummy_point[0],
    {
        "measurement": "memory",
        "tags": {
            "host": "server01",
            "region": "us-west"
        },
        "time": "2009-11-10T23:01:35Z",
        "fields": {
            "value": 33.0
        }
    }
]

if not using_pypy:
    dummy_pointDF = {
        "measurement": "cpu_load_short",
        "tags": {"host": "server01",
                 "region": "us-west"},
        "dataframe": pd.DataFrame(
            [[0.64]], columns=['value'],
            index=pd.to_datetime(["2009-11-10T23:00:00Z"]))
    }
    dummy_pointsDF = [{
        "measurement": "cpu_load_short",
        "tags": {"host": "server01", "region": "us-west"},
        "dataframe": pd.DataFrame(
            [[0.64]], columns=['value'],
            index=pd.to_datetime(["2009-11-10T23:00:00Z"])),
    }, {
        "measurement": "memory",
        "tags": {"host": "server01", "region": "us-west"},
        "dataframe": pd.DataFrame(
            [[33]], columns=['value'],
            index=pd.to_datetime(["2009-11-10T23:01:35Z"])
        )
    }]


dummy_point_without_timestamp = [
    {
        "measurement": "cpu_load_short",
        "tags": {
            "host": "server02",
            "region": "us-west"
        },
        "fields": {
            "value": 0.64
        }
    }
]


@skipServerTests
class SimpleTests(SingleTestCaseWithServerMixin,
                  unittest.TestCase):

    influxdb_template_conf = os.path.join(THIS_DIR, 'influxdb.conf.template')

    def test_fresh_server_no_db(self):
        self.assertEqual([], self.cli.get_list_database())

    def test_create_database(self):
        self.assertIsNone(self.cli.create_database('new_db_1'))
        self.assertIsNone(self.cli.create_database('new_db_2'))
        self.assertEqual(
            self.cli.get_list_database(),
            [{'name': 'new_db_1'}, {'name': 'new_db_2'}]
        )

    def test_drop_database(self):
        self.test_create_database()
        self.assertIsNone(self.cli.drop_database('new_db_1'))
        self.assertEqual([{'name': 'new_db_2'}], self.cli.get_list_database())

    def test_query_fail(self):
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.query('select column_one from foo')
        self.assertIn('database not found: db',
                      ctx.exception.content)

    def test_query_fail_ignore_errors(self):
        result = self.cli.query('select column_one from foo',
                                raise_errors=False)
        self.assertEqual(result.error, 'database not found: db')

    def test_create_user(self):
        self.cli.create_user('test_user', 'secret_password')
        rsp = list(self.cli.query("SHOW USERS")['results'])
        self.assertIn({'user': 'test_user', 'admin': False},
                      rsp)

    def test_create_user_admin(self):
        self.cli.create_user('test_user', 'secret_password', True)
        rsp = list(self.cli.query("SHOW USERS")['results'])
        self.assertIn({'user': 'test_user', 'admin': True},
                      rsp)

    def test_create_user_blank_password(self):
        self.cli.create_user('test_user', '')
        rsp = list(self.cli.query("SHOW USERS")['results'])
        self.assertIn({'user': 'test_user', 'admin': False},
                      rsp)

    def test_get_list_users_empty(self):
        rsp = self.cli.get_list_users()
        self.assertEqual([], rsp)

    def test_get_list_users(self):
        self.cli.query("CREATE USER test WITH PASSWORD 'test'")
        rsp = self.cli.get_list_users()

        self.assertEqual(
            [{'user': 'test', 'admin': False}],
            rsp
        )

    def test_create_user_blank_username(self):
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.create_user('', 'secret_password')
        self.assertIn('username required',
                      ctx.exception.content)
        rsp = list(self.cli.query("SHOW USERS")['results'])
        self.assertEqual(rsp, [])

    def test_drop_user(self):
        self.cli.query("CREATE USER test WITH PASSWORD 'test'")
        self.cli.drop_user('test')
        users = list(self.cli.query("SHOW USERS")['results'])
        self.assertEqual(users, [])

    def test_drop_user_nonexisting(self):
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.drop_user('test')
        self.assertIn('user not found',
                      ctx.exception.content)

    def test_drop_user_invalid(self):
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.drop_user('very invalid')
        self.assertEqual(400, ctx.exception.code)
        self.assertIn('{"error":"error parsing query: '
                      'found invalid, expected',
                      ctx.exception.content)

    @unittest.skip("Broken as of 0.9.0")
    def test_revoke_admin_privileges(self):
        self.cli.create_user('test', 'test', admin=True)
        self.assertEqual([{'user': 'test', 'admin': True}],
                         self.cli.get_list_users())
        self.cli.revoke_admin_privileges('test')
        self.assertEqual([{'user': 'test', 'admin': False}],
                         self.cli.get_list_users())

    def test_revoke_admin_privileges_invalid(self):
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.revoke_admin_privileges('')
        self.assertEqual(400, ctx.exception.code)
        self.assertIn('{"error":"error parsing query: ',
                      ctx.exception.content)

    def test_grant_privilege(self):
        self.cli.create_user('test', 'test')
        self.cli.create_database('testdb')
        self.cli.grant_privilege('all', 'testdb', 'test')
        # TODO: when supported by InfluxDB, check if privileges are granted

    def test_grant_privilege_invalid(self):
        self.cli.create_user('test', 'test')
        self.cli.create_database('testdb')
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.grant_privilege('', 'testdb', 'test')
        self.assertEqual(400, ctx.exception.code)
        self.assertIn('{"error":"error parsing query: ',
                      ctx.exception.content)

    def test_revoke_privilege(self):
        self.cli.create_user('test', 'test')
        self.cli.create_database('testdb')
        self.cli.revoke_privilege('all', 'testdb', 'test')
        # TODO: when supported by InfluxDB, check if privileges are revoked

    def test_revoke_privilege_invalid(self):
        self.cli.create_user('test', 'test')
        self.cli.create_database('testdb')
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.revoke_privilege('', 'testdb', 'test')
        self.assertEqual(400, ctx.exception.code)
        self.assertIn('{"error":"error parsing query: ',
                      ctx.exception.content)

    def test_invalid_port_fails(self):
        with self.assertRaises(ValueError):
            InfluxDBClient('host', '80/redir', 'username', 'password')


@skipServerTests
class CommonTests(ManyTestCasesWithServerMixin,
                  unittest.TestCase):

    influxdb_template_conf = os.path.join(THIS_DIR, 'influxdb.conf.template')

    def test_write(self):
        self.assertIs(True, self.cli.write(
            {'points': dummy_point},
            params={'db': 'db'},
        ))

    def test_write_check_read(self):
        self.test_write()
        time.sleep(1)
        rsp = self.cli.query('SELECT * FROM cpu_load_short', database='db')
        self.assertListEqual([{'value': 0.64, 'time': '2009-11-10T23:00:00Z',
                               "host": "server01", "region": "us-west"}],
                             list(rsp.get_points()))

    def test_write_points(self):
        self.assertIs(True, self.cli.write_points(dummy_point))

    @skipIfPYpy
    def test_write_points_DF(self):
        self.assertIs(
            True,
            self.cliDF.write_points(
                dummy_pointDF['dataframe'],
                dummy_pointDF['measurement'],
                dummy_pointDF['tags']
            )
        )

    def test_write_points_check_read(self):
        self.test_write_points()
        time.sleep(1)  # same as test_write_check_read()
        rsp = self.cli.query('SELECT * FROM cpu_load_short')

        self.assertEqual(
            list(rsp),
            [[
                {'value': 0.64,
                 'time': '2009-11-10T23:00:00Z',
                 "host": "server01",
                 "region": "us-west"}
            ]]
        )

        rsp2 = list(rsp.get_points())
        self.assertEqual(len(rsp2), 1)
        pt = rsp2[0]

        self.assertEqual(
            pt,
            {'time': '2009-11-10T23:00:00Z',
             'value': 0.64,
             "host": "server01",
             "region": "us-west"}
        )

    @unittest.skip("Broken as of 0.9.0")
    def test_write_points_check_read_DF(self):
        self.test_write_points_DF()
        time.sleep(1)  # same as test_write_check_read()

        rsp = self.cliDF.query('SELECT * FROM cpu_load_short')
        assert_frame_equal(
            rsp['cpu_load_short'],
            dummy_pointDF['dataframe']
        )

        # Query with Tags
        rsp = self.cliDF.query(
            "SELECT * FROM cpu_load_short GROUP BY *")
        assert_frame_equal(
            rsp[('cpu_load_short',
                 (('host', 'server01'), ('region', 'us-west')))],
            dummy_pointDF['dataframe']
        )

    def test_write_multiple_points_different_series(self):
        self.assertIs(True, self.cli.write_points(dummy_points))
        time.sleep(1)
        rsp = self.cli.query('SELECT * FROM cpu_load_short')
        lrsp = list(rsp)

        self.assertEqual(
            [[
                {'value': 0.64,
                 'time': '2009-11-10T23:00:00Z',
                 "host": "server01",
                 "region": "us-west"}
            ]],
            lrsp
        )

        rsp = list(self.cli.query('SELECT * FROM memory'))

        self.assertEqual(
            rsp,
            [[
                {'value': 33,
                 'time': '2009-11-10T23:01:35Z',
                 "host": "server01",
                 "region": "us-west"}
            ]]
        )

    @unittest.skip("Broken as of 0.9.0")
    def test_write_multiple_points_different_series_DF(self):
        for i in range(2):
            self.assertIs(
                True, self.cliDF.write_points(
                    dummy_pointsDF[i]['dataframe'],
                    dummy_pointsDF[i]['measurement'],
                    dummy_pointsDF[i]['tags']))
        time.sleep(1)
        rsp = self.cliDF.query('SELECT * FROM cpu_load_short')

        assert_frame_equal(
            rsp['cpu_load_short'],
            dummy_pointsDF[0]['dataframe']
        )

        rsp = self.cliDF.query('SELECT * FROM memory')
        assert_frame_equal(
            rsp['memory'],
            dummy_pointsDF[1]['dataframe']
        )

    def test_write_points_batch(self):
        dummy_points = [
            {"measurement": "cpu_usage", "tags": {"unit": "percent"},
             "time": "2009-11-10T23:00:00Z", "fields": {"value": 12.34}},
            {"measurement": "network", "tags": {"direction": "in"},
             "time": "2009-11-10T23:00:00Z", "fields": {"value": 123.00}},
            {"measurement": "network", "tags": {"direction": "out"},
             "time": "2009-11-10T23:00:00Z", "fields": {"value": 12.00}}
        ]
        self.cli.write_points(points=dummy_points,
                              tags={"host": "server01",
                                    "region": "us-west"},
                              batch_size=2)
        time.sleep(5)
        net_in = self.cli.query("SELECT value FROM network "
                                "WHERE direction='in'").raw
        net_out = self.cli.query("SELECT value FROM network "
                                 "WHERE direction='out'").raw
        cpu = self.cli.query("SELECT value FROM cpu_usage").raw
        self.assertIn(123, net_in['series'][0]['values'][0])
        self.assertIn(12, net_out['series'][0]['values'][0])
        self.assertIn(12.34, cpu['series'][0]['values'][0])

    def test_query(self):
        self.assertIs(True, self.cli.write_points(dummy_point))

    @unittest.skip('Not implemented for 0.9')
    def test_query_chunked(self):
        cli = InfluxDBClient(database='db')
        example_object = {
            'points': [
                [1415206250119, 40001, 667],
                [1415206244555, 30001, 7],
                [1415206228241, 20001, 788],
                [1415206212980, 10001, 555],
                [1415197271586, 10001, 23]
            ],
            'name': 'foo',
            'columns': [
                'time',
                'sequence_number',
                'val'
            ]
        }
        del cli
        del example_object
        # TODO ?

    def test_delete_series_invalid(self):
        with self.assertRaises(InfluxDBClientError):
            self.cli.delete_series()

    def test_default_retention_policy(self):
        rsp = self.cli.get_list_retention_policies()
        self.assertEqual(
            [
                {'name': 'default',
                 'duration': '0',
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'default': True}
            ],
            rsp
        )

    def test_create_retention_policy_default(self):
        self.cli.create_retention_policy('somename', '1d', 1, default=True)
        self.cli.create_retention_policy('another', '2d', 1, default=False)
        rsp = self.cli.get_list_retention_policies()

        self.assertEqual(
            [
                {'duration': '0',
                 'default': False,
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'name': 'default'},
                {'duration': '24h0m0s',
                 'default': True,
                 'replicaN': 1,
                 'shardGroupDuration': u'1h0m0s',
                 'name': 'somename'},
                {'duration': '48h0m0s',
                 'default': False,
                 'replicaN': 1,
                 'shardGroupDuration': u'24h0m0s',
                 'name': 'another'}
            ],
            rsp
        )

    def test_create_retention_policy(self):
        self.cli.create_retention_policy('somename', '1d', 1)
        rsp = self.cli.get_list_retention_policies()
        self.assertEqual(
            [
                {'duration': '0',
                 'default': True,
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'name': 'default'},
                {'duration': '24h0m0s',
                 'default': False,
                 'replicaN': 1,
                 'shardGroupDuration': u'1h0m0s',
                 'name': 'somename'}
            ],
            rsp
        )

    def test_alter_retention_policy(self):
        self.cli.create_retention_policy('somename', '1d', 1)

        # Test alter duration
        self.cli.alter_retention_policy('somename', 'db',
                                        duration='4d')
        rsp = self.cli.get_list_retention_policies()
        self.assertEqual(
            [
                {'duration': '0',
                 'default': True,
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'name': 'default'},
                {'duration': '96h0m0s',
                 'default': False,
                 'replicaN': 1,
                 'shardGroupDuration': u'24h0m0s',
                 'name': 'somename'}
            ],
            rsp
        )

        # Test alter replication
        self.cli.alter_retention_policy('somename', 'db',
                                        replication=4)
        rsp = self.cli.get_list_retention_policies()
        self.assertEqual(
            [
                {'duration': '0',
                 'default': True,
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'name': 'default'},
                {'duration': '96h0m0s',
                 'default': False,
                 'replicaN': 4,
                 'shardGroupDuration': u'24h0m0s',
                 'name': 'somename'}
            ],
            rsp
        )

        # Test alter default
        self.cli.alter_retention_policy('somename', 'db',
                                        default=True)
        rsp = self.cli.get_list_retention_policies()
        self.assertEqual(
            [
                {'duration': '0',
                 'default': False,
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'name': 'default'},
                {'duration': '96h0m0s',
                 'default': True,
                 'replicaN': 4,
                 'shardGroupDuration': u'24h0m0s',
                 'name': 'somename'}
            ],
            rsp
        )

    def test_alter_retention_policy_invalid(self):
        self.cli.create_retention_policy('somename', '1d', 1)
        with self.assertRaises(InfluxDBClientError) as ctx:
            self.cli.alter_retention_policy('somename', 'db')
        self.assertEqual(400, ctx.exception.code)
        self.assertIn('{"error":"error parsing query: ',
                      ctx.exception.content)
        rsp = self.cli.get_list_retention_policies()
        self.assertEqual(
            [
                {'duration': '0',
                 'default': True,
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'name': 'default'},
                {'duration': '24h0m0s',
                 'default': False,
                 'replicaN': 1,
                 'shardGroupDuration': u'1h0m0s',
                 'name': 'somename'}
            ],
            rsp
        )

    def test_drop_retention_policy(self):
        self.cli.create_retention_policy('somename', '1d', 1)

        # Test drop retention
        self.cli.drop_retention_policy('somename', 'db')
        rsp = self.cli.get_list_retention_policies()
        self.assertEqual(
            [
                {'duration': '0',
                 'default': True,
                 'replicaN': 1,
                 'shardGroupDuration': u'168h0m0s',
                 'name': 'default'}
            ],
            rsp
        )

    def test_issue_143(self):
        pt = partial(point, 'a_serie_name', timestamp='2015-03-30T16:16:37Z')
        pts = [
            pt(value=15),
            pt(tags={'tag_1': 'value1'}, value=5),
            pt(tags={'tag_1': 'value2'}, value=10),
        ]
        self.cli.write_points(pts)
        time.sleep(1)
        rsp = list(self.cli.query('SELECT * FROM a_serie_name GROUP BY tag_1'))

        self.assertEqual(
            [
                [{'value': 15, 'time': '2015-03-30T16:16:37Z'}],
                [{'value': 5, 'time': '2015-03-30T16:16:37Z'}],
                [{'value': 10, 'time': '2015-03-30T16:16:37Z'}]
            ],
            rsp
        )

        # a slightly more complex one with 2 tags values:
        pt = partial(point, 'serie2', timestamp='2015-03-30T16:16:37Z')
        pts = [
            pt(tags={'tag1': 'value1', 'tag2': 'v1'}, value=0),
            pt(tags={'tag1': 'value1', 'tag2': 'v2'}, value=5),
            pt(tags={'tag1': 'value2', 'tag2': 'v1'}, value=10),
        ]
        self.cli.write_points(pts)
        time.sleep(1)
        rsp = self.cli.query('SELECT * FROM serie2 GROUP BY tag1,tag2')

        self.assertEqual(
            [
                [{'value': 0, 'time': '2015-03-30T16:16:37Z'}],
                [{'value': 5, 'time': '2015-03-30T16:16:37Z'}],
                [{'value': 10, 'time': '2015-03-30T16:16:37Z'}]
            ],
            list(rsp)
        )

        all_tag2_equal_v1 = list(rsp[None, {'tag2': 'v1'}])

        self.assertEqual(
            [{'value': 0, 'time': '2015-03-30T16:16:37Z'},
             {'value': 10, 'time': '2015-03-30T16:16:37Z'}],
            all_tag2_equal_v1,
        )

    def test_query_multiple_series(self):
        pt = partial(point, 'serie1', timestamp='2015-03-30T16:16:37Z')
        pts = [
            pt(tags={'tag1': 'value1', 'tag2': 'v1'}, value=0),
        ]
        self.cli.write_points(pts)

        pt = partial(point, 'serie2', timestamp='1970-03-30T16:16:37Z')
        pts = [
            pt(tags={'tag1': 'value1', 'tag2': 'v1'},
               value=0, data1=33, data2="bla"),
        ]
        self.cli.write_points(pts)


@skipServerTests
class UdpTests(ManyTestCasesWithServerMixin,
               unittest.TestCase):

    influxdb_udp_enabled = True
    influxdb_template_conf = os.path.join(THIS_DIR,
                                          'influxdb.conf.template')

    def test_write_points_udp(self):
        cli = InfluxDBClient(
            'localhost',
            self.influxd_inst.http_port,
            'root',
            '',
            database='db',
            use_udp=True,
            udp_port=self.influxd_inst.udp_port
        )
        cli.write_points(dummy_point)

        # The points are not immediately available after write_points.
        # This is to be expected because we are using udp (no response !).
        # So we have to wait some time,
        time.sleep(3)  # 3 sec seems to be a good choice.
        rsp = self.cli.query('SELECT * FROM cpu_load_short')

        self.assertEqual(
            # this is dummy_points :
            [
                {'value': 0.64,
                 'time': '2009-11-10T23:00:00Z',
                 "host": "server01",
                 "region": "us-west"}
            ],
            list(rsp['cpu_load_short'])
        )