openstack/barbican
Code Issues Proposed changes

Merge "Make a whole host of modules hacking 0.9.2 compliant"

Browse Source
This commit is contained in:
Jenkins 2014-09-03 00:44:01 +00:00 committed by Gerrit Code Review
commit 344fba25f6
15 changed files with 147 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
barbican/plugin/kmip_secret_store.py
View File

@ -69,9 +69,11 @@ class KMIPSecretStore(ss.SecretStoreBase):
KMIP_ALGORITHM_ENUM = "kmip_algorithm_enum" KMIP_ALGORITHM_ENUM = "kmip_algorithm_enum"
def __init__(self, conf=CONF): def __init__(self, conf=CONF):
"""Initializes KMIPSecretStore by creating a dictionary of mappings """Initializes KMIPSecretStore
between SecretStore enum values and pyKMIP enum values. Initializes
the KMIP client with credentials needed to connect to the KMIP server. Creates a dictionary of mappings between SecretStore enum values
and pyKMIP enum values. Initializes the KMIP client with credentials
needed to connect to the KMIP server.
""" """
super(KMIPSecretStore, self).__init__() super(KMIPSecretStore, self).__init__()
self.valid_alg_dict = { self.valid_alg_dict = {
@ -180,14 +182,14 @@ class KMIPSecretStore(ss.SecretStoreBase):
template_attribute = kmip_objects.TemplateAttribute( template_attribute = kmip_objects.TemplateAttribute(
attributes=attribute_list) attributes=attribute_list)
secret_features = {} secret_features = {
'key_format_type': enums.KeyFormatType.RAW,
secret_features['key_format_type'] = enums.KeyFormatType.RAW 'key_value': {
secret_features['key_value'] =\ 'bytes': self._convert_base64_to_byte_array(secret_dto.secret)
{'bytes': self._convert_base64_to_byte_array(secret_dto.secret)} },
secret_features['cryptographic_algorithm'] = algorithm_value 'cryptographic_algorithm': algorithm_value,
secret_features['cryptographic_length'] =\ 'cryptographic_length': secret_dto.key_spec.bit_length
secret_dto.key_spec.bit_length }
secret = secrets.SecretFactory().create_secret(object_type, secret = secrets.SecretFactory().create_secret(object_type,
secret_features) secret_features)
@ -242,14 +244,16 @@ class KMIPSecretStore(ss.SecretStoreBase):
secret_type = self._map_type_kmip_to_ss( secret_type = self._map_type_kmip_to_ss(
result.object_type.enum) result.object_type.enum)
if type(secret_block.key_value.key_value) == \
kmip_objects.KeyValueStruct: key_value_type = type(secret_block.key_value.key_value)
if key_value_type == kmip_objects.KeyValueStruct:
secret_value = self._convert_byte_array_to_base64( secret_value = self._convert_byte_array_to_base64(
secret_block.key_value.key_value.key_material.value) secret_block.key_value.key_value.key_material.value)
elif type(secret_block.key_value.key_value) == \
kmip_objects.KeyValueString: elif key_value_type == kmip_objects.KeyValueString:
secret_value = self._convert_byte_array_to_base64( secret_value = self._convert_byte_array_to_base64(
secret_block.key_value.key_value.value) secret_block.key_value.key_value.value)
secret_alg = self._map_algorithm_kmip_to_ss( secret_alg = self._map_algorithm_kmip_to_ss(
secret_block.cryptographic_algorithm.value) secret_block.cryptographic_algorithm.value)
secret_bit_length = secret_block.cryptographic_length.value secret_bit_length = secret_block.cryptographic_length.value
@ -259,7 +263,7 @@ class KMIPSecretStore(ss.SecretStoreBase):
ss.KeySpec(secret_alg, secret_bit_length), ss.KeySpec(secret_alg, secret_bit_length),
'content_type', 'content_type',
transport_key=None) transport_key=None)
# TODO(kaitlin-farr) remove 'content-type' # TODO(kaitlin-farr) remove 'content-type'
LOG.debug("SUCCESS: Key retrieved with uuid: %s", LOG.debug("SUCCESS: Key retrieved with uuid: %s",
uuid) uuid)
return ret_secret_dto return ret_secret_dto
@ -282,15 +286,15 @@ class KMIPSecretStore(ss.SecretStoreBase):
:returns: boolean indicating if secret can be generated :returns: boolean indicating if secret can be generated
""" """
alg_dict_entry = self.valid_alg_dict.get(key_spec.alg.lower()) alg_dict_entry = self.valid_alg_dict.get(key_spec.alg.lower())
if alg_dict_entry and key_spec.bit_length in\ if (alg_dict_entry and key_spec.bit_length in
alg_dict_entry.get(KMIPSecretStore.VALID_BIT_LENGTHS): alg_dict_entry.get(KMIPSecretStore.VALID_BIT_LENGTHS)):
return True return True
return False return False
def delete_secret(self, secret_metadata): def delete_secret(self, secret_metadata):
"""Deletes the secret whose metadata is included in the dictionary. """Deletes the secret whose metadata is included in the dictionary.
Returns nothing if successful, raises an exception if an error occurs
Returns nothing if successful, raises an exception if an error occurs
:param secret_metadata: Dictionary of key metadata, requires: :param secret_metadata: Dictionary of key metadata, requires:
{'key_uuid': <uuid of key>} {'key_uuid': <uuid of key>}
:raises: SecretGeneralException :raises: SecretGeneralException
@ -330,26 +334,30 @@ class KMIPSecretStore(ss.SecretStoreBase):
return self.generate_supports(key_spec) return self.generate_supports(key_spec)
def _convert_base64_to_byte_array(self, base64_secret): def _convert_base64_to_byte_array(self, base64_secret):
"""Converts a base64 string to a byte array. KMIP transports secret """Converts a base64 string to a byte array.
values as byte arrays, so the key values must be converted to a byte
array for storage. KMIP transports secret values as byte arrays, so the key values
must be converted to a byte array for storage.
:param base64_secret: base64 value of key :param base64_secret: base64 value of key
:returns: bytearray of secret :returns: bytearray of secret
""" """
return bytearray(base64.b64decode(base64_secret)) return bytearray(base64.b64decode(base64_secret))
def _convert_byte_array_to_base64(self, byte_array): def _convert_byte_array_to_base64(self, byte_array):
"""Converts a byte array to a base64 string. KMIP transports secret """Converts a byte array to a base64 string.
values as byte arrays, so the key values must be converted to base64
strings upon getting a stored secret. KMIP transports secret values as byte arrays, so the key values
must be converted to base64 strings upon getting a stored secret.
:param byte_array: bytearray of key value :param byte_array: bytearray of key value
:returns: base64 string :returns: base64 string
""" """
return base64.b64encode(byte_array) return base64.b64encode(byte_array)
def _create_cryptographic_algorithm_attribute(self, alg): def _create_cryptographic_algorithm_attribute(self, alg):
"""Creates a KMIP Cryptographic Algorithm attribute. This attribute """Creates a KMIP Cryptographic Algorithm attribute.
is used when telling the KMIP server what kind of key to generate.
This attribute is used when telling the KMIP server what kind of
key to generate.
:param algorithm: A SecretStore KeyAlgorithm enum value :param algorithm: A SecretStore KeyAlgorithm enum value
:returns: A KMIP Cryptographic Algorithm attribute :returns: A KMIP Cryptographic Algorithm attribute
""" """
@ -364,10 +372,11 @@ class KMIPSecretStore(ss.SecretStoreBase):
return algorithm return algorithm
def _create_usage_mask_attribute(self): def _create_usage_mask_attribute(self):
"""Creates a KMIP Usage Mask attribute. For now, we assume the key """Creates a KMIP Usage Mask attribute.
will only be used for encryption and decryption. This attribute is
used when telling the KMIP server what kind of key to generate or For now, we assume the key will only be used for encryption and
store. decryption. This attribute is used when telling the KMIP server
what kind of key to generate or store.
:returns: A KMIP Usage Mask attribute with values ENCRYPT and DECRYPT :returns: A KMIP Usage Mask attribute with values ENCRYPT and DECRYPT
""" """
attribute_type = enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK attribute_type = enums.AttributeType.CRYPTOGRAPHIC_USAGE_MASK
@ -382,8 +391,10 @@ class KMIPSecretStore(ss.SecretStoreBase):
return usage_mask return usage_mask
def _create_cryptographic_length_attribute(self, bit_length): def _create_cryptographic_length_attribute(self, bit_length):
"""Creates a KMIP Cryptographic Length attribute. This attribute is """Creates a KMIP Cryptographic Length attribute.
used when telling the KMIP server what kind of key to generate.
This attribute is used when telling the KMIP server what kind of
key to generate.
:param bit_length: Bit length of the secret's algorithm :param bit_length: Bit length of the secret's algorithm
:returns: KMIP Cryptographic Length attribute :returns: KMIP Cryptographic Length attribute
""" """
@ -397,8 +408,10 @@ class KMIPSecretStore(ss.SecretStoreBase):
return length return length
def _map_type_ss_to_kmip(self, object_type): def _map_type_ss_to_kmip(self, object_type):
"""Map SecretType to KMIP type enum. Returns None if the type is not """Map SecretType to KMIP type enum
supported. The KMIP plugin only supports symmetric keys for now.
Returns None if the type is not supported. The KMIP plugin only
supports symmetric keys for now.
:param object_type: SecretType enum value :param object_type: SecretType enum value
:returns: KMIP type enum if supported, None if not supported :returns: KMIP type enum if supported, None if not supported
""" """
@ -408,9 +421,10 @@ class KMIPSecretStore(ss.SecretStoreBase):
return None return None
def _map_type_kmip_to_ss(self, object_type): def _map_type_kmip_to_ss(self, object_type):
"""Map KMIP type enum to SecretType enum. Returns None if the """Map KMIP type enum to SecretType enum
type is not supported. The KMIP plugin only supports symmetric keys
for now. Returns None if the type is not supported. The KMIP plugin only
supports symmetric keys for now.
:param object_type: KMIP type enum :param object_type: KMIP type enum
:returns: SecretType enum if type is supported, None if not supported :returns: SecretType enum if type is supported, None if not supported
""" """
@ -420,8 +434,9 @@ class KMIPSecretStore(ss.SecretStoreBase):
return None return None
def _map_algorithm_ss_to_kmip(self, algorithm): def _map_algorithm_ss_to_kmip(self, algorithm):
"""Map SecretStore enum value to the KMIP algorithm enum. Returns None """Map SecretStore enum value to the KMIP algorithm enum
if the algorithm is not supported.
Returns None if the algorithm is not supported.
:param algorithm: SecretStore algorithm enum value :param algorithm: SecretStore algorithm enum value
:returns: KMIP algorithm enum value if supported, None if not :returns: KMIP algorithm enum value if supported, None if not
supported supported
@ -433,8 +448,9 @@ class KMIPSecretStore(ss.SecretStoreBase):
return None return None
def _map_algorithm_kmip_to_ss(self, algorithm): def _map_algorithm_kmip_to_ss(self, algorithm):
"""Map KMIP algorithm enum to SecretStore algorithm enum. Returns None """Map KMIP algorithm enum to SecretStore algorithm enum
if the algorithm is not supported.
Returns None if the algorithm is not supported.
:param algorithm: KMIP algorithm enum :param algorithm: KMIP algorithm enum
:returns: SecretStore algorithm enum value if supported, None if not :returns: SecretStore algorithm enum value if supported, None if not
supported supported

