2c5dcf9bf1
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
114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
# 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 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"}',
|
|
})
|
|
|
|
REQUEST_URL = 'https://127.0.0.1:5000/hi'
|
|
RESPONSE_BODY = '{"hi": "there"}'
|
|
|
|
|
|
def get_client():
|
|
cl = httpclient.HTTPClient(username="username", password="password",
|
|
tenant_id="tenant", auth_url="auth_test",
|
|
cacert="ca.pem", key="key.pem", cert="cert.pem")
|
|
return cl
|
|
|
|
|
|
def get_authed_client():
|
|
cl = get_client()
|
|
cl.management_url = "https://127.0.0.1:5000"
|
|
cl.auth_token = "token"
|
|
return cl
|
|
|
|
|
|
class ClientTest(utils.TestCase):
|
|
|
|
def setUp(self):
|
|
super(ClientTest, self).setUp()
|
|
self.request_patcher = mock.patch.object(requests, 'request',
|
|
self.mox.CreateMockAnything())
|
|
self.request_patcher.start()
|
|
|
|
def tearDown(self):
|
|
self.request_patcher.stop()
|
|
super(ClientTest, self).tearDown()
|
|
|
|
@mock.patch.object(session.requests.Session, 'request')
|
|
def test_get(self, MOCK_REQUEST):
|
|
MOCK_REQUEST.return_value = FAKE_RESPONSE
|
|
cl = get_authed_client()
|
|
|
|
resp, body = cl.get("/hi")
|
|
|
|
# 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')
|
|
|
|
# Automatic JSON parsing
|
|
self.assertEqual(body, {"hi": "there"})
|
|
|
|
@mock.patch.object(session.requests.Session, 'request')
|
|
def test_post(self, MOCK_REQUEST):
|
|
MOCK_REQUEST.return_value = FAKE_RESPONSE
|
|
cl = get_authed_client()
|
|
|
|
cl.post("/hi", body=[1, 2, 3])
|
|
|
|
# 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')
|
|
|
|
@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
|
|
|
|
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')
|