12aa8a9339
Currently Barbican is not using oslo.db to set up database connection but it's own implementation directly using sqlalchemy. Because of this the database parameters were not updated and these are based on the names in quite old oslo.db library. This change updates the database options so that the name of these parameters become consistent with oslo.db. This would help us replace current own implementation by oslo.db in the future. Change-Id: I36926e62842780068f7e66564233c121c37565d0
286 lines
11 KiB
Python
286 lines
11 KiB
Python
# 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.
|
|
|
|
"""
|
|
Supports database/repositories oriented unit testing.
|
|
|
|
Warning: Do not merge this content with the utils.py module, as doing so will
|
|
break the DevStack functional test discovery process.
|
|
"""
|
|
import datetime
|
|
import oslotest.base as oslotest
|
|
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy import event
|
|
|
|
from barbican.model import models
|
|
from barbican.model import repositories
|
|
from barbican.tests import fixture as barbican_fixture
|
|
|
|
|
|
@event.listens_for(Engine, "connect")
|
|
def set_foreign_key_constraint(dbapi_connection, connection_record):
|
|
# Ensure that foreign key constraints are enforced during tests
|
|
dbapi_connection.execute("PRAGMA foreign_keys=ON")
|
|
|
|
|
|
def setup_in_memory_db():
|
|
# Ensure we are using in-memory SQLite database, and creating tables.
|
|
repositories.CONF.set_override("connection", "sqlite:///:memory:",
|
|
"database")
|
|
repositories.CONF.set_override("db_auto_create", True)
|
|
repositories.CONF.set_override("debug", True)
|
|
|
|
# Ensure the connection is completely closed, so any previous in-memory
|
|
# database can be removed prior to starting the next test run.
|
|
repositories.hard_reset()
|
|
|
|
# Start the in-memory database, creating required tables.
|
|
repositories.start()
|
|
|
|
|
|
def in_memory_cleanup():
|
|
repositories.clear()
|
|
|
|
|
|
def get_session():
|
|
return repositories.get_session()
|
|
|
|
|
|
def create_project(external_id="my keystone id", session=None):
|
|
project = models.Project()
|
|
project.external_id = external_id
|
|
project_repo = repositories.get_project_repository()
|
|
project_repo.create_from(project, session=session)
|
|
return project
|
|
|
|
|
|
def create_order(project=None, session=None, secret=None, container=None):
|
|
if not project:
|
|
project = create_project(session=session)
|
|
|
|
order = models.Order()
|
|
order.project_id = project.id
|
|
|
|
if secret:
|
|
order.secret_id = secret.id
|
|
if container:
|
|
order.container_id = container.id
|
|
|
|
order_repo = repositories.get_order_repository()
|
|
order_repo.create_from(order, session=session)
|
|
return order
|
|
|
|
|
|
def create_secret(project=None, session=None):
|
|
secret = models.Secret()
|
|
secret.project_id = project.id
|
|
secret_repo = repositories.get_secret_repository()
|
|
secret_repo.create_from(secret, session=session)
|
|
return secret
|
|
|
|
|
|
def create_transport_key(plugin_name="plugin", transport_key="tkey",
|
|
session=None):
|
|
transport_key = models.TransportKey(plugin_name, transport_key)
|
|
transport_key_repo = repositories.get_transport_key_repository()
|
|
transport_key_repo.create_from(transport_key, session=session)
|
|
return transport_key
|
|
|
|
|
|
def create_secret_metadatum(secret=None, key="key", value="value",
|
|
session=None):
|
|
secret_meta = models.SecretStoreMetadatum(key, value)
|
|
secret_meta.secret_id = secret.id
|
|
secret_meta_repo = repositories.get_secret_meta_repository()
|
|
secret_meta_repo.create_from(secret_meta, session=session)
|
|
return secret_meta
|
|
|
|
|
|
def create_secret_user_metadatum(secret=None, key="user_key",
|
|
value="user_value", session=None):
|
|
secret_user_metadatum = models.SecretUserMetadatum(key, value)
|
|
secret_user_metadatum.secret_id = secret.id
|
|
secret_user_metadatum_repo = repositories.get_secret_user_meta_repository()
|
|
secret_user_metadatum_repo.create_from(secret_user_metadatum,
|
|
session=session)
|
|
return secret_user_metadatum
|
|
|
|
|
|
def create_container(project=None, session=None):
|
|
container = models.Container()
|
|
container.project_id = project.id
|
|
container_repo = repositories.get_container_repository()
|
|
container_repo.create_from(container, session=session)
|
|
return container
|
|
|
|
|
|
def create_container_secret(container=None, secret=None, session=None):
|
|
container_secret = models.ContainerSecret()
|
|
container_secret.container_id = container.id
|
|
container_secret.secret_id = secret.id
|
|
container_secret_repo = repositories.get_container_secret_repository()
|
|
container_secret_repo.create_from(container_secret, session=session)
|
|
return container_secret
|
|
|
|
|
|
def create_kek_datum(project=None, plugin_name="plugin", session=None):
|
|
kek_datum = models.KEKDatum()
|
|
kek_datum.plugin_name = plugin_name
|
|
kek_datum.project_id = project.id
|
|
kek_datum_repo = repositories.get_kek_datum_repository()
|
|
kek_datum_repo.create_from(kek_datum, session=session)
|
|
return kek_datum
|
|
|
|
|
|
def create_encrypted_datum(secret=None, kek_datum=None, session=None):
|
|
enc_datum = models.EncryptedDatum()
|
|
enc_datum.secret_id = secret.id
|
|
enc_datum.kek_id = kek_datum.id
|
|
enc_datum_repo = repositories.get_encrypted_datum_repository()
|
|
enc_datum_repo.create_from(enc_datum, session=session)
|
|
return enc_datum
|
|
|
|
|
|
def create_order_meta_datum(order=None, key="key", value="value",
|
|
session=None):
|
|
order_meta_datum = models.OrderBarbicanMetadatum(key, value)
|
|
order_meta_datum.order_id = order.id
|
|
order_meta_datum_repo = repositories.get_order_barbican_meta_repository()
|
|
order_meta_datum_repo.create_from(order_meta_datum, session=session)
|
|
return order_meta_datum
|
|
|
|
|
|
def create_order_retry(order=None, retry_task="", retry_args=[],
|
|
retry_kwargs={}, retry_at=None, session=None):
|
|
order_retry = models.OrderRetryTask()
|
|
order_retry.retry_task = retry_task
|
|
order_retry.retry_args = retry_args
|
|
order_retry.retry_kwargs = retry_kwargs
|
|
if not retry_at:
|
|
order_retry.retry_at = datetime.datetime.utcnow()
|
|
order_retry.order_id = order.id
|
|
order_retry_task_repo = repositories.get_order_retry_tasks_repository()
|
|
order_retry_task_repo.create_from(order_retry, session)
|
|
return order_retry
|
|
|
|
|
|
def create_order_plugin_metadatum(order=None, key="key", value="value",
|
|
session=None):
|
|
order_plugin_metadatum = models.OrderPluginMetadatum(key, value)
|
|
order_plugin_metadatum.order_id = order.id
|
|
order_plugin_repo = repositories.get_order_plugin_meta_repository()
|
|
order_plugin_repo.create_from(order_plugin_metadatum, session=session)
|
|
return order_plugin_metadatum
|
|
|
|
|
|
def create_container_consumer_meta(container=None, parsed_request=None,
|
|
session=None):
|
|
if not parsed_request:
|
|
parsed_request = {"name": "name", "URL": "URL"}
|
|
container_consumer_meta = models.ContainerConsumerMetadatum(
|
|
container_id=container.id,
|
|
project_id=container.project_id,
|
|
parsed_request=parsed_request,
|
|
)
|
|
cont_cons_meta_repo = repositories.get_container_consumer_repository()
|
|
cont_cons_meta_repo.create_from(container_consumer_meta, session=session)
|
|
return container_consumer_meta
|
|
|
|
|
|
def create_secret_consumer(secret=None, service="service",
|
|
resource_id="resource_id",
|
|
resource_type="resource_type", session=None):
|
|
consumer = models.SecretConsumerMetadatum(
|
|
secret_id=secret.id,
|
|
project_id=secret.project_id,
|
|
service=service,
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
)
|
|
repo = repositories.get_secret_consumer_repository()
|
|
repo.create_from(consumer, session=session)
|
|
return consumer
|
|
|
|
|
|
def create_certificate_authority(project=None, parsed_ca_in=None,
|
|
session=None):
|
|
if not parsed_ca_in:
|
|
parsed_ca_in = {'plugin_name': 'plugin_name',
|
|
'plugin_ca_id': 'plugin_ca_id',
|
|
'expiration:': 'expiration',
|
|
'creator_id': 'creator_id',
|
|
'project_id': project.id}
|
|
certificate_authority = models.CertificateAuthority(
|
|
parsed_ca_in=parsed_ca_in)
|
|
cert_auth_repo = repositories.get_ca_repository()
|
|
cert_auth_repo.create_from(certificate_authority, session=session)
|
|
return certificate_authority
|
|
|
|
|
|
def create_preferred_cert_authority(cert_authority, session=None):
|
|
preferred_cert_authority = models.PreferredCertificateAuthority(
|
|
ca_id=cert_authority.id,
|
|
project_id=cert_authority.project_id)
|
|
preferred_ca_repo = repositories.get_preferred_ca_repository()
|
|
preferred_ca_repo.create_from(preferred_cert_authority, session=session)
|
|
return preferred_cert_authority
|
|
|
|
|
|
def create_project_cert_authority(certificate_authority=None, session=None):
|
|
project_cert_authority = models.ProjectCertificateAuthority(
|
|
ca_id=certificate_authority.id,
|
|
project_id=certificate_authority.project_id)
|
|
project_cert_repo = repositories.get_project_ca_repository()
|
|
project_cert_repo.create_from(project_cert_authority, session=session)
|
|
return project_cert_authority
|
|
|
|
|
|
def create_project_quotas(project=None, parsed_project_quotas=None,
|
|
session=None):
|
|
project_quota = models.ProjectQuotas(
|
|
project_id=project.id,
|
|
parsed_project_quotas=parsed_project_quotas)
|
|
project_quota_repo = repositories.get_project_quotas_repository()
|
|
project_quota_repo.create_from(project_quota, session=session)
|
|
return project_quota
|
|
|
|
|
|
def create_acl_secret(secret=None, user_ids=[], session=None):
|
|
acl_secret = models.SecretACL(secret.id, "read")
|
|
acl_secret.secret_id = secret.id
|
|
acl_secret_repo = repositories.get_secret_acl_repository()
|
|
acl_secret_repo.create_from(acl_secret, session=session)
|
|
return acl_secret
|
|
|
|
|
|
class RepositoryTestCase(oslotest.BaseTestCase):
|
|
"""Base test case class for in-memory database unit tests.
|
|
|
|
Database/Repository oriented unit tests should *not* modify the global
|
|
state in the barbican/model/repositories.py module, as this can lead to
|
|
hard to debug errors. Instead only utilize methods in this fixture.
|
|
|
|
Also, database-oriented unit tests extending this class MUST NO INVOKE
|
|
the repositories.start()/clear()/hard_reset() methods!*, otherwise *VERY*
|
|
hard to debug 'Broken Pipe' errors could result!
|
|
"""
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.useFixture(barbican_fixture.StandardLogging())
|
|
self.useFixture(barbican_fixture.WarningsFixture())
|
|
setup_in_memory_db()
|
|
|
|
# Clean up once tests are completed.
|
|
self.addCleanup(in_memory_cleanup)
|