Source

pyramid_mixpanel / pyramid_mixpanel / tests / unittests / test_pyramid_mixpanel.py

Full commit
# -*- coding: utf-8 -*-
import json
import base64
import unittest
from decimal import Decimal

import mock


class TestPyramidMixpanel(unittest.TestCase):

    @mock.patch('pyramid_mixpanel.MixpanelUtility')
    def test_includeme(self, utility):
        from pyramid_mixpanel import includeme
        config = mock.Mock()

        includeme(config)

        config.registry.registerUtility.assert_called_once_with(
            utility.return_value
            )
        config.commit.assert_called_once_with()
        utility.assert_called_once_with(config.registry.settings)

    def test_get_mixpanel_utility(self):
        from pyramid_mixpanel import get_mixpanel_utility, IMixpanelUtility
        request = mock.Mock()

        result = get_mixpanel_utility(request)

        request.registry.getUtility.assert_called_once_with(IMixpanelUtility)
        self.assertEqual(result, request.registry.getUtility.return_value)

    @mock.patch('requests.get')
    def test_track(self, m_get):
        from pyramid_mixpanel import MixpanelUtility
        mu = MixpanelUtility({'mixpanel.token': 'token'})
        event = 'test_event'
        properties = {
            u'install_referrer': 'Direct Link',
            }

        mu.track(event, properties)

        url = "https://api.mixpanel.com/track"
        properties['token'] = 'token'
        params = {'event': event, 'properties': properties}
        data = base64.b64encode(json.dumps(params))
        m_get.assert_called_once_with(url, params={'data': data})

    @mock.patch('requests.get')
    def test_track_overriden_token(self, m_get):
        from pyramid_mixpanel import MixpanelUtility
        mu = MixpanelUtility({'mixpanel.token': 'config token'})
        event = 'test_event2'
        properties = {
            u'referrer': u'Direct Link',
            u'token': u'given token'
            }

        mu.track(event, properties)

        url = "https://api.mixpanel.com/track"
        params = {'event': event, 'properties': properties}
        data = base64.b64encode(json.dumps(params))
        m_get.assert_called_once_with(url, params={'data': data})

    @mock.patch('requests.get')
    @mock.patch('pyramid_mixpanel.log')
    def test_track_exception(self, m_log, m_get):
        from pyramid_mixpanel import MixpanelUtility
        mu = MixpanelUtility({'mixpanel.token': 'token'})

        mu.track('test_event', {'prop': Decimal(10)})

        self.assertFalse(m_get.called)
        self.assertTrue(m_log.exception.called)

    @mock.patch('requests.get')
    @mock.patch('pyramid_mixpanel.log')
    def test_track_http_error(self, m_log, m_get):
        from requests import HTTPError
        from pyramid_mixpanel import MixpanelUtility
        m_get.side_effect = HTTPError()
        m_get.side_effect.response = mock.Mock()
        m_get.side_effect.response.status_code = 400
        mu = MixpanelUtility({'mixpanel.token': 'token'})

        mu.track('test_event', {})

        self.assertTrue(m_log.exception.called)

    @mock.patch('requests.post')
    def test_people_track(self, m_post):
        from pyramid_mixpanel import MixpanelUtility
        mu = MixpanelUtility({u'mixpanel.token': u'token'})

        mu.people_track(u'some ID', {u'whatever': u'whatevs'})

        data = base64.b64encode(json.dumps({
            u'$token': u'token',
            u'$distinct_id': u'some ID',
            u'$set': {
                'whatever': u'whatevs',
            },
        }))
        m_post.assert_called_once_with('https://api.mixpanel.com/engage/',
                                       {'data': data})

    @mock.patch('requests.post')
    @mock.patch('pyramid_mixpanel.log')
    def test_people_track_exception(self, m_log, m_post):
        from pyramid_mixpanel import MixpanelUtility
        mu = MixpanelUtility({u'mixpanel.token': u'token'})

        mu.people_track(u'id', {u'money': Decimal('100.30')})

        self.assertFalse(m_post.called)
        self.assertTrue(m_log.exception.called)

    @mock.patch('requests.post')
    @mock.patch('pyramid_mixpanel.log')
    def test_people_track_http_error(self, m_log, m_post):
        from requests import HTTPError
        from pyramid_mixpanel import MixpanelUtility
        m_post.side_effect = HTTPError()
        m_post.side_effect.response = mock.Mock()
        m_post.side_effect.response.status_code = 400
        mu = MixpanelUtility({u'mixpanel.token': u'token'})

        mu.people_track(u'some ID', {u'whatever': u'whatevs'})

        self.assertTrue(m_log.exception.called)

    @mock.patch('requests.post')
    @mock.patch('pyramid_mixpanel.log')
    def test_people_track_mixpanel_error(self, m_log, m_post):
        from pyramid_mixpanel import MixpanelUtility
        m_post.return_value.content = '0'
        mu = MixpanelUtility({u'mixpanel.token': u'token'})

        mu.people_track(u'some ID', {u'whatever': u'whatevs'})

        self.assertTrue(m_log.error.called)

    @mock.patch('requests.post')
    @mock.patch('pyramid_mixpanel.log')
    def test_people_track_mixpanel_unknown_status(self, m_log, m_post):
        from pyramid_mixpanel import MixpanelUtility
        m_post.return_value.content = '42'
        mu = MixpanelUtility({u'mixpanel.token': u'token'})

        mu.people_track(u'some ID', {u'whatever': u'whatevs'})

        self.assertTrue(m_log.warning.called)