openstack/python-barbicanclient
Code Issues Proposed changes

Split several unit tests into separate tests

Browse Source 
Make some style changes

Add a __ne__() method for Secrets and Orders
This commit is contained in:
Arash Ghoreyshi
2013-06-11 16:04:04 -05:00
parent e0101cbc7f
commit 90d297c67f
4 changed files with 97 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
barbicanclient/client.py
View File

@@ -8,7 +8,7 @@ from barbicanclient.common import config
from barbicanclient.secrets import Secret
from barbicanclient.orders import Order
from barbicanclient.common import auth
from barbicanclient.openstack.common import log, timeutils
from barbicanclient.openstack.common import log
from barbicanclient.common.exceptions import ClientException
from barbicanclient.openstack.common.gettextutils import _
from openstack.common.timeutils import parse_isotime
@@ -111,7 +111,7 @@ class Connection(object):
"""
LOG.debug(_("Listing secrets"))
href = "{0}/{1}?limit=100".format(self._tenant, self.SECRETS_PATH)
LOG.debug("href: {}".format(href))
LOG.debug("href: {0}".format(href))
hdrs, body = self._perform_http(href=href, method='GET')
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
@@ -130,9 +130,9 @@ class Connection(object):
bit_length=None,
cypher_type=None,
expiration=None):
LOG.debug(_("Creating secret of mime_type {}").format(mime_type))
LOG.debug(_("Creating secret of mime_type {0}").format(mime_type))
href = "{0}/{1}".format(self._tenant, self.SECRETS_PATH)
LOG.debug(_("href: {}").format(href))
LOG.debug(_("href: {0}").format(href))
secret_dict = {}
secret_dict['mime_type'] = mime_type
secret_dict['plain_text'] = plain_text
@@ -143,10 +143,8 @@ class Connection(object):
secret_dict['bit_length'] = int(bit_length)
if expiration is not None:
secret_dict['expiration'] = parse_isotime(expiration)
secret_dict['created'] = str(timeutils.utcnow())
secret_dict['status'] = 'created'
self._remove_empty_keys(secret_dict)
LOG.debug(_("Request body: {}").format(secret_dict))
LOG.debug(_("Request body: {0}").format(secret_dict))
hdrs, body = self._perform_http(href=href,
method='POST',
request_body=json.dumps(secret_dict))
@@ -157,7 +155,7 @@ class Connection(object):
def delete_secret_by_id(self, secret_id):
href = "{0}/{1}/{2}".format(self._tenant, self.SECRETS_PATH, secret_id)
LOG.info(_("Deleting secret - Secret ID: {}").format(secret_id))
LOG.info(_("Deleting secret - Secret ID: {0}").format(secret_id))
return self.delete_secret(href)
def delete_secret(self, href):
@@ -165,7 +163,7 @@ class Connection(object):
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
def get_secret_by_id(self, secret_id):
LOG.debug(_("Getting secret - Secret ID: {}").format(secret_id))
LOG.debug(_("Getting secret - Secret ID: {0}").format(secret_id))
href = "{0}/{1}/{2}".format(self._tenant, self.SECRETS_PATH, secret_id)
return self.get_secret(href)
@@ -192,7 +190,7 @@ class Connection(object):
"""
LOG.debug(_("Listing orders"))
href = "{0}/{1}?limit=100".format(self._tenant, self.ORDERS_PATH)
LOG.debug("href: {}".format(href))
LOG.debug("href: {0}".format(href))
hdrs, body = self._perform_http(href=href, method='GET')
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
@@ -209,9 +207,9 @@ class Connection(object):
bit_length=None,
name=None,
cypher_type=None):
LOG.debug(_("Creating order of mime_type {}").format(mime_type))
LOG.debug(_("Creating order of mime_type {0}").format(mime_type))
href = "{0}/{1}".format(self._tenant, self.ORDERS_PATH)
LOG.debug("href: {}".format(href))
LOG.debug("href: {0}".format(href))
order_dict = {'secret': {}}
order_dict['secret']['name'] = name
order_dict['secret']['mime_type'] = mime_type
@@ -219,7 +217,7 @@ class Connection(object):
order_dict['secret']['bit_length'] = bit_length
order_dict['secret']['cypher_type'] = cypher_type
self._remove_empty_keys(order_dict['secret'])
LOG.debug(_("Request body: {}").format(order_dict['secret']))
LOG.debug(_("Request body: {0}").format(order_dict['secret']))
hdrs, body = self._perform_http(href=href,
method='POST',
request_body=json.dumps(order_dict))
@@ -229,7 +227,7 @@ class Connection(object):
return self.get_order(body['order_ref'])
def delete_order_by_id(self, order_id):
LOG.info(_("Deleting order - Order ID: {}").format(order_id))
LOG.info(_("Deleting order - Order ID: {0}").format(order_id))
href = "{0}/{1}/{2}".format(self._tenant, self.ORDERS_PATH, order_id)
return self.delete_order(href)
@@ -238,7 +236,7 @@ class Connection(object):
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
def get_order_by_id(self, order_id):
LOG.debug(_("Getting order - Order ID: {}").format(order_id))
LOG.debug(_("Getting order - Order ID: {0}").format(order_id))
href = "{0}/{1}/{2}".format(self._tenant, self.ORDERS_PATH, order_id)
return self.get_order(href)
@@ -274,7 +272,7 @@ class Connection(object):
# Check if the status code is 2xx class
if not response.ok:
LOG.error('Bad response: {}'.format(response.status_code))
LOG.error('Bad response: {0}'.format(response.status_code))
raise ClientException(href=href, method=method,
http_status=response.status_code,
http_response_content=response.content)