17
barbican/plugin/symantec.py
View File

@ -16,9 +16,8 @@
""" """
Barbican certificate processing plugins and support. Barbican certificate processing plugins and support.
""" """
from requests import exceptions as request_exceptions
from oslo.config import cfg from oslo.config import cfg
from requests import exceptions as request_exceptions
from symantecssl.core import Symantec from symantecssl.core import Symantec
from symantecssl import exceptions as symantec_exceptions from symantecssl import exceptions as symantec_exceptions
@ -121,24 +120,24 @@ class SymantecCertificatePlugin(cert.CertificatePluginBase):
raise NotImplementedError # pragma: no cover raise NotImplementedError # pragma: no cover
def supports(self, certificate_spec): def supports(self, certificate_spec):
"""Returns a boolean indicating if the plugin supports the """Indicates if the plugin supports the certificate type.
certificate type.
:param certificate_spec: Contains details on the certificate to :param certificate_spec: Contains details on the certificate to
generate the certificate order generate the certificate order
:returns: boolean indicating if the plugin supports the certificate :returns: boolean indicating if the plugin supports the certificate
type type
""" """
#TODO(chellygel): Research what certificate types are supported by # TODO(chellygel): Research what certificate types are supported by
# symantec. Returning True for testing purposes # symantec. Returning True for testing purposes
return True return True
def _ca_create_order(self, order_meta, plugin_meta): def _ca_create_order(self, order_meta, plugin_meta):
"""Creates an order with the Symantec CA. The PartnerOrderId """Creates an order with the Symantec CA.
and GeoTrustOrderId are returned and stored in plugin_meta.
PartnerCode and ProductCode are also stored in plugin_meta for The PartnerOrderId and GeoTrustOrderId are returned and stored in
future use. plugin_meta. PartnerCode and ProductCode are also stored in plugin_meta
for future use.
All required order parameters must be stored as a dict in All required order parameters must be stored as a dict in
order_meta. order_meta.

