Commits

Simon Law  committed 7bd4682

Move experiments.analytics to lean_analytics.

  • Participants
  • Parent commits 577f6f4

Comments (0)

Files changed (13)

File django_lean/experiments/analytics/__init__.py

-from django.conf import settings
-from django.core.urlresolvers import get_callable
-
-
-class IdentificationError(RuntimeError):
-    pass
-
-
-def get_all_analytics_names():
-    return getattr(settings, 'LEAN_ANALYTICS', ())
-
-def get_all_analytics():
-    if get_all_analytics.cache is None:
-        names = get_all_analytics_names()
-        get_all_analytics.cache = [get_callable(a)() for a in names]
-    return get_all_analytics.cache
-
-def reset_caches():
-    get_all_analytics.cache = None
-reset_caches()

File django_lean/experiments/analytics/base.py

-from django_lean.experiments.analytics import IdentificationError
-from django_lean.experiments.models import Participant
-from django_lean.experiments.utils import WebUser
-
-
-class BaseAnalytics(object):
-    def _id_from_session(self, session):
-        try:
-            return 'Session %s' % session.session_key
-        except AttributeError, e:
-            raise IdentificationError(e)
-
-    def _id_from_user(self, user):
-        try:
-            return 'User %d' % user.pk
-        except AttributeError, e:
-            raise IdentificationError(e)
-
-    def _compute_id(self, experiment_user):
-        if not experiment_user.is_anonymous():
-            return self._id_from_user(experiment_user.user)
-        return self._id_from_session(experiment_user.session)
-
-    def enroll(self, experiment, experiment_user, group_id):
-        self._submit(name='Enrolled In Experiment',
-                     properties={'Experiment': unicode(experiment),
-                                 'Group': dict(Participant.GROUPS)[group_id]},
-                     experiment_user=experiment_user)
-
-    def record(self, goal_record, experiment_user):
-        self._submit(name='Goal Recorded',
-                     properties={'Goal Type': unicode(goal_record.goal_type)},
-                     experiment_user=experiment_user)
-
-    def event(self, name, properties, request=None):
-        if request:
-            self._submit(name, properties, experiment_user=WebUser(request))
-
-    def _submit(self, name, properties, experiment_user=None):
-        raise NotImplementedError()

File django_lean/experiments/analytics/kissmetrics.py

-from django_kissmetrics.middleware import TrackingMiddleware
-
-from django_lean.experiments.analytics import IdentificationError
-from django_lean.experiments.analytics.base import BaseAnalytics
-
-
-class KissMetrics(BaseAnalytics):
-    def __init__(self, KM=None, middleware=None):
-        if middleware is None:
-            self.middleware = TrackingMiddleware()
-        if KM is None:
-            KM = self.middleware.KM
-        self.KM = KM
-
-    def _id_from_session(self, session):
-        id_from_session = self.middleware.id_from_session
-        try:
-            return id_from_session(session)
-        except AttributeError, e:
-            raise IdentificationError(e)
-
-    def _id_from_user(self, user):
-        id_from_user = self.middleware.id_from_user
-        try:
-            return id_from_user(user)
-        except AttributeError, e:
-            raise IdentificationError(e)
-
-    def _identify(self, experiment_user):
-        try:
-            self.KM.identify(self._compute_id(experiment_user))
-            return True
-        except IdentificationError:
-            # Ignore experiment_users who cannot be tied to sessions or users
-            return False
-
-    def _submit(self, name, properties, experiment_user=None):
-        if self._identify(experiment_user):
-            self.KM.record(action=name, props=properties)

File django_lean/experiments/analytics/mixpanel.py

-from __future__ import absolute_import
-
-import time
-
-from mixpanel.tasks import EventTracker
-
-from django_lean.experiments.analytics import IdentificationError
-from django_lean.experiments.analytics.base import BaseAnalytics
-
-
-class Mixpanel(BaseAnalytics):
-    def __init__(self, tracker=None, tracker_class=EventTracker):
-        if tracker is None:
-            tracker = EventTracker()
-        self.tracker = tracker
-        self.remote_addr = None
-        self.identity = None
-
-    def _identify(self, experiment_user):
-        self.identity = None
-        self.remote_addr = None
-        request = experiment_user.request
-        if hasattr(request, 'META'):
-            self.remote_addr = request.META.get('REMOTE_ADDR', None)
-        try:
-            self.identity = self._compute_id(experiment_user)
-            return True
-        except IdentificationError:
-            # Ignore experiment_users who cannot be tied to sessions or users
-            return False
-
-    def _properties(self, properties):
-        result = {'time': '%d' % time.mktime(time.gmtime())}
-        for key, attr in (('ip', 'remote_addr'), ('distinct_id', 'identity')):
-            if key not in properties:
-                value = getattr(self, attr)
-                if value:
-                    result[key] = value
-        result.update(properties)
-        return result
-
-    def _submit(self, name, properties, experiment_user=None):
-        if self._identify(experiment_user):
-            properties = self._properties(properties)
-            self.tracker.run(event_name=name, properties=properties)

