Changeset View
Changeset View
Standalone View
Standalone View
swh/core/tests/test_statsd.py
| Show First 20 Lines • Show All 99 Lines • ▼ Show 20 Lines | |||||
| class TestStatsd(unittest.TestCase): | class TestStatsd(unittest.TestCase): | ||||
| def setUp(self): | def setUp(self): | ||||
| """ | """ | ||||
| Set up a default Statsd instance and mock the socket. | Set up a default Statsd instance and mock the socket. | ||||
| """ | """ | ||||
| # | # | ||||
| self.statsd = Statsd() | self.statsd = Statsd() | ||||
| self.statsd.socket = FakeSocket() | self.statsd._socket = FakeSocket() | ||||
| def recv(self): | def recv(self): | ||||
| return self.statsd.socket.recv() | return self.statsd.socket.recv() | ||||
| def test_set(self): | def test_set(self): | ||||
| self.statsd.set('set', 123) | self.statsd.set('set', 123) | ||||
| assert self.recv() == 'set:123|s' | assert self.recv() == 'set:123|s' | ||||
| ▲ Show 20 Lines • Show All 90 Lines • ▼ Show 20 Lines | class TestStatsd(unittest.TestCase): | ||||
| def assert_almost_equal(self, a, b, delta): | def assert_almost_equal(self, a, b, delta): | ||||
| self.assertTrue( | self.assertTrue( | ||||
| 0 <= abs(a - b) <= delta, | 0 <= abs(a - b) <= delta, | ||||
| "%s - %s not within %s" % (a, b, delta) | "%s - %s not within %s" % (a, b, delta) | ||||
| ) | ) | ||||
| def test_socket_error(self): | def test_socket_error(self): | ||||
| self.statsd.socket = BrokenSocket() | self.statsd._socket = BrokenSocket() | ||||
| self.statsd.gauge('no error', 1) | self.statsd.gauge('no error', 1) | ||||
| assert True, 'success' | assert True, 'success' | ||||
| def test_socket_timeout(self): | def test_socket_timeout(self): | ||||
| self.statsd.socket = SlowSocket() | self.statsd._socket = SlowSocket() | ||||
| self.statsd.gauge('no error', 1) | self.statsd.gauge('no error', 1) | ||||
| assert True, 'success' | assert True, 'success' | ||||
| def test_timed(self): | def test_timed(self): | ||||
| """ | """ | ||||
| Measure the distribution of a function's run time. | Measure the distribution of a function's run time. | ||||
| """ | """ | ||||
| @self.statsd.timed('timed.test') | @self.statsd.timed('timed.test') | ||||
| ▲ Show 20 Lines • Show All 175 Lines • ▼ Show 20 Lines | def test_batched(self): | ||||
| self.statsd.timing('timer', 123) | self.statsd.timing('timer', 123) | ||||
| self.statsd.close_buffer() | self.statsd.close_buffer() | ||||
| self.assertEqual('page.views:123|g\ntimer:123|ms', self.recv()) | self.assertEqual('page.views:123|g\ntimer:123|ms', self.recv()) | ||||
| def test_context_manager(self): | def test_context_manager(self): | ||||
| fake_socket = FakeSocket() | fake_socket = FakeSocket() | ||||
| with Statsd() as statsd: | with Statsd() as statsd: | ||||
| statsd.socket = fake_socket | statsd._socket = fake_socket | ||||
| statsd.gauge('page.views', 123) | statsd.gauge('page.views', 123) | ||||
| statsd.timing('timer', 123) | statsd.timing('timer', 123) | ||||
| self.assertEqual('page.views:123|g\ntimer:123|ms', fake_socket.recv()) | self.assertEqual('page.views:123|g\ntimer:123|ms', fake_socket.recv()) | ||||
| def test_batched_buffer_autoflush(self): | def test_batched_buffer_autoflush(self): | ||||
| fake_socket = FakeSocket() | fake_socket = FakeSocket() | ||||
| with Statsd() as statsd: | with Statsd() as statsd: | ||||
| statsd.socket = fake_socket | statsd._socket = fake_socket | ||||
| for i in range(51): | for i in range(51): | ||||
| statsd.increment('mycounter') | statsd.increment('mycounter') | ||||
| self.assertEqual( | self.assertEqual( | ||||
| '\n'.join(['mycounter:1|c' for i in range(50)]), | '\n'.join(['mycounter:1|c' for i in range(50)]), | ||||
| fake_socket.recv(), | fake_socket.recv(), | ||||
| ) | ) | ||||
| self.assertEqual('mycounter:1|c', fake_socket.recv()) | self.assertEqual('mycounter:1|c', fake_socket.recv()) | ||||
| def test_module_level_instance(self): | def test_module_level_instance(self): | ||||
| from swh.core.statsd import statsd | from swh.core.statsd import statsd | ||||
| self.assertTrue(isinstance(statsd, Statsd)) | self.assertTrue(isinstance(statsd, Statsd)) | ||||
| def test_instantiating_does_not_connect(self): | def test_instantiating_does_not_connect(self): | ||||
| local_statsd = Statsd() | local_statsd = Statsd() | ||||
| self.assertEqual(None, local_statsd.socket) | self.assertEqual(None, local_statsd._socket) | ||||
| def test_accessing_socket_opens_socket(self): | def test_accessing_socket_opens_socket(self): | ||||
| local_statsd = Statsd() | local_statsd = Statsd() | ||||
| try: | try: | ||||
| self.assertIsNotNone(local_statsd.get_socket()) | self.assertIsNotNone(local_statsd.socket) | ||||
| finally: | finally: | ||||
| local_statsd.socket.close() | local_statsd.close_socket() | ||||
| def test_accessing_socket_multiple_times_returns_same_socket(self): | def test_accessing_socket_multiple_times_returns_same_socket(self): | ||||
| local_statsd = Statsd() | local_statsd = Statsd() | ||||
| fresh_socket = FakeSocket() | fresh_socket = FakeSocket() | ||||
| local_statsd.socket = fresh_socket | local_statsd._socket = fresh_socket | ||||
| self.assertEqual(fresh_socket, local_statsd.get_socket()) | self.assertEqual(fresh_socket, local_statsd.socket) | ||||
| self.assertNotEqual(FakeSocket(), local_statsd.get_socket()) | self.assertNotEqual(FakeSocket(), local_statsd.socket) | ||||
| def test_tags_from_environment(self): | def test_tags_from_environment(self): | ||||
| with preserve_envvars('STATSD_TAGS'): | with preserve_envvars('STATSD_TAGS'): | ||||
| os.environ['STATSD_TAGS'] = 'country:china,age:45' | os.environ['STATSD_TAGS'] = 'country:china,age:45' | ||||
| statsd = Statsd() | statsd = Statsd() | ||||
| statsd.socket = FakeSocket() | statsd._socket = FakeSocket() | ||||
| statsd.gauge('gt', 123.4) | statsd.gauge('gt', 123.4) | ||||
| self.assertEqual('gt:123.4|g|#age:45,country:china', | self.assertEqual('gt:123.4|g|#age:45,country:china', | ||||
| statsd.socket.recv()) | statsd.socket.recv()) | ||||
| def test_tags_from_environment_and_constant(self): | def test_tags_from_environment_and_constant(self): | ||||
| with preserve_envvars('STATSD_TAGS'): | with preserve_envvars('STATSD_TAGS'): | ||||
| os.environ['STATSD_TAGS'] = 'country:china,age:45' | os.environ['STATSD_TAGS'] = 'country:china,age:45' | ||||
| statsd = Statsd(constant_tags={'country': 'canada'}) | statsd = Statsd(constant_tags={'country': 'canada'}) | ||||
| statsd.socket = FakeSocket() | statsd._socket = FakeSocket() | ||||
| statsd.gauge('gt', 123.4) | statsd.gauge('gt', 123.4) | ||||
| self.assertEqual('gt:123.4|g|#age:45,country:canada', | self.assertEqual('gt:123.4|g|#age:45,country:canada', | ||||
| statsd.socket.recv()) | statsd.socket.recv()) | ||||
| def test_tags_from_environment_warning(self): | def test_tags_from_environment_warning(self): | ||||
| with preserve_envvars('STATSD_TAGS'): | with preserve_envvars('STATSD_TAGS'): | ||||
| os.environ['STATSD_TAGS'] = 'valid:tag,invalid_tag' | os.environ['STATSD_TAGS'] = 'valid:tag,invalid_tag' | ||||
| with pytest.warns(UserWarning) as record: | with pytest.warns(UserWarning) as record: | ||||
| ▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Lines | def test_envvar_port(self): | ||||
| os.environ['STATSD_PORT'] = '12345' | os.environ['STATSD_PORT'] = '12345' | ||||
| local_statsd = Statsd() | local_statsd = Statsd() | ||||
| self.assertEqual(local_statsd.host, 'localhost') | self.assertEqual(local_statsd.host, 'localhost') | ||||
| self.assertEqual(local_statsd.port, 12345) | self.assertEqual(local_statsd.port, 12345) | ||||
| def test_namespace_added(self): | def test_namespace_added(self): | ||||
| local_statsd = Statsd(namespace='test-namespace') | local_statsd = Statsd(namespace='test-namespace') | ||||
| local_statsd.socket = FakeSocket() | local_statsd._socket = FakeSocket() | ||||
| local_statsd.gauge('gauge', 123.4) | local_statsd.gauge('gauge', 123.4) | ||||
| assert local_statsd.socket.recv() == 'test-namespace.gauge:123.4|g' | assert local_statsd.socket.recv() == 'test-namespace.gauge:123.4|g' | ||||
| def test_contextmanager_empty(self): | def test_contextmanager_empty(self): | ||||
| with self.statsd: | with self.statsd: | ||||
| assert True, 'success' | assert True, 'success' | ||||
| Show All 14 Lines | |||||