105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
from django.conf import settings
|
|
from django.http import HttpResponse
|
|
from django.test.client import RequestFactory
|
|
|
|
import mock
|
|
from nose.tools import eq_
|
|
|
|
from django_statsd import statsd, get_client, middleware
|
|
|
|
import unittest
|
|
|
|
|
|
@mock.patch.object(middleware.statsd, 'incr')
|
|
class TestIncr(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.req = RequestFactory().get('/')
|
|
self.res = HttpResponse()
|
|
|
|
def test_graphite_response(self, incr):
|
|
gmw = middleware.GraphiteMiddleware()
|
|
gmw.process_response(self.req, self.res)
|
|
assert incr.called
|
|
|
|
def test_graphite_response_authenticated(self, incr):
|
|
self.req.user = mock.Mock()
|
|
self.req.user.is_authenticated.return_value = True
|
|
gmw = middleware.GraphiteMiddleware()
|
|
gmw.process_response(self.req, self.res)
|
|
eq_(incr.call_count, 2)
|
|
|
|
def test_graphite_exception(self, incr):
|
|
gmw = middleware.GraphiteMiddleware()
|
|
gmw.process_exception(self.req, None)
|
|
assert incr.called
|
|
|
|
def test_graphite_exception_authenticated(self, incr):
|
|
self.req.user = mock.Mock()
|
|
self.req.user.is_authenticated.return_value = True
|
|
gmw = middleware.GraphiteMiddleware()
|
|
gmw.process_exception(self.req, None)
|
|
eq_(incr.call_count, 2)
|
|
|
|
|
|
@mock.patch.object(middleware.statsd, 'timing')
|
|
class TestTiming(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.req = RequestFactory().get('/')
|
|
self.res = HttpResponse()
|
|
|
|
def test_request_timing(self, timing):
|
|
func = lambda x: x
|
|
gmw = middleware.GraphiteRequestTimingMiddleware()
|
|
gmw.process_view(self.req, func, tuple(), dict())
|
|
gmw.process_response(self.req, self.res)
|
|
eq_(timing.call_count, 3)
|
|
names = ['view.%s.%s.GET' % (func.__module__, func.__name__),
|
|
'view.%s.GET' % func.__module__,
|
|
'view.GET']
|
|
for expected, (args, kwargs) in zip(names, timing.call_args_list):
|
|
eq_(expected, args[0])
|
|
|
|
def test_request_timing_exception(self, timing):
|
|
func = lambda x: x
|
|
gmw = middleware.GraphiteRequestTimingMiddleware()
|
|
gmw.process_view(self.req, func, tuple(), dict())
|
|
gmw.process_exception(self.req, self.res)
|
|
eq_(timing.call_count, 3)
|
|
names = ['view.%s.%s.GET' % (func.__module__, func.__name__),
|
|
'view.%s.GET' % func.__module__,
|
|
'view.GET']
|
|
for expected, (args, kwargs) in zip(names, timing.call_args_list):
|
|
eq_(expected, args[0])
|
|
|
|
|
|
class TestClient(unittest.TestCase):
|
|
|
|
@mock.patch_object(settings, 'STATSD_CLIENT', 'statsd.client')
|
|
def test_normal(self):
|
|
eq_(get_client().__module__, 'statsd.client')
|
|
|
|
@mock.patch_object(settings, 'STATSD_CLIENT',
|
|
'django_statsd.clients.null')
|
|
def test_null(self):
|
|
eq_(get_client().__module__, 'django_statsd.clients.null')
|
|
|
|
@mock.patch_object(settings, 'STATSD_CLIENT',
|
|
'django_statsd.clients.toolbar')
|
|
def test_toolbar(self):
|
|
eq_(get_client().__module__, 'django_statsd.clients.toolbar')
|
|
|
|
@mock.patch_object(settings, 'STATSD_CLIENT',
|
|
'django_statsd.clients.toolbar')
|
|
def test_toolbar_send(self):
|
|
client = get_client()
|
|
eq_(client.cache, [])
|
|
client.incr('testing')
|
|
eq_(client.cache, [['testing', u'1c', 1]])
|
|
|
|
|
|
|
|
|
|
|