14
barbican/tests/__init__.py
View File

@ -1,14 +0,0 @@
# Copyright (c) 2013-2014 Rackspace, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

4
barbican/tests/common/test_validators.py
View File

@ -805,8 +805,8 @@ class WhenTestingRSAContainerValidator(testtools.TestCase):
self.assertEqual('secret_refs', exception.invalid_property) self.assertEqual('secret_refs', exception.invalid_property)
def test_should_raise_duplicate_secret_id_in_secret_refs(self): def test_should_raise_duplicate_secret_id_in_secret_refs(self):
self.container_req['secret_refs'][0]['secret_ref'] = \ self.container_req['secret_refs'][0]['secret_ref'] = (
self.container_req['secret_refs'][2]['secret_ref'] self.container_req['secret_refs'][2]['secret_ref'])
exception = self.assertRaises( exception = self.assertRaises(
excep.InvalidObject, excep.InvalidObject,

9
barbican/tests/model/test_models.py
View File

@ -14,6 +14,7 @@
# limitations under the License. # limitations under the License.
import datetime import datetime
import testtools import testtools
from barbican.model import models from barbican.model import models
@ -101,13 +102,13 @@ class WhenCreatingNewContainer(testtools.TestCase):
self.parsed_container['secret_refs'][2]['secret_ref']) self.parsed_container['secret_refs'][2]['secret_ref'])
def test_parse_secret_ref_uri(self): def test_parse_secret_ref_uri(self):
self.parsed_container['secret_refs'][0]['secret_ref'] =\ self.parsed_container['secret_refs'][0]['secret_ref'] = (
'http://localhost:9110/123/secrets/123456' 'http://localhost:9110/123/secrets/123456')
container = models.Container(self.parsed_container) container = models.Container(self.parsed_container)
self.assertEqual(container.container_secrets[0].secret_id, '123456') self.assertEqual(container.container_secrets[0].secret_id, '123456')
self.parsed_container['secret_refs'][0]['secret_ref'] =\ self.parsed_container['secret_refs'][0]['secret_ref'] = (
'http://localhost:9110/123/secrets/123456/' 'http://localhost:9110/123/secrets/123456/')
container = models.Container(self.parsed_container) container = models.Container(self.parsed_container)
self.assertEqual(container.container_secrets[0].secret_id, '123456') self.assertEqual(container.container_secrets[0].secret_id, '123456')

