openstack/deb-python-keystonemiddleware
Code Issues Proposed changes

Extract a base Session object

Browse Source 
A wrapper around a number of connection variables. This will be extended
later with principals such as Kerberos authentication and http sessions.

The intent is that this session object will become the basis for all
other client library communications in OpenStack (as keystone wants to
control things like authentication for everybody).

Change-Id: I8ee728c49d554659d7057ebf17d0f8ceea4d7d8e
Part of: blueprint auth-plugins
This commit is contained in:
Jamie Lennox
2013-09-30 10:35:55 +10:00
parent 2a09adcac9
commit d17fc7c101
3 changed files with 187 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
test_https.py
View File

@@ -17,13 +17,13 @@ import mock
import requests
from keystoneclient import httpclient
from keystoneclient import session
from keystoneclient.tests import utils
FAKE_RESPONSE = utils.TestResponse({
"status_code": 200,
"text": '{"hi": "there"}',
})
MOCK_REQUEST = mock.Mock(return_value=(FAKE_RESPONSE))
REQUEST_URL = 'https://127.0.0.1:5000/hi'
RESPONSE_BODY = '{"hi": "there"}'
@@ -55,56 +55,59 @@ class ClientTest(utils.TestCase):
self.request_patcher.stop()
super(ClientTest, self).tearDown()
def test_get(self):
@mock.patch.object(session.requests.Session, 'request')
def test_get(self, MOCK_REQUEST):
MOCK_REQUEST.return_value = FAKE_RESPONSE
cl = get_authed_client()
with mock.patch.object(requests, "request", MOCK_REQUEST):
resp, body = cl.get("/hi")
resp, body = cl.get("/hi")
# this may become too tightly couple later
mock_args, mock_kwargs = MOCK_REQUEST.call_args
# this may become too tightly couple later
mock_args, mock_kwargs = MOCK_REQUEST.call_args
self.assertEqual(mock_args[0], 'GET')
self.assertEqual(mock_args[1], REQUEST_URL)
self.assertEqual(mock_kwargs['headers']['X-Auth-Token'], 'token')
self.assertEqual(mock_kwargs['cert'], ('cert.pem', 'key.pem'))
self.assertEqual(mock_kwargs['verify'], 'ca.pem')
self.assertEqual(mock_args[0], 'GET')
self.assertEqual(mock_args[1], REQUEST_URL)
self.assertEqual(mock_kwargs['headers']['X-Auth-Token'], 'token')
self.assertEqual(mock_kwargs['cert'], ('cert.pem', 'key.pem'))
self.assertEqual(mock_kwargs['verify'], 'ca.pem')
# Automatic JSON parsing
self.assertEqual(body, {"hi": "there"})
# Automatic JSON parsing
self.assertEqual(body, {"hi": "there"})
def test_post(self):
@mock.patch.object(session.requests.Session, 'request')
def test_post(self, MOCK_REQUEST):
MOCK_REQUEST.return_value = FAKE_RESPONSE
cl = get_authed_client()
with mock.patch.object(requests, "request", MOCK_REQUEST):
cl.post("/hi", body=[1, 2, 3])
cl.post("/hi", body=[1, 2, 3])
# this may become too tightly couple later
mock_args, mock_kwargs = MOCK_REQUEST.call_args
# this may become too tightly couple later
mock_args, mock_kwargs = MOCK_REQUEST.call_args
self.assertEqual(mock_args[0], 'POST')
self.assertEqual(mock_args[1], REQUEST_URL)
self.assertEqual(mock_kwargs['data'], '[1, 2, 3]')
self.assertEqual(mock_kwargs['headers']['X-Auth-Token'], 'token')
self.assertEqual(mock_kwargs['cert'], ('cert.pem', 'key.pem'))
self.assertEqual(mock_kwargs['verify'], 'ca.pem')
self.assertEqual(mock_args[0], 'POST')
self.assertEqual(mock_args[1], REQUEST_URL)
self.assertEqual(mock_kwargs['data'], '[1, 2, 3]')
self.assertEqual(mock_kwargs['headers']['X-Auth-Token'], 'token')
self.assertEqual(mock_kwargs['cert'], ('cert.pem', 'key.pem'))
self.assertEqual(mock_kwargs['verify'], 'ca.pem')
def test_post_auth(self):
with mock.patch.object(requests, "request", MOCK_REQUEST):
cl = httpclient.HTTPClient(
username="username", password="password", tenant_id="tenant",
auth_url="auth_test", cacert="ca.pem", key="key.pem",
cert="cert.pem")
cl.management_url = "https://127.0.0.1:5000"
cl.auth_token = "token"
cl.post("/hi", body=[1, 2, 3])
@mock.patch.object(session.requests.Session, 'request')
def test_post_auth(self, MOCK_REQUEST):
MOCK_REQUEST.return_value = FAKE_RESPONSE
cl = httpclient.HTTPClient(
username="username", password="password", tenant_id="tenant",
auth_url="auth_test", cacert="ca.pem", key="key.pem",
cert="cert.pem")
cl.management_url = "https://127.0.0.1:5000"
cl.auth_token = "token"
cl.post("/hi", body=[1, 2, 3])
# this may become too tightly couple later
mock_args, mock_kwargs = MOCK_REQUEST.call_args
# this may become too tightly couple later
mock_args, mock_kwargs = MOCK_REQUEST.call_args
self.assertEqual(mock_args[0], 'POST')
self.assertEqual(mock_args[1], REQUEST_URL)
self.assertEqual(mock_kwargs['data'], '[1, 2, 3]')
self.assertEqual(mock_kwargs['headers']['X-Auth-Token'], 'token')
self.assertEqual(mock_kwargs['cert'], ('cert.pem', 'key.pem'))
self.assertEqual(mock_kwargs['verify'], 'ca.pem')
self.assertEqual(mock_args[0], 'POST')
self.assertEqual(mock_args[1], REQUEST_URL)
self.assertEqual(mock_kwargs['data'], '[1, 2, 3]')
self.assertEqual(mock_kwargs['headers']['X-Auth-Token'], 'token')
self.assertEqual(mock_kwargs['cert'], ('cert.pem', 'key.pem'))
self.assertEqual(mock_kwargs['verify'], 'ca.pem')

