# Copyright 2016 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

try:
    import pandas
except ImportError:
    HAVE_PANDAS = False
else:
    HAVE_PANDAS = True  # pragma: NO COVER

import unittest


PROJECT = 'my-project'

INSTANCE_NAMES = ['instance-1', 'instance-2']
INSTANCE_ZONES = ['us-east1-a', 'us-east1-b']
INSTANCE_IDS = ['1234567890123456789', '9876543210987654321']

METRIC_TYPE = 'compute.googleapis.com/instance/cpu/utilization'
METRIC_LABELS = list({'instance_name': name} for name in INSTANCE_NAMES)

RESOURCE_TYPE = 'gce_instance'
RESOURCE_LABELS = list({
    'project_id': PROJECT,
    'zone': zone,
    'instance_id': instance_id,
} for zone, instance_id in zip(INSTANCE_ZONES, INSTANCE_IDS))

METRIC_KIND = 'GAUGE'
VALUE_TYPE = 'DOUBLE'

TIMESTAMPS = [
    '2016-04-06T22:05:00.042Z',
    '2016-04-06T22:05:01.042Z',
    '2016-04-06T22:05:02.042Z',
]

DIMENSIONS = len(TIMESTAMPS), len(INSTANCE_NAMES)
VALUES = list(0.1 * i for i in range(DIMENSIONS[1]))
ARRAY = [VALUES] * DIMENSIONS[0]


def parse_timestamps():
    from google.api_core import datetime_helpers

    return [
        datetime_helpers.from_rfc3339(t).replace(tzinfo=None)
        for t in TIMESTAMPS]


def generate_query_results():
    from google.cloud.monitoring_v3 import types

    def P(timestamp, value):
        interval = types.TimeInterval()
        interval.start_time.FromJsonString(timestamp)
        interval.end_time.FromJsonString(timestamp)
        return types.Point(interval=interval, value={'double_value': value})

    for metric_labels, resource_labels, value in zip(
            METRIC_LABELS, RESOURCE_LABELS, VALUES):
        yield types.TimeSeries(
            metric=types.Metric(type=METRIC_TYPE, labels=metric_labels),
            resource=types.MonitoredResource(
                type=RESOURCE_TYPE, labels=resource_labels),
            metric_kind=METRIC_KIND,
            value_type=VALUE_TYPE,
            points=[P(t, value) for t in TIMESTAMPS],
        )