20
barbican/tests/plugin/crypto/test_crypto.py
View File

@ -19,7 +19,6 @@ from Crypto.PublicKey import DSA
from Crypto.PublicKey import RSA from Crypto.PublicKey import RSA
from Crypto.Util import asn1 from Crypto.Util import asn1
from cryptography import fernet from cryptography import fernet
import mock import mock
import six import six
import testtools import testtools
@ -94,6 +93,7 @@ class WhenTestingSimpleCryptoPlugin(testtools.TestCase):
def test_encrypt_with_unicode_kek_must_pass(self): def test_encrypt_with_unicode_kek_must_pass(self):
"""Test plan: """Test plan:
Generate a kek Generate a kek
Encrypt with master kek Encrypt with master kek
Convert to unicode Convert to unicode
@ -335,10 +335,11 @@ class WhenTestingSimpleCryptoPlugin(testtools.TestCase):
generate_dto = plugin.GenerateDTO('rsa', 1024, None, 'changeme') generate_dto = plugin.GenerateDTO('rsa', 1024, None, 'changeme')
kek_meta_dto = self._get_mocked_kek_meta_dto() kek_meta_dto = self._get_mocked_kek_meta_dto()
private_dto, public_dto, passwd_dto = \ private_dto, public_dto, passwd_dto = self.plugin.generate_asymmetric(
self.plugin.generate_asymmetric(generate_dto, generate_dto,
kek_meta_dto, kek_meta_dto,
mock.MagicMock()) mock.MagicMock()
)
decrypt_dto = plugin.DecryptDTO(private_dto.cypher_text) decrypt_dto = plugin.DecryptDTO(private_dto.cypher_text)
private_dto = self.plugin.decrypt(decrypt_dto, private_dto = self.plugin.decrypt(decrypt_dto,
kek_meta_dto, kek_meta_dto,
@ -352,10 +353,11 @@ class WhenTestingSimpleCryptoPlugin(testtools.TestCase):
generate_dto = plugin.GenerateDTO('dsa', 1024, None, None) generate_dto = plugin.GenerateDTO('dsa', 1024, None, None)
kek_meta_dto = self._get_mocked_kek_meta_dto() kek_meta_dto = self._get_mocked_kek_meta_dto()
private_dto, public_dto, passwd_dto = \ private_dto, public_dto, passwd_dto = self.plugin.generate_asymmetric(
self.plugin.generate_asymmetric(generate_dto, generate_dto,
kek_meta_dto, kek_meta_dto,
mock.MagicMock()) mock.MagicMock()
)
decrypt_dto = plugin.DecryptDTO(private_dto.cypher_text) decrypt_dto = plugin.DecryptDTO(private_dto.cypher_text)
private_dto = self.plugin.decrypt(decrypt_dto, private_dto = self.plugin.decrypt(decrypt_dto,

5
barbican/tests/plugin/crypto/test_p11_crypto.py
View File

@ -14,7 +14,6 @@
# limitations under the License. # limitations under the License.
import mock import mock
import testtools import testtools
from barbican.model import models from barbican.model import models
@ -130,8 +129,8 @@ class WhenTestingP11CryptoPlugin(testtools.TestCase):
14, 15, 16] 14, 15, 16]
iv = self.plugin._generate_iv() iv = self.plugin._generate_iv()
self.assertEqual(len(iv), self.plugin.block_size) self.assertEqual(len(iv), self.plugin.block_size)
self.session.generateRandom.\ self.session.generateRandom.assert_called_once_with(
assert_called_once_with(self.plugin.block_size) self.plugin.block_size)
def test_generate_iv_with_invalid_response_size(self): def test_generate_iv_with_invalid_response_size(self):
self.session.generateRandom.return_value = [1, 2, 3, 4, 5, 6, 7] self.session.generateRandom.return_value = [1, 2, 3, 4, 5, 6, 7]

48
barbican/tests/plugin/test_dogtag.py
View File

@ -13,12 +13,12 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import mock
import os import os
import tempfile import tempfile
import testtools
import mock
from requests import exceptions as request_exceptions from requests import exceptions as request_exceptions
import testtools
try: try:
import barbican.plugin.dogtag as dogtag_import import barbican.plugin.dogtag as dogtag_import
@ -206,7 +206,7 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
self.order_id = mock.MagicMock() self.order_id = mock.MagicMock()
self.profile_id = mock.MagicMock() self.profile_id = mock.MagicMock()
#request generated # request generated
self.request = mock.MagicMock() self.request = mock.MagicMock()
self.request_id_mock = mock.MagicMock() self.request_id_mock = mock.MagicMock()
self.request.request_id = self.request_id_mock self.request.request_id = self.request_id_mock
@ -214,7 +214,7 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
self.cert_id_mock = mock.MagicMock() self.cert_id_mock = mock.MagicMock()
self.request.cert_id = self.cert_id_mock self.request.cert_id = self.cert_id_mock
#cert generated # cert generated
self.cert = mock.MagicMock() self.cert = mock.MagicMock()
self.cert_encoded_mock = mock.MagicMock() self.cert_encoded_mock = mock.MagicMock()
self.cert.encoded = self.cert_encoded_mock self.cert.encoded = self.cert_encoded_mock
@ -228,8 +228,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
self.modified_request = mock.MagicMock() self.modified_request = mock.MagicMock()
self.modified_request_id_mock = mock.MagicMock() self.modified_request_id_mock = mock.MagicMock()
self.modified_request.request_id = self.modified_request_id_mock self.modified_request.request_id = self.modified_request_id_mock
self.modified_request.request_status = \ self.modified_request.request_status = (
dogtag_cert.CertRequestStatus.COMPLETE dogtag_cert.CertRequestStatus.COMPLETE)
self.modified_request.cert_id = self.cert_id_mock self.modified_request.cert_id = self.cert_id_mock
def tearDown(self): def tearDown(self):
@ -409,8 +409,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
order_meta = {dogtag_import.DogtagCAPlugin.PROFILE_ID: self.profile_id} order_meta = {dogtag_import.DogtagCAPlugin.PROFILE_ID: self.profile_id}
plugin_meta = {} plugin_meta = {}
self.certclient_mock.enroll_cert.side_effect = \ self.certclient_mock.enroll_cert.side_effect = (
pki.BadRequestException("bad request") pki.BadRequestException("bad request"))
result_dto = self.plugin.issue_certificate_request( result_dto = self.plugin.issue_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)
@ -427,8 +427,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
order_meta = {dogtag_import.DogtagCAPlugin.PROFILE_ID: self.profile_id} order_meta = {dogtag_import.DogtagCAPlugin.PROFILE_ID: self.profile_id}
plugin_meta = {} plugin_meta = {}
self.certclient_mock.enroll_cert.side_effect = \ self.certclient_mock.enroll_cert.side_effect = (
pki.PKIException("generic enrollment error") pki.PKIException("generic enrollment error"))
self.assertRaises( self.assertRaises(
cm.CertificateGeneralException, cm.CertificateGeneralException,
@ -442,8 +442,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
order_meta = {dogtag_import.DogtagCAPlugin.PROFILE_ID: self.profile_id} order_meta = {dogtag_import.DogtagCAPlugin.PROFILE_ID: self.profile_id}
plugin_meta = {} plugin_meta = {}
self.certclient_mock.enroll_cert.side_effect = \ self.certclient_mock.enroll_cert.side_effect = (
request_exceptions.RequestException() request_exceptions.RequestException())
result_dto = self.plugin.issue_certificate_request( result_dto = self.plugin.issue_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)
@ -478,8 +478,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
order_meta = mock.ANY order_meta = mock.ANY
plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID: plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID:
self.request_id_mock} self.request_id_mock}
self.certclient_mock.review_request.side_effect = \ self.certclient_mock.review_request.side_effect = (
pki.RequestNotFoundException("request_not_found") pki.RequestNotFoundException("request_not_found"))
result_dto = self.plugin.cancel_certificate_request( result_dto = self.plugin.cancel_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)
@ -496,8 +496,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID: plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID:
self.request_id_mock} self.request_id_mock}
self.certclient_mock.review_request.return_value = self.review_response self.certclient_mock.review_request.return_value = self.review_response
self.certclient_mock.cancel_request.side_effect = \ self.certclient_mock.cancel_request.side_effect = (
pki.ConflictingOperationException("conflicting_operation") pki.ConflictingOperationException("conflicting_operation"))
result_dto = self.plugin.cancel_certificate_request( result_dto = self.plugin.cancel_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)
@ -514,8 +514,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
order_meta = mock.ANY order_meta = mock.ANY
plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID: plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID:
self.request_id_mock} self.request_id_mock}
self.certclient_mock.review_request.side_effect = \ self.certclient_mock.review_request.side_effect = (
request_exceptions.RequestException("request_exception") request_exceptions.RequestException("request_exception"))
result_dto = self.plugin.cancel_certificate_request( result_dto = self.plugin.cancel_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)
@ -689,8 +689,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
order_meta = mock.ANY order_meta = mock.ANY
plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID: plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID:
self.request_id_mock} self.request_id_mock}
self.certclient_mock.review_request.side_effect = \ self.certclient_mock.review_request.side_effect = (
pki.RequestNotFoundException("request_not_found") pki.RequestNotFoundException("request_not_found"))
result_dto = self.plugin.modify_certificate_request( result_dto = self.plugin.modify_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)
@ -707,8 +707,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID: plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID:
self.request_id_mock} self.request_id_mock}
self.certclient_mock.review_request.return_value = self.review_response self.certclient_mock.review_request.return_value = self.review_response
self.certclient_mock.cancel_request.side_effect = \ self.certclient_mock.cancel_request.side_effect = (
pki.ConflictingOperationException("conflicting_operation") pki.ConflictingOperationException("conflicting_operation"))
result_dto = self.plugin.modify_certificate_request( result_dto = self.plugin.modify_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)
@ -725,8 +725,8 @@ class WhenTestingDogtagCAPlugin(testtools.TestCase):
order_meta = mock.ANY order_meta = mock.ANY
plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID: plugin_meta = {dogtag_import.DogtagCAPlugin.REQUEST_ID:
self.request_id_mock} self.request_id_mock}
self.certclient_mock.review_request.side_effect = \ self.certclient_mock.review_request.side_effect = (
request_exceptions.RequestException("request_exception") request_exceptions.RequestException("request_exception"))
result_dto = self.plugin.modify_certificate_request( result_dto = self.plugin.modify_certificate_request(
self.order_id, order_meta, plugin_meta) self.order_id, order_meta, plugin_meta)

