fixing vm_metrics tests
@@ -6,6 +6,7 @@ from unittest import mock
import vm_metrics
from unittest import TestCase
from google.cloud import monitoring_v3
+from google.api import distribution_pb2
import os
TEST_PATH = './testdata'
@@ -95,12 +96,70 @@ def dict_to_obj(dict1):
"""Converting dictionary into object."""
return json.loads(json.dumps(dict1), object_hook=MetricsResponseObject)
-
-def get_response_from_filename(filename):
+def create_metrics_value_by_type(value, value_type) -> float:
+ """Parses the value from a value object in API response.
+
+ Args:
+ value (object): The value object from API response
+ value_type (int) : Integer representing the value type of the object, refer
+ https://cloud.google.com/monitoring/api/ref_v3/rest/v3/TypedValue.
+ """
+ if value_type == 1:
+ return monitoring_v3.TypedValue({
+ "bool_value": value.bool_value
+ })
+ elif value_type == 2:
+ return monitoring_v3.TypedValue({
+ "int64_value": value.int64_value
+ })
+ elif value_type == 3:
+ return monitoring_v3.TypedValue({
+ "double_value": value.double_value
+ })
+ elif value_type == 4:
+ return monitoring_v3.TypedValue({
+ "string_value": value.string_value
+ })
+ elif value_type == 5:
+ explicit = distribution_pb2.Distribution.BucketOptions.Explicit()
+ explicit.bounds.append(value.distribution_value.bucket_options.explicit_buckets.bounds)
+ bucket_options = distribution_pb2.Distribution.BucketOptions()
+ bucket_options.explicit_buckets.CopyFrom(explicit)
+
+ distribution = distribution_pb2.Distribution()
+ distribution.count = value.distribution_value.count
+ distribution.mean = value.distribution_value.mean
+ distribution.sum_of_squared_deviation = value.distribution_value.sum_of_squared_deviation
+ distribution.bucket_options.CopyFrom(bucket_options)
+
+ return monitoring_v3.TypedValue({
+ "distribution_value": distribution
+ })
+ else:
+ raise Exception('Unhandled Value type')
+
+def get_response_from_filename(filename, ):
response_filepath = os.path.join(TEST_PATH, filename + '.json')
metrics_file = open(response_filepath, 'r')
metrics_response = dict_to_obj(json.load(metrics_file))
- return metrics_response
+ metrics_file.close()
+
+ # Timeseries doesn't have a pb2 file. Hence we cannot deserialize json into
+ # timeseries object. Explicitly creating a timeseries object by reading json
+ # data as dictionary.
+ time_series = monitoring_v3.TimeSeries()
+ time_series.metric.type = metrics_response.metric.type
+ time_series.value_type = metrics_response.value_type
+ time_series.resource.type = metrics_response.resource.type
+ for point in metrics_response.points:
+ interval = monitoring_v3.TimeInterval(
+ {"end_time": {"seconds": point.interval.end_time.seconds},
+ "start_time": {"seconds": point.interval.start_time.seconds}}
+ )
+ value = create_metrics_value_by_type(point.value, metrics_response.value_type)
+ point_val = monitoring_v3.Point({"interval": interval, "value": value})
+ time_series.points.append(point_val)
+ return time_series
class TestVmmetricsTest(unittest.TestCase):
