Use requests rather than urllib3

mixpanel/__init__.py

@@ -21,6 +21,8 @@
 import time
 import uuid
 
+import requests
+from requests.auth import HTTPBasicAuth
 import six
 from six.moves import range
 import urllib3
@@ -540,7 +542,7 @@ class Consumer(object):
 
     def __init__(self, events_url=None, people_url=None, import_url=None,
             request_timeout=None, groups_url=None, api_host="api.mixpanel.com",
-            retry_limit=4, retry_backoff_factor=0.25, verify_cert=False):
+            retry_limit=4, retry_backoff_factor=0.25, verify_cert=True):
         # TODO: With next major version, make the above args kwarg-only, and reorder them.
         self._endpoints = {
             'events': events_url or 'https://{}/track'.format(api_host),
@@ -549,31 +551,31 @@ def __init__(self, events_url=None, people_url=None, import_url=None,
             'imports': import_url or 'https://{}/import'.format(api_host),
         }
 
-        retry_args = {
-            "total": retry_limit,
-            "backoff_factor": retry_backoff_factor,
-            "status_forcelist": set(range(500, 600)),
-        }
+        self._verify_cert = verify_cert
+        self._request_timeout = request_timeout
 
         # Work around renamed argument in urllib3.
         if hasattr(urllib3.util.Retry.DEFAULT, "allowed_methods"):
             methods_arg = "allowed_methods"
         else:
             methods_arg = "method_whitelist"
 
-        retry_args[methods_arg] = {"POST"}
+        retry_args = {
+            "total": retry_limit,
+            "backoff_factor": retry_backoff_factor,
+            "status_forcelist": set(range(500, 600)),
+            methods_arg: {"POST"},
+        }
         retry_config = urllib3.Retry(**retry_args)
 
-        if not verify_cert:
-            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
-        cert_reqs = 'CERT_REQUIRED' if verify_cert else 'CERT_NONE'
-        self._http = urllib3.PoolManager(
-            retries=retry_config,
-            timeout=urllib3.Timeout(request_timeout),
-            cert_reqs=str(cert_reqs),
+        self._session = requests.Session()
+        adapter = requests.adapters.HTTPAdapter(
+            max_retries=retry_config,
         )
 
+        self._session.mount('http://', adapter)
+        self._session.mount('https://', adapter)
+
     def send(self, endpoint, json_message, api_key=None, api_secret=None):
         """Immediately record an event or a profile update.
 
@@ -594,38 +596,36 @@ def send(self, endpoint, json_message, api_key=None, api_secret=None):
         self._write_request(self._endpoints[endpoint], json_message, api_key, api_secret)
 
     def _write_request(self, request_url, json_message, api_key=None, api_secret=None):
-        data = {
-            'data': json_message,
-            'verbose': 1,
-            'ip': 0,
-        }
-
         if isinstance(api_key, tuple):
             # For compatibility with subclassers, allow the auth details to be
             # packed into the existing api_key param.
             api_key, api_secret = api_key
 
+        params = {
+            'data': json_message,
+            'verbose': 1,
+            'ip': 0,
+        }
         if api_key:
-            data.update({'api_key': api_key})
-
-        headers = None
+            params['api_key'] = api_key
 
+        basic_auth = None
         if api_secret is not None:
-            headers = urllib3.util.make_headers(basic_auth="{}:".format(api_secret))
+            basic_auth = HTTPBasicAuth(api_secret, '')
 
         try:
-            response = self._http.request(
-                'POST',
+            response = self._session.post(
                 request_url,
-                fields=data,
-                headers=headers,
-                encode_multipart=False, # URL-encode payload in POST body.
+                json=params,
+                auth=basic_auth,
+                timeout=self._request_timeout,
+                verify=self._verify_cert,
             )
         except Exception as e:
             six.raise_from(MixpanelException(e), e)
 
         try:
-            response_dict = json.loads(response.data.decode('utf-8'))
+            response_dict = response.json()
         except ValueError:
             raise MixpanelException('Cannot interpret Mixpanel server response: {0}'.format(response.data))
 
@@ -669,7 +669,7 @@ class BufferedConsumer(object):
     """
     def __init__(self, max_size=50, events_url=None, people_url=None, import_url=None,
             request_timeout=None, groups_url=None, api_host="api.mixpanel.com",
-            retry_limit=4, retry_backoff_factor=0.25, verify_cert=False):
+            retry_limit=4, retry_backoff_factor=0.25, verify_cert=True):
         self._consumer = Consumer(events_url, people_url, import_url, request_timeout,
             groups_url, api_host, retry_limit, retry_backoff_factor, verify_cert)
         self._buffers = {

requirements-testing.txt

@@ -1,3 +1,3 @@
-mock==1.3.0
-pytest==4.6.11
-typing; python_version >='3.4' and python_version <'3.5' # To work around CI fail.
+pytest~=4.6
+responses~=0.13.3
+typing; python_version>='3.4' and python_version<'3.5' # To work around CI fail.

setup.py

@@ -25,8 +25,9 @@ def find_version(*paths):
     author_email='dev@mixpanel.com',
     license='Apache',
     install_requires=[
-        'six >= 1.9.0',
-        'urllib3 >= 1.21.1',
+        'six>=1.9.0',
+        'requests>=2.4.2',
+        'urllib3',
     ],
 
     classifiers=[

test_mixpanel.py

@@ -1,16 +1,13 @@
 from __future__ import absolute_import, unicode_literals
-import base64
-import contextlib
 import datetime
 import decimal
 import json
 import time
 
-from mock import Mock, patch
 import pytest
+import responses
 import six
 from six.moves import range
-import urllib3
 
 import mixpanel
 
@@ -286,20 +283,24 @@ def test_people_set_created_date_datetime(self):
             }
         )]
 
