openstack/python-barbicanclient
Code Issues Proposed changes

Convert CRUD tests to requests-mock

Browse Source 
Use the requests-mock library to stub out the responses that are
returned from the socket rather than mocking out client specific
responses. This makes the tests more flexable as we just ensure that the
correct calls were made to the network and not that certain functions
were called.

Change-Id: I6d9851ef76285de29fc899241cff69f0b8d25af5
This commit is contained in:
Jamie Lennox
2015-01-19 17:54:53 +10:00
parent 3e86669c44
commit 928e561282
5 changed files with 225 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
barbicanclient/client.py
View File

@@ -151,7 +151,7 @@ class Client(object):
headers.update(self._default_headers)
resp = self._session.get(href, headers=headers)
self._check_status_code(resp)
return resp.content
return resp.text
def _delete(self, href, json=None):
headers = dict()

12
barbicanclient/test/test_client.py
View File

@@ -307,16 +307,16 @@ class BaseEntityResource(testtools.TestCase):
)
self.fail(msg)
def _setUp(self, entity):
def _setUp(self, entity, entity_id='abcd1234-eabc-5678-9abc-abcdef012345'):
super(BaseEntityResource, self).setUp()
self.responses = self.useFixture(fixture.Fixture())
self.endpoint = 'http://localhost:9311'
self.project_id = '1234567'
self.entity = entity
self.entity_base = self.endpoint + "/" + self.entity + "/"
self.entity_href = self.entity_base + \
'abcd1234-eabc-5678-9abc-abcdef012345'
self.entity_id = entity_id
self.entity_base = self.endpoint + "/v1/" + self.entity
self.entity_href = self.entity_base + "/" + self.entity_id
self.api = mock.MagicMock()
self.api._base_url = self.endpoint
self.client = client.Client(endpoint=self.endpoint,
project_id=self.project_id)

186
barbicanclient/test/test_client_containers.py
View File