File django_lean/experiments/models.py

 from django.db import models
 from django.core.exceptions import ObjectDoesNotExist
 
-from django_lean.experiments.analytics import get_all_analytics
 from django_lean.experiments.signals import goal_recorded, user_enrolled
 
 
     test_conversion = models.IntegerField()
     control_conversion = models.IntegerField()
     confidence = models.FloatField(null=True)
-
-
-def analytics_goalrecord(sender, goal_record, experiment_user, *args, **kwargs):
-    for analytics in get_all_analytics():
-        analytics.record(goal_record=goal_record,
-                         experiment_user=experiment_user)
-
-goal_recorded.connect(analytics_goalrecord, sender=GoalRecord)
-
-def analytics_enrolled(sender, experiment, experiment_user, group_id,
-                       *args, **kwargs):
-    for analytics in get_all_analytics():
-        analytics.enroll(experiment=experiment,
-                         experiment_user=experiment_user,
-                         group_id=group_id)
-
-user_enrolled.connect(analytics_enrolled)

File django_lean/experiments/tests/test_analytics.py

-from __future__ import with_statement
-
-from contextlib import contextmanager
-
-from django.conf import settings
-from django.contrib.auth.models import AnonymousUser, User
-from django.http import HttpRequest
-
-from django_lean.experiments.analytics import (get_all_analytics,
-                                               get_all_analytics_names,
-                                               reset_caches,
-                                               IdentificationError)
-from django_lean.experiments.analytics.base import BaseAnalytics
-from django_lean.experiments.models import (AnonymousVisitor, Experiment,
-                                            GoalRecord, GoalType, Participant)
-from django_lean.experiments.tests.utils import get_session, patch, TestCase
-from django_lean.experiments.utils import StaticUser, WebUser
-
-import mox
-
-
-class TestAnalytics(TestCase):
-    def test_get_all_analytics_names(self):
-        with patch(settings, 'LEAN_ANALYTICS', NotImplemented):
-            reset_caches()
-            self.assertEqual(get_all_analytics_names(), ())
-        with patch(settings, 'LEAN_ANALYTICS', []):
-            reset_caches()
-            self.assertEqual(get_all_analytics_names(), [])
-        base_name = '%s.%s' % (BaseAnalytics.__module__, BaseAnalytics.__name__)
-        with patch(settings, 'LEAN_ANALYTICS', [base_name]):
-            reset_caches()
-            self.assertEqual(get_all_analytics_names(), [base_name])
-
-    def test_get_all_analytics(self):
-        with patch(settings, 'LEAN_ANALYTICS', NotImplemented):
-            reset_caches()
-            self.assertEqual(get_all_analytics(), [])
-        with patch(settings, 'LEAN_ANALYTICS', []):
-            reset_caches()
-            self.assertEqual(get_all_analytics(), [])
-        base_name = '%s.%s' % (BaseAnalytics.__module__, BaseAnalytics.__name__)
-        with patch(settings, 'LEAN_ANALYTICS', [base_name]):
-            reset_caches()
-            self.assertEqual([a.__class__.__name__ for a in get_all_analytics()],
-                             [BaseAnalytics.__name__])
-
-
-#############
-# KISSMETRICS
-#############
-
-try:
-    import django_kissmetrics
-except ImportError:
-    if 'experiments.analytics.kissmetrics.KissMetrics' in \
-       get_all_analytics_names():
-        traceback.print_exc()
-else:
-    from django_lean.experiments.analytics.kissmetrics import KissMetrics
-
-
-    class TestKissMetrics(TestCase):
-        def setUp(self):
-            self.mox = mox.Mox()
-            self.analytics = KissMetrics()
-
-        def test_id_from_user(self):
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            self.assertEqual(self.analytics._id_from_user(user),
-                             'User %d' % user.pk)
-            self.assertRaises(IdentificationError,
-                              self.analytics._id_from_user, None)
-
-        def test_id_from_session(self):
-            # With real session
-            with self.web_user(AnonymousUser()) as experiment_user:
-                self.mox.ReplayAll()
-                session = experiment_user.session
-                self.assertEqual(
-                    self.analytics._id_from_session(experiment_user.session),
-                    'Session %s' % session.session_key
-                )
-                self.mox.VerifyAll()
-
-            # With dict as session
-            experiment_user = StaticUser()
-            self.assertRaises(IdentificationError,
-                              self.analytics._id_from_session,
-                              experiment_user.session)
-
-        def test_compute_id(self):
-            # With anonymous WebUser
-            with self.web_user(AnonymousUser()) as experiment_user:
-                session = experiment_user.session
-                self.mox.ReplayAll()
-                self.assertEqual(self.analytics._compute_id(experiment_user),
-                                 'Session %s' % session.session_key)
-                self.mox.VerifyAll()
-
-            # With authenticated WebUser
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            with self.web_user(user) as experiment_user:
-                self.mox.ReplayAll()
-                self.assertEqual(self.analytics._compute_id(experiment_user),
-                                 'User %d' % user.id)
-                self.mox.VerifyAll()
-
-            # With StaticUser
-            experiment_user = StaticUser()
-            self.assertRaises(IdentificationError,
-                              self.analytics._compute_id, experiment_user)
-
-        def test_identify(self):
-            # With anonymous WebUser
-            with self.web_user(AnonymousUser()) as experiment_user:
-                self.mox.ReplayAll()
-                self.assertTrue(self.analytics._identify(experiment_user))
-                self.mox.VerifyAll()
-
-            # With authenticated WebUser
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            with self.web_user(user) as experiment_user:
-                self.mox.ReplayAll()
-                self.assertTrue(self.analytics._identify(experiment_user))
-                self.mox.VerifyAll()
-
-            # With StaticUser
-            experiment_user = StaticUser()
-            self.assertFalse(self.analytics._identify(experiment_user))
-
-        def test_enroll(self):
-            experiment = Experiment.objects.create(name='Experiment')
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            KM = self.mox.CreateMockAnything()
-            analytics = KissMetrics(KM=KM)
-            with self.web_user(user) as experiment_user:
-                KM.identify(analytics._compute_id(experiment_user))
-                KM.record(action='Enrolled In Experiment',
-                          props={'Experiment': experiment.name,
-                                 'Group': 'Test'})
-                self.mox.ReplayAll()
-                analytics.enroll(experiment=experiment,
-                                 experiment_user=experiment_user,
-                                 group_id=Participant.TEST_GROUP)
-                self.mox.VerifyAll()
-
-        def test_record(self):
-            KM = self.mox.CreateMockAnything()
-            analytics = KissMetrics(KM=KM)
-            with self.web_user(AnonymousUser()) as experiment_user:
-                KM.identify(analytics._id_from_session(experiment_user.session))
-                KM.record(action='Goal Recorded',
-                          props={'Goal Type': 'Goal Type'})
-                self.mox.ReplayAll()
-                goal_type = GoalType.objects.create(name='Goal Type')
-                goal_record = GoalRecord.record(goal_name=goal_type.name,
-                                                experiment_user=experiment_user)
-                analytics.record(goal_record=goal_record,
-                                 experiment_user=experiment_user)
-                self.mox.VerifyAll()
-
-        def test_event(self):
-            KM = self.mox.CreateMockAnything()
-            analytics = KissMetrics(KM=KM)
-            with self.web_user(AnonymousUser()) as experiment_user:
-                KM.identify(analytics._id_from_session(experiment_user.session))
-                KM.record(action='Event', props={'Foo': 'Bar'})
-                self.mox.ReplayAll()
-                analytics.event(name='Event',
-                                properties={'Foo': 'Bar'},
-                                request=experiment_user.request)
-                self.mox.VerifyAll()
-
-        @contextmanager
-        def web_user(self, user):
-            session = get_session(None)
-            request = self.mox.CreateMock(HttpRequest)
-            request.user = user
-            request.session = session
-            experiment_user = WebUser(request)
-            experiment_user.get_or_create_anonymous_visitor()
-            yield experiment_user
-
-
-##########
-# MIXPANEL
-##########
-
-try:
-    import mixpanel
-except ImportError:
-    if 'experiments.analytics.mixpanel.Mixpanel' in \
-       get_all_analytics_names():
-        traceback.print_exc()
-else:
-    from django_lean.experiments.analytics.mixpanel import Mixpanel
-
-
-    class TestMixpanel(TestCase):
-        def setUp(self):
-            self.mox = mox.Mox()
-            self.analytics = Mixpanel()
-
-        def tearDown(self):
-            self.mox.UnsetStubs()
-
-        def test_id_from_user(self):
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            self.assertEqual(self.analytics._id_from_user(user),
-                             'User %d' % user.pk)
-            self.assertRaises(IdentificationError,
-                              self.analytics._id_from_user, None)
-
-        def test_id_from_session(self):
-            # With real session
-            with self.web_user(AnonymousUser()) as experiment_user:
-                self.mox.ReplayAll()
-                session = experiment_user.session
-                self.assertEqual(
-                    self.analytics._id_from_session(experiment_user.session),
-                    'Session %s' % session.session_key
-                )
-                self.mox.VerifyAll()
-
-            # With dict as session
-            experiment_user = StaticUser()
-            self.assertRaises(IdentificationError,
-                              self.analytics._id_from_session,
-                              experiment_user.session)
-
-        def test_compute_id(self):
-            # With anonymous WebUser
-            with self.web_user(AnonymousUser()) as experiment_user:
-                session = experiment_user.session
-                self.mox.ReplayAll()
-                self.assertEqual(self.analytics._compute_id(experiment_user),
-                                 'Session %s' % session.session_key)
-                self.mox.VerifyAll()
-
-            # With authenticated WebUser
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            with self.web_user(user) as experiment_user:
-                self.mox.ReplayAll()
-                self.assertEqual(self.analytics._compute_id(experiment_user),
-                                 'User %d' % user.id)
-                self.mox.VerifyAll()
-
-            # With StaticUser
-            experiment_user = StaticUser()
-            self.assertRaises(IdentificationError,
-                              self.analytics._compute_id, experiment_user)
-
-        def test_identify(self):
-            # With anonymous WebUser
-            with self.web_user(AnonymousUser()) as experiment_user:
-                self.mox.ReplayAll()
-                self.assertTrue(self.analytics._identify(experiment_user))
-                self.assertEqual(
-                    self.analytics.identity,
-                    'Session %s' % experiment_user.session.session_key
-                )
-                self.mox.VerifyAll()
-
-            # With authenticated WebUser
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            with self.web_user(user) as experiment_user:
-                self.mox.ReplayAll()
-                self.assertTrue(self.analytics._identify(experiment_user))
-                self.assertEqual(self.analytics.identity,
-                                 'User %s' % experiment_user.user.pk)
-                self.mox.VerifyAll()
-
-            # With StaticUser
-            experiment_user = StaticUser()
-            self.assertFalse(self.analytics._identify(experiment_user))
-            self.assertEqual(self.analytics.identity, None)
-
-        def test_enroll(self):
-            import time
-            experiment = Experiment.objects.create(name='Experiment')
-            user = User.objects.create_user('user', 'user@example.com', 'user')
-            tracker = self.mox.CreateMockAnything()
-            analytics = Mixpanel(tracker=tracker)
-            now = time.gmtime()
-            self.mox.StubOutWithMock(time, 'gmtime')
-            time.gmtime().AndReturn(now)
-            with self.web_user(user) as experiment_user:
-                properties = {'time': '%d' % time.mktime(now),
-                              'distinct_id': 'User %d' % user.pk,
-                              'Experiment': experiment.name,
-                              'Group': 'Test'}
-                tracker.run(event_name='Enrolled In Experiment',
-                            properties=properties)
-                self.mox.ReplayAll()
-                analytics.enroll(experiment=experiment,
-                                 experiment_user=experiment_user,
-                                 group_id=Participant.TEST_GROUP)
-                self.mox.VerifyAll()
-
-        def test_record(self):
-            import time
-            tracker = self.mox.CreateMockAnything()
-            analytics = Mixpanel(tracker=tracker)
-            now = time.gmtime()
-            self.mox.StubOutWithMock(time, 'gmtime')
-            time.gmtime().AndReturn(now)
-            with self.web_user(AnonymousUser()) as experiment_user:
-                properties = {
-                    'time': '%d' % time.mktime(now),
-                    'distinct_id': ('Session %s' %
-                                    experiment_user.session.session_key),
-                    'Goal Type': 'Goal Type'
-                }
-                tracker.run(event_name='Goal Recorded',
-                            properties=properties)
-                self.mox.ReplayAll()
-                goal_type = GoalType.objects.create(name='Goal Type')
-                goal_record = GoalRecord.record(goal_name=goal_type.name,
-                                                experiment_user=experiment_user)
-                analytics.record(goal_record=goal_record,
-                                 experiment_user=experiment_user)
-                self.mox.VerifyAll()
-
-        def test_event(self):
-            import time
-            tracker = self.mox.CreateMockAnything()
-            analytics = Mixpanel(tracker=tracker)
-            now = time.gmtime()
-            self.mox.StubOutWithMock(time, 'gmtime')
-            time.gmtime().AndReturn(now)
-            with self.web_user(AnonymousUser()) as experiment_user:
-                properties = {
-                    'time': '%d' % time.mktime(now),
-                    'distinct_id': ('Session %s' %
-                                    experiment_user.session.session_key),
-                    'Foo': 'Bar'
-                }
-                tracker.run(event_name='Event',
-                            properties=properties)
-                self.mox.ReplayAll()
-                analytics.event(name='Event',
-                                properties={'Foo': 'Bar'},
-                                request=experiment_user.request)
-                self.mox.VerifyAll()
-
-        @contextmanager
-        def web_user(self, user):
-            session = get_session(None)
-            request = self.mox.CreateMock(HttpRequest)
-            request.user = user
-            request.session = session
-            experiment_user = WebUser(request)
-            experiment_user.get_or_create_anonymous_visitor()
-            yield experiment_user