+    @responses.activate
     def test_alias(self):
         # More complicated since alias() forces a synchronous call.
-        mock_response = Mock()
-        mock_response.data = six.b('{"status": 1, "error": null}')
-        with patch('mixpanel.urllib3.PoolManager.request', return_value=mock_response) as req:
-            self.mp.alias('ALIAS', 'ORIGINAL ID')
-            assert self.consumer.log == []
-            assert req.call_count == 1
-            ((method, url), kwargs) = req.call_args
-
-            assert method == 'POST'
-            assert url == 'https://api.mixpanel.com/track'
-            expected_data = {"event":"$create_alias","properties":{"alias":"ALIAS","token":"12345","distinct_id":"ORIGINAL ID"}}
-            assert json.loads(kwargs["fields"]["data"]) == expected_data
+        responses.add(
+            responses.POST,
+            'https://api.mixpanel.com/track',
+            json={"status": 1, "error": None},
+            status=200,
+        )
+
+        self.mp.alias('ALIAS', 'ORIGINAL ID')
+        assert self.consumer.log == []
+        assert len(responses.calls) == 1
+        call = responses.calls[0]
+        assert call.request.method == "POST"
+        assert call.request.url == "https://api.mixpanel.com/track"
+        posted_data = json.loads(six.ensure_str(call.request.body))
+        assert json.loads(posted_data["data"]) == {"event":"$create_alias","properties":{"alias":"ALIAS","token":"12345","distinct_id":"ORIGINAL ID"}}
 
     def test_merge(self):
         self.mp.merge('my_good_api_key', 'd1', 'd2')
@@ -449,37 +450,49 @@ class TestConsumer:
     def setup_class(cls):
         cls.consumer = mixpanel.Consumer(request_timeout=30)
 
-    @contextlib.contextmanager
-    def _assertSends(self, expect_url, expect_data, consumer=None):
-        if consumer is None:
-            consumer = self.consumer
-
-        mock_response = Mock()
-        mock_response.data = six.b('{"status": 1, "error": null}')
-        with patch('mixpanel.urllib3.PoolManager.request', return_value=mock_response) as req:
-            yield
-
-            assert req.call_count == 1
-            (call_args, kwargs) = req.call_args
-            (method, url) = call_args
-            assert method == 'POST'
-            assert url == expect_url
-            assert kwargs["fields"] == expect_data
-
+    @responses.activate
     def test_send_events(self):
