# Path: castellan/common/exception.py
# _FATAL_EXCEPTION_FORMAT_ERRORS = False
# class RedirectException(Exception):
# class CastellanException(Exception):
# class Forbidden(CastellanException):
# class KeyManagerError(CastellanException):
# class ManagedObjectNotFoundError(CastellanException):
# class InvalidManagedObjectDictError(CastellanException):
# class UnknownManagedObjectTypeError(CastellanException):
# class AuthTypeInvalidError(CastellanException):
# class InsufficientCredentialDataError(CastellanException):
#     def __init__(self, url):
#     def __init__(self, message_arg=None, *args, **kwargs):
# 
# Path: castellan/common/objects/symmetric_key.py
# class SymmetricKey(key.Key):
#     def __init__(self, algorithm, bit_length, key,
#                  name=None, created=None, id=None):
#     def managed_type(cls):
#     def algorithm(self):
#     def format(self):
#     def get_encoded(self):
#     def bit_length(self):
#     def __eq__(self, other):
#     def __ne__(self, other):
# 
# Path: castellan/key_manager/barbican_key_manager.py
# _BARBICAN_OPT_GROUP = 'barbican'
# _BARBICAN_SERVICE_USER_OPT_GROUP = 'barbican_service_user'
# LOG = logging.getLogger(__name__)
# class BarbicanKeyManager(key_manager.KeyManager):
#     def __init__(self, configuration):
#     def _get_barbican_client(self, context):
#     def _get_keystone_auth(self, context):
#     def _get_barbican_endpoint(self, auth, sess):
#     def _create_base_url(self, auth, sess, endpoint):
#     def create_key(self, context, algorithm, length,
#                    expiration=None, name=None):
#     def create_key_pair(self, context, algorithm, length,
#                         expiration=None, name=None):
#     def _get_barbican_object(self, barbican_client, managed_object):
#     def _get_normalized_payload(self, encoded_bytes, secret_type):
#     def store(self, context, managed_object, expiration=None):
#     def _create_secret_ref(self, object_id):
#     def _get_active_order(self, barbican_client, order_ref):
#     def _retrieve_secret_uuid(self, secret_ref):
#     def _get_secret_data(self, secret):
#     def _get_castellan_object(self, secret, metadata_only=False):
#     def _get_secret(self, context, object_id):
#     def _is_secret_not_found_error(self, error):
#     def get(self, context, managed_object_id, metadata_only=False):
#     def delete(self, context, managed_object_id):
#     def list(self, context, object_type=None, metadata_only=False):
#     def list_options_for_discovery(self):
# 
# Path: castellan/tests/unit/key_manager/test_key_manager.py
# CONF = cfg.CONF
# class KeyManagerTestCase(base.TestCase):
# class DefaultKeyManagerImplTestCase(KeyManagerTestCase):
#     def _create_key_manager(self):
#     def setUp(self):
#     def _create_key_manager(self):
#     def test_default_key_manager(self):


import calendar
from unittest import mock
from barbicanclient import exceptions as barbican_exceptions
from keystoneauth1 import identity
from keystoneauth1 import service_token
from oslo_context import context
from oslo_utils import timeutils
from castellan.common import exception
from castellan.common.objects import symmetric_key as sym_key
from castellan.key_manager import barbican_key_manager
from castellan.tests.unit.key_manager import test_key_manager
                                    secret_name)

        # Define the return values
        secret = mock.Mock()
        self.create.return_value = secret
        secret.store.return_value = self.secret_ref

        # Store the Key
        returned_uuid = self.key_mgr.store(self.ctxt, _key)

        self.create.assert_called_once_with(algorithm='AES',
                                            bit_length=key_length,
                                            payload=secret_key,
                                            name=secret_name,
                                            secret_type='symmetric')
        self.assertEqual(self.key_id, returned_uuid)

    def test_store_null_context(self):
        self.key_mgr._barbican_client = None
        self.assertRaises(exception.Forbidden,
                          self.key_mgr.store, None, None)

    def test_store_with_error(self):
        self.mock_barbican.secrets.create = mock.Mock(
            side_effect=barbican_exceptions.HTTPClientError('test error'))
        secret_key = bytes(b'\x01\x02\xA0\xB3')
        key_length = len(secret_key) * 8
        _key = sym_key.SymmetricKey('AES',
                                    key_length,
                                    secret_key)

