tests.py

#!/usr/bin/env python
import base64
import contextlib
import json
import time
import unittest
import urlparse

try:
    from mock import Mock, patch
except ImportError:
    print 'mixpanel-python requires the mock package to run the test suite'
    raise

import mixpanel

class LogConsumer(object):
    def __init__(self):
        self.log = []

    def send(self, endpoint, event, api_key=None):
        if api_key:
            self.log.append((endpoint, json.loads(event), api_key))
        else:
            self.log.append((endpoint, json.loads(event)))

class MixpanelTestCase(unittest.TestCase):
    def setUp(self):
        self.TOKEN = '12345'
        self.consumer = LogConsumer()
        self.mp = mixpanel.Mixpanel('12345', consumer=self.consumer)
        self.mp._now = lambda : 1000.1

    def test_track(self):
        self.mp.track('ID', 'button press', {'size': 'big', 'color': 'blue'})
        self.assertEqual(self.consumer.log, [(
            'events', {
                'event': 'button press',
                'properties': {
                    'token': self.TOKEN,
                    'size': 'big',
                    'color': 'blue',
                    'distinct_id': 'ID',
                    'time': int(self.mp._now()),
                    'mp_lib': 'python',
                    '$lib_version': mixpanel.VERSION,
                }
            }
        )])

    def test_import_data(self):
        " Unit test for the `import_data` method. "
        timestamp = time.time()
        self.mp.import_data('MY_API_KEY', 'ID', 'button press', timestamp, {'size': 'big', 'color': 'blue'})
        self.assertEqual(self.consumer.log, [(
            'imports', {
                'event': 'button press',
                'properties': {
                    'token': self.TOKEN,
                    'size': 'big',
                    'color': 'blue',
                    'distinct_id': 'ID',
                    'time': int(timestamp),
                    'mp_lib': 'python',
                    '$lib_version': mixpanel.VERSION,
                },
            },
            'MY_API_KEY'
        )])

    def test_track_meta(self):
        self.mp.track('ID', 'button press', {'size': 'big', 'color': 'blue'},
            meta={'ip': 0})
        self.assertEqual(self.consumer.log, [(
            'events', {
                'event': 'button press',
                'properties': {
                    'token': self.TOKEN,
                    'size': 'big',
                    'color': 'blue',
                    'distinct_id': 'ID',
                    'time': int(self.mp._now()),
                    'mp_lib': 'python',
                    '$lib_version': mixpanel.VERSION,
                },
                'ip': 0,
            }
        )])

    def test_people_set(self):
        self.mp.people_set('amq', {'birth month': 'october', 'favorite color': 'purple'})
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$set': {
                    'birth month': 'october',
                    'favorite color': 'purple',
                },
            }
        )])

    def test_people_set_once(self):
        self.mp.people_set_once('amq', {'birth month': 'october', 'favorite color': 'purple'})
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$set_once': {
                    'birth month': 'october',
                    'favorite color': 'purple',
                },
            }
        )])

    def test_people_increment(self):
        self.mp.people_increment('amq', {'Albums Released': 1})
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$add': {
                    'Albums Released': 1,
                },
            }
        )])

    def test_people_append(self):
        self.mp.people_append('amq', {'birth month': 'october', 'favorite color': 'purple'})
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$append': {
                    'birth month': 'october',
                    'favorite color': 'purple',
                },
            }
        )])

    def test_people_union(self):
        self.mp.people_union('amq', {'Albums': [ 'Diamond Dogs'] })
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$union': {
                    'Albums': [ 'Diamond Dogs' ],
                },
            }
        )])

    def test_people_unset(self):
        self.mp.people_unset('amq', [ 'Albums', 'Singles' ])
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$unset': [ 'Albums', 'Singles' ],
            }
        )])

    def test_people_track_charge(self):
        self.mp.people_track_charge('amq', 12.65, { '$time': '2013-04-01T09:02:00' })
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$append': {
                    '$transactions': {
                        '$time': '2013-04-01T09:02:00',
                        '$amount': 12.65,
                    },
                },
            }
        )])

    def test_people_clear_charges(self):
        self.mp.people_clear_charges('amq')
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$unset': [ '$transactions' ],
            }
        )])

    def test_alias(self):
        mock_response = Mock()
        mock_response.read.return_value = '{"status":1, "error": null}'
        with patch('urllib2.urlopen', return_value = mock_response) as urlopen:
            self.mp.alias('ALIAS','ORIGINAL ID')
            self.assertEqual(self.consumer.log, [])

            self.assertEqual(urlopen.call_count, 1)
            ((request,),_) = urlopen.call_args

            self.assertEqual(request.get_full_url(), 'https://api.mixpanel.com/track')
            self.assertEqual(request.get_data(), 'ip=0&data=eyJldmVudCI6IiRjcmVhdGVfYWxpYXMiLCJwcm9wZXJ0aWVzIjp7ImFsaWFzIjoiQUxJQVMiLCJ0b2tlbiI6IjEyMzQ1IiwiZGlzdGluY3RfaWQiOiJPUklHSU5BTCBJRCJ9fQ%3D%3D&verbose=1')


    def test_people_meta(self):
        self.mp.people_set('amq', {'birth month': 'october', 'favorite color': 'purple'},
            meta={'$ip': 0, '$ignore_time': True})
        self.assertEqual(self.consumer.log, [(
            'people', {
                '$time': int(self.mp._now() * 1000),
                '$token': self.TOKEN,
                '$distinct_id': 'amq',
                '$set': {
                    'birth month': 'october',
                    'favorite color': 'purple',
                },
                '$ip': 0,
                '$ignore_time': True,
            }
        )])

