Replace mox3 with mock

In heatclient, some unit tests still use mox3,
which will make the test speed much slow, so I decide to remove it.
The modify file: heatclient/tests/unit/test_common_http.py

Change-Id: I220fabedb8e9fcc8eeea9d5e98b3ddeed68f8a43
Implements: bp remove-mox3-in-heatclient
This commit is contained in:
shizhihui 2016-11-16 13:13:49 +08:00
parent 38dc7e6fd7
commit 16d86599c3

View File

@ -12,15 +12,11 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import logging
import os
import socket import socket
from keystoneauth1 import adapter from keystoneauth1 import adapter
import mock import mock
import mox
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
import requests
import six import six
import testtools import testtools
@ -30,138 +26,108 @@ from heatclient import exc
from heatclient.tests.unit import fakes from heatclient.tests.unit import fakes
@mock.patch('heatclient.common.http.requests.request')
class HttpClientTest(testtools.TestCase): class HttpClientTest(testtools.TestCase):
# Patch os.environ to avoid required auth info. def test_http_raw_request(self, mock_request):
def setUp(self):
super(HttpClientTest, self).setUp()
self.m = mox.Mox()
self.m.StubOutWithMock(requests, 'request')
self.addCleanup(self.m.VerifyAll)
self.addCleanup(self.m.UnsetStubs)
def test_http_raw_request(self):
headers = {'Content-Type': 'application/octet-stream', headers = {'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient'} 'User-Agent': 'python-heatclient'}
# Record a 200 # Record a 200
mock_conn = http.requests.request('GET', 'http://example.com:8004', mock_request.return_value = fakes.FakeHTTPResponse(
allow_redirects=False, 200, 'OK',
headers=headers) {'content-type': 'application/octet-stream'},
mock_conn.AndReturn( '')
fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/octet-stream'},
''))
# Replay, create client, assert # Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
resp = client.raw_request('GET', '') resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
self.assertEqual('', ''.join([x for x in resp.content])) self.assertEqual('', ''.join([x for x in resp.content]))
mock_request.assert_called_with('GET', 'http://example.com:8004',
allow_redirects=False,
headers=headers)
def test_token_or_credentials(self): def test_token_or_credentials(self, mock_request):
# Record a 200 # Record a 200
fake200 = fakes.FakeHTTPResponse( fake200 = fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
{'content-type': 'application/octet-stream'}, {'content-type': 'application/octet-stream'},
'') '')
mock_request.return_value = fake200
# no token or credentials # no token or credentials
mock_conn = http.requests.request(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(fake200)
# credentials
mock_conn = http.requests.request(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Key': 'pass',
'X-Auth-User': 'user'})
mock_conn.AndReturn(fake200)
# token suppresses credentials
mock_conn = http.requests.request(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Token': 'abcd1234'})
mock_conn.AndReturn(fake200)
# Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
resp = client.raw_request('GET', '') resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
# credentials
client.username = 'user' client.username = 'user'
client.password = 'pass' client.password = 'pass'
resp = client.raw_request('GET', '') resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
# token suppresses credentials
client.auth_token = 'abcd1234' client.auth_token = 'abcd1234'
resp = client.raw_request('GET', '') resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
mock_request.assert_has_calls([
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient'}),
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Key': 'pass',
'X-Auth-User': 'user'}),
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Token': 'abcd1234'})
])
def test_include_pass(self): def test_include_pass(self, mock_request):
# Record a 200 # Record a 200
fake200 = fakes.FakeHTTPResponse( fake200 = fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
{'content-type': 'application/octet-stream'}, {'content-type': 'application/octet-stream'},
'') '')
mock_request.return_value = fake200
# no token or credentials # no token or credentials
mock_conn = http.requests.request(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(fake200)
# credentials
mock_conn = http.requests.request(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Key': 'pass',
'X-Auth-User': 'user'})
mock_conn.AndReturn(fake200)
# token suppresses credentials
mock_conn = http.requests.request(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Token': 'abcd1234',
'X-Auth-Key': 'pass',
'X-Auth-User': 'user'})
mock_conn.AndReturn(fake200)
# Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
resp = client.raw_request('GET', '') resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
# credentials
client.username = 'user' client.username = 'user'
client.password = 'pass' client.password = 'pass'
client.include_pass = True client.include_pass = True
resp = client.raw_request('GET', '') resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
# token suppresses credentials
client.auth_token = 'abcd1234' client.auth_token = 'abcd1234'
resp = client.raw_request('GET', '') resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
mock_request.assert_has_calls([
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient'}),
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Key': 'pass',
'X-Auth-User': 'user'}),
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient',
'X-Auth-Token': 'abcd1234',
'X-Auth-Key': 'pass',
'X-Auth-User': 'user'})
])
def test_not_include_pass(self): def test_not_include_pass(self, mock_request):
# Record a 200 # Record a 200
fake500 = fakes.FakeHTTPResponse( fake500 = fakes.FakeHTTPResponse(
500, 'ERROR', 500, 'ERROR',
@ -169,21 +135,20 @@ class HttpClientTest(testtools.TestCase):
'(HTTP 401)') '(HTTP 401)')
# no token or credentials # no token or credentials
mock_conn = http.requests.request( mock_request.return_value = fake500
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(fake500)
# Replay, create client, assert # Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
e = self.assertRaises(exc.HTTPUnauthorized, e = self.assertRaises(exc.HTTPUnauthorized,
client.raw_request, 'GET', '') client.raw_request, 'GET', '')
self.assertIn('Authentication failed', str(e)) self.assertIn('Authentication failed', str(e))
mock_request.assert_called_with(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/octet-stream',
'User-Agent': 'python-heatclient'})
def test_region_name(self): def test_region_name(self, mock_request):
# Record a 200 # Record a 200
fake200 = fakes.FakeHTTPResponse( fake200 = fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
@ -191,45 +156,55 @@ class HttpClientTest(testtools.TestCase):
'') '')
# Specify region name # Specify region name
mock_conn = http.requests.request( mock_request.return_value = fake200
# Replay, create client, assert
client = http.HTTPClient('http://example.com:8004')
client.region_name = 'RegionOne'
resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code)
mock_request.assert_called_with(
'GET', 'http://example.com:8004', 'GET', 'http://example.com:8004',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/octet-stream', headers={'Content-Type': 'application/octet-stream',
'X-Region-Name': 'RegionOne', 'X-Region-Name': 'RegionOne',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(fake200)
# Replay, create client, assert def test_http_json_request(self, mock_request):
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004')
client.region_name = 'RegionOne'
resp = client.raw_request('GET', '')
self.assertEqual(200, resp.status_code)
def test_http_json_request(self):
# Record a 200 # Record a 200
mock_conn = http.requests.request( mock_request.return_value = fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}')
# Replay, create client, assert
client = http.HTTPClient('http://example.com:8004')
resp, body = client.json_request('GET', '')
self.assertEqual(200, resp.status_code)
self.assertEqual({}, body)
mock_request.assert_called_with(
'GET', 'http://example.com:8004', 'GET', 'http://example.com:8004',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/json', headers={'Content-Type': 'application/json',
'Accept': 'application/json', 'Accept': 'application/json',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}'))
# Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004')
resp, body = client.json_request('GET', '')
self.assertEqual(200, resp.status_code)
self.assertEqual({}, body)
def test_http_json_request_argument_passed_to_requests(self): def test_http_json_request_argument_passed_to_requests(self, mock_request):
"""Check that we have sent the proper arguments to requests.""" """Check that we have sent the proper arguments to requests."""
# Record a 200 # Record a 200
mock_conn = http.requests.request( mock_request.return_value = fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}')
client = http.HTTPClient('http://example.com:8004')
client.verify_cert = True
client.cert_file = 'RANDOM_CERT_FILE'
client.key_file = 'RANDOM_KEY_FILE'
client.auth_url = 'http://AUTH_URL'
resp, body = client.json_request('GET', '', data='text')
self.assertEqual(200, resp.status_code)
self.assertEqual({}, body)
mock_request.assert_called_with(
'GET', 'http://example.com:8004', 'GET', 'http://example.com:8004',
allow_redirects=False, allow_redirects=False,
cert=('RANDOM_CERT_FILE', 'RANDOM_KEY_FILE'), cert=('RANDOM_CERT_FILE', 'RANDOM_KEY_FILE'),
@ -239,410 +214,350 @@ class HttpClientTest(testtools.TestCase):
'Accept': 'application/json', 'Accept': 'application/json',
'X-Auth-Url': 'http://AUTH_URL', 'X-Auth-Url': 'http://AUTH_URL',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}'))
# Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004')
client.verify_cert = True
client.cert_file = 'RANDOM_CERT_FILE'
client.key_file = 'RANDOM_KEY_FILE'
client.auth_url = 'http://AUTH_URL'
resp, body = client.json_request('GET', '', data='text')
self.assertEqual(200, resp.status_code)
self.assertEqual({}, body)
def test_http_json_request_w_req_body(self): def test_http_json_request_w_req_body(self, mock_request):
# Record a 200 # Record a 200
mock_conn = http.requests.request( mock_request.return_value = fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}')
# Replay, create client, assert
client = http.HTTPClient('http://example.com:8004')
resp, body = client.json_request('GET', '', body='test-body')
self.assertEqual(200, resp.status_code)
mock_request.assert_called_with(
'GET', 'http://example.com:8004', 'GET', 'http://example.com:8004',
body='test-body', body='test-body',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/json', headers={'Content-Type': 'application/json',
'Accept': 'application/json', 'Accept': 'application/json',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}'))
# Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004')
resp, body = client.json_request('GET', '', body='test-body')
self.assertEqual(200, resp.status_code)
self.assertEqual({}, body)
def test_http_json_request_non_json_resp_cont_type(self): def test_http_json_request_non_json_resp_cont_type(self, mock_request):
# Record a 200 # Record a 200i
mock_conn = http.requests.request( mock_request.return_value = fakes.FakeHTTPResponse(
'GET', 'http://example.com:8004', body='test-body', 200, 'OK',
allow_redirects=False, {'content-type': 'not/json'},
headers={'Content-Type': 'application/json', '{}')
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'not/json'},
{}))
# Replay, create client, assert # Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
resp, body = client.json_request('GET', '', body='test-body') resp, body = client.json_request('GET', '', body='test-body')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
self.assertIsNone(body) self.assertIsNone(body)
mock_request.assert_called_with(
def test_http_json_request_invalid_json(self):
# Record a 200
mock_conn = http.requests.request(
'GET', 'http://example.com:8004', 'GET', 'http://example.com:8004',
body='test-body',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/json', headers={'Content-Type': 'application/json',
'Accept': 'application/json', 'Accept': 'application/json',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( def test_http_json_request_invalid_json(self, mock_request):
200, 'OK', # Record a 200
{'content-type': 'application/json'}, mock_request.return_value = fakes.FakeHTTPResponse(
'invalid-json')) 200, 'OK',
{'content-type': 'application/json'},
'invalid-json')
# Replay, create client, assert # Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
resp, body = client.json_request('GET', '') resp, body = client.json_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
self.assertEqual('invalid-json', body) self.assertEqual('invalid-json', body)
mock_request.assert_called_with(
def test_http_manual_redirect_delete(self): 'GET', 'http://example.com:8004',
mock_conn = http.requests.request(
'DELETE', 'http://example.com:8004/foo',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/json', headers={'Content-Type': 'application/json',
'Accept': 'application/json', 'Accept': 'application/json',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
def test_http_manual_redirect_delete(self, mock_request):
mock_request.side_effect = [
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
302, 'Found', 302, 'Found',
{'location': 'http://example.com:8004/foo/bar'}, {'location': 'http://example.com:8004/foo/bar'},
'')) ''),
mock_conn = http.requests.request(
'DELETE', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
{'content-type': 'application/json'}, {'content-type': 'application/json'},
'{}')) 'invalid-json')
]
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004/foo') client = http.HTTPClient('http://example.com:8004/foo')
resp, body = client.json_request('DELETE', '') resp, body = client.json_request('DELETE', '')
self.assertEqual(200, resp.status_code) mock_request.assert_has_calls([
mock.call('DELETE', 'http://example.com:8004/foo',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'}),
mock.call('DELETE', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
])
def test_http_manual_redirect_post(self): def test_http_manual_redirect_post(self, mock_request):
mock_conn = http.requests.request( mock_request.side_effect = [
'POST', 'http://example.com:8004/foo',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
302, 'Found', 302, 'Found',
{'location': 'http://example.com:8004/foo/bar'}, {'location': 'http://example.com:8004/foo/bar'},
'')) ''),
mock_conn = http.requests.request(
'POST', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
{'content-type': 'application/json'}, {'content-type': 'application/json'},
'{}')) 'invalid-json')
]
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004/foo') client = http.HTTPClient('http://example.com:8004/foo')
resp, body = client.json_request('POST', '') resp, body = client.json_request('POST', '')
self.assertEqual(200, resp.status_code) mock_request.assert_has_calls([
mock.call('POST', 'http://example.com:8004/foo',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'}),
mock.call('POST', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
])
def test_http_manual_redirect_put(self): def test_http_manual_redirect_put(self, mock_request):
mock_conn = http.requests.request( mock_request.side_effect = [
'PUT', 'http://example.com:8004/foo',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
302, 'Found', 302, 'Found',
{'location': 'http://example.com:8004/foo/bar'}, {'location': 'http://example.com:8004/foo/bar'},
'')) ''),
mock_conn = http.requests.request(
'PUT', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
{'content-type': 'application/json'}, {'content-type': 'application/json'},
'{}')) 'invalid-json')
]
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004/foo') client = http.HTTPClient('http://example.com:8004/foo')
resp, body = client.json_request('PUT', '') resp, body = client.json_request('PUT', '')
self.assertEqual(200, resp.status_code) mock_request.assert_has_calls([
mock.call('PUT', 'http://example.com:8004/foo',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'}),
mock.call('PUT', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
])
def test_http_manual_redirect_put_uppercase(self): def test_http_manual_redirect_put_uppercase(self, mock_request):
mock_conn = http.requests.request( mock_request.side_effect = [
'PUT', 'http://EXAMPLE.com:8004/foo',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
302, 'Found', 302, 'Found',
{'location': 'http://example.com:8004/foo/bar'}, {'location': 'http://example.com:8004/foo/bar'},
'')) ''),
mock_conn = http.requests.request(
'PUT', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
{'content-type': 'application/json'}, {'content-type': 'application/json'},
'{}')) 'invalid-json')
]
self.m.ReplayAll()
client = http.HTTPClient('http://EXAMPLE.com:8004/foo') client = http.HTTPClient('http://EXAMPLE.com:8004/foo')
resp, body = client.json_request('PUT', '') resp, body = client.json_request('PUT', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
def test_http_manual_redirect_error_without_location(self): mock_request.assert_has_calls([
mock_conn = http.requests.request( mock.call('PUT', 'http://EXAMPLE.com:8004/foo',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'}),
mock.call('PUT', 'http://example.com:8004/foo/bar',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
])
def test_http_manual_redirect_error_without_location(self, mock_request):
mock_request.return_value = fakes.FakeHTTPResponse(
302, 'Found',
{},
'')
client = http.HTTPClient('http://example.com:8004/foo')
self.assertRaises(exc.InvalidEndpoint,
client.json_request, 'DELETE', '')
mock_request.assert_called_once_with(
'DELETE', 'http://example.com:8004/foo', 'DELETE', 'http://example.com:8004/foo',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/json', headers={'Content-Type': 'application/json',
'Accept': 'application/json', 'Accept': 'application/json',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse(
302, 'Found',
{},
''))
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004/foo')
self.assertRaises(exc.InvalidEndpoint,
client.json_request, 'DELETE', '')
def test_http_json_request_redirect(self): def test_http_json_request_redirect(self, mock_request):
# Record the 302 # Record the 302
mock_conn = http.requests.request( mock_request.side_effect = [
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
302, 'Found', 302, 'Found',
{'location': 'http://example.com:8004'}, {'location': 'http://example.com:8004'},
'')) ''),
# Record the following 200
mock_conn = http.requests.request(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( fakes.FakeHTTPResponse(
200, 'OK', 200, 'OK',
{'content-type': 'application/json'}, {'content-type': 'application/json'},
'{}')) '{}')
# Replay, create client, assert ]
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
resp, body = client.json_request('GET', '') resp, body = client.json_request('GET', '')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
self.assertEqual({}, body) self.assertEqual({}, body)
mock_request.assert_has_calls([
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'}),
mock.call('GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
])
def test_http_404_json_request(self): def test_http_404_json_request(self, mock_request):
# Record a 404 # Record a 404
mock_conn = http.requests.request( mock_request.return_value = fakes.FakeHTTPResponse(
'GET', 'http://example.com:8004', 404, 'OK', {'content-type': 'application/json'},
allow_redirects=False, '{}')
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse(
404, 'OK', {'content-type': 'application/json'},
'{}'))
# Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
e = self.assertRaises(exc.HTTPNotFound, client.json_request, 'GET', '') e = self.assertRaises(exc.HTTPNotFound, client.json_request, 'GET', '')
# Assert that the raised exception can be converted to string # Assert that the raised exception can be converted to string
self.assertIsNotNone(str(e)) self.assertIsNotNone(str(e))
mock_request.assert_called_with(
def test_http_300_json_request(self):
# Record a 300
mock_conn = http.requests.request(
'GET', 'http://example.com:8004', 'GET', 'http://example.com:8004',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/json', headers={'Content-Type': 'application/json',
'Accept': 'application/json', 'Accept': 'application/json',
'User-Agent': 'python-heatclient'}) 'User-Agent': 'python-heatclient'})
mock_conn.AndReturn(
fakes.FakeHTTPResponse( def test_http_300_json_request(self, mock_request):
300, 'OK', {'content-type': 'application/json'}, # Record a 300
'{}')) mock_request.return_value = fakes.FakeHTTPResponse(
# Replay, create client, assert 300, 'OK', {'content-type': 'application/json'},
self.m.ReplayAll() '{}')
# Assert that the raised exception can be converted to string
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
e = self.assertRaises( e = self.assertRaises(
exc.HTTPMultipleChoices, client.json_request, 'GET', '') exc.HTTPMultipleChoices, client.json_request, 'GET', '')
# Assert that the raised exception can be converted to string
self.assertIsNotNone(str(e)) self.assertIsNotNone(str(e))
mock_request.assert_called_with(
'GET', 'http://example.com:8004',
allow_redirects=False,
headers={'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'python-heatclient'})
def test_fake_json_request(self): def test_fake_json_request(self, mock_request):
headers = {'User-Agent': 'python-heatclient'} headers = {'User-Agent': 'python-heatclient'}
mock_conn = http.requests.request('GET', 'fake://example.com:8004/', mock_request.side_effect = [socket.gaierror]
allow_redirects=False,
headers=headers)
mock_conn.AndRaise(socket.gaierror)
self.m.ReplayAll()
client = http.HTTPClient('fake://example.com:8004') client = http.HTTPClient('fake://example.com:8004')
self.assertRaises(exc.InvalidEndpoint, self.assertRaises(exc.InvalidEndpoint,
client._http_request, "/", "GET") client._http_request, "/", "GET")
def test_debug_curl_command(self): mock_request.assert_called_with('GET', 'fake://example.com:8004/',
self.m.StubOutWithMock(logging.Logger, 'debug') allow_redirects=False,
headers=headers)
ssl_connection_params = {'ca_file': 'TEST_CA', def test_debug_curl_command(self, mock_request):
'cert_file': 'TEST_CERT', with mock.patch('logging.Logger.debug') as mock_logging_debug:
'key_file': 'TEST_KEY',
'insecure': 'TEST_NSA'}
headers = {'key': 'value'} ssl_connection_params = {'ca_file': 'TEST_CA',
'cert_file': 'TEST_CERT',
'key_file': 'TEST_KEY',
'insecure': 'TEST_NSA'}
mock_logging_debug = logging.Logger.debug( headers = {'key': 'value'}
"curl -g -i -X GET -H 'key: value' --key TEST_KEY " mock_logging_debug.return_value = None
"--cert TEST_CERT --cacert TEST_CA " client = http.HTTPClient('http://foo')
"-k -d 'text' http://foo/bar" client.ssl_connection_params = ssl_connection_params
) client.log_curl_request('GET', '/bar', {'headers': headers,
mock_logging_debug.AndReturn(None) 'data': 'text'})
self.m.ReplayAll() mock_logging_debug.assert_called_with(
"curl -g -i -X GET -H 'key: value' --key TEST_KEY "
"--cert TEST_CERT --cacert TEST_CA "
"-k -d 'text' http://foo/bar"
)
client = http.HTTPClient('http://foo') def test_http_request_socket_error(self, mock_request):
client.ssl_connection_params = ssl_connection_params
client.log_curl_request('GET', '/bar', {'headers': headers,
'data': 'text'})
def test_http_request_socket_error(self):
headers = {'User-Agent': 'python-heatclient'} headers = {'User-Agent': 'python-heatclient'}
mock_conn = http.requests.request('GET', 'http://example.com:8004/', mock_request.side_effect = [socket.error]
allow_redirects=False,
headers=headers)
mock_conn.AndRaise(socket.error)
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
self.assertRaises(exc.CommunicationError, self.assertRaises(exc.CommunicationError,
client._http_request, "/", "GET") client._http_request, "/", "GET")
mock_request.assert_called_with('GET', 'http://example.com:8004/',
allow_redirects=False,
headers=headers)
def test_http_request_socket_timeout(self): def test_http_request_socket_timeout(self, mock_request):
headers = {'User-Agent': 'python-heatclient'} headers = {'User-Agent': 'python-heatclient'}
mock_conn = http.requests.request('GET', 'http://example.com:8004/', mock_request.side_effect = [socket.timeout]
allow_redirects=False,
headers=headers)
mock_conn.AndRaise(socket.timeout)
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004') client = http.HTTPClient('http://example.com:8004')
self.assertRaises(exc.CommunicationError, self.assertRaises(exc.CommunicationError,
client._http_request, "/", "GET") client._http_request, "/", "GET")
mock_request.assert_called_with('GET', 'http://example.com:8004/',
allow_redirects=False,
headers=headers)
def test_http_request_specify_timeout(self): def test_http_request_specify_timeout(self, mock_request):
mock_conn = http.requests.request( mock_request.return_value = fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}')
client = http.HTTPClient('http://example.com:8004', timeout='123')
resp, body = client.json_request('GET', '')
self.assertEqual(200, resp.status_code)
mock_request.assert_called_with(
'GET', 'http://example.com:8004', 'GET', 'http://example.com:8004',
allow_redirects=False, allow_redirects=False,
headers={'Content-Type': 'application/json', headers={'Content-Type': 'application/json',
'Accept': 'application/json', 'Accept': 'application/json',
'User-Agent': 'python-heatclient'}, 'User-Agent': 'python-heatclient'},
timeout=float(123)) timeout=float(123))
mock_conn.AndReturn(
fakes.FakeHTTPResponse(
200, 'OK',
{'content-type': 'application/json'},
'{}'))
# Replay, create client, assert
self.m.ReplayAll()
client = http.HTTPClient('http://example.com:8004', timeout='123')
resp, body = client.json_request('GET', '')
self.assertEqual(200, resp.status_code)
self.assertEqual({}, body)
def test_get_system_ca_file(self): def test_get_system_ca_file(self, mock_request):
chosen = '/etc/ssl/certs/ca-certificates.crt' chosen = '/etc/ssl/certs/ca-certificates.crt'
self.m.StubOutWithMock(os.path, 'exists') with mock.patch('os.path.exists') as mock_os:
os.path.exists(chosen).AndReturn(chosen) mock_os.return_value = chosen
self.m.ReplayAll() ca = http.get_system_ca_file()
self.assertEqual(chosen, ca)
mock_os.assert_called_once_with(chosen)
ca = http.get_system_ca_file() def test_insecure_verify_cert_None(self, mock_request):
self.assertEqual(chosen, ca)
def test_insecure_verify_cert_None(self):
client = http.HTTPClient('https://foo', insecure=True) client = http.HTTPClient('https://foo', insecure=True)
self.assertFalse(client.verify_cert) self.assertFalse(client.verify_cert)
def test_passed_cert_to_verify_cert(self): def test_passed_cert_to_verify_cert(self, mock_request):
client = http.HTTPClient('https://foo', ca_file="NOWHERE") client = http.HTTPClient('https://foo', ca_file="NOWHERE")
self.assertEqual("NOWHERE", client.verify_cert) self.assertEqual("NOWHERE", client.verify_cert)
self.m.StubOutWithMock(http, 'get_system_ca_file') with mock.patch('heatclient.common.http.get_system_ca_file') as gsf:
http.get_system_ca_file().AndReturn("SOMEWHERE") gsf.return_value = "SOMEWHERE"
self.m.ReplayAll() client = http.HTTPClient('https://foo')
client = http.HTTPClient('https://foo') self.assertEqual("SOMEWHERE", client.verify_cert)
self.assertEqual("SOMEWHERE", client.verify_cert)
def test_curl_log_i18n_headers(self): @mock.patch('logging.Logger.debug', return_value=None)
self.m.StubOutWithMock(logging.Logger, 'debug') def test_curl_log_i18n_headers(self, mock_log, mock_request):
kwargs = {'headers': {'Key': b'foo\xe3\x8a\x8e'}} kwargs = {'headers': {'Key': b'foo\xe3\x8a\x8e'}}
mock_logging_debug = logging.Logger.debug(
u"curl -g -i -X GET -H 'Key: foo㊎' http://somewhere"
)
mock_logging_debug.AndReturn(None)
self.m.ReplayAll()
client = http.HTTPClient('http://somewhere') client = http.HTTPClient('http://somewhere')
client.log_curl_request("GET", '', kwargs=kwargs) client.log_curl_request("GET", '', kwargs=kwargs)
mock_log.assert_called_once_with(
u"curl -g -i -X GET -H 'Key: foo㊎' http://somewhere")
class SessionClientTest(testtools.TestCase): class SessionClientTest(testtools.TestCase):