File django_lean/experiments/tests/utils.py

 from django.utils.importlib import import_module
 from django.utils.functional import LazyObject
 
-from django_lean.experiments.analytics import reset_caches
 from django_lean.experiments.loader import ExperimentLoader
 from django_lean.experiments.models import Participant
+from django_lean.lean_analytics import reset_caches
 
 
 def get_session(session_key):

File django_lean/lean_analytics/__init__.py

+from django.conf import settings
+from django.core.urlresolvers import get_callable
+
+
+class IdentificationError(RuntimeError):
+    pass
+
+
+def get_all_analytics_names():
+    return getattr(settings, 'LEAN_ANALYTICS', ())
+
+def get_all_analytics():
+    if get_all_analytics.cache is None:
+        names = get_all_analytics_names()
+        get_all_analytics.cache = [get_callable(a)() for a in names]
+    return get_all_analytics.cache
+
+def reset_caches():
+    get_all_analytics.cache = None
+reset_caches()

File django_lean/lean_analytics/base.py

+from django_lean.experiments.models import Participant
+from django_lean.experiments.utils import WebUser
+from django_lean.lean_analytics import IdentificationError
+
+
+class BaseAnalytics(object):
+    def _id_from_session(self, session):
+        try:
+            return 'Session %s' % session.session_key
+        except AttributeError, e:
+            raise IdentificationError(e)
+
+    def _id_from_user(self, user):
+        try:
+            return 'User %d' % user.pk
+        except AttributeError, e:
+            raise IdentificationError(e)
+
+    def _compute_id(self, experiment_user):
+        if not experiment_user.is_anonymous():
+            return self._id_from_user(experiment_user.user)
+        return self._id_from_session(experiment_user.session)
+
+    def enroll(self, experiment, experiment_user, group_id):
+        self._submit(name='Enrolled In Experiment',
+                     properties={'Experiment': unicode(experiment),
+                                 'Group': dict(Participant.GROUPS)[group_id]},
+                     experiment_user=experiment_user)
+
+    def record(self, goal_record, experiment_user):
+        self._submit(name='Goal Recorded',
+                     properties={'Goal Type': unicode(goal_record.goal_type)},
+                     experiment_user=experiment_user)
+
+    def event(self, name, properties, request=None):
+        if request:
+            self._submit(name, properties, experiment_user=WebUser(request))
+
+    def _submit(self, name, properties, experiment_user=None):
+        raise NotImplementedError()