2
barbican/tests/plugin/test_store_crypto.py
View File

@ -12,10 +12,10 @@
# implied. # implied.
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import testtools
from barbican.plugin.interface import secret_store from barbican.plugin.interface import secret_store
from barbican.plugin import store_crypto from barbican.plugin import store_crypto
import testtools
class WhenStoreCryptoAdapterPlugin(testtools.TestCase): class WhenStoreCryptoAdapterPlugin(testtools.TestCase):

14
barbican/tests/queue/__init__.py
View File

@ -1,14 +0,0 @@
# Copyright (c) 2013-2014 Rackspace, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

4
barbican/tests/queue/test_server.py
View File

@ -33,8 +33,8 @@ class WhenUsingBeginOrderTask(utils.BaseTestCase):
self.tasks.process_order(context=None, self.tasks.process_order(context=None,
order_id=self.order_id, order_id=self.order_id,
keystone_id=self.keystone_id) keystone_id=self.keystone_id)
mock_begin_order.return_value.process\ mock_begin_order.return_value.process.assert_called_with(
.assert_called_with(self.order_id, self.keystone_id) self.order_id, self.keystone_id)
class WhenUsingTaskServer(utils.BaseTestCase): class WhenUsingTaskServer(utils.BaseTestCase):

14
barbican/tests/tasks/__init__.py
View File