140
test_session.py Normal file
View File

@@ -0,0 +1,140 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Licensed 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 httpretty
import mock
from keystoneclient import exceptions
from keystoneclient import session as client_session
from keystoneclient.tests import utils
class SessionTests(utils.TestCase):
TEST_URL = 'http://127.0.0.1:5000/'
@httpretty.activate
def test_get(self):
session = client_session.Session()
self.stub_url(httpretty.GET, body='response')
resp = session.get(self.TEST_URL)
self.assertEqual(httpretty.GET, httpretty.last_request().method)
self.assertEqual(resp.text, 'response')
self.assertTrue(resp.ok)
@httpretty.activate
def test_post(self):
session = client_session.Session()
self.stub_url(httpretty.POST, body='response')
resp = session.post(self.TEST_URL, json={'hello': 'world'})
self.assertEqual(httpretty.POST, httpretty.last_request().method)
self.assertEqual(resp.text, 'response')
self.assertTrue(resp.ok)
self.assertRequestBodyIs(json={'hello': 'world'})
@httpretty.activate
def test_head(self):
session = client_session.Session()
self.stub_url(httpretty.HEAD)
resp = session.head(self.TEST_URL)
self.assertEqual(httpretty.HEAD, httpretty.last_request().method)
self.assertTrue(resp.ok)
self.assertRequestBodyIs('')
@httpretty.activate
def test_put(self):
session = client_session.Session()
self.stub_url(httpretty.PUT, body='response')
resp = session.put(self.TEST_URL, json={'hello': 'world'})
self.assertEqual(httpretty.PUT, httpretty.last_request().method)
self.assertEqual(resp.text, 'response')
self.assertTrue(resp.ok)
self.assertRequestBodyIs(json={'hello': 'world'})
@httpretty.activate
def test_delete(self):
session = client_session.Session()
self.stub_url(httpretty.DELETE, body='response')
resp = session.delete(self.TEST_URL)
self.assertEqual(httpretty.DELETE, httpretty.last_request().method)
self.assertTrue(resp.ok)
self.assertEqual(resp.text, 'response')
@httpretty.activate
def test_patch(self):
session = client_session.Session()
self.stub_url(httpretty.PATCH, body='response')
resp = session.patch(self.TEST_URL, json={'hello': 'world'})
self.assertEqual(httpretty.PATCH, httpretty.last_request().method)
self.assertTrue(resp.ok)
self.assertEqual(resp.text, 'response')
self.assertRequestBodyIs(json={'hello': 'world'})
@httpretty.activate
def test_user_agent(self):
session = client_session.Session(user_agent='test-agent')
self.stub_url(httpretty.GET, body='response')
resp = session.get(self.TEST_URL)
self.assertTrue(resp.ok)
self.assertRequestHeaderEqual('User-Agent', 'test-agent')
resp = session.get(self.TEST_URL, headers={'User-Agent': 'new-agent'})
self.assertTrue(resp.ok)
self.assertRequestHeaderEqual('User-Agent', 'new-agent')
resp = session.get(self.TEST_URL, headers={'User-Agent': 'new-agent'},
user_agent='overrides-agent')
self.assertTrue(resp.ok)
self.assertRequestHeaderEqual('User-Agent', 'overrides-agent')
@httpretty.activate
def test_http_session_opts(self):
session = client_session.Session(cert='cert.pem', timeout=5,
verify='certs')
FAKE_RESP = utils.TestResponse({'status_code': 200, 'text': 'resp'})
RESP = mock.Mock(return_value=FAKE_RESP)
with mock.patch.object(session.session, 'request', RESP) as mocked:
session.post(self.TEST_URL, data='value')
mock_args, mock_kwargs = mocked.call_args
self.assertEqual(mock_args[0], 'POST')
self.assertEqual(mock_args[1], self.TEST_URL)
self.assertEqual(mock_kwargs['data'], 'value')
self.assertEqual(mock_kwargs['cert'], 'cert.pem')
self.assertEqual(mock_kwargs['verify'], 'certs')
self.assertEqual(mock_kwargs['timeout'], 5)
@httpretty.activate
def test_not_found(self):
session = client_session.Session()
self.stub_url(httpretty.GET, status=404)
self.assertRaises(exceptions.NotFound, session.get, self.TEST_URL)
@httpretty.activate
def test_server_error(self):
session = client_session.Session()
self.stub_url(httpretty.GET, status=500)
self.assertRaises(exceptions.InternalServerError,
session.get, self.TEST_URL)

4
test_shell.py
View File

@@ -25,6 +25,7 @@ import testtools
from testtools import matchers
from keystoneclient import exceptions
from keystoneclient import session
from keystoneclient import shell as openstack_shell
from keystoneclient.tests import utils
from keystoneclient.v2_0 import shell as shell_v2_0
@@ -423,7 +424,8 @@ class ShellTest(utils.TestCase):
'endpoints': [],
})
request_mock = mock.MagicMock(return_value=response_mock)
with mock.patch('requests.request', request_mock):
with mock.patch.object(session.requests.Session, 'request',
request_mock):
shell(('--timeout 2 --os-token=blah --os-endpoint=blah'
' --os-auth-url=blah.com endpoint-list'))
request_mock.assert_called_with(mock.ANY, mock.ANY,