# -*- coding: utf-8 -*-
# Licensed to the Apache Software Foundation (ASF) under one or more§
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import sys
import ssl
from mock import Mock, call
from libcloud.test import unittest
from libcloud.common.base import Connection
from libcloud.common.base import LoggingConnection
from libcloud.httplib_ssl import LibcloudBaseConnection
from libcloud.httplib_ssl import LibcloudHTTPConnection
[docs]class BaseConnectionClassTestCase(unittest.TestCase):
[docs] def test_parse_proxy_url(self):
conn = LibcloudBaseConnection()
proxy_url = 'http://127.0.0.1:3128'
result = conn._parse_proxy_url(proxy_url=proxy_url)
self.assertEqual(result[0], 'http')
self.assertEqual(result[1], '127.0.0.1')
self.assertEqual(result[2], 3128)
self.assertEqual(result[3], None)
self.assertEqual(result[4], None)
proxy_url = 'http://user1:pass1@127.0.0.1:3128'
result = conn._parse_proxy_url(proxy_url=proxy_url)
self.assertEqual(result[0], 'http')
self.assertEqual(result[1], '127.0.0.1')
self.assertEqual(result[2], 3128)
self.assertEqual(result[3], 'user1')
self.assertEqual(result[4], 'pass1')
proxy_url = 'https://127.0.0.1:3128'
expected_msg = 'Only http proxies are supported'
self.assertRaisesRegexp(ValueError, expected_msg,
conn._parse_proxy_url,
proxy_url=proxy_url)
proxy_url = 'http://127.0.0.1'
expected_msg = 'proxy_url must be in the following format'
self.assertRaisesRegexp(ValueError, expected_msg,
conn._parse_proxy_url,
proxy_url=proxy_url)
proxy_url = 'http://@127.0.0.1:3128'
expected_msg = 'URL is in an invalid format'
self.assertRaisesRegexp(ValueError, expected_msg,
conn._parse_proxy_url,
proxy_url=proxy_url)
proxy_url = 'http://user@127.0.0.1:3128'
expected_msg = 'URL is in an invalid format'
self.assertRaisesRegexp(ValueError, expected_msg,
conn._parse_proxy_url,
proxy_url=proxy_url)
[docs] def test_constructor(self):
conn = LibcloudHTTPConnection(host='localhost', port=80)
self.assertEqual(conn.proxy_scheme, None)
self.assertEqual(conn.proxy_host, None)
self.assertEqual(conn.proxy_port, None)
proxy_url = 'http://127.0.0.3:3128'
conn.set_http_proxy(proxy_url=proxy_url)
self.assertEqual(conn.proxy_scheme, 'http')
self.assertEqual(conn.proxy_host, '127.0.0.3')
self.assertEqual(conn.proxy_port, 3128)
proxy_url = 'http://127.0.0.4:3128'
conn = LibcloudHTTPConnection(host='localhost', port=80,
proxy_url=proxy_url)
self.assertEqual(conn.proxy_scheme, 'http')
self.assertEqual(conn.proxy_host, '127.0.0.4')
self.assertEqual(conn.proxy_port, 3128)
os.environ['http_proxy'] = proxy_url
proxy_url = 'http://127.0.0.5:3128'
conn = LibcloudHTTPConnection(host='localhost', port=80,
proxy_url=proxy_url)
self.assertEqual(conn.proxy_scheme, 'http')
self.assertEqual(conn.proxy_host, '127.0.0.5')
self.assertEqual(conn.proxy_port, 3128)
[docs]class ConnectionClassTestCase(unittest.TestCase):
[docs] def setUp(self):
self.originalConnect = Connection.connect
self.originalResponseCls = Connection.responseCls
Connection.connect = Mock()
Connection.responseCls = Mock()
Connection.allow_insecure = True
[docs] def tearDown(self):
Connection.connect = self.originalConnect
Connection.responseCls = Connection.responseCls
Connection.allow_insecure = True
[docs] def test_dont_allow_insecure(self):
Connection.allow_insecure = True
Connection(secure=False)
Connection.allow_insecure = False
expected_msg = (r'Non https connections are not allowed \(use '
'secure=True\)')
self.assertRaisesRegexp(ValueError, expected_msg, Connection,
secure=False)
[docs] def test_content_length(self):
con = Connection()
con.connection = Mock()
# GET method
# No data, no content length should be present
con.request('/test', method='GET', data=None)
call_kwargs = con.connection.request.call_args[1]
self.assertTrue('Content-Length' not in call_kwargs['headers'])
# '' as data, no content length should be present
con.request('/test', method='GET', data='')
call_kwargs = con.connection.request.call_args[1]
self.assertTrue('Content-Length' not in call_kwargs['headers'])
# 'a' as data, content length should be present (data in GET is not
# correct, but anyways)
con.request('/test', method='GET', data='a')
call_kwargs = con.connection.request.call_args[1]
self.assertEqual(call_kwargs['headers']['Content-Length'], '1')
# POST, PUT method
# No data, content length should be present
for method in ['POST', 'PUT', 'post', 'put']:
con.request('/test', method=method, data=None)
call_kwargs = con.connection.request.call_args[1]
self.assertEqual(call_kwargs['headers']['Content-Length'], '0')
# '' as data, content length should be present
for method in ['POST', 'PUT', 'post', 'put']:
con.request('/test', method=method, data='')
call_kwargs = con.connection.request.call_args[1]
self.assertEqual(call_kwargs['headers']['Content-Length'], '0')
# No data, raw request, do not touch Content-Length if present
for method in ['POST', 'PUT', 'post', 'put']:
con.request('/test', method=method, data=None,
headers={'Content-Length': '42'}, raw=True)
putheader_call_list = con.connection.putheader.call_args_list
self.assertIn(call('Content-Length', '42'), putheader_call_list)
# '' as data, raw request, do not touch Content-Length if present
for method in ['POST', 'PUT', 'post', 'put']:
con.request('/test', method=method, data=None,
headers={'Content-Length': '42'}, raw=True)
putheader_call_list = con.connection.putheader.call_args_list
self.assertIn(call('Content-Length', '42'), putheader_call_list)
# 'a' as data, content length should be present
for method in ['POST', 'PUT', 'post', 'put']:
con.request('/test', method=method, data='a')
call_kwargs = con.connection.request.call_args[1]
self.assertEqual(call_kwargs['headers']['Content-Length'], '1')
[docs] def test_cache_busting(self):
params1 = {'foo1': 'bar1', 'foo2': 'bar2'}
params2 = [('foo1', 'bar1'), ('foo2', 'bar2')]
con = Connection()
con.connection = Mock()
con.pre_connect_hook = Mock()
con.pre_connect_hook.return_value = {}, {}
con.cache_busting = False
con.request(action='/path', params=params1)
args, kwargs = con.pre_connect_hook.call_args
self.assertFalse('cache-busting' in args[0])
self.assertEqual(args[0], params1)
con.request(action='/path', params=params2)
args, kwargs = con.pre_connect_hook.call_args
self.assertFalse('cache-busting' in args[0])
self.assertEqual(args[0], params2)
con.cache_busting = True
con.request(action='/path', params=params1)
args, kwargs = con.pre_connect_hook.call_args
self.assertTrue('cache-busting' in args[0])
con.request(action='/path', params=params2)
args, kwargs = con.pre_connect_hook.call_args
self.assertTrue('cache-busting' in args[0][len(params2)])
[docs] def test_context_is_reset_after_request_has_finished(self):
context = {'foo': 'bar'}
def responseCls(connection, response):
connection.called = True
self.assertEqual(connection.context, context)
con = Connection()
con.called = False
con.connection = Mock()
con.responseCls = responseCls
con.set_context(context)
self.assertEqual(con.context, context)
con.request('/')
# Context should have been reset
self.assertTrue(con.called)
self.assertEqual(con.context, {})
# Context should also be reset if a method inside request throws
con = Connection()
con.connection = Mock()
con.set_context(context)
self.assertEqual(con.context, context)
con.connection.request = Mock(side_effect=ssl.SSLError())
try:
con.request('/')
except ssl.SSLError:
pass
self.assertEqual(con.context, {})
con.connection = Mock()
con.set_context(context)
self.assertEqual(con.context, context)
con.responseCls = Mock(side_effect=ValueError())
try:
con.request('/')
except ValueError:
pass
self.assertEqual(con.context, {})
[docs] def test_log_curl(self):
url = '/test/path'
body = None
headers = {}
con = LoggingConnection()
con.protocol = 'http'
con.host = 'example.com'
con.port = 80
for method in ['GET', 'POST', 'PUT', 'DELETE']:
cmd = con._log_curl(method=method, url=url, body=body,
headers=headers)
self.assertEqual(cmd, 'curl -i -X %s --compress http://example.com:80/test/path' %
(method))
# Should use --head for head requests
cmd = con._log_curl(method='HEAD', url=url, body=body, headers=headers)
self.assertEqual(cmd, 'curl -i --head --compress http://example.com:80/test/path')
if __name__ == '__main__':
sys.exit(unittest.main())