1
"""Tests for metrics context manager."""
3
from mocker import ANY, MockerTestCase
4
from txstatsd.metrics import Measure
7
class MeasureTest(MockerTestCase):
8
"""Test case for the L{Measure} context manager."""
10
def test_measure(self):
12
operation_name = 'fake_function'
13
meter = self.mocker.mock()
14
meter.increment(operation_name)
15
meter.timing(operation_name, ANY)
16
meter.increment(operation_name + '.done')
17
meter.decrement(operation_name)
25
with Measure('test_prefix', 'fake_function', meter):
27
self.assertEqual([0], result)
29
def test_measure_timing(self):
30
"""Test the timing works."""
31
operation_name = 'fake_function'
32
MockTime = self.mocker.replace('time.time') # pylint: disable=C0103
33
self.expect(MockTime()).result(10)
34
self.expect(MockTime()).result(15)
35
meter = self.mocker.mock()
36
meter.increment(operation_name)
37
meter.timing(operation_name, 5)
38
meter.increment(operation_name + '.done')
39
meter.decrement(operation_name)
45
with Measure('test_prefix', 'fake_function', meter):
48
def test_measure_handle_exceptions(self):
49
"""Test exceptions."""
51
class TestException(Exception):
52
"""Exception used to test the wrapper."""
55
operation_name = 'fake_function'
56
meter = self.mocker.mock()
57
meter.increment(operation_name)
58
meter.increment(operation_name + '.error')
59
meter.decrement(operation_name)
63
"""Fake Method that raises an exception."""
67
with Measure('test_prefix', 'fake_function', meter):
72
self.fail('measure context manager did not reraise exception.')
1
"""Tests for the Metrics convenience class."""
3
from unittest import TestCase
5
from txstatsd.metrics.metrics import Metrics
8
class FakeStatsDClient(object):
11
"""Connect to the StatsD server."""
15
"""Disconnect from the StatsD server."""
18
def write(self, data):
19
"""Send the metric to the StatsD server."""
23
class TestMetrics(TestCase):
26
self.connection = FakeStatsDClient()
27
self.metrics = Metrics(self.connection, 'txstatsd.tests')
30
"""Test reporting of a gauge metric sample."""
31
self.metrics.gauge('gauge', 102)
32
self.assertEqual(self.connection.data,
33
'txstatsd.tests.gauge:102|g')
35
def test_counter(self):
36
"""Test the increment and decrement operations."""
37
self.metrics.increment('counter', 18)
38
self.assertEqual(self.connection.data,
39
'txstatsd.tests.counter:18|c')
40
self.metrics.decrement('counter', 9)
41
self.assertEqual(self.connection.data,
42
'txstatsd.tests.counter:-9|c')
44
def test_timing(self):
45
"""Test the timing operation."""
46
self.metrics.timing('timing', 101123)
47
self.assertEqual(self.connection.data,
48
'txstatsd.tests.timing:101123|ms')
50
def test_empty_namespace(self):
51
"""Test reporting of an empty namespace."""
52
self.metrics.namespace = None
53
self.metrics.gauge('gauge', 213)
54
self.assertEqual(self.connection.data,
57
self.metrics.namespace = ''
58
self.metrics.gauge('gauge', 413)
59
self.assertEqual(self.connection.data,
added
removed
txstatsd/tests/test_metrics.py
