openstack/python-barbicanclient
Code Issues Proposed changes

Implement secret and order list paging

Browse Source 
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
This commit is contained in:
Arash Ghoreyshi
2013-06-14 13:05:33 -05:00
parent 0240a76654
commit a5d9ca5ec6
5 changed files with 151 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
barbicanclient/client.py
View File

@@ -103,22 +103,46 @@ class Connection(object):
self._token = value
self._session.headers['X-Auth-Token'] = value
def list_secrets(self):
def list_secrets(self, limit=20, offset=0):
"""
Returns the list of secrets for the auth'd tenant
Returns a tuple containing three items: a list of secrets pertaining
to the given offset and limit, a reference to the previous set of
secrets, and a reference to the next set of secrets. Either of the
references may be None.
"""
LOG.debug(_("Listing secrets"))
href = "{0}/{1}?limit=100".format(self._tenant, self.SECRETS_PATH)
LOG.debug(_("Listing secrets - offset: {0}, limit: {1}").format(offset,
limit))
href = "{0}/{1}?limit={2}&offset={3}".format(self._tenant,
self.SECRETS_PATH,
limit, offset)
return self.list_secrets_by_href(href)
def list_secrets_by_href(self, href):
"""
Returns a tuple containing three items: a list of secrets pertaining
to the offset and limit within href, a reference to the previous set
of secrets, and a reference to the next set of secrets. Either of the
references may be None.
"""
LOG.debug(_("Listing secrets by 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))
secrets_dict = body['secrets']
secrets = []
for s in secrets_dict:
secrets.append(Secret(self._conn, s))
secrets = [Secret(self._conn, s) for s in secrets_dict]
return secrets
if 'previous' in body:
prev_ref = body['previous']
else:
prev_ref = None
if 'next' in body:
next_ref = body['next']
else:
next_ref = None
return secrets, prev_ref, next_ref
def create_secret(self,
mime_type,
@@ -182,22 +206,46 @@ class Connection(object):
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
return body
def list_orders(self):
def list_orders(self, limit=20, offset=0):
"""
Returns the list of orders
Returns a tuple containing three items: a list of orders pertaining
to the given offset and limit, a reference to the previous set of
orders, and a reference to the next set of orders. Either of the
references may be None.
"""
LOG.debug(_("Listing orders"))
href = "{0}/{1}?limit=100".format(self._tenant, self.ORDERS_PATH)
LOG.debug(_("Listing orders - offset: {0}, limit: {1}").format(offset,
limit))
href = "{0}/{1}?limit={2}&offset={3}".format(self._tenant,
self.ORDERS_PATH,
limit, offset)
return self.list_orders_by_href(href)
def list_orders_by_href(self, href):
"""
Returns a tuple containing three items: a list of orders pertaining
to the offset and limit within href, a reference to the previous set
of orders, and a reference to the next set of orders. Either of the
references may be None.
"""
LOG.debug(_("Listing orders by 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))
orders_dict = body['orders']
orders = []
for o in orders_dict:
orders.append(Order(self._conn, o))
orders = [Order(self._conn, o) for o in orders_dict]
return orders
if 'previous' in body:
prev_ref = body['previous']
else:
prev_ref = None
if 'next' in body:
next_ref = body['next']
else:
next_ref = None
return orders, prev_ref, next_ref
def create_order(self,
mime_type,

13
cloudkeyp → keep
View File

@@ -5,7 +5,7 @@ import argparse
from barbicanclient import client
class CloudKeyp:
class Keep:
def __init__(self):
self.parser = argparse.ArgumentParser(description='Access the Barbican'
' key management sevice.')
@@ -70,9 +70,12 @@ class CloudKeyp:
def create(self, args):
if self.args.type == 'secret':
secret = self.conn.create_secret(self.args.mime_type,
self.args.plain_text, self.args.name,
self.args.algorithm, self.args.bit_length,
self.args.cypher_type, self.args.expiration)
self.args.plain_text,
self.args.name,
self.args.algorithm,
self.args.bit_length,
self.args.cypher_type,
self.args.expiration)
print secret.secret_ref
def execute(self):
@@ -86,7 +89,7 @@ class CloudKeyp:
def main():
CloudKeyp()
Keep()
if __name__ == '__main__':

3
setup.py
View File

@@ -54,5 +54,6 @@ setuptools.setup(
'Operating System :: OS Independent',
'Programming Language :: Python :: 2.7',
'Environment :: No Input/Output (Daemon)',
]
],
scripts = ['keep']
)

88
tests/client_test.py
View File

@@ -14,7 +14,7 @@
# limitations under the License.
import json
import unittest
import unittest2 as unittest
from mock import MagicMock
@@ -146,10 +146,13 @@ class WhenTestingConnection(unittest.TestCase):
body0 = {'secrets': []}
secrets = []
self.request.return_value.content = json.dumps(body0)
self.assertTrue(self._are_equivalent(secrets,
self.connection.list_secrets()))
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',
@@ -162,13 +165,22 @@ class WhenTestingConnection(unittest.TestCase):
'7-4090-bbef-bbb6025e5e7b',
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}]}
'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)
self.assertTrue(self._are_equivalent(secrets,
self.connection.list_secrets()))
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',
@@ -181,29 +193,43 @@ class WhenTestingConnection(unittest.TestCase):
'7-4090-bbef-bbb6025e5e7b',
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}]}
'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)
self.assertTrue(self._are_equivalent(secrets,
self.connection.list_secrets()))
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)
self.assertTrue(self._are_equivalent(orders,
self.connection.list_orders()))
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',
@@ -217,13 +243,21 @@ class WhenTestingConnection(unittest.TestCase):
'algorithm': None,
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}}]}
'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)
self.assertTrue(self._are_equivalent(orders,
self.connection.list_orders()))
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',
@@ -237,19 +271,34 @@ class WhenTestingConnection(unittest.TestCase):
'algorithm': None,
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'}}]}
'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)
self.assertTrue(self._are_equivalent(orders,
self.connection.list_orders()))
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()
@@ -272,8 +321,11 @@ class WhenTestingConnection(unittest.TestCase):
def test_should_raise_exception(self):
self._setup_request()
self.request.return_value.ok = False
with self.assertRaises(ClientException):
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'}

8
tox.ini
View File

@@ -4,7 +4,7 @@
# and then run "tox" from this directory.
[tox]
envlist = py26, py27, py33, pep8
envlist = py26, py27
[testenv]
setenv = VIRTUAL_ENV={envdir}
@@ -33,3 +33,9 @@ downloadcache = ~/cache/pip
ignore = F,H
show-source = True
exclude = .venv,.tox,dist,doc,*egg
[testenv:py26]
commands = nosetests {posargs:--with-xcoverage --all-modules --cover-inclusive --traverse-namespace --with-xunit --cover-package=barbican}
[testenv:py27]
commands = nosetests {posargs:--with-xcoverage --all-modules --cover-inclusive --traverse-namespace --with-xunit --cover-package=barbican}