openstack/python-barbicanclient
Code Issues Proposed changes

Make create_secret() and create_order() return objects rather than dicts

Browse Source 
and get metadata before returning

Create unittests for those two functions

Turn debug log messages off by default
This commit is contained in:
Arash Ghoreyshi
2013-06-07 14:24:26 -05:00
parent 421a12726d
commit 8aea8b7fe6
4 changed files with 96 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
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
from barbicanclient.openstack.common import log, timeutils
from barbicanclient.common.exceptions import ClientException
from barbicanclient.openstack.common.gettextutils import _
from openstack.common.timeutils import parse_isotime
@@ -43,6 +43,7 @@ class Connection(object):
self._endpoint = (kwargs.get('endpoint')
or 'https://barbican.api.rackspacecloud.com/v1/')
self._cacert = kwargs.get('cacert')
self.log = LOG
self.connect(token=token)
@@ -143,17 +144,18 @@ class Connection(object):
secret_dict['bit_length'] = int(bit_length)
if expiration is not None:
secret_dict['expiration'] = parse_isotime(expiration)
#(secret_dict.pop(k) for k in secret_dict.keys() if secret_dict[k] is None)
secret_dict['created'] = str(timeutils.utcnow())
secret_dict['status'] = 'created'
self._remove_empty_keys(secret_dict)
#LOG.critical("DICT: {}".format(secret_dict))
LOG.debug(_("Request body: {}").format(secret_dict))
hdrs, body = self._perform_http(href=href,
method='POST',
request_body=json.dumps(secret_dict))
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
return body['secret_ref']
#return Secret(self, body)
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
#LOG.critical("MYSECRETTTTT: {}".format(self.get_secret(body['secret_ref']).created))
return self.get_secret(body['secret_ref'])
def delete_secret_by_id(self, secret_id):
href = "{0}/{1}/{2}".format(self._tenant, self.SECRETS_PATH, secret_id)
@@ -205,9 +207,9 @@ class Connection(object):
def create_order(self,
mime_type,
name=None,
algorithm=None,
bit_length=None,
name=None,
cypher_type=None):
LOG.debug(_("Creating order of mime_type {}").format(mime_type))
href = "{0}/{1}".format(self._tenant, self.ORDERS_PATH)
@@ -223,7 +225,10 @@ class Connection(object):
hdrs, body = self._perform_http(href=href,
method='POST',
request_body=json.dumps(order_dict))
return body['order_ref']
LOG.debug(_("Response - headers: {0}\nbody: {1}").format(hdrs, body))
return self.get_order(body['order_ref'])
def delete_order_by_id(self, order_id):
LOG.info(_("Deleting order - Order ID: {}").format(order_id))
@@ -269,7 +274,6 @@ class Connection(object):
response = self.request(method=method, url=url, data=request_body,
headers=headers)
LOG.critical("Response: {}".format(response.content))
# Check if the status code is 2xx class
if not response.ok:
LOG.error('Bad response: {}'.format(response.status_code))

6
barbicanclient/secrets.py
View File

@@ -11,10 +11,10 @@ class Secret(object):
Builds a secret object from a json representation. Includes the
connection object for subtasks.
"""
self._connection = connection
self.connection = connection
self.secret_ref = secret_dict['secret_ref']
self.created = parse_isotime(secret_dict['created'])
self.status = secret_dict['status']
self.created = parse_isotime(secret_dict.get('created'))
self.status = secret_dict.get('status')
self.algorithm = secret_dict.get('algorithm')
self.bit_length = secret_dict.get('bit_length')

2
etc/barbicanclient.conf
View File

@@ -3,7 +3,7 @@
verbose = True
# Show debugging output in logs (sets DEBUG log level output)
debug = True
debug = False
# Log to this file. Make sure you do not set the same log
# file for both the API and registry servers!

85
tests/client_test.py
View File

@@ -14,8 +14,11 @@
# limitations under the License.
import unittest
import json
from mock import MagicMock
from barbicanclient.openstack.common.timeutils import parse_isotime
from barbicanclient.client import Connection
@@ -28,7 +31,6 @@ def suite():
class WhenTestingConnection(unittest.TestCase):
def setUp(self):
self.auth_endpoint = 'https://keystone.com/v2'
self.user = 'user'
@@ -40,13 +42,25 @@ class WhenTestingConnection(unittest.TestCase):
self.authenticate = MagicMock()
self.authenticate.return_value = (self.endpoint, self.auth_token)
self.request = MagicMock()
self.request.return_value.content = {"secret_ref": "http://localhost:9"
+ "311/v1/None/secrets/ea1bb4e5-"
+ "e769-4d2c-8e58-cbbb17f1a3de"}
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 = Connection(self.auth_endpoint, self.user, self.key,
self.tenant, token=self.auth_token,
authenticate=self.authenticate,)
authenticate=self.authenticate,
request=self.request)
def test_should_connect_with_token(self):
self.assertFalse(self.authenticate.called)
@@ -75,7 +89,66 @@ class WhenTestingConnection(unittest.TestCase):
self.assertEqual(self.endpoint, self.connection._endpoint)
def test_should_create_secret(self):
pass
body = {'status': 'ACTIVE',
'content_types': {'default': 'text/plain'},
'updated': '2013-06-07T16:13:38.889857',
'cypher_type': None,
'name': 'test_secret',
'algorithm': None,
'created': '2013-06-07T16:13:38.889851',
'secret_ref': 'http://localhost:9311/v1/None/secrets/e6e7d'
'b5e-3738-408e-aaba-05a7177cade5',
'expiration': None,
'bit_length': None,
'mime_type': 'text/plain'
}
self.request.return_value.content = json.dumps(body)
secret = self.connection.create_secret('text/plain',
'Test secret',
name='test_secret',
algorithm=None,
bit_length=None,
cypher_type=None,
expiration=None)
self.assertEqual(body['secret_ref'], secret.secret_ref)
self.assertEqual(self.connection, secret.connection)
self.assertEqual(body['status'], secret.status)
self.assertEqual(body['name'], secret.name)
self.assertEqual(body['mime_type'], secret.mime_type)
self.assertEqual(parse_isotime(body['created']), secret.created)
self.assertEqual(parse_isotime(body['updated']), secret.updated)
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": None,
"bit_length": 256,
"mime_type": "text/plain"
},
"order_ref": "http://localhost:9311/v1/12345/orders/003f2b91-"
"2f53-4c0a-a0f3-33796671efc3"
}
self.request.return_value.content = json.dumps(body)
order = self.connection.create_order('text/plain',
name='test_secret',
bit_length=256,
algorithm='aes',
cypher_type='CDC')
self.assertEqual(self.connection, order.connection)
self.assertEqual(body['secret_ref'], order.secret_ref)
self.assertEqual(body['status'], order.status)
self.assertEqual(parse_isotime(body['created']), order.created)
self.assertEqual(parse_isotime(body['updated']), order.updated)
self.assertEqual(body['secret'], order.secret)
if __name__ == '__main__':
unittest.main()