@ -1,14 +0,0 @@
# Copyright (c) 2013-2014 Rackspace, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

4
barbican/tests/tasks/test_certificate_resources.py
View File

@ -101,8 +101,8 @@ class WhenIssuingCertificateRequests(testtools.TestCase):
) )
def test_should_return_ca_unavailable_for_request(self): def test_should_return_ca_unavailable_for_request(self):
self.result.status = cert_man.CertificateStatus.\ self.result.status = (
CA_UNAVAILABLE_FOR_REQUEST cert_man.CertificateStatus.CA_UNAVAILABLE_FOR_REQUEST)
cert_res.issue_certificate_request(self.order_model, cert_res.issue_certificate_request(self.order_model,
self.tenant_model, self.tenant_model,

61
barbican/tests/tasks/test_resources.py
View File

@ -54,8 +54,8 @@ class WhenBeginningOrder(testtools.TestCase):
self.order.secret_bit_length = self.secret_bit_length self.order.secret_bit_length = self.secret_bit_length
self.order.secret_mode = self.secret_mode self.order.secret_mode = self.secret_mode
self.order.secret_expiration = self.secret_expiration self.order.secret_expiration = self.secret_expiration
self.order.secret_payload_content_type = self\ self.order.secret_payload_content_type = (
.secret_payload_content_type self.secret_payload_content_type)
self.order_repo = mock.MagicMock() self.order_repo = mock.MagicMock()
self.order_repo.get.return_value = self.order self.order_repo.get.return_value = self.order
@ -88,19 +88,18 @@ class WhenBeginningOrder(testtools.TestCase):
self.resource.process(self.order.id, self.keystone_id) self.resource.process(self.order.id, self.keystone_id)
self.order_repo.get \ self.order_repo.get.assert_called_once_with(
.assert_called_once_with(entity_id=self.order.id, entity_id=self.order.id, keystone_id=self.keystone_id)
keystone_id=self.keystone_id)
self.assertEqual(self.order.status, models.States.ACTIVE) self.assertEqual(self.order.status, models.States.ACTIVE)
secret_info = self.order.to_dict_fields()['secret'] secret_info = self.order.to_dict_fields()['secret']
mock_generate_secret\ mock_generate_secret.assert_called_once_with(
.assert_called_once_with( secret_info,
secret_info, secret_info.get('payload_content_type',
secret_info.get('payload_content_type', 'application/octet-stream'),
'application/octet-stream'), self.tenant,
self.tenant, mock.ANY mock.ANY
) )
def test_should_raise_during_retrieval(self): def test_should_raise_during_retrieval(self):
# Force an error during the order retrieval phase. # Force an error during the order retrieval phase.
@ -244,19 +243,18 @@ class WhenBeginningKeyTypeOrder(testtools.TestCase):
mock_generate_secret.return_value = self.secret mock_generate_secret.return_value = self.secret
self.resource.process(self.order.id, self.keystone_id) self.resource.process(self.order.id, self.keystone_id)
self.order_repo.get \ self.order_repo.get.assert_called_once_with(
.assert_called_once_with(entity_id=self.order.id, entity_id=self.order.id, keystone_id=self.keystone_id)
keystone_id=self.keystone_id)
self.assertEqual(self.order.status, models.States.ACTIVE) self.assertEqual(self.order.status, models.States.ACTIVE)
secret_info = self.order.to_dict_fields()['meta'] secret_info = self.order.to_dict_fields()['meta']
mock_generate_secret\ mock_generate_secret.assert_called_once_with(
.assert_called_once_with( secret_info,
secret_info, secret_info.get('payload_content_type',
secret_info.get('payload_content_type', 'application/octet-stream'),
'application/octet-stream'), self.tenant,
self.tenant, mock.ANY mock.ANY
) )
def test_should_fail_during_retrieval(self): def test_should_fail_during_retrieval(self):
# Force an error during the order retrieval phase. # Force an error during the order retrieval phase.
@ -396,19 +394,18 @@ class WhenBeginningAsymmetricTypeOrder(testtools.TestCase):
mock_generate_asymmetric_secret.return_value = self.container mock_generate_asymmetric_secret.return_value = self.container
self.resource.process(self.order.id, self.keystone_id) self.resource.process(self.order.id, self.keystone_id)
self.order_repo.get \ self.order_repo.get.assert_called_once_with(
.assert_called_once_with(entity_id=self.order.id, entity_id=self.order.id, keystone_id=self.keystone_id)
keystone_id=self.keystone_id)
self.assertEqual(self.order.status, models.States.ACTIVE) self.assertEqual(self.order.status, models.States.ACTIVE)
secret_info = self.order.to_dict_fields()['meta'] secret_info = self.order.to_dict_fields()['meta']
mock_generate_asymmetric_secret\ mock_generate_asymmetric_secret.assert_called_once_with(
.assert_called_once_with( secret_info,
secret_info, secret_info.get('payload_content_type',
secret_info.get('payload_content_type', 'application/octet-stream'),
'application/octet-stream'), self.tenant,
self.tenant, mock.ANY mock.ANY
) )
def test_should_fail_during_retrieval(self): def test_should_fail_during_retrieval(self):
# Force an error during the order retrieval phase. # Force an error during the order retrieval phase.

6
barbican/tests/test_middleware_auth.py
View File

@ -33,14 +33,14 @@ expected_response = {"v1": "current", "build": "0.1.34dev"}
# uuid tokens are smaller and easier to test with # uuid tokens are smaller and easier to test with
# assume there is a "demo" user with only member role # assume there is a "demo" user with only member role
# curl -XPOST -d '{"auth":{"passwordCredentials":{"username": "demo", \ # curl -XPOST -d '{"auth":{"passwordCredentials":{"username": "demo",
# "password": "secret"}, "tenantName": "demo"}}' \ # "password": "secret"}, "tenantName": "demo"}}'
# -H "Content-type: application/json" http://localhost:35357/v2.0/tokens # -H "Content-type: application/json" http://localhost:35357/v2.0/tokens
# #
# pull out the token_id from above and use in ping_barbican # pull out the token_id from above and use in ping_barbican
# #
#TODO(malini) flesh this out # TODO(malini) flesh this out
def get_demo_token(password): def get_demo_token(password):
pass pass