<commit_before>import base64
import json
import unittest

from libsaas.executors import test_executor
from libsaas.services import mixpanel


class MixpanelTestCase(unittest.TestCase):

    def setUp(self):
        self.executor = test_executor.use()
        self.executor.set_response(b'1', 200, {})

        self.service = mixpanel.Mixpanel('my-token', 'api-key', 'api-secret')

    def serialize(self, data):
        return base64.b64encode(json.dumps(data).encode('utf-8'))

    def expect(self, uri, params_expected=None, subdomain=None, b64params=()):
        if not subdomain:
            domain = 'mixpanel.com'
        else:
            domain = '{0}.mixpanel.com'.format(subdomain)

        if subdomain != 'api':
            uri = 'api/2.0/{0}'.format(uri)

        self.assertEqual(self.executor.request.uri,
                         'http://{0}/{1}'.format(domain, uri))
        if not params_expected:
            return

        params_used = self.executor.request.params.copy()

        if 'api_key' in params_used:
            params_expected['api_key'] = 'api-key'

        params_used.pop('sig', None)
        params_used.pop('expire', None)

        for name in b64params:
            params_used[name] = json.loads(base64.b64decode(params_used[name]))

        self.assertEqual(params_used, params_expected)

    def test_track(self):
        ret = self.service.track('login', {'user': 'foo', 'important': True,
                                           'widgets': ['foo', 'bar']}, ip=True)
        data = self.serialize({'event': 'login',<commit_msg>mixpanel/tests: make sure to pass unicode to json.loads<commit_after>import base64
import json
import unittest

from libsaas import port
from libsaas.executors import test_executor
from libsaas.services import mixpanel


class MixpanelTestCase(unittest.TestCase):

    def setUp(self):
        self.executor = test_executor.use()
        self.executor.set_response(b'1', 200, {})

        self.service = mixpanel.Mixpanel('my-token', 'api-key', 'api-secret')

    def serialize(self, data):
        return base64.b64encode(json.dumps(data).encode('utf-8'))

    def expect(self, uri, params_expected=None, subdomain=None, b64params=()):
        if not subdomain:
            domain = 'mixpanel.com'
        else:
            domain = '{0}.mixpanel.com'.format(subdomain)

        if subdomain != 'api':
            uri = 'api/2.0/{0}'.format(uri)

        self.assertEqual(self.executor.request.uri,
                         'http://{0}/{1}'.format(domain, uri))
        if not params_expected:
            return

        params_used = self.executor.request.params.copy()

        if 'api_key' in params_used:
            params_expected['api_key'] = 'api-key'

        params_used.pop('sig', None)
        params_used.pop('expire', None)

        for name in b64params:
            decoded = base64.b64decode(params_used[name])
            params_used[name] = json.loads(port.to_u(decoded))

        self.assertEqual(params_used, params_expected)

    def test_track(self):
        ret = self.service.track('login', {'user': 'foo', 'important': True,
                                           'widgets': ['foo', 'bar']}, ip=True)
        data = self.serialize({'event': 'login',
{'fraction_non_alphanumeric': 0.13491087891825446, 'fraction_numerical': 0.015365703749231715, 'mean_word_length': 2.4627659574468086, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7111537', 'n_tokens_mistral': 1024, 'n_tokens_neox': 977, 'n_words': 196}