-        with self._assertSends('https://api.mixpanel.com/track', {"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'}):
-            self.consumer.send('events', '{"foo":"bar"}')
-
+        responses.add(
+            responses.POST,
+            'https://api.mixpanel.com/track',
+            json={"status": 1, "error": None},
+            status=200,
+            match=[responses.json_params_matcher({"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'})],
+        )
+        self.consumer.send('events', '{"foo":"bar"}')
+
+    @responses.activate
     def test_send_people(self):
-        with self._assertSends('https://api.mixpanel.com/engage', {"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'}):
-            self.consumer.send('people', '{"foo":"bar"}')
-
+        responses.add(
+            responses.POST,
+            'https://api.mixpanel.com/engage',
+            json={"status": 1, "error": None},
+            status=200,
+            match=[responses.json_params_matcher({"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'})],
+        )
+        self.consumer.send('people', '{"foo":"bar"}')
+
+    @responses.activate
     def test_consumer_override_api_host(self):
-        consumer = mixpanel.Consumer(api_host="api-eu.mixpanel.com")
-        with self._assertSends('https://api-eu.mixpanel.com/track', {"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'}, consumer=consumer):
-            consumer.send('events', '{"foo":"bar"}')
-        with self._assertSends('https://api-eu.mixpanel.com/engage', {"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'}, consumer=consumer):
-            consumer.send('people', '{"foo":"bar"}')
+        consumer = mixpanel.Consumer(api_host="api-zoltan.mixpanel.com")
+
+        responses.add(
+            responses.POST,
+            'https://api-zoltan.mixpanel.com/track',
+            json={"status": 1, "error": None},
+            status=200,
+            match=[responses.json_params_matcher({"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'})],
+        )
+        consumer.send('events', '{"foo":"bar"}')
+
+        responses.add(
+            responses.POST,
+            'https://api-zoltan.mixpanel.com/engage',
+            json={"status": 1, "error": None},
+            status=200,
+            match=[responses.json_params_matcher({"ip": 0, "verbose": 1, "data": '{"foo":"bar"}'})],
+        )
+        consumer.send('people', '{"foo":"bar"}')
 
     def test_unknown_endpoint(self):
         with pytest.raises(mixpanel.MixpanelException):
@@ -520,17 +533,23 @@ def test_unknown_endpoint_raises_on_send(self):
         with pytest.raises(mixpanel.MixpanelException):
             self.consumer.send('unknown', '1')
 
+    @responses.activate
     def test_useful_reraise_in_flush_endpoint(self):
-        error_mock = Mock()
-        error_mock.data = six.b('{"status": 0, "error": "arbitrary error"}')
+        responses.add(
+            responses.POST,
+            'https://api.mixpanel.com/track',
+            json={"status": 0, "error": "arbitrary error"},
+            status=200,
+        )
+
         broken_json = '{broken JSON'
         consumer = mixpanel.BufferedConsumer(2)
-        with patch('mixpanel.urllib3.PoolManager.request', return_value=error_mock):
-            consumer.send('events', broken_json)
-            with pytest.raises(mixpanel.MixpanelException) as excinfo:
-                consumer.flush()
-            assert excinfo.value.message == '[%s]' % broken_json
-            assert excinfo.value.endpoint == 'events'
+        consumer.send('events', broken_json)
+
+        with pytest.raises(mixpanel.MixpanelException) as excinfo:
+            consumer.flush()
+        assert excinfo.value.message == '[%s]' % broken_json
+        assert excinfo.value.endpoint == 'events'
 
     def test_send_remembers_api_key(self):
         self.consumer.send('imports', '"Event"', api_key='MY_API_KEY')
@@ -554,27 +573,40 @@ def setup_class(cls):
         cls.mp = mixpanel.Mixpanel(cls.TOKEN)
         cls.mp._now = lambda: 1000
 
-    @contextlib.contextmanager
-    def _assertRequested(self, expect_url, expect_data):
-        res = Mock()
-        res.data = six.b('{"status": 1, "error": null}')
-        with patch('mixpanel.urllib3.PoolManager.request', return_value=res) as req:
-            yield
-
-            assert req.call_count == 1
-            ((method, url,), data) = req.call_args
-            data = data["fields"]["data"]
-            assert method == 'POST'
-            assert url == expect_url
-            payload = json.loads(data)
-            assert payload == expect_data
-
+    @responses.activate
     def test_track_functional(self):
-        expect_data = {'event': 'button_press', 'properties': {'size': 'big', 'color': 'blue', 'mp_lib': 'python', 'token': '12345', 'distinct_id': 'player1', '$lib_version': mixpanel.__version__, 'time': 1000, '$insert_id': 'xyz1200'}}
-        with self._assertRequested('https://api.mixpanel.com/track', expect_data):
-            self.mp.track('player1', 'button_press', {'size': 'big', 'color': 'blue', '$insert_id': 'xyz1200'})
-
+        responses.add(
+            responses.POST,
+            'https://api.mixpanel.com/track',
+            json={"status": 1, "error": None},
+            status=200,
+        )
+
+        self.mp.track('player1', 'button_press', {'size': 'big', 'color': 'blue', '$insert_id': 'xyz1200'})
+        body = six.ensure_str(responses.calls[0].request.body)
+        wrapper = json.loads(body)
+        data = json.loads(wrapper["data"])
+        del wrapper["data"]
+
+        assert {"ip": 0, "verbose": 1} == wrapper
+        expected_data = {'event': 'button_press', 'properties': {'size': 'big', 'color': 'blue', 'mp_lib': 'python', 'token': '12345', 'distinct_id': 'player1', '$lib_version': mixpanel.__version__, 'time': 1000, '$insert_id': 'xyz1200'}}
+        assert expected_data == data
+
+    @responses.activate
     def test_people_set_functional(self):
-        expect_data = {'$distinct_id': 'amq', '$set': {'birth month': 'october', 'favorite color': 'purple'}, '$time': 1000, '$token': '12345'}
-        with self._assertRequested('https://api.mixpanel.com/engage', expect_data):
-            self.mp.people_set('amq', {'birth month': 'october', 'favorite color': 'purple'})
+        responses.add(
+            responses.POST,
+            'https://api.mixpanel.com/engage',
+            json={"status": 1, "error": None},
+            status=200,
+        )
+
+        self.mp.people_set('amq', {'birth month': 'october', 'favorite color': 'purple'})
+        body = six.ensure_str(responses.calls[0].request.body)
+        wrapper = json.loads(body)
+        data = json.loads(wrapper["data"])
+        del wrapper["data"]
+
+        assert {"ip": 0, "verbose": 1} == wrapper
+        expected_data = {'$distinct_id': 'amq', '$set': {'birth month': 'october', 'favorite color': 'purple'}, '$time': 1000, '$token': '12345'}
+        assert expected_data == data