Update Order models
This CR updates the Order models to use the revamped Orders API. There are two concrete Order classes: KeyOrder and AsymmetricOrder which can be used to create typed orders with types 'key' and 'asymmetric' respectively. Change-Id: I0bfb221d1ee00ae3e26be77afc32c7da9cbc59ca
This commit is contained in:
Douglas Mendizabal
@@ -15,9 +15,10 @@
|
||||
"""
|
||||
Base utilities to build API operation managers.
|
||||
"""
|
||||
import six
|
||||
import uuid
|
||||
|
||||
import six
|
||||
|
||||
|
||||
def filter_empty_keys(dictionary):
|
||||
return dict(((k, v) for k, v in dictionary.items() if v))
|
||||
|
||||
@@ -12,10 +12,12 @@
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
import abc
|
||||
import functools
|
||||
import logging
|
||||
|
||||
from oslo.utils.timeutils import parse_isotime
|
||||
import six
|
||||
|
||||
from barbicanclient import base
|
||||
from barbicanclient import formatter
|
||||
@@ -33,7 +35,7 @@ def immutable_after_save(func):
|
||||
return wrapper
|
||||
|
||||
|
||||
class OrderFormatter(formatter.EntityFormatter):
|
||||
class KeyOrderFormatter(formatter.EntityFormatter):
|
||||
|
||||
columns = ("Order href",
|
||||
"Secret href",
|
||||
@@ -54,85 +56,115 @@ class OrderFormatter(formatter.EntityFormatter):
|
||||
return data
|
||||
|
||||
|
||||
class Order(OrderFormatter):
|
||||
class AsymmetricOrderFormatter(formatter.EntityFormatter):
|
||||
|
||||
columns = ("Order href",
|
||||
"Container href",
|
||||
"Created",
|
||||
"Status",
|
||||
"Error code",
|
||||
"Error message"
|
||||
)
|
||||
|
||||
def _get_formatted_data(self):
|
||||
data = (self.order_ref,
|
||||
self.container_ref,
|
||||
self.created,
|
||||
self.status,
|
||||
self.error_status_code,
|
||||
self.error_reason
|
||||
)
|
||||
return data
|
||||
|
||||
|
||||
@six.add_metaclass(abc.ABCMeta)
|
||||
class Order(object):
|
||||
"""
|
||||
Orders are used to request the generation of a Secret in Barbican.
|
||||
Base order object to hold common functionality
|
||||
|
||||
This should be considered an abstract class that should not be
|
||||
instantiated directly.
|
||||
"""
|
||||
_entity = 'orders'
|
||||
|
||||
def __init__(self, api, name=None, algorithm=None, bit_length=None,
|
||||
mode=None, payload_content_type='application/octet-stream',
|
||||
order_ref=None, secret_ref=None, status=None,
|
||||
created=None, updated=None, expiration=None,
|
||||
error_status_code=None, error_reason=None, secret=None,
|
||||
meta=None, type=None):
|
||||
def __init__(self, api, type, status=None, created=None, updated=None,
|
||||
meta=None, order_ref=None, error_status_code=None,
|
||||
error_reason=None):
|
||||
super(Order, self).__init__()
|
||||
|
||||
self._api = api
|
||||
self._order_ref = order_ref
|
||||
self._type = type
|
||||
self._meta = meta
|
||||
if order_ref:
|
||||
self._error_status_code = error_status_code
|
||||
self._error_reason = error_reason
|
||||
self._status = status
|
||||
self._created = created
|
||||
self._updated = updated
|
||||
if self._created:
|
||||
self._created = parse_isotime(self._created)
|
||||
if self._updated:
|
||||
self._updated = parse_isotime(self._updated)
|
||||
self._secret_ref = secret_ref
|
||||
self._secret = secret
|
||||
self._status = status
|
||||
|
||||
if created:
|
||||
self._created = parse_isotime(created)
|
||||
else:
|
||||
self._error_status_code = None
|
||||
self._error_reason = None
|
||||
self._status = None
|
||||
self._created = None
|
||||
|
||||
if updated:
|
||||
self._updated = parse_isotime(updated)
|
||||
else:
|
||||
self._updated = None
|
||||
self._secret_ref = None
|
||||
self._secret = base.filter_empty_keys({
|
||||
'name': name,
|
||||
'algorithm': algorithm,
|
||||
'bit_length': bit_length,
|
||||
'mode': mode,
|
||||
'payload_content_type': payload_content_type,
|
||||
'expiration': expiration
|
||||
})
|
||||
if self._secret.get("expiration"):
|
||||
self._secret['expiration'] = parse_isotime(
|
||||
self._secret.get('expiration'))
|
||||
|
||||
self._order_ref = order_ref
|
||||
|
||||
self._meta = base.filter_empty_keys(meta)
|
||||
|
||||
self._error_status_code = error_status_code
|
||||
self._error_reason = error_reason
|
||||
|
||||
if 'expiration' in self._meta.keys():
|
||||
self._meta['expiration'] = parse_isotime(self._meta['expiration'])
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self._secret.get('name')
|
||||
return self._meta.get('name')
|
||||
|
||||
@property
|
||||
def expiration(self):
|
||||
return self._secret.get('expiration')
|
||||
@name.setter
|
||||
@immutable_after_save
|
||||
def name(self, value):
|
||||
self._meta['name'] = value
|
||||
|
||||
@property
|
||||
def algorithm(self):
|
||||
return self._secret.get('algorithm')
|
||||
return self._meta.get('algorithm')
|
||||
|
||||
@algorithm.setter
|
||||
@immutable_after_save
|
||||
def algorithm(self, value):
|
||||
self._meta['algorithm'] = value
|
||||
|
||||
@property
|
||||
def bit_length(self):
|
||||
return self._secret.get('bit_length')
|
||||
return self._meta.get('bit_length')
|
||||
|
||||
@bit_length.setter
|
||||
@immutable_after_save
|
||||
def bit_length(self, value):
|
||||
self._meta['bit_length'] = value
|
||||
|
||||
@property
|
||||
def mode(self):
|
||||
return self._secret.get('mode')
|
||||
def expiration(self):
|
||||
return self._meta.get('expiration')
|
||||
|
||||
@expiration.setter
|
||||
@immutable_after_save
|
||||
def expiration(self, value):
|
||||
self._meta['expiration'] = value
|
||||
|
||||
@property
|
||||
def payload_content_type(self):
|
||||
return self._secret.get('payload_content_type')
|
||||
return self._meta.get('payload_content_type')
|
||||
|
||||
@payload_content_type.setter
|
||||
@immutable_after_save
|
||||
def payload_content_type(self, value):
|
||||
self._meta['payload_content_type'] = value
|
||||
|
||||
@property
|
||||
def order_ref(self):
|
||||
return self._order_ref
|
||||
|
||||
@property
|
||||
def secret_ref(self):
|
||||
return self._secret_ref
|
||||
|
||||
@property
|
||||
def created(self):
|
||||
return self._created
|
||||
@@ -153,60 +185,10 @@ class Order(OrderFormatter):
|
||||
def error_reason(self):
|
||||
return self._error_reason
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
return self._type
|
||||
|
||||
@property
|
||||
def meta(self):
|
||||
return self._meta
|
||||
|
||||
@name.setter
|
||||
@immutable_after_save
|
||||
def name(self, value):
|
||||
self._secret['name'] = value
|
||||
|
||||
@expiration.setter
|
||||
@immutable_after_save
|
||||
def expiration(self, value):
|
||||
self._secret['expiration'] = value
|
||||
|
||||
@algorithm.setter
|
||||
@immutable_after_save
|
||||
def algorithm(self, value):
|
||||
self._secret['algorithm'] = value
|
||||
|
||||
@bit_length.setter
|
||||
@immutable_after_save
|
||||
def bit_length(self, value):
|
||||
self._secret['bit_length'] = value
|
||||
|
||||
@mode.setter
|
||||
@immutable_after_save
|
||||
def mode(self, value):
|
||||
self._secret['mode'] = value
|
||||
|
||||
@payload_content_type.setter
|
||||
@immutable_after_save
|
||||
def payload_content_type(self, value):
|
||||
self._secret['payload_content_type'] = value
|
||||
|
||||
@type.setter
|
||||
@immutable_after_save
|
||||
def type(self, value):
|
||||
self._type = value
|
||||
|
||||
@meta.setter
|
||||
@immutable_after_save
|
||||
def meta(self, value):
|
||||
self._meta = value
|
||||
|
||||
@immutable_after_save
|
||||
def submit(self):
|
||||
order_dict = dict({
|
||||
'secret': self._secret
|
||||
})
|
||||
LOG.debug("Request body: {0}".format(order_dict.get('secret')))
|
||||
order_dict = {'type': self._type, 'meta': self._meta}
|
||||
LOG.debug("Request body: {0}".format(order_dict))
|
||||
response = self._api._post(self._entity, order_dict)
|
||||
if response:
|
||||
self._order_ref = response.get('order_ref')
|
||||
@@ -219,12 +201,87 @@ class Order(OrderFormatter):
|
||||
else:
|
||||
raise LookupError("Order is not yet stored.")
|
||||
|
||||
|
||||
class KeyOrder(Order, KeyOrderFormatter):
|
||||
_type = 'key'
|
||||
|
||||
def __init__(self, api, name=None, algorithm=None, bit_length=None,
|
||||
mode=None, expiration=None, payload_content_type=None,
|
||||
status=None, created=None, updated=None, order_ref=None,
|
||||
secret_ref=None, error_status_code=None, error_reason=None):
|
||||
super(KeyOrder, self).__init__(
|
||||
api, self._type, status=status, created=created, updated=updated,
|
||||
meta={
|
||||
'name': name, 'algorithm': algorithm, 'bit_length': bit_length,
|
||||
'expiration': expiration,
|
||||
'payload_content_type': payload_content_type
|
||||
}, order_ref=order_ref, error_status_code=error_status_code,
|
||||
error_reason=error_reason)
|
||||
self._secret_ref = secret_ref
|
||||
if mode:
|
||||
self._meta['mode'] = mode
|
||||
|
||||
@property
|
||||
def mode(self):
|
||||
return self._meta.get('mode')
|
||||
|
||||
@property
|
||||
def secret_ref(self):
|
||||
return self._secret_ref
|
||||
|
||||
@mode.setter
|
||||
@immutable_after_save
|
||||
def mode(self, value):
|
||||
self._meta['mode'] = value
|
||||
|
||||
def __repr__(self):
|
||||
return 'Order(order_ref={0})'.format(self.order_ref)
|
||||
return 'KeyOrder(order_ref={0})'.format(self.order_ref)
|
||||
|
||||
|
||||
class AsymmetricOrder(Order, AsymmetricOrderFormatter):
|
||||
_type = 'asymmetric'
|
||||
|
||||
def __init__(self, api, name=None, algorithm=None, bit_length=None,
|
||||
pass_phrase=None, expiration=None, payload_content_type=None,
|
||||
status=None, created=None, updated=None, order_ref=None,
|
||||
container_ref=None, error_status_code=None,
|
||||
error_reason=None):
|
||||
super(AsymmetricOrder, self).__init__(
|
||||
api, self._type, status=status, created=created, updated=updated,
|
||||
meta={
|
||||
'name': name, 'algorithm': algorithm, 'bit_length': bit_length,
|
||||
'expiration': expiration,
|
||||
'payload_content_type': payload_content_type
|
||||
}, order_ref=order_ref, error_status_code=error_status_code,
|
||||
error_reason=error_reason)
|
||||
self._container_ref = container_ref
|
||||
if pass_phrase:
|
||||
self._meta['pass_phrase'] = pass_phrase
|
||||
|
||||
@property
|
||||
def container_ref(self):
|
||||
return self._container_ref
|
||||
|
||||
@property
|
||||
def pass_phrase(self):
|
||||
return self._meta.get('pass_phrase')
|
||||
|
||||
@pass_phrase.setter
|
||||
@immutable_after_save
|
||||
def pass_phrase(self, value):
|
||||
self._meta['pass_phrase'] = value
|
||||
|
||||
def __repr__(self):
|
||||
return 'AsymmetricOrder(order_ref={0})'.format(self.order_ref)
|
||||
|
||||
|
||||
class OrderManager(base.BaseEntityManager):
|
||||
|
||||
_order_type_to_class_map = {
|
||||
'key': KeyOrder,
|
||||
'asymmetric': AsymmetricOrder
|
||||
}
|
||||
|
||||
def __init__(self, api):
|
||||
super(OrderManager, self).__init__(api, 'orders')
|
||||
|
||||
@@ -233,30 +290,67 @@ class OrderManager(base.BaseEntityManager):
|
||||
Get an Order
|
||||
|
||||
:param order_ref: Full HATEOAS reference to an Order
|
||||
:returns: Order
|
||||
:returns: An instance of the appropriate subtype of Order
|
||||
"""
|
||||
LOG.debug("Getting order - Order href: {0}".format(order_ref))
|
||||
base.validate_ref(order_ref, 'Order')
|
||||
response = self._api._get(order_ref)
|
||||
return Order(api=self._api, **response)
|
||||
try:
|
||||
response = self._api._get(order_ref)
|
||||
except AttributeError:
|
||||
raise LookupError(
|
||||
'Order {0} could not be found.'.format(order_ref)
|
||||
)
|
||||
return self._create_typed_order(response)
|
||||
|
||||
def create(self, name=None, payload_content_type=None,
|
||||
algorithm=None, bit_length=None, mode=None, expiration=None):
|
||||
def _create_typed_order(self, response):
|
||||
resp_type = response.pop('type').lower()
|
||||
order_type = self._order_type_to_class_map.get(resp_type)
|
||||
|
||||
response.update(response.pop('meta'))
|
||||
|
||||
if order_type is KeyOrder:
|
||||
return KeyOrder(self._api, **response)
|
||||
elif order_type is AsymmetricOrder:
|
||||
return AsymmetricOrder(self._api, **response)
|
||||
else:
|
||||
raise TypeError('Unknown Order type "{0}"'.format(order_type))
|
||||
|
||||
def create_key(self, name=None, algorithm=None, bit_length=None, mode=None,
|
||||
payload_content_type=None, expiration=None):
|
||||
"""
|
||||
Create an Order
|
||||
Create an Order for a Symmetric Key
|
||||
|
||||
:param name: A friendly name for the secret
|
||||
:param payload_content_type: The format/type of the secret data
|
||||
:param name: A friendly name for the secret to be created
|
||||
:param algorithm: The algorithm associated with this secret key
|
||||
:param bit_length: The bit length of this secret key
|
||||
:param mode: The algorithm mode used with this secret key
|
||||
:param payload_content_type: The format/type of the secret data
|
||||
:param expiration: The expiration time of the secret in ISO 8601 format
|
||||
:returns: Order
|
||||
:returns: KeyOrder
|
||||
"""
|
||||
return Order(api=self._api, name=name,
|
||||
payload_content_type=payload_content_type,
|
||||
algorithm=algorithm, bit_length=bit_length, mode=mode,
|
||||
expiration=expiration)
|
||||
return KeyOrder(api=self._api, name=name,
|
||||
algorithm=algorithm, bit_length=bit_length, mode=mode,
|
||||
payload_content_type=payload_content_type,
|
||||
expiration=expiration)
|
||||
|
||||
def create_asymmetric(self, name=None, algorithm=None, bit_length=None,
|
||||
pass_phrase=None, payload_content_type=None,
|
||||
expiration=None):
|
||||
"""
|
||||
Create an Order for an Asymmetric Key
|
||||
|
||||
:param name: A friendly name for the container to be created
|
||||
:param algorithm: The algorithm associated with this secret key
|
||||
:param bit_length: The bit length of this secret key
|
||||
:param pass_phrase: Optional passphrase
|
||||
:param payload_content_type: The format/type of the secret data
|
||||
:param expiration: The expiration time of the secret in ISO 8601 format
|
||||
:returns AsymmetricOrder
|
||||
"""
|
||||
return AsymmetricOrder(api=self._api, name=name, algorithm=algorithm,
|
||||
bit_length=bit_length, pass_phrase=pass_phrase,
|
||||
payload_content_type=payload_content_type,
|
||||
expiration=expiration)
|
||||
|
||||
def delete(self, order_ref):
|
||||
"""
|
||||
@@ -282,4 +376,6 @@ class OrderManager(base.BaseEntityManager):
|
||||
params = {'limit': limit, 'offset': offset}
|
||||
response = self._api._get(href, params)
|
||||
|
||||
return [Order(api=self._api, **o) for o in response.get('orders', [])]
|
||||
return [
|
||||
self._create_typed_order(o) for o in response.get('orders', [])
|
||||
]
|
||||
|
||||
@@ -12,116 +12,78 @@
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
import json
|
||||
|
||||
import mock
|
||||
from oslo.utils import timeutils
|
||||
import testtools
|
||||
|
||||
from barbicanclient import orders, base
|
||||
from barbicanclient.test import test_client
|
||||
from barbicanclient.test import test_client_secrets as test_secrets
|
||||
|
||||
|
||||
class OrderData(object):
|
||||
def __init__(self):
|
||||
self.created = str(timeutils.utcnow())
|
||||
|
||||
self.secret = test_secrets.SecretData()
|
||||
self.status = 'ACTIVE'
|
||||
self.order_dict = {'created': self.created,
|
||||
'status': self.status,
|
||||
'secret': self.secret.get_dict()}
|
||||
|
||||
def get_dict(self, order_ref, secret_ref=None):
|
||||
order = self.order_dict
|
||||
order['order_ref'] = order_ref
|
||||
if secret_ref:
|
||||
order['secret_ref'] = secret_ref
|
||||
return order
|
||||
|
||||
|
||||
class WhenTestingOrders(test_client.BaseEntityResource):
|
||||
|
||||
class OrdersTestCase(testtools.TestCase):
|
||||
def setUp(self):
|
||||
self._setUp('orders')
|
||||
super(OrdersTestCase, self).setUp()
|
||||
self.secret_ref = ("http://localhost:9311/v1/secrets/"
|
||||
"a2292306-6da0-4f60-bd8a-84fc8d692716")
|
||||
self.order_ref = ("http://localhost:9311/v1/orders/"
|
||||
"d0460cc4-2876-4493-b7de-fc5c812883cc")
|
||||
self.key_order_data = """{{
|
||||
"status": "ACTIVE",
|
||||
"secret_ref": "{0}",
|
||||
"updated": "2014-10-21T17:15:50.871596",
|
||||
"meta": {{
|
||||
"name": "secretname",
|
||||
"algorithm": "aes",
|
||||
"payload_content_type": "application/octet-stream",
|
||||
"mode": "cbc",
|
||||
"bit_length": 256,
|
||||
"expiration": "2015-02-28T19:14:44.180394"
|
||||
}},
|
||||
"created": "2014-10-21T17:15:50.824202",
|
||||
"type": "key",
|
||||
"order_ref": "{1}"
|
||||
}}""".format(self.secret_ref, self.order_ref)
|
||||
self.api = mock.MagicMock()
|
||||
self.api._base_url = 'http://localhost:9311/v1'
|
||||
self.manager = orders.OrderManager(api=self.api)
|
||||
|
||||
self.order = OrderData()
|
||||
def _get_order_args(self, order_data):
|
||||
order_args = json.loads(order_data)
|
||||
order_args.update(order_args.pop('meta'))
|
||||
order_args.pop('type')
|
||||
return order_args
|
||||
|
||||
self.manager = orders.OrderManager(self.api)
|
||||
|
||||
def test_should_entity_str(self):
|
||||
order = self.order.get_dict(self.entity_href)
|
||||
class WhenTestingKeyOrders(OrdersTestCase):
|
||||
|
||||
def test_should_include_errors_in_str(self):
|
||||
order_args = self._get_order_args(self.key_order_data)
|
||||
error_code = 500
|
||||
error_reason = 'Something is broken'
|
||||
order_obj = orders.Order(api=None, error_status_code=error_code,
|
||||
error_reason=error_reason, **order)
|
||||
self.assertIn(self.order.status, str(order_obj))
|
||||
order_obj = orders.KeyOrder(api=None, error_status_code=error_code,
|
||||
error_reason=error_reason, **order_args)
|
||||
self.assertIn(str(error_code), str(order_obj))
|
||||
self.assertIn(error_reason, str(order_obj))
|
||||
|
||||
def test_should_entity_repr(self):
|
||||
order = self.order.get_dict(self.entity_href)
|
||||
order_obj = orders.Order(api=None, **order)
|
||||
self.assertIn('order_ref=' + self.entity_href, repr(order_obj))
|
||||
|
||||
def test_should_submit_via_constructor(self):
|
||||
self.api._post.return_value = {'order_ref': self.entity_href}
|
||||
|
||||
order = self.manager.create(
|
||||
name=self.order.secret.name,
|
||||
algorithm=self.order.secret.algorithm,
|
||||
payload_content_type=self.order.secret.content
|
||||
)
|
||||
order_href = order.submit()
|
||||
|
||||
self.assertEqual(self.entity_href, order_href)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._post.call_args
|
||||
entity_resp = args[0]
|
||||
self.assertEqual(self.entity, entity_resp)
|
||||
|
||||
# Verify that correct information was sent in the call.
|
||||
order_req = args[1]
|
||||
self.assertEqual(self.order.secret.name, order_req['secret']['name'])
|
||||
self.assertEqual(self.order.secret.algorithm,
|
||||
order_req['secret']['algorithm'])
|
||||
self.assertEqual(self.order.secret.payload_content_type,
|
||||
order_req['secret']['payload_content_type'])
|
||||
|
||||
def test_should_submit_via_attributes(self):
|
||||
self.api._post.return_value = {'order_ref': self.entity_href}
|
||||
|
||||
order = self.manager.create()
|
||||
order.name = self.order.secret.name
|
||||
order.algorithm = self.order.secret.algorithm
|
||||
order.payload_content_type = self.order.secret.content
|
||||
order_href = order.submit()
|
||||
|
||||
self.assertEqual(self.entity_href, order_href)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._post.call_args
|
||||
entity_resp = args[0]
|
||||
self.assertEqual(self.entity, entity_resp)
|
||||
|
||||
# Verify that correct information was sent in the call.
|
||||
order_req = args[1]
|
||||
self.assertEqual(self.order.secret.name, order_req['secret']['name'])
|
||||
self.assertEqual(self.order.secret.algorithm,
|
||||
order_req['secret']['algorithm'])
|
||||
self.assertEqual(self.order.secret.payload_content_type,
|
||||
order_req['secret']['payload_content_type'])
|
||||
def test_should_include_order_ref_in_repr(self):
|
||||
order_args = self._get_order_args(self.key_order_data)
|
||||
order_obj = orders.KeyOrder(api=None, **order_args)
|
||||
self.assertIn('order_ref=' + self.order_ref, repr(order_obj))
|
||||
|
||||
def test_should_be_immutable_after_submit(self):
|
||||
self.api._post.return_value = {'order_ref': self.entity_href}
|
||||
self.api._post.return_value = {'order_ref': self.order_ref}
|
||||
|
||||
order = self.manager.create(
|
||||
name=self.order.secret.name,
|
||||
algorithm=self.order.secret.algorithm,
|
||||
payload_content_type=self.order.secret.content
|
||||
order = self.manager.create_key(
|
||||
name='name',
|
||||
algorithm='algorithm',
|
||||
payload_content_type='payload_content_type'
|
||||
)
|
||||
order_href = order.submit()
|
||||
|
||||
self.assertEqual(self.entity_href, order_href)
|
||||
self.assertEqual(self.order_ref, order_href)
|
||||
|
||||
# Verify that attributes are immutable after store.
|
||||
attributes = [
|
||||
@@ -135,8 +97,57 @@ class WhenTestingOrders(test_client.BaseEntityResource):
|
||||
except base.ImmutableException:
|
||||
pass
|
||||
|
||||
def test_should_submit_via_constructor(self):
|
||||
self.api._post.return_value = {'order_ref': self.order_ref}
|
||||
|
||||
order = self.manager.create_key(
|
||||
name='name',
|
||||
algorithm='algorithm',
|
||||
payload_content_type='payload_content_type'
|
||||
)
|
||||
order_href = order.submit()
|
||||
|
||||
self.assertEqual(self.order_ref, order_href)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._post.call_args
|
||||
entity_resp = args[0]
|
||||
self.assertEqual('orders', entity_resp)
|
||||
|
||||
# Verify that correct information was sent in the call.
|
||||
order_req = args[1]
|
||||
self.assertEqual('name', order_req['meta']['name'])
|
||||
self.assertEqual('algorithm',
|
||||
order_req['meta']['algorithm'])
|
||||
self.assertEqual('payload_content_type',
|
||||
order_req['meta']['payload_content_type'])
|
||||
|
||||
def test_should_submit_via_attributes(self):
|
||||
self.api._post.return_value = {'order_ref': self.order_ref}
|
||||
|
||||
order = self.manager.create_key()
|
||||
order.name = 'name'
|
||||
order.algorithm = 'algorithm'
|
||||
order.payload_content_type = 'payload_content_type'
|
||||
order_href = order.submit()
|
||||
|
||||
self.assertEqual(self.order_ref, order_href)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._post.call_args
|
||||
entity_resp = args[0]
|
||||
self.assertEqual('orders', entity_resp)
|
||||
|
||||
# Verify that correct information was sent in the call.
|
||||
order_req = args[1]
|
||||
self.assertEqual('name', order_req['meta']['name'])
|
||||
self.assertEqual('algorithm',
|
||||
order_req['meta']['algorithm'])
|
||||
self.assertEqual('payload_content_type',
|
||||
order_req['meta']['payload_content_type'])
|
||||
|
||||
def test_should_not_be_able_to_set_generated_attributes(self):
|
||||
order = self.manager.create()
|
||||
order = self.manager.create_key()
|
||||
|
||||
# Verify that generated attributes cannot be set.
|
||||
attributes = [
|
||||
@@ -150,46 +161,80 @@ class WhenTestingOrders(test_client.BaseEntityResource):
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
def test_should_get(self):
|
||||
self.api._get.return_value = self.order.get_dict(self.entity_href)
|
||||
|
||||
order = self.manager.get(order_ref=self.entity_href)
|
||||
self.assertIsInstance(order, orders.Order)
|
||||
self.assertEqual(self.entity_href, order.order_ref)
|
||||
class WhenTestingAsymmetricOrders(OrdersTestCase):
|
||||
|
||||
def test_should_be_immutable_after_submit(self):
|
||||
self.api._post.return_value = {'order_ref': self.order_ref}
|
||||
|
||||
order = self.manager.create_asymmetric(
|
||||
name='name',
|
||||
algorithm='algorithm',
|
||||
payload_content_type='payload_content_type'
|
||||
)
|
||||
order_href = order.submit()
|
||||
|
||||
self.assertEqual(self.order_ref, order_href)
|
||||
|
||||
# Verify that attributes are immutable after store.
|
||||
attributes = [
|
||||
"name", "expiration", "algorithm", "bit_length", "pass_phrase",
|
||||
"payload_content_type"
|
||||
]
|
||||
for attr in attributes:
|
||||
try:
|
||||
setattr(order, attr, "test")
|
||||
self.fail(
|
||||
"{0} didn't raise an ImmutableException exception".format(
|
||||
attr
|
||||
)
|
||||
)
|
||||
except base.ImmutableException:
|
||||
pass
|
||||
|
||||
|
||||
class WhenTestingOrderManager(OrdersTestCase):
|
||||
|
||||
def test_should_get(self):
|
||||
self.api._get.return_value = json.loads(self.key_order_data)
|
||||
|
||||
order = self.manager.get(order_ref=self.order_ref)
|
||||
self.assertIsInstance(order, orders.KeyOrder)
|
||||
self.assertEqual(self.order_ref, order.order_ref)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._get.call_args
|
||||
url = args[0]
|
||||
self.assertEqual(self.entity_href, url)
|
||||
|
||||
def test_should_delete(self):
|
||||
self.manager.delete(order_ref=self.entity_href)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._delete.call_args
|
||||
url = args[0]
|
||||
self.assertEqual(self.entity_href, url)
|
||||
self.assertEqual(self.order_ref, url)
|
||||
|
||||
def test_should_get_list(self):
|
||||
order_resp = self.order.get_dict(self.entity_href)
|
||||
self.api._get.return_value = {"orders":
|
||||
[order_resp for v in range(3)]}
|
||||
self.api._get.return_value = {
|
||||
"orders": [json.loads(self.key_order_data) for _ in range(3)]
|
||||
}
|
||||
|
||||
orders_list = self.manager.list(limit=10, offset=5)
|
||||
self.assertTrue(len(orders_list) == 3)
|
||||
self.assertIsInstance(orders_list[0], orders.Order)
|
||||
self.assertEqual(self.entity_href, orders_list[0].order_ref)
|
||||
self.assertIsInstance(orders_list[0], orders.KeyOrder)
|
||||
self.assertEqual(self.order_ref, orders_list[0].order_ref)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._get.call_args
|
||||
url = args[0]
|
||||
self.assertEqual(self.entity_base[:-1], url)
|
||||
self.assertEqual(self.api._base_url + '/orders', url)
|
||||
|
||||
# Verify that correct information was sent in the call.
|
||||
params = args[1]
|
||||
self.assertEqual(10, params['limit'])
|
||||
self.assertEqual(5, params['offset'])
|
||||
|
||||
def test_should_delete(self):
|
||||
self.manager.delete(order_ref=self.order_ref)
|
||||
|
||||
# Verify the correct URL was used to make the call.
|
||||
args, kwargs = self.api._delete.call_args
|
||||
url = args[0]
|
||||
self.assertEqual(self.order_ref, url)
|
||||
|
||||
def test_should_fail_delete_no_href(self):
|
||||
self.assertRaises(ValueError, self.manager.delete, None)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user