File django_lean/lean_analytics/kissmetrics.py

+from django_kissmetrics.middleware import TrackingMiddleware
+
+from django_lean.lean_analytics import IdentificationError
+from django_lean.lean_analytics.base import BaseAnalytics
+
+
+class KissMetrics(BaseAnalytics):
+    def __init__(self, KM=None, middleware=None):
+        if middleware is None:
+            self.middleware = TrackingMiddleware()
+        if KM is None:
+            KM = self.middleware.KM
+        self.KM = KM
+
+    def _id_from_session(self, session):
+        id_from_session = self.middleware.id_from_session
+        try:
+            return id_from_session(session)
+        except AttributeError, e:
+            raise IdentificationError(e)
+
+    def _id_from_user(self, user):
+        id_from_user = self.middleware.id_from_user
+        try:
+            return id_from_user(user)
+        except AttributeError, e:
+            raise IdentificationError(e)
+
+    def _identify(self, experiment_user):
+        try:
+            self.KM.identify(self._compute_id(experiment_user))
+            return True
+        except IdentificationError:
+            # Ignore experiment_users who cannot be tied to sessions or users
+            return False
+
+    def _submit(self, name, properties, experiment_user=None):
+        if self._identify(experiment_user):
+            self.KM.record(action=name, props=properties)