# Path: castellan/common/exception.py
# _FATAL_EXCEPTION_FORMAT_ERRORS = False
# class RedirectException(Exception):
# class CastellanException(Exception):
# class Forbidden(CastellanException):
# class KeyManagerError(CastellanException):
# class ManagedObjectNotFoundError(CastellanException):
# class InvalidManagedObjectDictError(CastellanException):
# class UnknownManagedObjectTypeError(CastellanException):
# class AuthTypeInvalidError(CastellanException):
# class InsufficientCredentialDataError(CastellanException):
#     def __init__(self, url):
#     def __init__(self, message_arg=None, *args, **kwargs):
# 
# Path: castellan/common/objects/symmetric_key.py
# class SymmetricKey(key.Key):
#     def __init__(self, algorithm, bit_length, key,
#                  name=None, created=None, id=None):
#     def managed_type(cls):
#     def algorithm(self):
#     def format(self):
#     def get_encoded(self):
#     def bit_length(self):
#     def __eq__(self, other):
#     def __ne__(self, other):
# 
# Path: castellan/key_manager/barbican_key_manager.py
# _BARBICAN_OPT_GROUP = 'barbican'
# _BARBICAN_SERVICE_USER_OPT_GROUP = 'barbican_service_user'
# LOG = logging.getLogger(__name__)
# class BarbicanKeyManager(key_manager.KeyManager):
#     def __init__(self, configuration):
#     def _get_barbican_client(self, context):
#     def _get_keystone_auth(self, context):
#     def _get_barbican_endpoint(self, auth, sess):
#     def _create_base_url(self, auth, sess, endpoint):
#     def create_key(self, context, algorithm, length,
#                    expiration=None, name=None):
#     def create_key_pair(self, context, algorithm, length,
#                         expiration=None, name=None):
#     def _get_barbican_object(self, barbican_client, managed_object):
#     def _get_normalized_payload(self, encoded_bytes, secret_type):
#     def store(self, context, managed_object, expiration=None):
#     def _create_secret_ref(self, object_id):
#     def _get_active_order(self, barbican_client, order_ref):
#     def _retrieve_secret_uuid(self, secret_ref):
#     def _get_secret_data(self, secret):
#     def _get_castellan_object(self, secret, metadata_only=False):
#     def _get_secret(self, context, object_id):
#     def _is_secret_not_found_error(self, error):
#     def get(self, context, managed_object_id, metadata_only=False):
#     def delete(self, context, managed_object_id):
#     def list(self, context, object_type=None, metadata_only=False):
#     def list_options_for_discovery(self):
# 
# Path: castellan/tests/unit/key_manager/test_key_manager.py
# CONF = cfg.CONF
# class KeyManagerTestCase(base.TestCase):
# class DefaultKeyManagerImplTestCase(KeyManagerTestCase):
#     def _create_key_manager(self):
#     def setUp(self):
#     def _create_key_manager(self):
#     def test_default_key_manager(self):


# Path: castellan/tests/unit/key_manager/test_barbican_key_manager.py
import calendar
from unittest import mock
from barbicanclient import exceptions as barbican_exceptions
from keystoneauth1 import identity
from keystoneauth1 import service_token
from oslo_context import context
from oslo_utils import timeutils
from castellan.common import exception
from castellan.common.objects import symmetric_key as sym_key
from castellan.key_manager import barbican_key_manager
from castellan.tests.unit.key_manager import test_key_manager

                                    secret_name)

        # Define the return values
        secret = mock.Mock()
        self.create.return_value = secret
        secret.store.return_value = self.secret_ref

        # Store the Key
        returned_uuid = self.key_mgr.store(self.ctxt, _key)

        self.create.assert_called_once_with(algorithm='AES',
                                            bit_length=key_length,
                                            payload=secret_key,
                                            name=secret_name,
                                            secret_type='symmetric')
        self.assertEqual(self.key_id, returned_uuid)

    def test_store_null_context(self):
        self.key_mgr._barbican_client = None
        self.assertRaises(exception.Forbidden,
                          self.key_mgr.store, None, None)

    def test_store_with_error(self):
        self.mock_barbican.secrets.create = mock.Mock(
            side_effect=barbican_exceptions.HTTPClientError('test error'))
        secret_key = bytes(b'\x01\x02\xA0\xB3')
        key_length = len(secret_key) * 8
        _key = sym_key.SymmetricKey('AES',
                                    key_length,
                                    secret_key)

        self.assertRaises(exception.KeyManagerError,