3
barbicanclient/orders.py
View File

@@ -40,3 +40,6 @@ class Order(object):
def __eq__(self, other):
return isinstance(other, Order) and self.__dict__ == other.__dict__
def __ne__(self, other):
return not self.__eq__(other)

3
barbicanclient/secrets.py
View File

@@ -43,3 +43,6 @@ class Secret(object):
def __eq__(self, other):
return isinstance(other, Secret) and self.__dict__ == other.__dict__
def __ne__(self, other):
return not self.__eq__(other)

127
tests/client_test.py
View File

@@ -13,13 +13,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import unittest
import json
import unittest
from mock import MagicMock
from barbicanclient.client import Connection, Order, Secret
from barbicanclient.common.exceptions import ClientException
import barbicanclient.client as client
def suite():
@@ -38,6 +38,7 @@ class WhenTestingConnection(unittest.TestCase):
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)
@@ -57,22 +58,22 @@ class WhenTestingConnection(unittest.TestCase):
'req-6c19d09e-1167-445c-b435-d6b0818b59b9'
}
self.request.return_value.ok = True
self.connection = Connection(self.auth_endpoint, self.user, self.key,
self.tenant, token=self.auth_token,
authenticate=self.authenticate,
request=self.request)
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 = Connection(self.auth_endpoint,
self.user,
self.key,
self.tenant,
authenticate=self.authenticate,
endpoint=self.endpoint
)
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,
@@ -103,7 +104,7 @@ class WhenTestingConnection(unittest.TestCase):
'mime_type': 'text/plain'
}
secret = Secret(self.connection, body)
secret = client.Secret(self.connection, body)
self.request.return_value.content = json.dumps(body)
created = self.connection.create_secret('text/plain',
'Test secret',
@@ -132,7 +133,7 @@ class WhenTestingConnection(unittest.TestCase):
"2f53-4c0a-a0f3-33796671efc3"
}
order = Order(self.connection, body)
order = client.Order(self.connection, body)
self.request.return_value.content = json.dumps(body)
created = self.connection.create_order('text/plain',
name='test_secret',
@@ -141,12 +142,13 @@ class WhenTestingConnection(unittest.TestCase):
cypher_type='CDC')
self.assertEqual(order, created)
def test_should_list_secrets(self):
def test_list_no_secrets(self):
body0 = {'secrets': []}
secrets = []
self.request.return_value.content = json.dumps(body0)
self.assertEquals(secrets, self.connection.list_secrets())
def test_list_single_secret(self):
body1 = {'secrets': [{'status': 'ACTIVE',
'content_types': {'default': 'text/plain'},
'updated': '2013-06-03T21:16:58.349230',
@@ -160,34 +162,44 @@ class WhenTestingConnection(unittest.TestCase):
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}]}
secrets.append(Secret(self.connection, body1['secrets'][0]))
secrets = [client.Secret(self.connection, body1['secrets'][0])]
self.request.return_value.content = json.dumps(body1)
self.assertEquals(secrets, self.connection.list_secrets())
body2 = {'secrets': [{'status': 'ACTIVE',
'content_types': {'default': 'text/plain'},
'updated': '2013-07-03T21:17:58.349230',
'cypher_type': None,
'name': 'test_2',
'algorithm': 'aes',
'created': '2013-06-03T21:16:58.349222',
'secret_ref': 'http://localhost:9311/v1/'
'None/secrets/bbd2036f-730'
'7-4090-bbef-bbb6025eabcd',
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}]}
secrets.append(Secret(self.connection, body2['secrets'][0]))
def test_list_multiple_secrets(self):
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'}]}
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'
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)
self.assertEquals(secrets, self.connection.list_secrets())
def test_should_list_orders(self):
def test_list_no_orders(self):
body0 = {'orders': []}
orders = []
self.request.return_value.content = json.dumps(body0)
self.assertEquals(orders, self.connection.list_orders())
def test_list_single_order(self):
body1 = {'orders': [{'status': 'PENDING',
'updated': '2013-06-05T15:15:30.904760',
'created': '2013-06-05T15:15:30.904752',
@@ -202,50 +214,65 @@ class WhenTestingConnection(unittest.TestCase):
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}}]}
orders.append(Order(self.connection, body1['orders'][0]))
orders = [client.Order(self.connection, body1['orders'][0])]
self.request.return_value.content = json.dumps(body1)
self.assertEquals(orders, self.connection.list_orders())
body2 = {'orders': [{'status': 'ACTIVE',
'updated': '2013-07-05T15:15:30.904938',
'created': '2013-07-05T15:15:30.904752',
def test_list_multiple_orders(self):
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-3051656382fj',
'-45b3-bc60-3051656d5168',
'secret_ref': 'http://localhost:9311/'
'v1/None/secrets/????',
'secret': {'cypher_type': None,
'name': 'test_2',
'name': 'test_1',
'algorithm': None,
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}}]}
orders.append(Order(self.connection, body2['orders'][0]))
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'
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)
self.assertEquals(orders, self.connection.list_orders())
def test_should_perform_http(self):
href = 'http://localhost:9311/v1/12345/orders'
self.request.return_value.headers = {'Accept': 'application/json'}
self.request.return_value.content = ''
headers, body = self.connection._perform_http('GET', href)
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(self.request.return_value.content, body)
self.assertEqual(json.loads(self.request.return_value.content), body)
self.request.return_value.content = '{"test": "response"}'
headers, body = self.connection._perform_http('GET', href,
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)
headers, body = self.connection._perform_http('GET', href,
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
with self.assertRaises(ClientException):
self.connection._perform_http('GET', href)
self.connection._perform_http('GET', self.href)
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'
if __name__ == '__main__':