Add proxy support for SSE transport

requirements-test.txt

@@ -0,0 +1,3 @@
+-r requirements.txt
+
+httmock

requirements.txt

@@ -1,5 +1,4 @@
 certifi
-httmock
 protobuf>=3.0.0
 pyformance>=0.3.1
 requests>=2.7.0

setup.py

@@ -26,8 +26,8 @@
     long_description=long_description,
     zip_safe=True,
     packages=find_packages(),
-    test_requires=test_requirements,
     install_requires=requirements,
+    tests_require=test_requirements,
     classifiers=[
         'Operating System :: OS Independent',
         'Programming Language :: Python',

signalfx/signalflow/__init__.py

@@ -16,8 +16,9 @@ class SignalFlowClient(object):
     def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
                  timeout=constants.DEFAULT_TIMEOUT,
                  transport=ws.WebSocketTransport,
-                 compress=True):
-        self._transport = transport(token, endpoint, timeout, compress)
+                 compress=True, proxy_url=None):
+        self._transport = transport(token, endpoint, timeout, compress,
+                                    proxy_url)
         self._computations = set([])
 
     def __enter__(self):

signalfx/signalflow/sse.py

@@ -24,7 +24,8 @@ class SSETransport(transport._SignalFlowTransport):
     _SIGNALFLOW_ENDPOINT = 'v2/signalflow'
 
     def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
-                 timeout=constants.DEFAULT_TIMEOUT, compress=True):
+                 timeout=constants.DEFAULT_TIMEOUT, compress=True,
+                 proxy_url=None):
         super(SSETransport, self).__init__(token, endpoint, timeout)
         pool_args = {
             'url': self._endpoint,
@@ -43,7 +44,12 @@ def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
                 'ca_certs': certifi.where()    # Path to the Certifi bundle.
             })
 
-        self._http = urllib3.connectionpool.connection_from_url(**pool_args)
+        if proxy_url:
+            proxy_manager = urllib3.poolmanager.proxy_from_url(proxy_url)
+            endpoint = pool_args.pop('url')
+            self._http = proxy_manager.connection_from_url(endpoint, pool_kwargs=pool_args)
+        else:
+            self._http = urllib3.connectionpool.connection_from_url(**pool_args)
 
     def __str__(self):
         return 'sse+{0}'.format(self._endpoint)

signalfx/signalflow/ws.py

@@ -29,7 +29,11 @@ class WebSocketTransport(transport._SignalFlowTransport, WebSocketClient):
     _SIGNALFLOW_WEBSOCKET_ENDPOINT = 'v2/signalflow/connect'
 
     def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
-                 timeout=constants.DEFAULT_TIMEOUT, compress=True):
+                 timeout=constants.DEFAULT_TIMEOUT, compress=True,
+                 proxy_url=None):
+        if proxy_url:
+            raise NotImplementedError('Websocket transport cannot be proxied!')
+
         ws_endpoint = '{0}/{1}'.format(
             endpoint.replace('http', 'ws', 1),
             WebSocketTransport._SIGNALFLOW_WEBSOCKET_ENDPOINT)

tox.ini

@@ -8,7 +8,7 @@ envlist = py27,py35,flake8
 [testenv]
 usedevelop = True
 passenv = ORG_NAME_FOR_SIGNALFX_API_TOKEN_IS_SFx_Test SIGNALFX_API_TOKEN
-deps = -r{toxinidir}/requirements.txt
+deps = -r{toxinidir}/requirements-test.txt
 ignore_errors = True
 commands = python {toxinidir}/tests/test_data_reporting.py
            python {toxinidir}/tests/live_tests.py --metric_name 'MET' --tag_name 'TG' --key 'K' --value 'V'