Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import unittest
from opencensus.metrics import label_value
from opencensus.metrics.export import point, time_series, value
START_TIMESTAMP = '2018-10-09T22:33:44.012345Z'
LABEL_VALUE1 = label_value.LabelValue('value one')
LABEL_VALUE2 = label_value.LabelValue('价值二')
LABEL_VALUES = (LABEL_VALUE1, LABEL_VALUE2)
POINTS = (point.Point(
value.ValueLong(1), "2018-10-09T23:33:44.012345Z"),
point.Point(
value.ValueLong(2), "2018-10-10T00:33:44.012345Z"),
point.Point(
value.ValueLong(3), "2018-10-10T01:33:44.012345Z"),
point.Point(
value.ValueLong(4), "2018-10-10T02:33:44.012345Z"),
point.Point(
value.ValueLong(5), "2018-10-10T03:33:44.012345Z"))
class TestTimeSeries(unittest.TestCase):
def test_init(self):
ts = time_series.TimeSeries(LABEL_VALUES, POINTS, START_TIMESTAMP)
self.assertEqual(ts.start_timestamp, START_TIMESTAMP)
self.assertEqual(ts.label_values, LABEL_VALUES)
self.assertEqual(ts.points, POINTS)
def test_init_invalid(self):
time_series.TimeSeries(LABEL_VALUES, POINTS, None)
def test_new_aggregation_data_explicit(self):
measure = mock.Mock(spec=measure_module.MeasureInt)
sum_aggregation = aggregation_module.SumAggregation(sum=1)
agg_data = sum_aggregation.new_aggregation_data(measure)
self.assertEqual(1, agg_data.sum_data)
self.assertEqual(value.ValueLong, agg_data.value_type)
def setUp(self):
self.double_value = value_module.ValueDouble(55.5)
self.long_value = value_module.ValueLong(9876543210)
self.timestamp = '2018-10-06T17:57:57.936475Z'
value_at_percentile = [summary_module.ValueAtPercentile(99.5, 10.2)]
snapshot = summary_module.Snapshot(10, 87.07, value_at_percentile)
self.summary = summary_module.Summary(10, 6.6, snapshot)
self.summary_value = value_module.ValueSummary(self.summary)
self.distribution_value = value_module.ValueDistribution(
100,
1000.0,
10.0,
value_module.BucketOptions(
value_module.Explicit(list(range(1, 10)))),
[value_module.Bucket(10, None) for ii in range(10)],
)
from opencensus.metrics.export import point, time_series, value
START_TIMESTAMP = '2018-10-09T22:33:44.012345Z'
LABEL_VALUE1 = label_value.LabelValue('value one')
LABEL_VALUE2 = label_value.LabelValue('价值二')
LABEL_VALUES = (LABEL_VALUE1, LABEL_VALUE2)
POINTS = (point.Point(
value.ValueLong(1), "2018-10-09T23:33:44.012345Z"),
point.Point(
value.ValueLong(2), "2018-10-10T00:33:44.012345Z"),
point.Point(
value.ValueLong(3), "2018-10-10T01:33:44.012345Z"),
point.Point(
value.ValueLong(4), "2018-10-10T02:33:44.012345Z"),
point.Point(
value.ValueLong(5), "2018-10-10T03:33:44.012345Z"))
class TestTimeSeries(unittest.TestCase):
def test_init(self):
ts = time_series.TimeSeries(LABEL_VALUES, POINTS, START_TIMESTAMP)
self.assertEqual(ts.start_timestamp, START_TIMESTAMP)
self.assertEqual(ts.label_values, LABEL_VALUES)
self.assertEqual(ts.points, POINTS)
def test_init_invalid(self):
time_series.TimeSeries(LABEL_VALUES, POINTS, None)
with self.assertRaises(ValueError):
time_series.TimeSeries(None, POINTS, START_TIMESTAMP)
with self.assertRaises(ValueError):
time_series.TimeSeries(LABEL_VALUES, None, START_TIMESTAMP)
def test_add_sample_int(self):
sum_data = 1
value = 3
sum_aggregation_data = aggregation_data_module.SumAggregationData(
value_type=value_module.ValueLong, sum_data=sum_data)
sum_aggregation_data.add_sample(value, None, None)
self.assertEqual(4, sum_aggregation_data.sum_data)
def test_to_type_class(self):
self.assertEqual(
metric_descriptor.MetricDescriptorType.to_type_class(
metric_descriptor.MetricDescriptorType.GAUGE_INT64),
value.ValueLong)
with self.assertRaises(ValueError):
metric_descriptor.MetricDescriptorType.to_type_class(10)
def test_get_metric(self):
derived_cumulative = cumulative.DerivedLongCumulative(
Mock(), Mock(), Mock(), [])
mock_fn = Mock()
mock_fn.return_value = 123
derived_cumulative.create_default_time_series(mock_fn)
now1 = Mock()
[ts] = derived_cumulative.get_metric(now1).time_series
[ts_point] = ts.points
self.assertEqual(ts_point.timestamp, now1)
self.assertEqual(ts_point.value.value, 123)
self.assertIsInstance(ts_point.value, value_module.ValueLong)
def to_point(self, timestamp):
"""Get a Point conversion of this aggregation.
:type timestamp: :class: `datetime.datetime`
:param timestamp: The time to report the point as having been recorded.
:rtype: :class: `opencensus.metrics.export.point.Point`
:return: a :class: `opencensus.metrics.export.value.ValueLong`-valued
Point with value equal to `count_data`.
"""
return point.Point(value.ValueLong(self.count_data), timestamp)
def to_point_value(self):
"""Get a point value conversion of the current value.
:rtype: :class:`opencensus.metrics.export.value.ValueLong`
:return: A converted `ValueLong`.
"""
return value_module.ValueLong(self.value)