Codebase list python-castellan / e2603c8
Merge "Update Barbican functional tests" Jenkins authored 8 years ago Gerrit Code Review committed 8 years ago
4 changed file(s) with 265 addition(s) and 94 deletion(s). Raw diff Collapse all Expand all
+33
-74
castellan/tests/functional/key_manager/test_barbican_key_manager.py less more
2020
2121 import uuid
2222
23 from barbicanclient import exceptions as barbican_exceptions
2423 from keystoneclient.v3 import client
24 from oslo_config import cfg
2525 from oslo_context import context
26 from oslotest import base
2627
2728 from castellan.common import exception
28 from castellan.common.objects import symmetric_key
2929 from castellan.key_manager import barbican_key_manager
3030 from castellan.tests.functional import config
3131 from castellan.tests.functional.key_manager import test_key_manager
3434 CONF = config.get_config()
3535
3636
37 class BarbicanKeyManagerTestCase(test_key_manager.KeyManagerTestCase):
37 class BarbicanKeyManagerTestCase(test_key_manager.KeyManagerTestCase,
38 base.BaseTestCase):
3839
3940 def _create_key_manager(self):
40 return barbican_key_manager.BarbicanKeyManager()
41 return barbican_key_manager.BarbicanKeyManager(cfg.CONF)
4142
4243 def setUp(self):
4344 super(BarbicanKeyManagerTestCase, self).setUp()
4849 keystone_client = client.Client(username=username,
4950 password=password,
5051 project_name=project_name,
51 auth_url=auth_url)
52 auth_url=auth_url,
53 project_domain_id='default')
54 project_list = keystone_client.projects.list(name=project_name)
55
5256 self.ctxt = context.RequestContext(
53 auth_token=keystone_client.auth_token)
57 auth_token=keystone_client.auth_token,
58 tenant=project_list[0].id)
5459
5560 def tearDown(self):
5661 super(BarbicanKeyManagerTestCase, self).tearDown()
57
58 def test_create_key(self):
59 key_uuid = self.key_mgr.create_key(self.ctxt,
60 algorithm='AES',
61 length=256)
62 self.addCleanup(self.key_mgr.delete_key, self.ctxt, key_uuid)
63 self.assertIsNotNone(key_uuid)
6462
6563 def test_create_null_context(self):
6664 self.assertRaises(exception.Forbidden,
6765 self.key_mgr.create_key, None, 'AES', 256)
6866
69 def test_delete_symmetric_key(self):
70 key_uuid = self.key_mgr.create_key(self.ctxt,
71 algorithm='AES',
72 length=256)
73 self.key_mgr.delete_key(self.ctxt, key_uuid)
74 try:
75 self.key_mgr.get_key(self.ctxt, key_uuid)
76 except barbican_exceptions.HTTPClientError as e:
77 self.assertEqual(404, e.status_code)
78 else:
79 self.fail('No exception when deleting non-existent key')
67 def test_create_key_pair_null_context(self):
68 self.assertRaises(exception.Forbidden,
69 self.key_mgr.create_key_pair, None, 'RSA', 2048)
8070
8171 def test_delete_null_context(self):
82 key_uuid = self.key_mgr.create_key(self.ctxt,
83 algorithm='AES',
84 length=256)
85 self.addCleanup(self.key_mgr.delete_key, self.ctxt, key_uuid)
72 key_uuid = self._get_valid_object_uuid(
73 test_key_manager._get_test_symmetric_key())
74 self.addCleanup(self.key_mgr.delete, self.ctxt, key_uuid)
8675 self.assertRaises(exception.Forbidden,
87 self.key_mgr.delete_key, None, key_uuid)
76 self.key_mgr.delete, None, key_uuid)
8877
89 def test_delete_null_key(self):
78 def test_delete_null_object(self):
9079 self.assertRaises(exception.KeyManagerError,
91 self.key_mgr.delete_key, self.ctxt, None)
80 self.key_mgr.delete, self.ctxt, None)
9281
93 def test_delete_unknown_key(self):
94 bad_key_uuid = str(uuid.uuid4())
95 self.assertRaises(barbican_exceptions.HTTPClientError,
96 self.key_mgr.delete_key, self.ctxt, bad_key_uuid)
97
98 def test_get_key(self):
99 secret_key = b'\x01\x02\xA0\xB3'
100 key = symmetric_key.SymmetricKey('AES', secret_key)
101
102 uuid = self.key_mgr.store_key(self.ctxt, key)
103 self.addCleanup(self.key_mgr.delete_key, self.ctxt, uuid)
104
105 retrieved_key = self.key_mgr.get_key(self.ctxt, uuid)
106 self.assertEqual(key.get_encoded(), retrieved_key.get_encoded())
82 def test_delete_unknown_object(self):
83 unknown_uuid = str(uuid.uuid4())
84 self.assertRaises(exception.ManagedObjectNotFoundError,
85 self.key_mgr.delete, self.ctxt, unknown_uuid)
10786
10887 def test_get_null_context(self):
109 key_uuid = self.key_mgr.create_key(self.ctxt,
110 algorithm='AES',
111 length=256)
88 key_uuid = self._get_valid_object_uuid(
89 test_key_manager._get_test_symmetric_key())
11290 self.assertRaises(exception.Forbidden,
113 self.key_mgr.get_key, None, key_uuid)
91 self.key_mgr.get, None, key_uuid)
11492
115 def test_get_null_key(self):
116 key_uuid = self.key_mgr.create_key(self.ctxt,
117 algorithm='AES',
118 length=256)
119 self.addCleanup(self.key_mgr.delete_key, self.ctxt, key_uuid)
93 def test_get_null_object(self):
12094 self.assertRaises(exception.KeyManagerError,
121 self.key_mgr.get_key, self.ctxt, None)
95 self.key_mgr.get, self.ctxt, None)
12296
12397 def test_get_unknown_key(self):
124 key_uuid = self.key_mgr.create_key(self.ctxt,
125 algorithm='AES',
126 length=256)
127 self.addCleanup(self.key_mgr.delete_key, self.ctxt, key_uuid)
12898 bad_key_uuid = str(uuid.uuid4())
129 self.assertRaises(barbican_exceptions.HTTPClientError,
130 self.key_mgr.get_key, self.ctxt, bad_key_uuid)
131
132 def test_store(self):
133 secret_key = b'\x01\x02\xA0\xB3'
134 key = symmetric_key.SymmetricKey('AES', secret_key)
135
136 uuid = self.key_mgr.store_key(self.ctxt, key)
137 self.addCleanup(self.key_mgr.delete_key, self.ctxt, uuid)
138
139 retrieved_key = self.key_mgr.get_key(self.ctxt, uuid)
140 self.assertEqual(key.get_encoded(), retrieved_key.get_encoded())
99 self.assertRaises(exception.ManagedObjectNotFoundError,
100 self.key_mgr.get, self.ctxt, bad_key_uuid)
141101
142102 def test_store_null_context(self):
143 secret_key = b'\x01\x02\xA0\xB3'
144 key = symmetric_key.SymmetricKey('AES', secret_key)
103 key = test_key_manager._get_test_symmetric_key()
145104
146105 self.assertRaises(exception.Forbidden,
147 self.key_mgr.store_key, None, key)
106 self.key_mgr.store, None, key)
+127
-4
castellan/tests/functional/key_manager/test_key_manager.py less more
1313 # under the License.
1414
1515 """
16 Test cases for the key manager.
16 Test cases for a key manager.
17
18 These test cases should pass against any key manager.
1719 """
1820
19 from castellan.tests import base
21 from castellan.common import exception
22 from castellan.common.objects import opaque_data
23 from castellan.common.objects import passphrase
24 from castellan.common.objects import private_key
25 from castellan.common.objects import public_key
26 from castellan.common.objects import symmetric_key
27 from castellan.common.objects import x_509
28 from castellan.tests import utils
2029
2130
22 class KeyManagerTestCase(base.TestCase):
31 def _get_test_symmetric_key():
32 key_bytes = bytes(utils.get_symmetric_key())
33 bit_length = 128
34 key = symmetric_key.SymmetricKey('AES', bit_length, key_bytes)
35 return key
36
37
38 def _get_test_public_key():
39 key_bytes = bytes(utils.get_public_key_der())
40 bit_length = 2048
41 key = public_key.PublicKey('RSA', bit_length, key_bytes)
42 return key
43
44
45 def _get_test_private_key():
46 key_bytes = bytes(utils.get_private_key_der())
47 bit_length = 2048
48 key = private_key.PrivateKey('RSA', bit_length, key_bytes)
49 return key
50
51
52 def _get_test_certificate():
53 data = bytes(utils.get_certificate_der())
54 cert = x_509.X509(data)
55 return cert
56
57
58 def _get_test_opaque_data():
59 data = bytes(b'opaque data')
60 opaque_object = opaque_data.OpaqueData(data)
61 return opaque_object
62
63
64 def _get_test_passphrase():
65 data = bytes(b'passphrase')
66 passphrase_object = passphrase.Passphrase(data)
67 return passphrase_object
68
69
70 @utils.parameterized_test_case
71 class KeyManagerTestCase(object):
2372
2473 def _create_key_manager(self):
2574 raise NotImplementedError()
2675
2776 def setUp(self):
2877 super(KeyManagerTestCase, self).setUp()
78 self.key_mgr = self._create_key_manager()
2979
30 self.key_mgr = self._create_key_manager()
80 def _get_valid_object_uuid(self, managed_object):
81 object_uuid = self.key_mgr.store(self.ctxt, managed_object)
82 self.assertIsNotNone(object_uuid)
83 return object_uuid
84
85 def test_create_key(self):
86 key_uuid = self.key_mgr.create_key(self.ctxt,
87 algorithm='AES',
88 length=256)
89 self.addCleanup(self.key_mgr.delete, self.ctxt, key_uuid)
90 self.assertIsNotNone(key_uuid)
91
92 def test_create_key_pair(self):
93 private_key_uuid, public_key_uuid = self.key_mgr.create_key_pair(
94 self.ctxt,
95 algorithm='RSA',
96 length=2048)
97
98 self.addCleanup(self.key_mgr.delete, self.ctxt, private_key_uuid)
99 self.addCleanup(self.key_mgr.delete, self.ctxt, public_key_uuid)
100
101 self.assertIsNotNone(private_key_uuid)
102 self.assertIsNotNone(public_key_uuid)
103 self.assertNotEqual(private_key_uuid, public_key_uuid)
104
105 @utils.parameterized_dataset({
106 'symmetric_key': [_get_test_symmetric_key()],
107 'public_key': [_get_test_public_key()],
108 'private_key': [_get_test_private_key()],
109 'certificate': [_get_test_certificate()],
110 'passphrase': [_get_test_passphrase()],
111 'opaque_data': [_get_test_opaque_data()],
112 })
113 def test_delete(self, managed_object):
114 object_uuid = self._get_valid_object_uuid(managed_object)
115 self.key_mgr.delete(self.ctxt, object_uuid)
116 try:
117 self.key_mgr.get(self.ctxt, object_uuid)
118 except exception.ManagedObjectNotFoundError:
119 pass
120 else:
121 self.fail('No exception when deleting non-existent key')
122
123 @utils.parameterized_dataset({
124 'symmetric_key': [_get_test_symmetric_key()],
125 'public_key': [_get_test_public_key()],
126 'private_key': [_get_test_private_key()],
127 'certificate': [_get_test_certificate()],
128 'passphrase': [_get_test_passphrase()],
129 'opaque_data': [_get_test_opaque_data()],
130 })
131 def test_get(self, managed_object):
132 uuid = self._get_valid_object_uuid(managed_object)
133 self.addCleanup(self.key_mgr.delete, self.ctxt, uuid)
134
135 retrieved_object = self.key_mgr.get(self.ctxt, uuid)
136 self.assertEqual(managed_object.get_encoded(),
137 retrieved_object.get_encoded())
138
139 @utils.parameterized_dataset({
140 'symmetric_key': [_get_test_symmetric_key()],
141 'public_key': [_get_test_public_key()],
142 'private_key': [_get_test_private_key()],
143 'certificate': [_get_test_certificate()],
144 'passphrase': [_get_test_passphrase()],
145 'opaque_data': [_get_test_opaque_data()],
146 })
147 def test_store(self, managed_object):
148 uuid = self.key_mgr.store(self.ctxt, managed_object)
149 self.addCleanup(self.key_mgr.delete, self.ctxt, uuid)
150
151 retrieved_object = self.key_mgr.get(self.ctxt, uuid)
152 self.assertEqual(managed_object.get_encoded(),
153 retrieved_object.get_encoded())
+2
-16
castellan/tests/unit/key_manager/test_barbican_key_manager.py less more
170170 self.key_mgr.delete(self.ctxt, self.key_id)
171171 self.delete.assert_called_once_with(self.secret_ref)
172172
173 def test_delete_none_key(self):
173 def test_delete_unknown_key(self):
174174 self.assertRaises(exception.KeyManagerError,
175175 self.key_mgr.delete, self.ctxt, None)
176
177 def test_delete_unkown_key(self):
178 side_effect = barbican_exceptions.HTTPClientError('key not found')
179 side_effect.status_code = 404
180 self.mock_barbican.secrets.delete = mock.Mock(side_effect=side_effect)
181 self.assertRaises(exception.ManagedObjectNotFoundError,
182 self.key_mgr.delete, self.ctxt, self.key_id)
183176
184177 def test_delete_with_error(self):
185178 self.mock_barbican.secrets.delete = mock.Mock(
206199 self.assertRaises(exception.Forbidden,
207200 self.key_mgr.get, None, self.key_id)
208201
209 def test_get_none_key(self):
202 def test_get_unknown_key(self):
210203 self.assertRaises(exception.KeyManagerError,
211204 self.key_mgr.get, self.ctxt, None)
212
213 def test_get_unknown_key(self):
214 side_effect = barbican_exceptions.HTTPClientError('key not found')
215 side_effect.status_code = 404
216 self.mock_barbican.secrets.get = mock.Mock(side_effect=side_effect)
217 self.assertRaises(exception.ManagedObjectNotFoundError,
218 self.key_mgr.get, self.ctxt, self.key_id)
219205
220206 def test_get_with_error(self):
221207 self.mock_barbican.secrets.get = mock.Mock(
+103
-0
castellan/tests/utils.py less more
1414
1515
1616 """These utility functions are borrowed from Barbican's testing utilities."""
17
18 import functools
19 import types
20
21 import six
22
23
24 def construct_new_test_function(original_func, name, build_params):
25 """Builds a new test function based on parameterized data.
26
27 :param original_func: The original test function that is used as a template
28 :param name: The fullname of the new test function
29 :param build_params: A dictionary or list containing args or kwargs
30 for the new test
31 :return: A new function object
32 """
33 new_func = types.FunctionType(
34 six.get_function_code(original_func),
35 six.get_function_globals(original_func),
36 name=name,
37 argdefs=six.get_function_defaults(original_func)
38 )
39
40 for key, val in six.iteritems(original_func.__dict__):
41 if key != 'build_data':
42 new_func.__dict__[key] = val
43
44 # Support either an arg list or kwarg dict for our data
45 build_args = build_params if isinstance(build_params, list) else []
46 build_kwargs = build_params if isinstance(build_params, dict) else {}
47
48 # Build a test wrapper to execute with our kwargs
49 def test_wrapper(func, test_args, test_kwargs):
50 @functools.wraps(func)
51 def wrapper(self):
52 return func(self, *test_args, **test_kwargs)
53 return wrapper
54
55 return test_wrapper(new_func, build_args, build_kwargs)
56
57
58 def process_parameterized_function(name, func_obj, build_data):
59 """Build lists of functions to add and remove to a test case."""
60 to_remove = []
61 to_add = []
62
63 for subtest_name, params in six.iteritems(build_data):
64 # Build new test function
65 func_name = '{0}_{1}'.format(name, subtest_name)
66 new_func = construct_new_test_function(func_obj, func_name, params)
67
68 # Mark the new function as needed to be added to the class
69 to_add.append((func_name, new_func))
70
71 # Mark key for removal
72 to_remove.append(name)
73
74 return to_remove, to_add
75
76
77 def parameterized_test_case(cls):
78 """Class decorator to process parameterized tests
79
80 This allows for parameterization to be used for potentially any
81 unittest compatible runner; including testr and py.test.
82 """
83 tests_to_remove = []
84 tests_to_add = []
85 for key, val in six.iteritems(vars(cls)):
86 # Only process tests with build data on them
87 if key.startswith('test_') and val.__dict__.get('build_data'):
88 to_remove, to_add = process_parameterized_function(
89 name=key,
90 func_obj=val,
91 build_data=val.__dict__.get('build_data')
92 )
93 tests_to_remove.extend(to_remove)
94 tests_to_add.extend(to_add)
95
96 # Add all new test functions
97 [setattr(cls, name, func) for name, func in tests_to_add]
98
99 # Remove all old test function templates (if they still exist)
100 [delattr(cls, key) for key in tests_to_remove if hasattr(cls, key)]
101 return cls
102
103
104 def parameterized_dataset(build_data):
105 """Simple decorator to mark a test method for processing."""
106 def decorator(func):
107 func.__dict__['build_data'] = build_data
108 return func
109 return decorator
17110
18111
19112 def get_certificate_der():
210303 b'\x01\x15\xcd\x52\x83\x3f\x06\x67\xfd\xa1\x2d\x2b\x07\xba\x32'
211304 b'\x62\x21\x07\x2f\x02\x03\x01\x00\x01')
212305 return key_der
306
307
308 def get_symmetric_key():
309 """Returns symmetric key bytes
310
311 16 bytes that were randomly generated. Form a 128 bit key.
312 """
313 symmetric_key = (
314 b'\x92\xcf\x1e\xd9\x54\xea\x30\x70\xd8\xc2\x48\xae\xc1\xc8\x72\xa3')
315 return symmetric_key