class ConsumerTestCase(unittest.TestCase):
    def setUp(self):
        self.consumer = mixpanel.Consumer(request_timeout=30)

    @contextlib.contextmanager
    def _assertSends(self, expect_url, expect_data):
        mock_response = Mock()
        mock_response.read.return_value = '{"status":1, "error": null}'
        with patch('urllib2.urlopen', return_value = mock_response) as urlopen:
            yield

            self.assertEqual(urlopen.call_count, 1)

            (call_args, kwargs) = urlopen.call_args
            (request,) = call_args
            timeout = kwargs.get('timeout', None)

            self.assertEqual(request.get_full_url(), expect_url)
            self.assertEqual(request.get_data(), expect_data)
            self.assertEqual(timeout, self.consumer._request_timeout)

    def test_send_events(self):
        with self._assertSends('https://api.mixpanel.com/track', 'ip=0&data=IkV2ZW50Ig%3D%3D&verbose=1'):
            self.consumer.send('events', '"Event"')

    def test_send_people(self):
        with self._assertSends('https://api.mixpanel.com/engage','ip=0&data=IlBlb3BsZSI%3D&verbose=1'):
            self.consumer.send('people', '"People"')

class BufferedConsumerTestCase(unittest.TestCase):
    def setUp(self):
        self.MAX_LENGTH = 10
        self.consumer = mixpanel.BufferedConsumer(self.MAX_LENGTH)
        self.mock = Mock()
        self.mock.read.return_value = '{"status":1, "error": null}'

    def test_buffer_hold_and_flush(self):
        with patch('urllib2.urlopen', return_value = self.mock) as urlopen:
            self.consumer.send('events', '"Event"')
            self.assertTrue(not self.mock.called)
            self.consumer.flush()

            self.assertEqual(urlopen.call_count, 1)

            (call_args, kwargs) = urlopen.call_args
            (request,) = call_args
            timeout = kwargs.get('timeout', None)

            self.assertEqual(request.get_full_url(), 'https://api.mixpanel.com/track')
            self.assertEqual(request.get_data(), 'ip=0&data=WyJFdmVudCJd&verbose=1')
            self.assertIsNone(timeout)

    def test_buffer_fills_up(self):
        with patch('urllib2.urlopen', return_value = self.mock) as urlopen:
            for i in xrange(self.MAX_LENGTH - 1):
                self.consumer.send('events', '"Event"')
                self.assertTrue(not self.mock.called)

            self.consumer.send('events', '"Last Event"')

            self.assertEqual(urlopen.call_count, 1)
            ((request,),_) = urlopen.call_args
            self.assertEqual(request.get_full_url(), 'https://api.mixpanel.com/track')
            self.assertEqual(request.get_data(), 'ip=0&data=WyJFdmVudCIsIkV2ZW50IiwiRXZlbnQiLCJFdmVudCIsIkV2ZW50IiwiRXZlbnQiLCJFdmVudCIsIkV2ZW50IiwiRXZlbnQiLCJMYXN0IEV2ZW50Il0%3D&verbose=1')

class FunctionalTestCase(unittest.TestCase):
    def setUp(self):
        self.TOKEN = '12345'
        self.mp = mixpanel.Mixpanel(self.TOKEN)
        self.mp._now = lambda : 1000

    @contextlib.contextmanager
    def _assertRequested(self, expect_url, expect_data):
        mock_response = Mock()
        mock_response.read.return_value = '{"status":1, "error": null}'
        with patch('urllib2.urlopen', return_value = mock_response) as urlopen:
            yield

            self.assertEqual(urlopen.call_count, 1)
            ((request,),_) = urlopen.call_args
            self.assertEqual(request.get_full_url(), expect_url)
            data = urlparse.parse_qs(request.get_data())
            self.assertEqual(len(data['data']), 1)
            payload_encoded = data['data'][0]
            payload_json = base64.b64decode(payload_encoded)
            payload = json.loads(payload_json)
            self.assertEqual(payload, expect_data)

    def test_track_functional(self):
        # XXX this includes $lib_version, which means the test breaks
        # every time we release.
        expect_data = {u'event': {u'color': u'blue', u'size': u'big'}, u'properties': {u'mp_lib': u'python', u'token': u'12345', u'distinct_id': u'button press', u'$lib_version': unicode(mixpanel.VERSION), u'time': 1000}}
        with self._assertRequested('https://api.mixpanel.com/track', expect_data):
            self.mp.track('button press', {'size': 'big', 'color': 'blue'})

    def test_people_set_functional(self):
        expect_data = {u'$distinct_id': u'amq', u'$set': {u'birth month': u'october', u'favorite color': u'purple'}, u'$time': 1000000, u'$token': u'12345'}
        with self._assertRequested('https://api.mixpanel.com/engage', expect_data):
             self.mp.people_set('amq', {'birth month': 'october', 'favorite color': 'purple'})

if __name__ == "__main__":
    unittest.main()