Skip to content

E
event-tracking
Commit 8bec259e by Gabe Mulley
Options

remove the sample django middleware from the project 

It was basically serving as an example, however, that can be included in the docs (or not at all).  Actual usage of this middleware was cumbersome since it is effectively frozen in the library.  It is expected that most people would want to customize this behavior dramatically.
parent b22a3929
Showing with 0 additions and 311 deletions
eventtracking/django/middleware.py deleted 100644 → 0
"""Middleware that can be included to add context to tracked events."""
from __future__ import absolute_import
from contextlib import contextmanager
import re
import logging
from eventtracking import tracker
from django.conf import settings
logger = logging.getLogger(__name__) # pylint: disable=invalid-name
@contextmanager
def failures_only_in_debug():
"""
Raises an error only when the Django setting DEBUG evaluates to True.
Otherwise it simply logs the exception information and swallows the error.
"""
try:
yield
except Exception: # pylint: disable=broad-except
if not getattr(settings, 'DEBUG', False):
logger.warn('Exception raised in safe block. Swallowing the error.', exc_info=True)
else:
raise
class TrackRequestContextMiddleware(object):
"""
Adds a bunch of data pulled out of the request to the tracker context.
Any application code (middleware, views etc) that execute after this
middleware will include a bunch of information from the request in the
context of any events they emit.
"""
CONTEXT_NAME = 'django.context'
def process_request(self, request):
"""Adds request variables to the Django tracker context."""
with failures_only_in_debug():
context = {
'session': self.get_session_key(request),
'user_primary_key': self.get_user_primary_key(request),
'username': self.get_username(request)
}
header_context_key_map = {
'REMOTE_ADDR': 'ip',
'SERVER_NAME': 'host',
'HTTP_USER_AGENT': 'agent',
'PATH_INFO': 'path'
}
for header_name, context_key in header_context_key_map.iteritems():
context[context_key] = request.META.get(header_name, '')
tracker.get_tracker().enter_context(self.CONTEXT_NAME, context)
def get_session_key(self, request):
"""Gets the Django session key from the request or an empty string if it isn't found"""
try:
return request.session.session_key
except AttributeError:
return ''
def get_user_primary_key(self, request):
"""Gets the primary key of the logged in Django user"""
try:
return request.user.pk
except AttributeError:
return ''
def get_username(self, request):
"""Gets the username of the logged in Django user"""
try:
return request.user.username
except AttributeError:
return ''
def process_response(self, request, response): # pylint: disable=unused-argument
"""Remove the request variable context from the tracker context stack"""
with failures_only_in_debug():
try:
tracker.get_tracker().exit_context(self.CONTEXT_NAME)
except KeyError:
# If an error occurred processing some other middleware it is possible
# this method could be called without process_request having been
# called, in that case, don't raise an exception here.
pass
return response
class TrackRequestMiddleware(object):
"""
Track all requests processed by Django.
Supported Settings::
`TRACKING_IGNORE_URL_PATTERNS`: A list of regular expressions that are executed
on `request.path_info`, if any of them matches then no event will be emitted
for the request. Defaults to `[]`.
`TRACKING_HTTP_REQUEST_EVENT_TYPE`: The event type for the events emitted on
every request. Defaults to "http.request."
Example event::
{
"event_type": "http.request"
# ...
"data": {
"method": "GET",
"query": {
"arbitrary_get_parameter_name": "arbitrary_get_parameter_value",
# ... (all GET parameters will be included here)
},
"body": {
"arbitrary_post_parameter_name": "arbitrary_post_parameter_value",
# ... (all POST parameters will be included here)
}
}
}
"""
def process_response(self, request, response):
"""Emit an event for every request"""
with failures_only_in_debug():
if not self._should_process_request(request):
return response
event_type = getattr(settings, 'TRACKING_HTTP_REQUEST_EVENT_TYPE', 'http.request')
event = {
'method': request.method,
'query': self._remove_sensitive_request_variables(request.GET),
'body': self._remove_sensitive_request_variables(request.POST)
}
tracker.emit(event_type, event)
return response
def _remove_sensitive_request_variables(self, variable_dict):
"""Remove passwords and other sensitive data from the dictionary"""
cleaned_dict = dict()
for key, value in variable_dict.iteritems():
if 'password' in key:
value = ''
cleaned_dict[key] = value
return cleaned_dict
def _should_process_request(self, request):
"""Ignore requests for paths that match a set of ignored patterns"""
ignored_url_patterns = getattr(settings, 'TRACKING_IGNORE_URL_PATTERNS', [])
for pattern in ignored_url_patterns:
# Note we are explicitly relying on python's internal caching of
# compiled regular expressions here.
if re.match(pattern, request.path_info):
return False
return True
eventtracking/django/tests/test_middleware.py deleted 100644 → 0
"""Tests for the middleware"""
from __future__ import absolute_import
from mock import patch
from mock import sentinel
from django.test import TestCase
from django.test.client import RequestFactory
from django.test.utils import override_settings
from eventtracking.django.middleware import TrackRequestContextMiddleware
from eventtracking.django.middleware import TrackRequestMiddleware
class TestTrackRequestContextMiddleware(TestCase):
"""Test middleware that adds context to every request"""
def setUp(self):
get_tracker_patcher = patch('eventtracking.django.middleware.tracker.get_tracker')
mock_get_tracker = get_tracker_patcher.start()
self.addCleanup(get_tracker_patcher.stop)
self.mock_tracker = mock_get_tracker.return_value
self.middleware = TrackRequestContextMiddleware()
self.request_factory = RequestFactory()
self.request = self.request_factory.get('/somewhere')
def test_simple_request(self):
self.middleware.process_request(self.request)
self.mock_tracker.enter_context.assert_called_once_with(
'django.context',
{
'username': '',
'user_primary_key': '',
'ip': '127.0.0.1',
'agent': '',
'host': 'testserver',
'session': '',
'path': '/somewhere'
}
)
resp = self.middleware.process_response(self.request, sentinel.response)
self.mock_tracker.exit_context.assert_called_once_with('django.context')
self.assertEquals(resp, sentinel.response)
def test_response_without_request(self):
self.mock_tracker.exit_context.side_effect = KeyError
resp = self.middleware.process_response(self.request, sentinel.response)
self.assertEquals(resp, sentinel.response)
class TestTrackRequestMiddleware(TestCase):
"""Test the middleware that tracks every request"""
def setUp(self):
self.track_middleware = TrackRequestMiddleware()
self.request_factory = RequestFactory()
track_patcher = patch('eventtracking.django.middleware.tracker.emit')
self.mock_track = track_patcher.start()
self.addCleanup(track_patcher.stop)
def test_normal_request(self):
request = self.request_factory.get('/somewhere')
self.track_middleware.process_response(request, None)
self.assert_event_was_emitted()
def assert_event_was_emitted(self):
"""Fail if no event was emitted"""
self.assertTrue(self.mock_track.called)
@override_settings(TRACKING_IGNORE_URL_PATTERNS=[])
def test_reading_filtered_urls_from_settings(self):
request = self.request_factory.get('/event')
self.track_middleware.process_response(request, None)
self.assert_event_was_emitted()
@override_settings(TRACKING_IGNORE_URL_PATTERNS=[r'^/some/excluded.*'])
def test_anchoring_of_patterns_at_beginning(self):
request = self.request_factory.get('/excluded')
self.track_middleware.process_response(request, None)
self.assert_event_was_emitted()
self.mock_track.reset_mock()
request = self.request_factory.get('/some/excluded/url')
self.track_middleware.process_response(request, None)
self.assert_event_was_not_emitted()
def assert_event_was_not_emitted(self):
"""Fail if an event was emitted"""
self.assertFalse(self.mock_track.called)
@override_settings(DEBUG=False)
def test_does_not_fail(self):
self.mock_track.side_effect = Exception
request = self.request_factory.get('/anywhere')
self.track_middleware.process_response(request, None)
# Ensure the exception isn't propogated out to the caller
@override_settings(DEBUG=True)
def test_fails_in_debug_mode(self):
self.mock_track.side_effect = Exception
request = self.request_factory.get('/anywhere')
with self.assertRaises(Exception):
self.track_middleware.process_response(request, None)
def test_get_request_parameters_are_included(self):
test_request_params = dict()
for param in ['foo', 'bar', 'baz']:
test_request_params[param] = param
request = self.request_factory.get('/anywhere', test_request_params)
self.track_middleware.process_response(request, None)
self.assert_exactly_one_event_emitted_with({'method': 'GET', 'body': {}, 'query': test_request_params})
def assert_exactly_one_event_emitted_with(self, data):
"""Ensure exactly one event was emitted with the given data"""
self.mock_track.assert_called_once_with('http.request', data)
def test_passwords_stripped_from_request(self):
test_request_params = dict()
expected_event_data = dict()
for param in ['password', 'newpassword', 'new_password', 'oldpassword', 'old_password']:
test_request_params[param] = sentinel.password
expected_event_data[param] = ''
test_request_params['other_field'] = 'other value'
expected_event_data['other_field'] = test_request_params['other_field']
request = self.request_factory.get('/anywhere', test_request_params)
self.track_middleware.process_response(request, None)
self.assert_exactly_one_event_emitted_with({'method': 'GET', 'body': {}, 'query': expected_event_data})
@override_settings(TRACKING_HTTP_REQUEST_EVENT_TYPE='custom.event')
def test_setting_override_event_type(self):
request = self.request_factory.get('/anywhere')
self.track_middleware.process_response(request, None)
self.mock_track.assert_called_once_with('custom.event', {'method': 'GET', 'body': {}, 'query': {}})
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