@unittest.skipUnless(HAVE_PANDAS, 'No pandas')
class Test__build_dataframe(unittest.TestCase):

    def _call_fut(self, *args, **kwargs):
        from google.cloud.monitoring_v3._dataframe import _build_dataframe

        return _build_dataframe(*args, **kwargs)

    def test_both_label_and_labels_illegal(self):
        with self.assertRaises(ValueError):
            self._call_fut([], label='instance_name', labels=['zone'])

    def test_empty_labels_illegal(self):
        with self.assertRaises(ValueError):
            self._call_fut([], labels=[])

    def test_simple_label(self):
        iterable = generate_query_results()
        dataframe = self._call_fut(iterable, label='instance_name')

        self.assertEqual(dataframe.shape, DIMENSIONS)
        self.assertEqual(dataframe.values.tolist(), ARRAY)

        expected_headers = [(instance_name,)
                            for instance_name in INSTANCE_NAMES]
        self.assertEqual(list(dataframe.columns), expected_headers)
        self.assertIsNone(dataframe.columns.name)

        self.assertEqual(list(dataframe.index), parse_timestamps())
        self.assertIsNone(dataframe.index.name)

    def test_multiple_labels(self):
        NAMES = ['resource_type', 'instance_id']

        iterable = generate_query_results()
        dataframe = self._call_fut(iterable, labels=NAMES)

        self.assertEqual(dataframe.shape, DIMENSIONS)
        self.assertEqual(dataframe.values.tolist(), ARRAY)

        expected_headers = [(RESOURCE_TYPE, instance_id)
                            for instance_id in INSTANCE_IDS]
        self.assertEqual(list(dataframe.columns), expected_headers)
        self.assertEqual(dataframe.columns.names, NAMES)
        self.assertIsNone(dataframe.columns.name)

        self.assertEqual(list(dataframe.index), parse_timestamps())
        self.assertIsNone(dataframe.index.name)

    def test_multiple_labels_with_just_one(self):
        NAME = 'instance_id'
        NAMES = [NAME]

        iterable = generate_query_results()
        dataframe = self._call_fut(iterable, labels=NAMES)

        self.assertEqual(dataframe.shape, DIMENSIONS)
        self.assertEqual(dataframe.values.tolist(), ARRAY)

        expected_headers = [(instance_id,) for instance_id in INSTANCE_IDS]
        self.assertEqual(list(dataframe.columns), expected_headers)
        self.assertEqual(dataframe.columns.names, NAMES)
        self.assertIsNone(dataframe.columns.name)

        self.assertEqual(list(dataframe.index), parse_timestamps())
        self.assertIsNone(dataframe.index.name)

    def test_smart_labels(self):
        NAMES = ['resource_type', 'project_id',
                 'zone', 'instance_id',
                 'instance_name']

        iterable = generate_query_results()
        dataframe = self._call_fut(iterable)

        self.assertEqual(dataframe.shape, DIMENSIONS)
        self.assertEqual(dataframe.values.tolist(), ARRAY)

        expected_headers = [
            (RESOURCE_TYPE, PROJECT, zone, instance_id, instance_name)
            for zone, instance_id, instance_name
            in zip(INSTANCE_ZONES, INSTANCE_IDS, INSTANCE_NAMES)]
        self.assertEqual(list(dataframe.columns), expected_headers)
        self.assertEqual(dataframe.columns.names, NAMES)
        self.assertIsNone(dataframe.columns.name)

        self.assertEqual(list(dataframe.index), parse_timestamps())
        self.assertIsNone(dataframe.index.name)

    def test_empty_table_simple_label(self):
        dataframe = self._call_fut([], label='instance_name')
        self.assertEqual(dataframe.shape, (0, 0))
        self.assertIsNone(dataframe.columns.name)
        self.assertIsNone(dataframe.index.name)
        self.assertIsInstance(dataframe.index, pandas.DatetimeIndex)

    def test_empty_table_multiple_labels(self):
        NAMES = ['resource_type', 'instance_id']
        dataframe = self._call_fut([], labels=NAMES)
        self.assertEqual(dataframe.shape, (0, 0))
        self.assertEqual(dataframe.columns.names, NAMES)
        self.assertIsNone(dataframe.columns.name)
        self.assertIsNone(dataframe.index.name)
        self.assertIsInstance(dataframe.index, pandas.DatetimeIndex)

    def test_empty_table_multiple_labels_with_just_one(self):
        NAME = 'instance_id'
        NAMES = [NAME]
        dataframe = self._call_fut([], labels=NAMES)
        self.assertEqual(dataframe.shape, (0, 0))
        self.assertEqual(dataframe.columns.names, NAMES)
        self.assertIsNone(dataframe.columns.name)
        self.assertIsNone(dataframe.index.name)
        self.assertIsInstance(dataframe.index, pandas.DatetimeIndex)

    def test_empty_table_smart_labels(self):
        NAME = 'resource_type'
        NAMES = [NAME]
        dataframe = self._call_fut([])
        self.assertEqual(dataframe.shape, (0, 0))
        self.assertEqual(dataframe.columns.names, NAMES)
        self.assertIsNone(dataframe.columns.name)
        self.assertIsNone(dataframe.index.name)
        self.assertIsInstance(dataframe.index, pandas.DatetimeIndex)


class Test__sorted_resource_labels(unittest.TestCase):

    def _call_fut(self, labels):
        from google.cloud.monitoring_v3._dataframe import (
            _sorted_resource_labels)

        return _sorted_resource_labels(labels)

    def test_empty(self):
        self.assertEqual(self._call_fut([]), [])

    def test_sorted(self):
        from google.cloud.monitoring_v3._dataframe import TOP_RESOURCE_LABELS

        EXPECTED = TOP_RESOURCE_LABELS + ('other-1', 'other-2')
        self.assertSequenceEqual(self._call_fut(EXPECTED), EXPECTED)

    def test_reversed(self):
        from google.cloud.monitoring_v3._dataframe import TOP_RESOURCE_LABELS

        EXPECTED = TOP_RESOURCE_LABELS + ('other-1', 'other-2')
        INPUT = list(reversed(EXPECTED))
        self.assertSequenceEqual(self._call_fut(INPUT), EXPECTED)