File django_lean/lean_analytics/mixpanel.py

+from __future__ import absolute_import
+
+import time
+
+from mixpanel.tasks import EventTracker
+
+from django_lean.lean_analytics import IdentificationError
+from django_lean.lean_analytics.base import BaseAnalytics
+
+
+class Mixpanel(BaseAnalytics):
+    def __init__(self, tracker=None, tracker_class=EventTracker):
+        if tracker is None:
+            tracker = EventTracker()
+        self.tracker = tracker
+        self.remote_addr = None
+        self.identity = None
+
+    def _identify(self, experiment_user):
+        self.identity = None
+        self.remote_addr = None
+        request = experiment_user.request
+        if hasattr(request, 'META'):
+            self.remote_addr = request.META.get('REMOTE_ADDR', None)
+        try:
+            self.identity = self._compute_id(experiment_user)
+            return True
+        except IdentificationError:
+            # Ignore experiment_users who cannot be tied to sessions or users
+            return False
+
+    def _properties(self, properties):
+        result = {'time': '%d' % time.mktime(time.gmtime())}
+        for key, attr in (('ip', 'remote_addr'), ('distinct_id', 'identity')):
+            if key not in properties:
+                value = getattr(self, attr)
+                if value:
+                    result[key] = value
+        result.update(properties)
+        return result
+
+    def _submit(self, name, properties, experiment_user=None):
+        if self._identify(experiment_user):
+            properties = self._properties(properties)
+            self.tracker.run(event_name=name, properties=properties)