@@ -12,6 +12,9 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import mock
from oslo_utils import timeutils
@@ -89,14 +92,13 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self._setUp('containers')
self.container = ContainerData()
self.api.secrets.Secret.return_value = self.container.secret
self.manager = containers.ContainerManager(self.api)
self.consumers_post_resource = (
self.entity_href.replace(self.endpoint + '/', '') + '/consumers'
)
self.consumers_delete_resource = (
self.entity_href + '/consumers'
)
self.client.secrets = mock.MagicMock()
self.client.secrets.get.return_value = self.container.secret
self.client.secrets._api = self.client
self.manager = self.client.containers
self.consumers_post_resource = self.entity_href + '/consumers/'
self.consumers_delete_resource = self.entity_href + '/consumers'
def test_should_generic_container_str(self):
container_obj = self.manager.create(name=self.container.name)
@@ -131,7 +133,8 @@ class WhenTestingContainers(test_client.BaseEntityResource):
repr(container_obj))
def test_should_store_generic_via_constructor(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create(
name=self.container.name,
@@ -141,19 +144,19 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_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)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
container_req = args[1]
container_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.container.name, container_req['name'])
self.assertEqual(self.container.type, container_req['type'])
self.assertEqual(self.container.generic_secret_refs_json,
container_req['secret_refs'])
def test_should_store_generic_via_attributes(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create()
container.name = self.container.name
@@ -163,19 +166,19 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_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)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
container_req = args[1]
container_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.container.name, container_req['name'])
self.assertEqual(self.container.type, container_req['type'])
self.assertItemsEqual(self.container.generic_secret_refs_json,
container_req['secret_refs'])
self.assertEqual(self.container.generic_secret_refs_json,
container_req['secret_refs'])
def test_should_store_certificate_via_attributes(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create_certificate()
container.name = self.container.name
@@ -188,19 +191,19 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_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)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
container_req = args[1]
container_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.container.name, container_req['name'])
self.assertEqual('certificate', container_req['type'])
self.assertItemsEqual(self.container.certificate_secret_refs_json,
container_req['secret_refs'])
def test_should_store_certificate_via_constructor(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create_certificate(
name=self.container.name,
@@ -213,19 +216,19 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_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)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
container_req = args[1]
container_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.container.name, container_req['name'])
self.assertEqual('certificate', container_req['type'])
self.assertItemsEqual(self.container.certificate_secret_refs_json,
container_req['secret_refs'])
def test_should_store_rsa_via_attributes(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create_rsa()
container.name = self.container.name
@@ -237,19 +240,19 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_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)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
container_req = args[1]
container_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.container.name, container_req['name'])
self.assertEqual('rsa', container_req['type'])
self.assertItemsEqual(self.container.rsa_secret_refs_json,
container_req['secret_refs'])
def test_should_store_rsa_via_constructor(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create_rsa(
name=self.container.name,
@@ -262,19 +265,19 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, container_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)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
container_req = args[1]
container_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.container.name, container_req['name'])
self.assertEqual('rsa', container_req['type'])
self.assertItemsEqual(self.container.rsa_secret_refs_json,
container_req['secret_refs'])
def test_should_get_secret_refs_when_created_using_secret_objects(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create(
name=self.container.name,
@@ -285,8 +288,11 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.container.generic_secret_refs)
def test_should_reload_attributes_after_store(self):
self.api._post.return_value = {'container_ref': self.entity_href}
self.api._get.return_value = self.container.get_dict(self.entity_href)
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
data = self.container.get_dict(self.entity_href)
self.responses.get(self.entity_href, json=data)
container = self.manager.create(
name=self.container.name,
@@ -321,7 +327,8 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container.add(self.container.secret.name, self.container.secret)
def test_should_be_immutable_after_store(self):
self.api._post.return_value = {'container_ref': self.entity_href}
data = {'container_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
container = self.manager.create(
name=self.container.name,
@@ -359,31 +366,27 @@ class WhenTestingContainers(test_client.BaseEntityResource):
pass
def test_should_get_generic_container(self):
self.api._get.return_value = self.container.get_dict(self.entity_href)
data = self.container.get_dict(self.entity_href)
self.responses.get(self.entity_href, json=data)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_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)
self.assertEqual(self.entity_href, self.responses.last_request.url)
self.assertIsNotNone(container.secrets)
def test_should_get_certificate_container(self):
self.api._get.return_value = self.container.get_dict(
self.entity_href, type='certificate'
)
data = self.container.get_dict(self.entity_href, type='certificate')
self.responses.get(self.entity_href, json=data)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_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)
self.assertEqual(self.entity_href, self.responses.last_request.url)
# Verify the returned type is correct
self.assertIsInstance(container, containers.CertificateContainer)
@@ -393,17 +396,15 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertIsNotNone(container.intermediates)
def test_should_get_rsa_container(self):
self.api._get.return_value = self.container.get_dict(self.entity_href,
type='rsa')
data = self.container.get_dict(self.entity_href, type='rsa')
self.responses.get(self.entity_href, json=data)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_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)
self.assertEqual(self.entity_href, self.responses.last_request.url)
# Verify the returned type is correct
self.assertIsInstance(container, containers.RSAContainer)
@@ -412,15 +413,17 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertIsNotNone(container.private_key_passphrase)
def test_should_delete_from_manager(self):
self.responses.delete(self.entity_href, status_code=204)
self.manager.delete(container_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.entity_href, self.responses.last_request.url)
def test_should_delete_from_object(self):
self.api._get.return_value = self.container.get_dict(self.entity_href)
data = self.container.get_dict(self.entity_href)
m = self.responses.get(self.entity_href, json=data)
n = self.responses.delete(self.entity_href, status_code=204)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsNotNone(container.container_ref)
@@ -428,15 +431,20 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container.delete()
# 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.assertTrue(m.called)
self.assertTrue(n.called)
# Verify that the Container no longer has a container_ref
self.assertIsNone(container.container_ref)
def test_should_store_after_delete_from_object(self):
self.api._get.return_value = self.container.get_dict(self.entity_href)
data = self.container.get_dict(self.entity_href)
self.responses.get(self.entity_href, json=data)
data = self.container.get_dict(self.entity_href)
self.responses.post(self.entity_base + '/', json=data)
m = self.responses.delete(self.entity_href, status_code=204)
container = self.manager.get(container_ref=self.entity_href)
self.assertIsNotNone(container.container_ref)
@@ -444,9 +452,7 @@ class WhenTestingContainers(test_client.BaseEntityResource):
container.delete()
# 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.entity_href, m.last_request.url)
# Verify that the Container no longer has a container_ref
self.assertIsNone(container.container_ref)
@@ -458,8 +464,8 @@ class WhenTestingContainers(test_client.BaseEntityResource):
def test_should_get_list(self):
container_resp = self.container.get_dict(self.entity_href)
self.api._get.return_value = {"containers":
[container_resp for v in range(3)]}
data = {"containers": [container_resp for v in range(3)]}
self.responses.get(self.entity_base, json=data)
containers_list = self.manager.list(limit=10, offset=5)
self.assertTrue(len(containers_list) == 3)
@@ -467,14 +473,12 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, containers_list[0].container_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.entity_base,
self.responses.last_request.url.split('?')[0])
# Verify that correct information was sent in the call.
params = args[1]
self.assertEqual(10, params['limit'])
self.assertEqual(5, params['offset'])
self.assertEqual(['10'], self.responses.last_request.qs['limit'])
self.assertEqual(['5'], self.responses.last_request.qs['offset'])
def test_should_fail_get_invalid_container(self):
self.assertRaises(ValueError, self.manager.get,
@@ -484,9 +488,10 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertRaises(ValueError, self.manager.delete, None)
def test_should_register_consumer(self):
self.api._post.return_value = self.container.get_dict(
self.entity_href, consumers=[self.container.consumer]
)
data = self.container.get_dict(self.entity_href,
consumers=[self.container.consumer])
self.responses.post(self.entity_href + '/consumers/', json=data)
container = self.manager.register_consumer(
self.entity_href, self.container.consumer.get('name'),
self.container.consumer.get('URL')
@@ -494,27 +499,26 @@ class WhenTestingContainers(test_client.BaseEntityResource):
self.assertIsInstance(container, containers.Container)
self.assertEqual(self.entity_href, container.container_ref)
args, kwargs = self.api._post.call_args
url, body = args[0], args[1]
self.assertEqual(self.consumers_post_resource, url)
body = json.loads(self.responses.last_request.text)
self.assertEqual(self.consumers_post_resource,
self.responses.last_request.url)
self.assertEqual(self.container.consumer, body)
self.assertEqual([self.container.consumer], container.consumers)
def test_should_remove_consumer(self):
self.responses.delete(self.entity_href + '/consumers', status_code=204)
self.manager.remove_consumer(
self.entity_href, self.container.consumer.get('name'),
self.container.consumer.get('URL')
)
args, kwargs = self.api._delete.call_args
url = args[0]
body = kwargs['json']
self.assertEqual(self.consumers_delete_resource, url)
body = json.loads(self.responses.last_request.text)
self.assertEqual(self.consumers_delete_resource,
self.responses.last_request.url)
self.assertEqual(self.container.consumer, body)
def test_should_get_total(self):
self.api._get.return_value = {'total': 1}
self.responses.get(self.entity_base, json={'total': 1})
total = self.manager.total()
self.assertEqual(total, 1)

92
barbicanclient/test/test_client_orders.py
View File

@@ -23,13 +23,13 @@ from barbicanclient.test import test_client
from barbicanclient.test import test_client_secrets as test_secrets
class OrdersTestCase(testtools.TestCase):
class OrdersTestCase(test_client.BaseEntityResource):
def setUp(self):
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._setUp('orders', entity_id='d0460cc4-2876-4493-b7de-fc5c812883cc')
self.secret_ref = (self.endpoint +
'/secrets/a2292306-6da0-4f60-bd8a-84fc8d692716')
self.key_order_data = """{{
"status": "ACTIVE",
"secret_ref": "{0}",
@@ -45,10 +45,9 @@ class OrdersTestCase(testtools.TestCase):
"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)
}}""".format(self.secret_ref, self.entity_href)
self.manager = self.client.orders
def _get_order_args(self, order_data):
order_args = json.loads(order_data)
@@ -71,10 +70,11 @@ class WhenTestingKeyOrders(OrdersTestCase):
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))
self.assertIn('order_ref=' + self.entity_href, repr(order_obj))
def test_should_be_immutable_after_submit(self):
self.api._post.return_value = {'order_ref': self.order_ref}
data = {'order_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
order = self.manager.create_key(
name='name',
@@ -83,7 +83,7 @@ class WhenTestingKeyOrders(OrdersTestCase):
)
order_href = order.submit()
self.assertEqual(self.order_ref, order_href)
self.assertEqual(self.entity_href, order_href)
# Verify that attributes are immutable after store.
attributes = [
@@ -98,7 +98,8 @@ class WhenTestingKeyOrders(OrdersTestCase):
pass
def test_should_submit_via_constructor(self):
self.api._post.return_value = {'order_ref': self.order_ref}
data = {'order_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
order = self.manager.create_key(
name='name',
@@ -107,15 +108,14 @@ class WhenTestingKeyOrders(OrdersTestCase):
)
order_href = order.submit()
self.assertEqual(self.order_ref, order_href)
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('orders', entity_resp)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
order_req = args[1]
order_req = json.loads(self.responses.last_request.text)
self.assertEqual('name', order_req['meta']['name'])
self.assertEqual('algorithm',
order_req['meta']['algorithm'])
@@ -123,7 +123,8 @@ class WhenTestingKeyOrders(OrdersTestCase):
order_req['meta']['payload_content_type'])
def test_should_submit_via_attributes(self):
self.api._post.return_value = {'order_ref': self.order_ref}
data = {'order_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
order = self.manager.create_key()
order.name = 'name'
@@ -131,15 +132,14 @@ class WhenTestingKeyOrders(OrdersTestCase):
order.payload_content_type = 'payload_content_type'
order_href = order.submit()
self.assertEqual(self.order_ref, order_href)
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('orders', entity_resp)
self.assertEqual(self.entity_base + '/',
self.responses.last_request.url)
# Verify that correct information was sent in the call.
order_req = args[1]
order_req = json.loads(self.responses.last_request.text)
self.assertEqual('name', order_req['meta']['name'])
self.assertEqual('algorithm',
order_req['meta']['algorithm'])
@@ -165,7 +165,8 @@ class WhenTestingKeyOrders(OrdersTestCase):
class WhenTestingAsymmetricOrders(OrdersTestCase):
def test_should_be_immutable_after_submit(self):
self.api._post.return_value = {'order_ref': self.order_ref}
data = {'order_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
order = self.manager.create_asymmetric(
name='name',
@@ -174,7 +175,7 @@ class WhenTestingAsymmetricOrders(OrdersTestCase):
)
order_href = order.submit()
self.assertEqual(self.order_ref, order_href)
self.assertEqual(self.entity_href, order_href)
# Verify that attributes are immutable after store.
attributes = [
@@ -196,49 +197,44 @@ class WhenTestingAsymmetricOrders(OrdersTestCase):
class WhenTestingOrderManager(OrdersTestCase):
def test_should_get(self):
self.api._get.return_value = json.loads(self.key_order_data)
self.responses.get(self.entity_href, text=self.key_order_data)
order = self.manager.get(order_ref=self.order_ref)
order = self.manager.get(order_ref=self.entity_href)
self.assertIsInstance(order, orders.KeyOrder)
self.assertEqual(self.order_ref, order.order_ref)
self.assertEqual(self.entity_href, 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.order_ref, url)
self.assertEqual(self.entity_href, self.responses.last_request.url)
def test_should_get_list(self):
self.api._get.return_value = {
"orders": [json.loads(self.key_order_data) for _ in range(3)]
}
data = {"orders": [json.loads(self.key_order_data) for _ in range(3)]}
self.responses.get(self.entity_base, json=data)
orders_list = self.manager.list(limit=10, offset=5)
self.assertTrue(len(orders_list) == 3)
self.assertIsInstance(orders_list[0], orders.KeyOrder)
self.assertEqual(self.order_ref, orders_list[0].order_ref)
self.assertEqual(self.entity_href, 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.api._base_url + '/orders', url)
self.assertEqual(self.entity_base,
self.responses.last_request.url.split('?')[0])
# Verify that correct information was sent in the call.
params = args[1]
self.assertEqual(10, params['limit'])
self.assertEqual(5, params['offset'])
self.assertEqual(['10'], self.responses.last_request.qs['limit'])
self.assertEqual(['5'], self.responses.last_request.qs['offset'])
def test_should_delete(self):
self.manager.delete(order_ref=self.order_ref)
self.responses.delete(self.entity_href, status_code=204)
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.order_ref, url)
self.assertEqual(self.entity_href, self.responses.last_request.url)
def test_should_fail_delete_no_href(self):
self.assertRaises(ValueError, self.manager.delete, None)
def test_should_get_total(self):
self.api._get.return_value = {'total': 1}
self.responses.get(self.entity_base, json={'total': 1})
total = self.manager.total()
self.assertEqual(total, 1)

163
barbicanclient/test/test_client_secrets.py
View File

@@ -13,7 +13,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from oslo_utils import timeutils
import json
from oslo.utils import timeutils
from barbicanclient.test import test_client
from barbicanclient import secrets, base
@@ -48,8 +50,7 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self._setUp('secrets')
self.secret = SecretData()
self.manager = secrets.SecretManager(self.api)
self.manager = self.client.secrets
def test_should_entity_str(self):
secret_obj = self.manager.create(name=self.secret.name)
@@ -60,7 +61,8 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertIn('name="{0}"'.format(self.secret.name), repr(secret_obj))
def test_should_store_via_constructor(self):
self.api._post.return_value = {'secret_ref': self.entity_href}
data = {'secret_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
secret = self.manager.create(name=self.secret.name,
payload=self.secret.payload,
@@ -68,20 +70,16 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
secret_href = secret.store()
self.assertEqual(self.entity_href, secret_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.
secret_req = args[1]
secret_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.secret.name, secret_req['name'])
self.assertEqual(self.secret.payload, secret_req['payload'])
self.assertEqual(self.secret.payload_content_type,
secret_req['payload_content_type'])
def test_should_store_via_attributes(self):
self.api._post.return_value = {'secret_ref': self.entity_href}
data = {'secret_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
secret = self.manager.create()
secret.name = self.secret.name
@@ -90,20 +88,16 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
secret_href = secret.store()
self.assertEqual(self.entity_href, secret_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.
secret_req = args[1]
secret_req = json.loads(self.responses.last_request.text)
self.assertEqual(self.secret.name, secret_req['name'])
self.assertEqual(self.secret.payload, secret_req['payload'])
self.assertEqual(self.secret.payload_content_type,
secret_req['payload_content_type'])
def test_should_be_immutable_after_submit(self):
self.api._post.return_value = {'secret_ref': self.entity_href}
data = {'secret_ref': self.entity_href}
self.responses.post(self.entity_base + '/', json=data)
secret = self.manager.create(name=self.secret.name,
payload=self.secret.payload,
@@ -139,26 +133,29 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
pass
def test_should_get_lazy(self):
self.api._get.return_value = self.secret.get_dict(self.entity_href)
data = self.secret.get_dict(self.entity_href)
m = self.responses.get(self.entity_href, json=data)
secret = self.manager.get(secret_ref=self.entity_href)
self.assertIsInstance(secret, secrets.Secret)
self.assertEqual(self.entity_href, secret.secret_ref)
# Verify GET wasn't called yet
self.assertFalse(self.api._get.called)
self.assertFalse(m.called)
# Check an attribute to trigger lazy-load
self.assertEqual(self.secret.name, secret.name)
# Verify the correct URL was used to make the GET call
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
self.assertEqual(self.entity_href, m.last_request.url)
def test_should_get_payload_only(self):
self.api._get.return_value = self.secret.get_dict(self.entity_href)
self.api._get_raw.return_value = self.secret.payload
m = self.responses.get(self.entity_href,
request_headers={'Accept': 'application/json'},
json=self.secret.get_dict(self.entity_href))
n = self.responses.get(self.entity_href,
request_headers={'Accept': 'text/plain'},
text=self.secret.payload)
secret = self.manager.get(
secret_ref=self.entity_href,
@@ -168,64 +165,68 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, secret.secret_ref)
# Verify `get` wasn't called yet (metadata)
self.assertFalse(self.api._get.called)
self.assertFalse(m.called)
# Verify `get_raw` wasn't called yet (payload)
self.assertFalse(self.api._get_raw.called)
self.assertFalse(n.called)
# GET payload (with payload_content_type)
self.assertEqual(self.secret.payload, secret.payload)
# Verify `get` still wasn't called (metadata)
self.assertFalse(self.api._get.called)
self.assertFalse(m.called)
# Verify `get_raw` was called (payload)
self.assertTrue(self.api._get_raw.called)
self.assertTrue(n.called)
# Verify the correct URL was used to make the `get_raw` call
args, kwargs = self.api._get_raw.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
self.assertEqual(self.entity_href, n.last_request.url)
def test_should_fetch_metadata_to_get_payload_if_no_content_type_set(self):
content_types_dict = {'default': 'application/octet-stream'}
self.api._get.return_value = self.secret.get_dict(
self.entity_href, content_types_dict=content_types_dict)
self.api._get_raw.return_value = self.secret.payload
data = self.secret.get_dict(self.entity_href,
content_types_dict=content_types_dict)
m = self.responses.get(self.entity_href,
request_headers={'Accept': 'application/json'},
json=data)
request_headers = {'Accept': 'application/octet-stream'}
n = self.responses.get(self.entity_href,
request_headers=request_headers,
text=self.secret.payload)
secret = self.manager.get(secret_ref=self.entity_href)
self.assertIsInstance(secret, secrets.Secret)
self.assertEqual(self.entity_href, secret.secret_ref)
# Verify `get` wasn't called yet (metadata)
self.assertFalse(self.api._get.called)
self.assertFalse(m.called)
# Verify `get_raw` wasn't called yet (payload)
self.assertFalse(self.api._get_raw.called)
self.assertFalse(n.called)
# GET payload (with no payload_content_type) trigger lazy-load
self.assertEqual(self.secret.payload, secret.payload)
# Verify `get` was called (metadata)
self.assertTrue(self.api._get.called)
self.assertTrue(m.called)
# Verify `get_raw` was called (payload)
self.assertTrue(self.api._get_raw.called)
self.assertTrue(n.called)
# Verify the correct URL was used to make the `get` calls
args, kwargs = self.api._get.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
args, kwargs = self.api._get_raw.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
self.assertEqual(self.entity_href, m.last_request.url)
self.assertEqual(self.entity_href, n.last_request.url)
def test_should_decrypt_with_content_type(self):
self.api._get.return_value = self.secret.get_dict(self.entity_href)
decrypted = 'decrypted text here'
self.api._get_raw.return_value = decrypted
request_headers = {'Accept': 'application/octet-stream'}
m = self.responses.get(self.entity_href,
request_headers=request_headers,
text=decrypted)
secret = self.manager.get(
secret_ref=self.entity_href,
@@ -235,51 +236,44 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertEqual(decrypted, secret_payload)
# Verify the correct URL was used to make the call.
args, kwargs = self.api._get_raw.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
# Verify that correct information was sent in the call.
headers = args[1]
self.assertEqual('application/octet-stream', headers['Accept'])
self.assertEqual(self.entity_href, m.last_request.url)
def test_should_decrypt_without_content_type(self):
content_types_dict = {'default': 'application/octet-stream'}
self.api._get.return_value = self.secret.get_dict(self.entity_href,
content_types_dict)
json = self.secret.get_dict(self.entity_href, content_types_dict)
m = self.responses.get(self.entity_href,
request_headers={'Accept': 'application/json'},
json=json)
decrypted = 'decrypted text here'
self.api._get_raw.return_value = decrypted
request_headers = {'Accept': 'application/octet-stream'}
n = self.responses.get(self.entity_href,
request_headers=request_headers,
text=decrypted)
secret = self.manager.get(secret_ref=self.entity_href)
secret_payload = secret.payload
self.assertEqual(decrypted, secret_payload)
# 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)
self.assertEqual(self.entity_href, m.last_request.url)
# Verify the correct URL was used to make the call.
args, kwargs = self.api._get_raw.call_args
url = args[0]
self.assertEqual(self.entity_href, url)
# Verify that correct information was sent in the call.
headers = args[1]
self.assertEqual('application/octet-stream', headers['Accept'])
self.assertEqual(self.entity_href, n.last_request.url)
def test_should_delete(self):
self.responses.delete(self.entity_href, status_code=204)
self.manager.delete(secret_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.entity_href, self.responses.last_request.url)
def test_should_get_list(self):
secret_resp = self.secret.get_dict(self.entity_href)
self.api._get.return_value = {"secrets":
[secret_resp for v in range(3)]}
data = {"secrets": [secret_resp for v in range(3)]}
m = self.responses.get(self.entity_base, json=data)
secrets_list = self.manager.list(limit=10, offset=5)
self.assertTrue(len(secrets_list) == 3)
@@ -287,22 +281,22 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertEqual(self.entity_href, secrets_list[0].secret_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.entity_base,
m.last_request.url.split('?')[0])
# Verify that correct information was sent in the call.
params = args[1]
self.assertEqual(10, params['limit'])
self.assertEqual(5, params['offset'])
self.assertEqual(['10'], m.last_request.qs['limit'])
self.assertEqual(['5'], m.last_request.qs['offset'])
def test_should_fail_get_invalid_secret(self):
self.assertRaises(ValueError, self.manager.get,
**{'secret_ref': '12345'})
def test_should_fail_decrypt_no_content_types(self):
self.api._get.return_value = self.secret.get_dict(self.entity_href)
data = self.secret.get_dict(self.entity_href)
self.responses.get(self.entity_href, json=data)
secret = self.manager.get(secret_ref=self.entity_href)
try:
secret.payload
self.fail("didn't raise a ValueError exception")
@@ -311,8 +305,9 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
def test_should_fail_decrypt_no_default_content_type(self):
content_types_dict = {'no-default': 'application/octet-stream'}
self.api._get.return_value = self.secret.get_dict(self.entity_href,
content_types_dict)
data = self.secret.get_dict(self.entity_href, content_types_dict)
self.responses.get(self.entity_href, json=data)
secret = self.manager.get(secret_ref=self.entity_href)
try:
secret.payload
@@ -324,6 +319,6 @@ class WhenTestingSecrets(test_client.BaseEntityResource):
self.assertRaises(ValueError, self.manager.delete, None)
def test_should_get_total(self):
self.api._get.return_value = {'total': 1}
self.responses.get(self.entity_base, json={'total': 1})
total = self.manager.total()
self.assertEqual(total, 1)