Skip to content

Commit

Permalink
Simplify opencensus quickstart tests. (#2990)
Browse files Browse the repository at this point in the history
* Simplify opencensus tests.

* Fix lint error

* Only check stdout.

* Remember this is python, not java.
  • Loading branch information
kurtisvg authored Feb 26, 2020
1 parent fc80b56 commit ff9682e
Show file tree
Hide file tree
Showing 2 changed files with 8 additions and 65 deletions.
3 changes: 2 additions & 1 deletion opencensus/metrics_quickstart.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,12 +62,13 @@ def main():
# Record 100 fake latency values between 0 and 5 seconds.
for num in range(100):
ms = random() * 5 * 1000
print("Latency {}: {}".format(num, ms))

mmap = stats.stats.stats_recorder.new_measurement_map()
mmap.measure_float_put(LATENCY_MS, ms)
mmap.record()

print("Fake latency recorded ({}: {})".format(num, ms))

# Keep the thread alive long enough for the exporter to export at least
# once.
time.sleep(65)
Expand Down
70 changes: 6 additions & 64 deletions opencensus/metrics_quickstart_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,69 +12,11 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from unittest import mock
import unittest
import metrics_quickstart

from opencensus.ext.stackdriver import stats_exporter
from opencensus.stats import aggregation
from opencensus.stats import measure
from opencensus.stats import stats
from opencensus.stats import view


class TestMetricsQuickstart(unittest.TestCase):
"""Sanity checks for the OpenCensus metrics quickstart examples.
These tests check that a few basic features of the library work as expected
in order for the demo to run. See the opencensus and
opencensus-ext-stackdriver source for actual library tests.
"""
def test_measure_creation(self):
measure.MeasureFloat(
"task_latency",
"The task latency in milliseconds",
"ms")

def test_view_creation(self):
test_view = view.View(
"task_latency_distribution",
"The distribution of the task latencies",
[],
mock.Mock(),
aggregation.DistributionAggregation([1.0, 2.0, 3.0]))
# Check that metric descriptor conversion doesn't crash
test_view.get_metric_descriptor()

# Don't modify global stats
@mock.patch('opencensus.ext.stackdriver.stats_exporter.stats.stats',
stats._Stats())
def test_measurement_map_record(self):
mock_measure = mock.Mock()
mock_measure_name = mock.Mock()
mock_measure.name = mock_measure_name
mock_view = mock.Mock()
mock_view.columns = []
mock_view.measure = mock_measure

stats.stats.view_manager.register_view(mock_view)

mmap = stats.stats.stats_recorder.new_measurement_map()
mmap.measure_float_put(mock_measure, 1.0)
mmap.record()

# Reaching into the stats internals here to check that recording the
# measurement map affects view data.
m2vd = (stats.stats.view_manager.measure_to_view_map
._measure_to_view_data_list_map)
self.assertIn(mock_measure_name, m2vd)
[view_data] = m2vd[mock_measure_name]
agg_data = view_data.tag_value_aggregation_data_map[tuple()]
agg_data.add_sample.assert_called_once()

@mock.patch('opencensus.ext.stackdriver.stats_exporter'
'.monitoring_v3.MetricServiceClient')
def test_new_stats_exporter(self, mock_client):
transport = stats_exporter.new_stats_exporter()
self.assertIsNotNone(transport)
self.assertIsNotNone(transport.options)
self.assertIsNotNone(transport.options.project_id)
def test_quickstart_main(capsys):
# Run the quickstart, making sure that it runs successfully
metrics_quickstart.main()
output = capsys.readouterr()
assert "Fake latency recorded" in output.out

0 comments on commit ff9682e

Please sign in to comment.