a5d9ca5ec6
Update the unit tests to reflect all changes made rename the command line tool to "keep" and add it to setup.py Add nosetest to tox
344 lines
16 KiB
Python
344 lines
16 KiB
Python
# Copyright (c) 2013 Rackspace, Inc.
|
|
#
|
|
# 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 json
|
|
import unittest2 as unittest
|
|
|
|
from mock import MagicMock
|
|
|
|
from barbicanclient import client
|
|
from barbicanclient.common.exceptions import ClientException
|
|
|
|
|
|
def suite():
|
|
suite = unittest.TestSuite()
|
|
|
|
suite.addTest(WhenTestingConnection())
|
|
|
|
return suite
|
|
|
|
|
|
class WhenTestingConnection(unittest.TestCase):
|
|
def setUp(self):
|
|
self.auth_endpoint = 'https://keystone.com/v2'
|
|
self.user = 'user'
|
|
self.key = 'key'
|
|
self.tenant = 'tenant'
|
|
self.endpoint = 'http://localhost:9311/v1/'
|
|
self.auth_token = 'token'
|
|
self.href = 'http://localhost:9311/v1/12345/orders'
|
|
|
|
self.authenticate = MagicMock()
|
|
self.authenticate.return_value = (self.endpoint, self.auth_token)
|
|
self.request = MagicMock()
|
|
self.request.return_value.content = json.dumps(
|
|
{
|
|
"secret_ref": "http://localhost:9311/None/secrets"
|
|
"/8502cea9-9d35-46d7-96f5-80e43905e4c5"
|
|
}
|
|
)
|
|
self.request.return_value.headers = {
|
|
'content-length': '92',
|
|
'content-type': 'application/json; charset=utf-8',
|
|
'location': 'http://localhost:9311/None/'
|
|
'secrets/8502cea9-9d35-46d7-96f5-80e43905e4c5',
|
|
'x-openstack-request-id':
|
|
'req-6c19d09e-1167-445c-b435-d6b0818b59b9'
|
|
}
|
|
self.request.return_value.ok = True
|
|
self.connection = client.Connection(self.auth_endpoint, self.user,
|
|
self.key, self.tenant,
|
|
token=self.auth_token,
|
|
authenticate=self.authenticate,
|
|
request=self.request)
|
|
|
|
def test_should_connect_with_token(self):
|
|
self.assertFalse(self.authenticate.called)
|
|
|
|
def test_should_connect_without_token(self):
|
|
self.connection = client.Connection(self.auth_endpoint,
|
|
self.user,
|
|
self.key,
|
|
self.tenant,
|
|
authenticate=self.authenticate,
|
|
endpoint=self.endpoint)
|
|
self.authenticate\
|
|
.assert_called_once_with(self.auth_endpoint,
|
|
self.user,
|
|
self.key,
|
|
self.tenant,
|
|
endpoint=self.endpoint,
|
|
cacert=None
|
|
)
|
|
self.assertEqual(self.auth_token, self.connection.auth_token)
|
|
self.assertEqual(self.auth_endpoint, self.connection._auth_endpoint)
|
|
self.assertEqual(self.user, self.connection._user)
|
|
self.assertEqual(self.key, self.connection._key)
|
|
self.assertEqual(self.tenant, self.connection._tenant)
|
|
self.assertEqual(self.endpoint, self.connection._endpoint)
|
|
|
|
def test_should_create_secret(self):
|
|
body = {'status': 'ACTIVE',
|
|
'content_types': {'default': 'text/plain'},
|
|
'updated': '2013-06-07T16:13:38.889857',
|
|
'cypher_type': 'CDC',
|
|
'name': 'test_secret',
|
|
'algorithm': 'aes',
|
|
'created': '2013-06-07T16:13:38.889851',
|
|
'secret_ref': 'http://localhost:9311/v1/None/secrets/e6e7d'
|
|
'b5e-3738-408e-aaba-05a7177cade5',
|
|
'expiration': '2015-06-07T16:13:38.889851',
|
|
'bit_length': 256,
|
|
'mime_type': 'text/plain'
|
|
}
|
|
|
|
secret = client.Secret(self.connection, body)
|
|
self.request.return_value.content = json.dumps(body)
|
|
created = self.connection.create_secret('text/plain',
|
|
'Test secret',
|
|
name='test_secret',
|
|
algorithm=None,
|
|
bit_length=None,
|
|
cypher_type=None,
|
|
expiration=None)
|
|
self.assertTrue(self._are_equivalent(secret, created))
|
|
|
|
def test_should_create_order(self):
|
|
body = {"status": "ACTIVE",
|
|
"secret_ref": "http://localhost:9311/v1/12345/secrets/5706054"
|
|
"9-2fcf-46eb-92bb-bf49fcf5d089",
|
|
"updated": "2013-06-07T19:00:37.338386",
|
|
"created": "2013-06-07T19:00:37.298704",
|
|
"secret": {
|
|
"cypher_type": "CDC",
|
|
"name": "test_secret",
|
|
"algorithm": "aes",
|
|
"expiration": "2015-06-07T19:00:37.298704",
|
|
"bit_length": 256,
|
|
"mime_type": "text/plain"
|
|
},
|
|
"order_ref": "http://localhost:9311/v1/12345/orders/003f2b91-"
|
|
"2f53-4c0a-a0f3-33796671efc3"
|
|
}
|
|
|
|
order = client.Order(self.connection, body)
|
|
self.request.return_value.content = json.dumps(body)
|
|
created = self.connection.create_order('text/plain',
|
|
name='test_secret',
|
|
bit_length=256,
|
|
algorithm='aes',
|
|
cypher_type='CDC')
|
|
self.assertTrue(self._are_equivalent(order, created))
|
|
|
|
def test_list_no_secrets(self):
|
|
body0 = {'secrets': []}
|
|
secrets = []
|
|
self.request.return_value.content = json.dumps(body0)
|
|
secret_list, prev_ref, next_ref = self.connection.list_secrets(0, 0)
|
|
self.assertTrue(self._are_equivalent(secrets, secret_list))
|
|
self.assertIsNone(prev_ref)
|
|
self.assertIsNone(next_ref)
|
|
|
|
def test_list_single_secret(self):
|
|
limit = 1
|
|
body1 = {'secrets': [{'status': 'ACTIVE',
|
|
'content_types': {'default': 'text/plain'},
|
|
'updated': '2013-06-03T21:16:58.349230',
|
|
'cypher_type': None,
|
|
'name': 'test_1',
|
|
'algorithm': None,
|
|
'created': '2013-06-03T21:16:58.349222',
|
|
'secret_ref': 'http://localhost:9311/v1/'
|
|
'None/secrets/bbd2036f-730'
|
|
'7-4090-bbef-bbb6025e5e7b',
|
|
'expiration': None,
|
|
'bit_length': None,
|
|
'mime_type': 'text/plain'}],
|
|
'next': "{0}/{1}?limit={2}&offset={2}".format(self.connection.
|
|
_tenant,
|
|
self.connection.
|
|
SECRETS_PATH,
|
|
limit)}
|
|
secrets = [client.Secret(self.connection, body1['secrets'][0])]
|
|
self.request.return_value.content = json.dumps(body1)
|
|
secret_list, prev_ref, next_ref = self.connection.list_secrets(limit,
|
|
0)
|
|
self.assertTrue(self._are_equivalent(secrets, secret_list))
|
|
self.assertIsNone(prev_ref)
|
|
self.assertEqual(body1['next'], next_ref)
|
|
|
|
def test_list_multiple_secrets(self):
|
|
limit = 2
|
|
body1 = {'secrets': [{'status': 'ACTIVE',
|
|
'content_types': {'default': 'text/plain'},
|
|
'updated': '2013-06-03T21:16:58.349230',
|
|
'cypher_type': None,
|
|
'name': 'test_1',
|
|
'algorithm': None,
|
|
'created': '2013-06-03T21:16:58.349222',
|
|
'secret_ref': 'http://localhost:9311/v1/'
|
|
'None/secrets/bbd2036f-730'
|
|
'7-4090-bbef-bbb6025e5e7b',
|
|
'expiration': None,
|
|
'bit_length': None,
|
|
'mime_type': 'text/plain'}],
|
|
'previous': "{0}/{1}?limit={2}&offset={2}".format(
|
|
self.connection._tenant,
|
|
self.connection.
|
|
SECRETS_PATH,
|
|
limit)}
|
|
|
|
body2 = body1
|
|
body2['secrets'][0]['name'] = 'test_2'
|
|
body2['secrets'][0]['secret_ref'] = 'http://localhost:9311/v1/No'\
|
|
+ 'ne/secrets/bbd2036f-7307-'\
|
|
+ '4090-bbef-bbb6025eabcd'
|
|
body2['previous'] = 'http://localhost:9311/v1/None/secrets/19106'\
|
|
+ 'b6e-4ef1-48d1-8950-170c1a5838e1'
|
|
body2['next'] = None
|
|
|
|
secrets = [client.Secret(self.connection, b['secrets'][0])
|
|
for b in (body1, body2)]
|
|
body2['secrets'].insert(0, body1['secrets'][0])
|
|
self.request.return_value.content = json.dumps(body2)
|
|
secret_list, prev_ref, next_ref = self.connection.list_secrets(limit,
|
|
1)
|
|
self.assertTrue(self._are_equivalent(secrets, secret_list))
|
|
self.assertEqual(body2['previous'], prev_ref)
|
|
self.assertIsNone(next_ref)
|
|
|
|
def test_list_no_orders(self):
|
|
body0 = {'orders': []}
|
|
orders = []
|
|
self.request.return_value.content = json.dumps(body0)
|
|
order_list, prev_ref, next_ref = self.connection.list_orders(0, 0)
|
|
self.assertTrue(self._are_equivalent(orders, order_list))
|
|
self.assertIsNone(prev_ref)
|
|
self.assertIsNone(next_ref)
|
|
|
|
def test_list_single_order(self):
|
|
limit = 1
|
|
body1 = {'orders': [{'status': 'PENDING',
|
|
'updated': '2013-06-05T15:15:30.904760',
|
|
'created': '2013-06-05T15:15:30.904752',
|
|
'order_ref': 'http://localhost:9311/v1/'
|
|
'None/orders/9f651441-3ccd'
|
|
'-45b3-bc60-3051656d5168',
|
|
'secret_ref': 'http://localhost:9311/'
|
|
'v1/None/secrets/????',
|
|
'secret': {'cypher_type': None,
|
|
'name': 'test_1',
|
|
'algorithm': None,
|
|
'expiration': None,
|
|
'bit_length': None,
|
|
'mime_type': 'text/plain'}}],
|
|
'next': "{0}/{1}?limit={2}&offset={2}".format(self.connection.
|
|
_tenant,
|
|
self.connection.
|
|
ORDERS_PATH,
|
|
limit)}
|
|
orders = [client.Order(self.connection, body1['orders'][0])]
|
|
self.request.return_value.content = json.dumps(body1)
|
|
order_list, prev_ref, next_ref = self.connection.list_orders(limit, 0)
|
|
self.assertTrue(self._are_equivalent(orders, order_list))
|
|
self.assertIsNone(prev_ref)
|
|
self.assertEqual(body1['next'], next_ref)
|
|
|
|
def test_list_multiple_orders(self):
|
|
limit = 2
|
|
body1 = {'orders': [{'status': 'PENDING',
|
|
'updated': '2013-06-05T15:15:30.904760',
|
|
'created': '2013-06-05T15:15:30.904752',
|
|
'order_ref': 'http://localhost:9311/v1/'
|
|
'None/orders/9f651441-3ccd'
|
|
'-45b3-bc60-3051656d5168',
|
|
'secret_ref': 'http://localhost:9311/'
|
|
'v1/None/secrets/????',
|
|
'secret': {'cypher_type': None,
|
|
'name': 'test_1',
|
|
'algorithm': None,
|
|
'expiration': None,
|
|
'bit_length': None,
|
|
'mime_type': 'text/plain'}}],
|
|
'previous': "{0}/{1}?limit={2}&offset={2}".format(
|
|
self.connection._tenant,
|
|
self.connection.
|
|
SECRETS_PATH,
|
|
limit)}
|
|
body2 = body1
|
|
body2['orders'][0]['order_ref'] = 'http://localhost:9311/v1/No'\
|
|
+ 'ne/orders/9f651441-3ccd-4'\
|
|
+ '5b3-bc60-3051656382fj'
|
|
body2['orders'][0]['secret']['name'] = 'test_2'
|
|
|
|
body2['orders'][0]['name'] = 'test_2'
|
|
body2['orders'][0]['secret_ref'] = 'http://localhost:9311/v1/No'\
|
|
+ 'ne/secrets/bbd2036f-7307-'\
|
|
+ '4090-bbef-bbb6025eabcd'
|
|
body2['previous'] = 'http://localhost:9311/v1/None/orders/19106'\
|
|
+ 'b6e-4ef1-48d1-8950-170c1a5838e1'
|
|
body2['next'] = None
|
|
|
|
orders = [client.Order(self.connection, b['orders'][0])
|
|
for b in (body1, body2)]
|
|
body2['orders'].insert(0, body1['orders'][0])
|
|
self.request.return_value.content = json.dumps(body2)
|
|
order_list, prev_ref, next_ref = self.connection.list_orders(limit, 1)
|
|
self.assertTrue(self._are_equivalent(orders, order_list))
|
|
self.assertEqual(body2['previous'], prev_ref)
|
|
self.assertIsNone(next_ref)
|
|
|
|
def test_should_get_response(self):
|
|
self._setup_request()
|
|
headers, body = self.connection._perform_http('GET', self.href)
|
|
self.assertEqual(self.request.return_value.headers, headers)
|
|
self.assertEqual(json.loads(self.request.return_value.content), body)
|
|
|
|
def test_should_parse_json(self):
|
|
self._setup_request()
|
|
headers, body = self.connection._perform_http('GET', self.href,
|
|
parse_json=True)
|
|
self.assertEqual(json.loads(self.request.return_value.content), body)
|
|
|
|
def test_should_not_parse_json(self):
|
|
self._setup_request()
|
|
headers, body = self.connection._perform_http('GET', self.href,
|
|
parse_json=False)
|
|
self.assertEqual(self.request.return_value.content, body)
|
|
|
|
def test_should_raise_exception(self):
|
|
self._setup_request()
|
|
self.request.return_value.ok = False
|
|
self.request.return_value.status_code = 404
|
|
with self.assertRaises(ClientException) as e:
|
|
self.connection._perform_http('GET', self.href)
|
|
exception = e.exception
|
|
self.assertEqual(404, exception.http_status)
|
|
|
|
def _setup_request(self):
|
|
self.request.return_value.headers = {'Accept': 'application/json'}
|
|
self.request.return_value.content = '{"test": "response"}'
|
|
self.href = 'http://localhost:9311/v1/12345/orders'
|
|
|
|
def _are_equivalent(self, a, b):
|
|
if isinstance(a, list) and isinstance(b, list):
|
|
return all([self._are_equivalent(x, y) for x, y in zip(a, b)])
|
|
else:
|
|
return (a.__dict__ == b.__dict__)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|