Commit 9f1658b0 authored by David Douard's avatar David Douard
Browse files

[test/statsd] add tests for cubicweb.statsd_logger

parent c96dd92e480e
......@@ -88,19 +88,32 @@ def setup(bucket, address):
_socket = socket.socket(family, socket.SOCK_DGRAM)
def teardown():
"""Unconfigure the statsd endpoint
This is most likely only useful for unit tests"""
global _bucket, _address, _socket
_bucket = 'cubicweb'
_address = None
_socket = None
def statsd_c(context, n=1):
if _address is not None:
_socket.sendto('{0}.{1}:{2}|c\n'.format(_bucket, context, n), _address)
_socket.sendto('{0}.{1}:{2}|c\n'.format(_bucket, context, n).encode(),
_address)
def statsd_g(context, value):
if _address is not None:
_socket.sendto('{0}.{1}:{2}|g\n'.format(_bucket, context, value), _address)
_socket.sendto('{0}.{1}:{2}|g\n'.format(_bucket, context, value).encode(),
_address)
def statsd_t(context, value):
if _address is not None:
_socket.sendto('{0}.{1}:{2:.4f}|ms\n'.format(_bucket, context, value), _address)
_socket.sendto('{0}.{1}:{2:.4f}|ms\n'.format(_bucket, context, value).encode(),
_address)
class statsd_timeit(object):
......@@ -126,7 +139,7 @@ class statsd_timeit(object):
finally:
dt = 1000 * (time.time() - t0)
msg = '{0}.{1}:{2:.4f}|ms\n{0}.{1}:1|c\n'.format(
_bucket, self.__name__, dt)
_bucket, self.__name__, dt).encode()
_socket.sendto(msg, _address)
def __get__(self, obj, objtype):
......@@ -147,5 +160,5 @@ def statsd_timethis(ctxmsg):
if _address is not None:
dt = 1000 * (time.time() - t0)
msg = '{0}.{1}:{2:.4f}|ms\n{0}.{1}:1|c\n'.format(
_bucket, ctxmsg, dt)
_bucket, ctxmsg, dt).encode()
_socket.sendto(msg, _address)
# -*- coding: utf-8 -*-
# copyright 2018 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
#
# This file is part of CubicWeb.
#
# CubicWeb is free software: you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# CubicWeb is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with CubicWeb. If not, see <http://www.gnu.org/licenses/>.
"""unit tests for module cubicweb.statsd_logger"""
import threading
import socket
import time
import re
from unittest import TestCase
from cubicweb import statsd_logger as statsd
UDP_PORT = None
RUNNING = True
SOCK = socket.socket(socket.AF_INET,
socket.SOCK_DGRAM)
SOCK.settimeout(0.1)
STATSD = None
DATA = []
def statsd_rcv():
while RUNNING:
try:
data, addr = SOCK.recvfrom(1024)
if data:
rcv = [row.strip().decode() for row in data.splitlines()]
DATA.extend(rcv)
except socket.timeout:
pass
def setUpModule(*args):
global UDP_PORT, STATSD
SOCK.bind(('127.0.0.1', 0))
UDP_PORT = SOCK.getsockname()[1]
STATSD = threading.Thread(target=statsd_rcv)
STATSD.start()
statsd.setup('test', ('127.0.0.1', UDP_PORT))
def tearDownModule(*args):
global RUNNING
RUNNING = False
STATSD.join()
statsd.teardown()
class StatsdTC(TestCase):
def setUp(self):
super(StatsdTC, self).setUp()
DATA[:] = []
def check_received(self, value):
for i in range(10):
if value in DATA:
break
time.sleep(0.01)
else:
self.assertIn(value, DATA)
def check_received_ms(self, value):
value = re.compile(value.replace('?', '\d'))
for i in range(10):
if [x for x in DATA if value.match(x)]:
break
time.sleep(0.01)
else:
self.assertTrue([x for x in DATA if value.match(x)], DATA)
def test_statsd_c(self):
statsd.statsd_c('context')
self.check_received('test.context:1|c')
statsd.statsd_c('context', 10)
self.check_received('test.context:10|c')
def test_statsd_g(self):
statsd.statsd_g('context', 42)
self.check_received('test.context:42|g')
statsd.statsd_g('context', 'Igorrr')
self.check_received('test.context:Igorrr|g')
def test_statsd_t(self):
statsd.statsd_t('context', 1)
self.check_received('test.context:1.0000|ms')
statsd.statsd_t('context', 10)
self.check_received('test.context:10.0000|ms')
statsd.statsd_t('context', 0.12344)
self.check_received('test.context:0.1234|ms')
statsd.statsd_t('context', 0.12345)
self.check_received('test.context:0.1235|ms')
def test_decorator(self):
@statsd.statsd_timeit
def measure_me_please():
"some nice function"
return 42
self.assertEqual(measure_me_please.__doc__,
"some nice function")
measure_me_please()
self.check_received_ms('test.measure_me_please:0.0???|ms')
self.check_received('test.measure_me_please:1|c')
def test_context_manager(self):
with statsd.statsd_timethis('cm'):
time.sleep(0.1)
self.check_received_ms('test.cm:100.????|ms')
self.check_received('test.cm:1|c')
if __name__ == '__main__':
from unittest import main
main()
......@@ -106,6 +106,7 @@ cubicweb/test/unittest_rqlrewrite.py
cubicweb/test/unittest_rset.py
cubicweb/test/unittest_rtags.py
cubicweb/test/unittest_schema.py
cubicweb/test/unittest_statsd.py
cubicweb/test/unittest_toolsutils.py
cubicweb/test/unittest_wfutils.py
cubicweb/web/application.py
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment