Andriy Kornatskyy avatar Andriy Kornatskyy committed eed1618

Introduced httpclient module; added unit tests.

Comments (0)

Files changed (3)

src/wheezy/core/comp.py

     parsedate = lambda s: time.strptime(s, "%a, %d %b %Y %H:%M:%S GMT")  # noqa
 
 if PY3:  # pragma: nocover
+    from http.cookies import SimpleCookie
+else:  # pragma: nocover
+    from Cookie import SimpleCookie  # noqa
+
+if PY3:  # pragma: nocover
+    from urllib.parse import urlencode
+    from urllib.parse import urljoin
+    from urllib.parse import urlparse
     from urllib.parse import urlsplit
     from urllib.parse import urlunsplit
 else:  # pragma: nocover
+    from urllib import urlencode  # noqa
+    from urlparse import urljoin  # noqa
+    from urlparse import urlparse  # noqa
     from urlparse import urlsplit  # noqa
     from urlparse import urlunsplit  # noqa
 
 
         def json_loads(s, **kw):  # noqa
             raise NotImplementedError('JSON decoder is required.')
+
+
+if PY3 and PY_MINOR >= 3:  # pragma: nocover
+    from decimal import Decimal
+else:  # pragma: nocover
+    try:
+        from cdecimal import Decimal  # noqa
+    except ImportError:
+        from decimal import Decimal  # noqa

src/wheezy/core/httpclient.py

+"""
+"""
+
+from httplib import HTTPConnection
+
+from wheezy.core.collections import attrdict
+from wheezy.core.collections import defaultdict
+from wheezy.core.comp import Decimal
+from wheezy.core.comp import SimpleCookie
+from wheezy.core.comp import json_loads
+from wheezy.core.comp import urlencode
+from wheezy.core.comp import urljoin
+from wheezy.core.comp import urlparse
+from wheezy.core.comp import urlsplit
+
+
+class HTTPClient(object):
+
+    def __init__(self, url, headers=None):
+        """ HTTP client sends HTTP requests to server in order to accomplish
+            application specific use cases, e.g. remote web server API, etc.
+        """
+        r = urlparse(url)
+        self.connection = HTTPConnection(r.netloc)
+        self.default_headers = headers and headers or {}
+        self.path = r.path
+        self.method = None
+        self.headers = None
+        self.cookies = {}
+        self.status_code = 0
+        self.content = None
+        self.__json = None
+
+    @property
+    def json(self):
+        """ Returns a json response.
+        """
+        if self.__json is None:
+            assert 'application/json' in self.headers['content-type'][0]
+            self.__json = json_loads(self.content,
+                                     object_hook=attrdict,
+                                     parse_float=Decimal)
+        return self.__json
+
+    def get(self, path, **kwargs):
+        """ Sends GET HTTP request.
+        """
+        return self.go(path, 'GET', **kwargs)
+
+    def ajax_get(self, path, **kwargs):
+        """ Sends GET HTTP AJAX request.
+        """
+        return self.ajax_go(path, 'GET', **kwargs)
+
+    def head(self, path, **kwargs):
+        """ Sends HEAD HTTP request.
+        """
+        return self.go(path, 'HEAD', **kwargs)
+
+    def post(self, path, **kwargs):
+        """ Sends POST HTTP request.
+        """
+        return self.go(path, 'POST', **kwargs)
+
+    def ajax_post(self, path, **kwargs):
+        """ Sends POST HTTP AJAX request.
+        """
+        return self.ajax_go(path, 'POST', **kwargs)
+
+    def follow(self):
+        """ Follows HTTP redirect (e.g. status code 302).
+        """
+        sc = self.status_code
+        assert sc in [207, 301, 302, 303, 307]
+        location = self.headers['location'][0]
+        scheme, netloc, path, query, fragment = urlsplit(location)
+        method = sc == 307 and self.method or 'GET'
+        return self.go(path, method)
+
+    def ajax_go(self, path=None, method='GET', params=None, headers=None):
+        """ Sends HTTP AJAX request to web server.
+        """
+        headers = headers or {}
+        headers['X-Requested-With'] = 'XMLHttpRequest'
+        return self.go(path, method, params, headers)
+
+    def go(self, path=None, method='GET', params=None, headers=None):
+        """ Sends HTTP request to web server.
+        """
+        self.method = method
+        headers = headers and dict(self.default_headers,
+                                   **headers) or dict(self.default_headers)
+        if self.cookies:
+            headers['Cookie'] = '; '.join(
+                '%s=%s' % cookie for cookie in self.cookies.items())
+        path = urljoin(self.path, path)
+        body = ''
+        if params:
+            if method == 'GET':
+                path += '?' + urlencode(params, doseq=True)
+            else:
+                body = urlencode(params, doseq=True)
+                headers['Content-Type'] = 'application/x-www-form-urlencoded'
+
+        self.status_code = 0
+        self.content = None
+        self.__json = None
+
+        self.connection.request(method, path, body, headers)
+        r = self.connection.getresponse()
+        self.content = r.read().decode('utf-8')
+        self.connection.close()
+
+        self.status_code = r.status
+        self.headers = defaultdict(list)
+        for name, value in r.getheaders():
+            self.headers[name].append(value)
+        for cookie_string in self.headers['set-cookie']:
+            cookies = SimpleCookie(cookie_string)
+            for name in cookies:
+                value = cookies[name].value
+                if value:
+                    self.cookies[name] = value
+                elif name in self.cookies:
+                    del self.cookies[name]
+        return self.status_code