File django_lean/lean_analytics/models.py

+from django_lean.experiments.models import GoalRecord
+from django_lean.experiments.signals import goal_recorded, user_enrolled
+from django_lean.lean_analytics import get_all_analytics
+
+
+def analytics_goalrecord(sender, goal_record, experiment_user, *args, **kwargs):
+    for analytics in get_all_analytics():
+        analytics.record(goal_record=goal_record,
+                         experiment_user=experiment_user)
+
+goal_recorded.connect(analytics_goalrecord, sender=GoalRecord)
+
+def analytics_enrolled(sender, experiment, experiment_user, group_id,
+                       *args, **kwargs):
+    for analytics in get_all_analytics():
+        analytics.enroll(experiment=experiment,
+                         experiment_user=experiment_user,
+                         group_id=group_id)
+
+user_enrolled.connect(analytics_enrolled)

File django_lean/lean_analytics/tests.py

+from __future__ import with_statement
+
+from contextlib import contextmanager
+
+from django.conf import settings
+from django.contrib.auth.models import AnonymousUser, User
+from django.http import HttpRequest
+
+from django_lean.experiments.models import (AnonymousVisitor, Experiment,
+                                            GoalRecord, GoalType, Participant)
+from django_lean.experiments.tests.utils import get_session, patch, TestCase
+from django_lean.experiments.utils import StaticUser, WebUser
+from django_lean.lean_analytics import (get_all_analytics,
+                                        get_all_analytics_names,
+                                        reset_caches,
+                                        IdentificationError)
+from django_lean.lean_analytics.base import BaseAnalytics
+
+import mox
+
+class TestAnalytics(TestCase):
+    def test_get_all_analytics_names(self):
+        with patch(settings, 'LEAN_ANALYTICS', NotImplemented):
+            reset_caches()
+            self.assertEqual(get_all_analytics_names(), ())
+        with patch(settings, 'LEAN_ANALYTICS', []):
+            reset_caches()
+            self.assertEqual(get_all_analytics_names(), [])
+        base_name = '%s.%s' % (BaseAnalytics.__module__, BaseAnalytics.__name__)
+        with patch(settings, 'LEAN_ANALYTICS', [base_name]):
+            reset_caches()
+            self.assertEqual(get_all_analytics_names(), [base_name])
+
+    def test_get_all_analytics(self):
+        with patch(settings, 'LEAN_ANALYTICS', NotImplemented):
+            reset_caches()
+            self.assertEqual(get_all_analytics(), [])
+        with patch(settings, 'LEAN_ANALYTICS', []):
+            reset_caches()
+            self.assertEqual(get_all_analytics(), [])
+        base_name = '%s.%s' % (BaseAnalytics.__module__, BaseAnalytics.__name__)
+        with patch(settings, 'LEAN_ANALYTICS', [base_name]):
+            reset_caches()
+            self.assertEqual([a.__class__.__name__ for a in get_all_analytics()],
+                             [BaseAnalytics.__name__])
+
+
+#############
+# KISSMETRICS
+#############
+
+try:
+    import django_kissmetrics
+except ImportError:
+    if 'django_lean.lean_analytics.kissmetrics.KissMetrics' in \
+       get_all_analytics_names():
+        traceback.print_exc()
+else:
+    from django_lean.lean_analytics.kissmetrics import KissMetrics
+
+
+    class TestKissMetrics(TestCase):
+        def setUp(self):
+            self.mox = mox.Mox()
+            self.analytics = KissMetrics()
+
+        def test_id_from_user(self):
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            self.assertEqual(self.analytics._id_from_user(user),
+                             'User %d' % user.pk)
+            self.assertRaises(IdentificationError,
+                              self.analytics._id_from_user, None)
+
+        def test_id_from_session(self):
+            # With real session
+            with self.web_user(AnonymousUser()) as experiment_user:
+                self.mox.ReplayAll()
+                session = experiment_user.session
+                self.assertEqual(
+                    self.analytics._id_from_session(experiment_user.session),
+                    'Session %s' % session.session_key
+                )
+                self.mox.VerifyAll()
+
+            # With dict as session
+            experiment_user = StaticUser()
+            self.assertRaises(IdentificationError,
+                              self.analytics._id_from_session,
+                              experiment_user.session)
+
+        def test_compute_id(self):
+            # With anonymous WebUser
+            with self.web_user(AnonymousUser()) as experiment_user:
+                session = experiment_user.session
+                self.mox.ReplayAll()
+                self.assertEqual(self.analytics._compute_id(experiment_user),
+                                 'Session %s' % session.session_key)
+                self.mox.VerifyAll()
+
+            # With authenticated WebUser
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            with self.web_user(user) as experiment_user:
+                self.mox.ReplayAll()
+                self.assertEqual(self.analytics._compute_id(experiment_user),
+                                 'User %d' % user.id)
+                self.mox.VerifyAll()
+
+            # With StaticUser
+            experiment_user = StaticUser()
+            self.assertRaises(IdentificationError,
+                              self.analytics._compute_id, experiment_user)
+
+        def test_identify(self):
+            # With anonymous WebUser
+            with self.web_user(AnonymousUser()) as experiment_user:
+                self.mox.ReplayAll()
+                self.assertTrue(self.analytics._identify(experiment_user))
+                self.mox.VerifyAll()
+
+            # With authenticated WebUser
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            with self.web_user(user) as experiment_user:
+                self.mox.ReplayAll()
+                self.assertTrue(self.analytics._identify(experiment_user))
+                self.mox.VerifyAll()
+
+            # With StaticUser
+            experiment_user = StaticUser()
+            self.assertFalse(self.analytics._identify(experiment_user))
+
+        def test_enroll(self):
+            experiment = Experiment.objects.create(name='Experiment')
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            KM = self.mox.CreateMockAnything()
+            analytics = KissMetrics(KM=KM)
+            with self.web_user(user) as experiment_user:
+                KM.identify(analytics._compute_id(experiment_user))
+                KM.record(action='Enrolled In Experiment',
+                          props={'Experiment': experiment.name,
+                                 'Group': 'Test'})
+                self.mox.ReplayAll()
+                analytics.enroll(experiment=experiment,
+                                 experiment_user=experiment_user,
+                                 group_id=Participant.TEST_GROUP)
+                self.mox.VerifyAll()
+
+        def test_record(self):
+            KM = self.mox.CreateMockAnything()
+            analytics = KissMetrics(KM=KM)
+            with self.web_user(AnonymousUser()) as experiment_user:
+                KM.identify(analytics._id_from_session(experiment_user.session))
+                KM.record(action='Goal Recorded',
+                          props={'Goal Type': 'Goal Type'})
+                self.mox.ReplayAll()
+                goal_type = GoalType.objects.create(name='Goal Type')
+                goal_record = GoalRecord.record(goal_name=goal_type.name,
+                                                experiment_user=experiment_user)
+                analytics.record(goal_record=goal_record,
+                                 experiment_user=experiment_user)
+                self.mox.VerifyAll()
+
+        def test_event(self):
+            KM = self.mox.CreateMockAnything()
+            analytics = KissMetrics(KM=KM)
+            with self.web_user(AnonymousUser()) as experiment_user:
+                KM.identify(analytics._id_from_session(experiment_user.session))
+                KM.record(action='Event', props={'Foo': 'Bar'})
+                self.mox.ReplayAll()
+                analytics.event(name='Event',
+                                properties={'Foo': 'Bar'},
+                                request=experiment_user.request)
+                self.mox.VerifyAll()
+
+        @contextmanager
+        def web_user(self, user):
+            session = get_session(None)
+            request = self.mox.CreateMock(HttpRequest)
+            request.user = user
+            request.session = session
+            experiment_user = WebUser(request)
+            experiment_user.get_or_create_anonymous_visitor()
+            yield experiment_user
+
+
+##########
+# MIXPANEL
+##########
+
+try:
+    import mixpanel
+except ImportError:
+    if 'django_lean.lean_analytics.mixpanel.Mixpanel' in \
+       get_all_analytics_names():
+        traceback.print_exc()
+else:
+    from django_lean.lean_analytics.mixpanel import Mixpanel
+
+
+    class TestMixpanel(TestCase):
+        def setUp(self):
+            self.mox = mox.Mox()
+            self.analytics = Mixpanel()
+
+        def tearDown(self):
+            self.mox.UnsetStubs()
+
+        def test_id_from_user(self):
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            self.assertEqual(self.analytics._id_from_user(user),
+                             'User %d' % user.pk)
+            self.assertRaises(IdentificationError,
+                              self.analytics._id_from_user, None)
+
+        def test_id_from_session(self):
+            # With real session
+            with self.web_user(AnonymousUser()) as experiment_user:
+                self.mox.ReplayAll()
+                session = experiment_user.session
+                self.assertEqual(
+                    self.analytics._id_from_session(experiment_user.session),
+                    'Session %s' % session.session_key
+                )
+                self.mox.VerifyAll()
+
+            # With dict as session
+            experiment_user = StaticUser()
+            self.assertRaises(IdentificationError,
+                              self.analytics._id_from_session,
+                              experiment_user.session)
+
+        def test_compute_id(self):
+            # With anonymous WebUser
+            with self.web_user(AnonymousUser()) as experiment_user:
+                session = experiment_user.session
+                self.mox.ReplayAll()
+                self.assertEqual(self.analytics._compute_id(experiment_user),
+                                 'Session %s' % session.session_key)
+                self.mox.VerifyAll()
+
+            # With authenticated WebUser
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            with self.web_user(user) as experiment_user:
+                self.mox.ReplayAll()
+                self.assertEqual(self.analytics._compute_id(experiment_user),
+                                 'User %d' % user.id)
+                self.mox.VerifyAll()
+
+            # With StaticUser
+            experiment_user = StaticUser()
+            self.assertRaises(IdentificationError,
+                              self.analytics._compute_id, experiment_user)
+
+        def test_identify(self):
+            # With anonymous WebUser
+            with self.web_user(AnonymousUser()) as experiment_user:
+                self.mox.ReplayAll()
+                self.assertTrue(self.analytics._identify(experiment_user))
+                self.assertEqual(
+                    self.analytics.identity,
+                    'Session %s' % experiment_user.session.session_key
+                )
+                self.mox.VerifyAll()
+
+            # With authenticated WebUser
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            with self.web_user(user) as experiment_user:
+                self.mox.ReplayAll()
+                self.assertTrue(self.analytics._identify(experiment_user))
+                self.assertEqual(self.analytics.identity,
+                                 'User %s' % experiment_user.user.pk)
+                self.mox.VerifyAll()
+
+            # With StaticUser
+            experiment_user = StaticUser()
+            self.assertFalse(self.analytics._identify(experiment_user))
+            self.assertEqual(self.analytics.identity, None)
+
+        def test_enroll(self):
+            import time
+            experiment = Experiment.objects.create(name='Experiment')
+            user = User.objects.create_user('user', 'user@example.com', 'user')
+            tracker = self.mox.CreateMockAnything()
+            analytics = Mixpanel(tracker=tracker)
+            now = time.gmtime()
+            self.mox.StubOutWithMock(time, 'gmtime')
+            time.gmtime().AndReturn(now)
+            with self.web_user(user) as experiment_user:
+                properties = {'time': '%d' % time.mktime(now),
+                              'distinct_id': 'User %d' % user.pk,
+                              'Experiment': experiment.name,
+                              'Group': 'Test'}
+                tracker.run(event_name='Enrolled In Experiment',
+                            properties=properties)
+                self.mox.ReplayAll()
+                analytics.enroll(experiment=experiment,
+                                 experiment_user=experiment_user,
+                                 group_id=Participant.TEST_GROUP)
+                self.mox.VerifyAll()
+
+        def test_record(self):
+            import time
+            tracker = self.mox.CreateMockAnything()
+            analytics = Mixpanel(tracker=tracker)
+            now = time.gmtime()
+            self.mox.StubOutWithMock(time, 'gmtime')
+            time.gmtime().AndReturn(now)
+            with self.web_user(AnonymousUser()) as experiment_user:
+                properties = {
+                    'time': '%d' % time.mktime(now),
+                    'distinct_id': ('Session %s' %
+                                    experiment_user.session.session_key),
+                    'Goal Type': 'Goal Type'
+                }
+                tracker.run(event_name='Goal Recorded',
+                            properties=properties)
+                self.mox.ReplayAll()
+                goal_type = GoalType.objects.create(name='Goal Type')
+                goal_record = GoalRecord.record(goal_name=goal_type.name,
+                                                experiment_user=experiment_user)
+                analytics.record(goal_record=goal_record,
+                                 experiment_user=experiment_user)
+                self.mox.VerifyAll()
+
+        def test_event(self):
+            import time
+            tracker = self.mox.CreateMockAnything()
+            analytics = Mixpanel(tracker=tracker)
+            now = time.gmtime()
+            self.mox.StubOutWithMock(time, 'gmtime')
+            time.gmtime().AndReturn(now)
+            with self.web_user(AnonymousUser()) as experiment_user:
+                properties = {
+                    'time': '%d' % time.mktime(now),
+                    'distinct_id': ('Session %s' %
+                                    experiment_user.session.session_key),
+                    'Foo': 'Bar'
+                }
+                tracker.run(event_name='Event',
+                            properties=properties)
+                self.mox.ReplayAll()
+                analytics.event(name='Event',
+                                properties={'Foo': 'Bar'},
+                                request=experiment_user.request)
+                self.mox.VerifyAll()
+
+        @contextmanager
+        def web_user(self, user):
+            session = get_session(None)
+            request = self.mox.CreateMock(HttpRequest)
+            request.user = user
+            request.session = session
+            experiment_user = WebUser(request)
+            experiment_user.get_or_create_anonymous_visitor()
+            yield experiment_user