Jason R. Coombs avatar Jason R. Coombs committed f5cd875

Added support to ProxyInfo object for detection of proxy from environment variables

Comments (0)

Files changed (2)

python2/httplib2/__init__.py

 import time
 import random
 import errno
-# remove depracated warning in python2.6
 try:
     from hashlib import sha1 as _sha, md5 as _md5
 except ImportError:
+    # prior to Python 2.5, these were separate modules
     import sha
     import md5
     _sha = sha.new
 
 class ProxyInfo(object):
     """Collect information required to use a proxy."""
+    bypass_hosts = ()
+
     def __init__(self, proxy_type, proxy_host, proxy_port,
         proxy_rdns=None, proxy_user=None, proxy_pass=None):
         """The parameter proxy_type must be set to one of socks.PROXY_TYPE_XXX
     def isgood(self):
         return (self.proxy_host != None) and (self.proxy_port != None)
 
+    @classmethod
+    def from_environment(cls, method='http'):
+        """
+        Read proxy info from the environment variables.
+        """
+        if method not in ['http', 'https']: return
+
+        env_var = method+'_proxy'
+        url = os.environ.get(env_var, os.environ.get(env_var.upper()))
+        if not url: return
+        pi = cls.from_url(url, method)
+
+        no_proxy = os.environ.get('no_proxy', '')
+        bypass_hosts = no_proxy.split(',') if no_proxy else []
+
+        pi.bypass_hosts = bypass_hosts
+        return pi
+
+    @classmethod
+    def from_url(cls, url, method='http'):
+        """
+        Construct a ProxyInfo from a URL
+        """
+        url = urlparse.urlparse(url)
+        ident, sep, host_port = url.netloc.rpartition('@')
+        username, sep, password = ident.partition(':')
+        host, sep, port = host_port.partition(':')
+        if port:
+            port = int(port)
+        else:
+            port = dict(https=443, http=80)[method]
+        proxy_type = 3 # socks.PROXY_TYPE_HTTP
+        return cls(
+            proxy_type = proxy_type,
+            proxy_host = host,
+            proxy_port = port,
+            proxy_user = username or None,
+            proxy_pass = password or None,
+        )
+
 
 class HTTPConnectionWithTimeout(httplib.HTTPConnection):
     """

python2/httplib2test.py

         end2end = httplib2._get_end2end_headers(response)
         self.assertEquals(0, len(end2end))
 
+
+class TestProxyInfo(unittest.TestCase):
+    def setUp(self):
+        self.orig_env = dict(os.environ)
+
+    def tearDown(self):
+        os.environ.clear()
+        os.environ.update(self.orig_env)
+
+    def test_from_url(self):
+        pi = httplib2.ProxyInfo.from_url('http://myproxy.example.com')
+        self.assertEquals(pi.proxy_host, 'myproxy.example.com')
+        self.assertEquals(pi.proxy_port, 80)
+        self.assertEquals(pi.proxy_user, None)
+
+    def test_from_url_ident(self):
+        pi = httplib2.ProxyInfo.from_url('http://zoidberg:fish@someproxy:99')
+        self.assertEquals(pi.proxy_host, 'someproxy')
+        self.assertEquals(pi.proxy_port, 99)
+        self.assertEquals(pi.proxy_user, 'zoidberg')
+        self.assertEquals(pi.proxy_pass, 'fish')
+
+    def test_from_env(self):
+        os.environ['http_proxy'] = 'http://myproxy.example.com:8080'
+        pi = httplib2.ProxyInfo.from_environment()
+        self.assertEquals(pi.proxy_host, 'myproxy.example.com')
+        self.assertEquals(pi.proxy_port, 8080)
+        self.assertEquals(pi.bypass_hosts, [])
+
+    def test_from_env_no_proxy(self):
+        os.environ['http_proxy'] = 'http://myproxy.example.com:80'
+        os.environ['https_proxy'] = 'http://myproxy.example.com:81'
+        os.environ['no_proxy'] = 'localhost,otherhost.domain.local'
+        pi = httplib2.ProxyInfo.from_environment('https')
+        self.assertEquals(pi.proxy_host, 'myproxy.example.com')
+        self.assertEquals(pi.proxy_port, 81)
+        self.assertEquals(pi.bypass_hosts, ['localhost',
+            'otherhost.domain.local'])
+
+    def test_from_env_none(self):
+        os.environ.clear()
+        pi = httplib2.ProxyInfo.from_environment()
+        self.assertEquals(pi, None)
+
+
 if __name__ == '__main__':
     unittest.main()
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.