src/wheezy/core/tests/test_httpclient.py

+
+"""
+"""
+
+import unittest
+
+from mock import Mock
+from mock import patch
+
+
+class HTTPClientTestCase(unittest.TestCase):
+
+    def setUp(self):
+        from wheezy.core import __version__
+        from wheezy.core import httpclient
+        from wheezy.core.comp import ntob
+        self.patcher = patch.object(httpclient, 'HTTPConnection')
+        self.mock_c_class = self.patcher.start()
+        self.headers = [
+            ('date', 'Sat, 12 Oct 2013 18:29:13 GMT')
+        ]
+        self.mock_response = Mock()
+        self.mock_response.getheaders.return_value = self.headers
+        self.mock_response.read.return_value = ntob('', 'utf-8')
+        self.mock_c = Mock()
+        self.mock_c.getresponse.return_value = self.mock_response
+        self.mock_c_class.return_value = self.mock_c
+        self.client = httpclient.HTTPClient(
+            'http://localhost:8080/api/v1/',
+            headers={
+                'User-Agent': 'wheezy/%s' % __version__
+            })
+
+    def tearDown(self):
+        self.patcher.stop()
+
+    def test_init(self):
+        self.mock_c_class.assert_called_once_with('localhost:8080')
+        assert '/api/v1/' == self.client.path
+        assert {} == self.client.cookies
+        assert None == self.client.headers
+
+    def test_get(self):
+        self.mock_response.status = 200
+        assert 200 == self.client.get('auth/token')
+        assert self.mock_c.request.called
+        method, path, body, headers = self.mock_c.request.call_args[0]
+        assert 'GET' == method
+        assert '/api/v1/auth/token' == path
+        assert '' == body
+        assert self.client.default_headers == headers
+
+    def test_ajax_get(self):
+        self.client.ajax_get('auth/token')
+        method, path, body, headers = self.mock_c.request.call_args[0]
+        assert 'XMLHttpRequest' == headers['X-Requested-With']
+
+    def test_get_query(self):
+        self.client.get('auth/token', params={
+            'a': ['1']
+        })
+        method, path, body, headers = self.mock_c.request.call_args[0]
+        assert '/api/v1/auth/token?a=1' == path
+
+    def test_head(self):
+        self.client.head('auth/token')
+        method, path, body, headers = self.mock_c.request.call_args[0]
+        assert 'HEAD' == method
+
+    def test_post(self):
+        self.client.post('auth/token', params={
+            'a': ['1'],
+            'b': ['2']
+        })
+        method, path, body, headers = self.mock_c.request.call_args[0]
+        assert 'POST' == method
+        assert '/api/v1/auth/token' == path
+        assert 'a=1&b=2' == body
+        assert 'application/x-www-form-urlencoded' == headers['Content-Type']
+
+    def test_ajax_post(self):
+        self.client.ajax_post('auth/token', params={
+            'a': ['1']
+        })
+        assert self.mock_c.request.called
+        method, path, body, headers = self.mock_c.request.call_args[0]
+        assert 'XMLHttpRequest' == headers['X-Requested-With']
+
+    def test_follow(self):
+        self.mock_response.status = 303
+        self.headers.append(('location', 'http://localhost:8080/error/401'))
+        assert 303 == self.client.get('auth/token')
+        self.client.follow()
+        method, path, body, headers = self.mock_c.request.call_args[0]
+        assert 'GET' == method
+        assert '/error/401' == path
+
+    def test_cookies(self):
+        self.headers.append(('set-cookie', '_x=1; path=/; httponly'))
+        self.client.get('auth/token')
+        assert self.client.cookies
+        assert '1' == self.client.cookies['_x']
+
+        self.headers.append(('set-cookie', '_x=; path=/; httponly'))
+        self.client.get('auth/token')
+        assert not self.client.cookies
+
+    def test_assert_json(self):
+        """ Expecting json response but content type is not valid.
+        """
+        self.headers.append(('content-type', 'text/html; charset=UTF-8'))
+        self.client.get('auth/token')
+        self.assertRaises(AssertionError, lambda: self.client.json)
+
+    def test_json(self):
+        """ json response.
+        """
+        from wheezy.core.comp import ntob
+        self.headers.append(('content-type',
+                             'application/json; charset=UTF-8'))
+        self.mock_response.read.return_value = ntob('{}', 'utf-8')
+        self.client.get('auth/token')
+        assert {} == self.client.json
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.