text: # 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)
  