x/tuskar
Code Issues Proposed changes

SQLAlchemy Storage Driver

Browse Source 
This patch adds the local database driver for Tuskar template storage.
It includes some better testing of the base API which is easier against
a real driver.

Change-Id: Icc14e9920a611ec24754a44d02178e62a327854b
Implements: blueprint tripleo-juno-tuskar-template-storage
This commit is contained in:
Dougal Matthews 2014-08-07 08:26:11 +01:00
parent f4367afb5f
commit 8787800692
7 changed files with 837 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
etc/tuskar/tuskar.conf.sample
View File

@ -415,6 +415,14 @@
#matchmaker_heartbeat_ttl=600
#
# Options defined in tuskar.storage.drivers.sqlalchemy
#
# MySQL engine (string value)
#mysql_engine=InnoDB
[database]
#
@ -722,6 +730,6 @@
# Storage driver to store Deployment Plans and Heat
# Orchestration Templates (string value)
#driver=mock.Mock
#driver=tuskar.storage.drivers.sqlalchemy.SQLAlchemyDriver

54
tuskar/db/sqlalchemy/migrate_repo/versions/002_add_stored_file.py Normal file
View File

@ -0,0 +1,54 @@
#
# 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.
from sqlalchemy import Column, DateTime, Integer, MetaData, String, Table
from tuskar.openstack.common.gettextutils import _ # noqa
from tuskar.openstack.common import log as logging
LOG = logging.getLogger(__name__)
ENGINE = 'InnoDB'
CHARSET = 'utf8'
def upgrade(migrate_engine):
meta = MetaData()
meta.bind = migrate_engine
stored_file = Table(
'stored_file',
meta,
Column('uuid', String(length=36), primary_key=True, nullable=False),
Column('contents', String(), nullable=False),
Column('object_type', String(length=20), nullable=False),
Column('name', String(length=64), nullable=True),
Column('version', Integer(), nullable=True),
Column('created_at', DateTime),
Column('updated_at', DateTime),
mysql_engine=ENGINE,
mysql_charset=CHARSET,
)
try:
LOG.info(repr(stored_file))
stored_file.create()
except Exception:
LOG.info(repr(stored_file))
LOG.exception(_('Exception while creating table.'))
raise
def downgrade(migrate_engine):
raise NotImplementedError('Downgrade is unsupported.')

27
tuskar/db/sqlalchemy/models.py
View File

@ -210,3 +210,30 @@ class Overcloud(Base):
d['counts'] = count_dicts
return d
class StoredFile(Base):
"""Tuskar Stored File
The StoredFile model is used by the tuskar.storage package and more
specifically for the SQLAlchemy storage driver. Simply put it is a
collection of text files with some metadata.
"""
__tablename__ = "stored_file"
#: UUID's are used as the unique identifier.
uuid = Column(String(length=36), primary_key=True)
#: contents contains the full file contents as a string.
contents = Column(String(), nullable=False)
#: Object type flags the type of file that this is, i.e. template or
#: environment file.
object_type = Column(String(length=20), nullable=False)
#: Names provide a short human readable description of a file.
name = Column(String(length=64), nullable=True)
#: Versions are an automatic incrementing count.
version = Column(Integer(), nullable=True)

3
tuskar/storage/__init__.py
View File

@ -16,11 +16,10 @@ from oslo.config import cfg
from tuskar.openstack.common import log as logging
# TODO(dmatthew): Switch this to the default driver when it is added.
heat_opts = [
cfg.StrOpt(
'driver',
default='mock.Mock',
default='tuskar.storage.drivers.sqlalchemy.SQLAlchemyDriver',
help=('Storage driver to store Deployment Plans and Heat '
'Orchestration Templates')
)

335
tuskar/storage/drivers/sqlalchemy.py Normal file
View File

@ -0,0 +1,335 @@
# -*- encoding: utf-8 -*-
#
# 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.
from __future__ import absolute_import
from uuid import uuid4
from oslo.config import cfg
from sqlalchemy import func
from sqlalchemy.orm.exc import NoResultFound
from tuskar.db.sqlalchemy.models import StoredFile
from tuskar.openstack.common.db.sqlalchemy import session as db_session
from tuskar.storage.drivers.base import BaseDriver
from tuskar.storage.exceptions import NameAlreadyUsed
from tuskar.storage.exceptions import UnknownName
from tuskar.storage.exceptions import UnknownUUID
from tuskar.storage.exceptions import UnknownVersion
from tuskar.storage.models import StoredFile as StorageModel
sql_opts = [
cfg.StrOpt('mysql_engine',
default='InnoDB',
help='MySQL engine')
]
cfg.CONF.register_opts(sql_opts)
def get_session():
return db_session.get_session(sqlite_fk=True)
class SQLAlchemyDriver(BaseDriver):
def _generate_uuid(self):
return str(uuid4())
def _to_storage_model(self, store, result):
"""Convert a result from SQLAlchemy into an instance of the common
model used in the tuskar.storage.
:param store: Instance of the storage store
:type store: tuskat.storage.stores._BaseStore
:param result: Instance of the SQLAlchemy model as returned by a query.
:type result: tuskar.db.sqlalchemy.models.StoredFile
:return: Instance of the StoredFile class.
:rtype: tuskar.storage.models.StoredFile
"""
file_dict = result.as_dict()
file_dict.pop('object_type')
file_dict['store'] = store
return StorageModel(**file_dict)
def _upsert(self, store, stored_file):
session = get_session()
session.begin()
try:
session.add(stored_file)
session.commit()
return self._to_storage_model(store, stored_file)
finally:
session.close()
def _get_latest_version(self, store, name):
session = get_session()
try:
return session.query(
func.max(StoredFile.version)
).filter_by(
object_type=store.object_type, name=name
).scalar()
finally:
session.close()
def _create(self, store, name, contents, version):
stored_file = StoredFile(
uuid=self._generate_uuid(),
contents=contents,
object_type=store.object_type,
name=name,
version=version
)
return self._upsert(store, stored_file)
def create(self, store, name, contents):
"""Given the store, name and contents create a new file and return a
`StoredFile` instance representing it.
Some of the stored items such as environment files do not have names.
When working with these, name must be passed explicitly as None. This
is why the name has a type of "str or None" below.
:param store: The store class, used for routing the storage.
:type store: tuskar.storage.stores._BaseStore
:param name: name of the object to store (optional)
:type name: str or None
:param contents: String containing the file contents
:type contents: str
:return: StoredFile instance containing the file metadata and contents
:rtype: tuskar.storage.models.StoredFile
"""
if store.versioned:
version = 1
else:
version = None
if name is not None:
try:
self.retrieve_by_name(store, name)
msg = "A file with the name '{0}' already exists".format(name)
raise NameAlreadyUsed(msg)
except UnknownName:
pass
return self._create(store, name, contents, version)
def _retrieve(self, object_type, uuid):
session = get_session()
try:
return session.query(StoredFile).filter_by(
uuid=uuid,
object_type=object_type
).one()
except NoResultFound:
msg = "No results found for the UUID: {0}".format(uuid)
raise UnknownUUID(msg)
finally:
session.close()
def retrieve(self, store, uuid):
"""Returns the stored file for a given store that matches the provided
UUID.
:param store: The store class, used for routing the storage.
:type store: tuskar.storage.stores._BaseStore
:param uuid: UUID of the object to retrieve.
:type uuid: str
:return: StoredFile instance containing the file metadata and contents
:rtype: tuskar.storage.models.StoredFile
:raises: tuskar.storage.exceptions.UnknownUUID if the UUID can't be
found
"""
stored_file = self._retrieve(store.object_type, uuid)
return self._to_storage_model(store, stored_file)
def update(self, store, uuid, contents):
"""Given the store, uuid, name and contents update the existing stored
file and return an instance of StoredFile that reflects the updates.
Either name and/or contents can be provided. If they are not then they
will remain unchanged.
:param store: The store class, used for routing the storage.
:type store: tuskar.storage.stores._BaseStore
:param uuid: UUID of the object to update.
:type uuid: str
:param name: name of the object to store (optional)
:type name: str
:param contents: String containing the file contents (optional)
:type contents: str
:return: StoredFile instance containing the file metadata and contents
:rtype: tuskar.storage.models.StoredFile
:raises: tuskar.storage.exceptions.UnknownUUID if the UUID can't be
found
"""
stored_file = self._retrieve(store.object_type, uuid)
stored_file.contents = contents
if store.versioned:
version = self._get_latest_version(store, stored_file.name) + 1
return self._create(
store, stored_file.name, stored_file.contents, version)
return self._upsert(store, stored_file)
def delete(self, store, uuid):
"""Delete the stored file with the UUID under the given store.
:param store: The store class, used for routing the storage.
:type store: tuskar.storage.stores._BaseStore
:param uuid: UUID of the object to update.
:type uuid: str
:return: Returns nothing on success. Exceptions are expected for errors
:rtype: None
:raises: tuskar.storage.exceptions.UnknownUUID if the UUID can't be
found
"""
session = get_session()
session.begin()
stored_file = self._retrieve(store.object_type, uuid)
try:
session.delete(stored_file)
session.commit()
finally:
session.close()
def list(self, store, only_latest=False):
"""Return a list of all the stored objects for a given store.
Optionally only_latest can be set to True to return only the most
recent version of each objects (grouped by name).
:param store: The store class, used for routing the storage.
:type store: tuskar.storage.stores._BaseStore
:param only_latest: If set to True only the latest versions of each
object will be returned.
:type only_latest: bool
:return: List of StoredFile instances
:rtype: [tuskar.storage.models.StoredFile]
"""
object_type = store.object_type
session = get_session()
try:
files = session.query(StoredFile).filter_by(
object_type=object_type
)
if only_latest:
stmt = session.query(
StoredFile.uuid,
func.max(StoredFile.version).label("version")
).subquery()
files = files.filter(
StoredFile.version == stmt.c.version,
StoredFile.uuid == stmt.c.uuid
)
return [self._to_storage_model(store, file_) for file_ in files]
finally:
session.close()
def retrieve_by_name(self, store, name, version=None):
"""Returns the stored file for a given store that matches the provided
name and optionally version.
:param store: The store class, used for routing the storage.
:type store: tuskar.storage.stores._BaseStore
:param name: name of the object to retrieve.
:type name: str
:param version: Version of the object to retrieve. If the version isn't
provided, the latest will be returned.
:type version: int
:return: StoredFile instance containing the file metadata and contents
:rtype: tuskar.storage.models.StoredFile
:raises: tuskar.storage.exceptions.UnknownName if the name can't be
found
:raises: tuskar.storage.exceptions.UnknownVersion if the version can't
be found
"""
object_type = store.object_type
session = get_session()
try:
query = session.query(StoredFile).filter_by(
name=name,
object_type=object_type,
)
if version is not None:
query = query.filter_by(version=version)
else:
query = query.filter_by(
version=self._get_latest_version(store, name)
)
stored_file = query.one()
return self._to_storage_model(store, stored_file)
except NoResultFound:
name_query = session.query(StoredFile).filter_by(
name=name,
object_type=object_type,
)
if name_query.count() == 0:
msg = "No results found for the Name: {0}".format(name)
raise UnknownName(msg)
elif name_query.filter_by(version=version).count() == 0:
msg = "No results found for the Version: {0}".format(version)
raise UnknownVersion(msg)
raise
finally:
session.close()

279
tuskar/tests/storage/drivers/test_sqlalchemy.py Normal file
View File

@ -0,0 +1,279 @@
# -*- encoding: utf-8 -*-
#
# 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.
from functools import partial
from mock import Mock
from mock import patch
from sqlalchemy.orm.exc import NoResultFound
from tuskar.common import context as tuskar_context
from tuskar.storage.drivers.sqlalchemy import SQLAlchemyDriver
from tuskar.storage.exceptions import UnknownName
from tuskar.storage.exceptions import UnknownUUID
from tuskar.storage.exceptions import UnknownVersion
from tuskar.storage.stores import DeploymentPlanStore
from tuskar.storage.stores import TemplateStore
from tuskar.tests import base
class SQLAlchemyDriverTestCase(base.TestCase):
def setUp(self):
super(SQLAlchemyDriverTestCase, self).setUp()
self.context = tuskar_context.get_admin_context()
self.driver = SQLAlchemyDriver()
self.store = TemplateStore(self.driver)
@patch('tuskar.storage.drivers.sqlalchemy.SQLAlchemyDriver._generate_uuid')
def test_create(self, mock_uuid):
# Setup
expected_uuid = 'b4b85dc2-0b0a-48ed-a56c-e4d582fd1473'
mock_uuid.return_value = expected_uuid
# Test
result = self.driver.create(self.store, "swift.yaml", "YAML")
# Verify
self.assertEqual(result.uuid, expected_uuid)
self.assertEqual(result.version, 1)
@patch('tuskar.storage.drivers.sqlalchemy.SQLAlchemyDriver._generate_uuid')
def test_create_no_versioning(self, mock_uuid):
# Setup
store = DeploymentPlanStore(self.driver)
expected_uuid = 'b4b85dc2-0b0a-48ed-a56c-e4d582fd1473'
mock_uuid.return_value = expected_uuid
# Test
result = self.driver.create(store, "swift.yaml", "YAML")
# Verify
self.assertEqual(result.uuid, expected_uuid)
self.assertEqual(result.version, None)
@patch('tuskar.storage.drivers.sqlalchemy.SQLAlchemyDriver._generate_uuid')
def test_retrieve(self, mock_uuid):
# Setup
expected_uuid = 'b4b85dc2-0b0a-48ed-a56c-e4d582fd1473'
expected_name = "swift.yaml"
expected_contents = "YAML"
mock_uuid.return_value = expected_uuid
self.driver.create(self.store, expected_name, expected_contents)
# Test
result = self.driver.retrieve(self.store, expected_uuid)
# Verify
self.assertEqual(result.uuid, expected_uuid)
self.assertEqual(result.name, expected_name)
self.assertEqual(result.contents, expected_contents)
def test_retrieve_invalid(self):
# Setup
retrieve_call = partial(
self.driver.retrieve,
self.store, "uuid"
)
# Test & Verify
self.assertRaises(UnknownUUID, retrieve_call)
def test_update(self):
# Setup
expected_name = "swift.yaml"
original_contents = "YAML"
created = self.driver.create(
self.store, expected_name, original_contents)
# Test
new_contents = "YAML2"
updated = self.driver.update(self.store, created.uuid, new_contents)
# Verify
retrieved = self.driver.retrieve(self.store, created.uuid)
self.assertEqual(retrieved.uuid, created.uuid)
self.assertEqual(retrieved.name, expected_name)
# Original and retrieved have not been updated
self.assertEqual(retrieved.contents, original_contents)
self.assertEqual(created.version, 1)
self.assertEqual(retrieved.version, 1)
# Updated has a new version, and new contents
self.assertEqual(updated.contents, new_contents)
self.assertEqual(updated.version, 2)
def test_update_no_versioning(self):
# Setup
store = DeploymentPlanStore(self.driver)
expected_name = "swift.yaml"
original_contents = "YAML"
created = self.driver.create(store, expected_name, original_contents)
# Test
new_contents = "YAML2"
updated = self.driver.update(store, created.uuid, new_contents)
# Verify
self.assertEqual(updated.uuid, created.uuid)
self.assertEqual(updated.name, expected_name)
self.assertEqual("YAML2", updated.contents)
self.assertEqual(updated.version, None)
def test_update_invalid_uuid(self):
# Setup
update_call = partial(self.driver.update, self.store, "uuid", "YAML2")
# Test & Verify
self.assertRaises(UnknownUUID, update_call)
@patch('tuskar.storage.drivers.sqlalchemy.SQLAlchemyDriver._generate_uuid')
def test_delete(self, mock_uuid):
# Setup
expected_uuid = 'b4b85dc2-0b0a-48ed-a56c-e4d582fd1473'
expected_name = "swift.yaml"
contents = "YAML"
mock_uuid.return_value = expected_uuid
self.driver.create(self.store, expected_name, contents)
# Test
result = self.driver.delete(self.store, expected_uuid)
# Verify
self.assertEqual(None, result)
retrieve_call = partial(
self.driver.retrieve,
self.store, expected_uuid
)
self.assertRaises(UnknownUUID, retrieve_call)
def test_delete_invalid(self):
self.assertRaises(
UnknownUUID, self.driver.delete, self.store, "uuid")
def test_list(self):
name = "swift.yaml"
template = self.driver.create(self.store, name, "YAML1")
self.assertEqual(1, len(self.driver.list(self.store)))
self.driver.update(self.store, template.uuid, "YAML2")
self.assertEqual(2, len(self.driver.list(self.store)))
def test_list_only_latest(self):
name = "swift.yaml"
template = self.driver.create(self.store, name, "YAML1")
self.driver.update(self.store, template.uuid, "YAML2")
listed = self.driver.list(self.store, only_latest=True)
self.assertEqual(1, len(listed))
def test_retrieve_by_name(self):
# Setup
create_result = self.driver.create(self.store, "name", "YAML")
self.driver.update(self.store, create_result.uuid, "YAML2")
# Test
retrieved = self.driver.retrieve_by_name(self.store, "name")
# Verify
self.assertNotEqual(create_result.uuid, retrieved.uuid)
self.assertEqual(retrieved.contents, "YAML2")
self.assertEqual(retrieved.version, 2)
def test_retrieve_by_name_version(self):
name = "swift.yaml"
# Setup
first = self.driver.create(self.store, name, "YAML1")
second = self.driver.update(self.store, first.uuid, "YAML2")
third = self.driver.update(self.store, first.uuid, "YAML3")
# Test
retrieved_first = self.driver.retrieve_by_name(self.store, name, 1)
retrieved_second = self.driver.retrieve_by_name(self.store, name, 2)
retrieved_third = self.driver.retrieve_by_name(self.store, name, 3)
# Verify
self.assertEqual(3, len(self.driver.list(self.store)))
self.assertEqual(retrieved_first.uuid, first.uuid)
self.assertEqual(1, retrieved_first.version)
self.assertEqual("YAML1", retrieved_first.contents)
self.assertEqual(retrieved_second.uuid, second.uuid)
self.assertEqual(2, retrieved_second.version)
self.assertEqual("YAML2", retrieved_second.contents)
self.assertEqual(retrieved_third.uuid, third.uuid)
self.assertEqual(3, retrieved_third.version)
self.assertEqual("YAML3", retrieved_third.contents)
def test_retrieve_by_name_invalid_name(self):
retrieve_by_name_call = partial(
self.driver.retrieve_by_name,
self.store, "name"
)
self.assertRaises(UnknownName, retrieve_by_name_call)
def test_retrieve_by_name_invalid_version(self):
self.driver.create(self.store, "name", "YAML")
retrieve_by_name_call = partial(
self.driver.retrieve_by_name,
self.store, "name", 2
)
self.assertRaises(UnknownVersion, retrieve_by_name_call)
def test_retrieve_by_name_other_error(self):
"""Verify that a NoResultFound exception is re-raised (and not
lost/squashed) if it isn't detected to be due to a missing name or
version that wasn't found.
"""
self.driver.create(self.store, "name", "YAML")
with patch('tuskar.storage.drivers.sqlalchemy.get_session') as mock:
query_mock = Mock()
query_mock.query.side_effect = NoResultFound()
mock.return_value = query_mock
retrieve_by_name_call = partial(
self.driver.retrieve_by_name,
self.store, "name"
)
self.assertRaises(NoResultFound, retrieve_by_name_call)

146
tuskar/tests/storage/test_stores.py
View File

@ -17,6 +17,7 @@ from functools import partial
from mock import Mock
from tuskar.storage.exceptions import NameAlreadyUsed
from tuskar.storage.models import StoredFile
from tuskar.storage.stores import _BaseStore
from tuskar.storage.stores import _NamedStore
@ -177,10 +178,10 @@ class EnvironmentFileTests(TestCase):
self.driver.create.assert_called_once_with(self.store, None, contents)
class DeploymentPlanTests(TestCase):
class DeploymentPlanMockedTests(TestCase):
def setUp(self):
super(DeploymentPlanTests, self).setUp()
super(DeploymentPlanMockedTests, self).setUp()
self.driver = Mock()
@ -230,10 +231,8 @@ class DeploymentPlanTests(TestCase):
self.driver.create.return_value = self._stored_file(name, contents)
result = self.store.create(
name, 'Template UUID', 'Environment UUID')
self.driver.create.assert_called_once_with(
self.store, name, contents)
result = self.store.create(name, 'Template UUID', 'Environment UUID')
self.driver.create.assert_called_once_with(self.store, name, contents)
self.assertEqual(result.name, name)
@ -250,15 +249,13 @@ class DeploymentPlanTests(TestCase):
self.driver.create.return_value = self._stored_file(name, contents)
self.template_store.create.return_value = Mock(uuid="UUID1")
result = self.store.create(
name, environment_uuid='Environment UUID')
result = self.store.create(name, environment_uuid='Environment UUID')
self.template_store.create.assert_called_once_with(
'deployment_plan name', '')
self.assertItemsEqual(self.environment_store.create.call_args_list, [])
self.driver.create.assert_called_once_with(
self.store, name, contents)
self.driver.create.assert_called_once_with(self.store, name, contents)
self.assertEqual(result.name, name)
self.template_store.retrieve.assert_called_once_with('UUID1')
@ -274,14 +271,12 @@ class DeploymentPlanTests(TestCase):
self.driver.create.return_value = self._stored_file(name, contents)
self.environment_store.create.return_value = Mock(uuid="UUID2")
result = self.store.create(
name, master_template_uuid='Template UUID')
result = self.store.create(name, master_template_uuid='Template UUID')
self.environment_store.create.assert_called_once_with('')
self.assertItemsEqual(self.template_store.create.call_args_list, [])
self.driver.create.assert_called_once_with(
self.store, name, contents)
self.driver.create.assert_called_once_with(self.store, name, contents)
self.assertEqual(result.name, name)
self.template_store.retrieve.assert_called_once_with('Template UUID')
@ -312,3 +307,126 @@ class DeploymentPlanTests(TestCase):
# test & verify
self.assertRaises(ValueError, update_call)
self.assertItemsEqual(self.driver.update.call_args_list, [])
class DeploymentPlanTests(TestCase):
def setUp(self):
super(DeploymentPlanTests, self).setUp()
self.template_store = TemplateStore()
self.environment_store = EnvironmentFileStore()
self.store = DeploymentPlanStore(
template_store=self.template_store,
environment_store=self.environment_store
)
self._create_plan()
def _create_plan(self):
contents = "Template Contents"
self.template = self.template_store.create("Template", contents)
self.env = self.environment_store.create("Environment Contents")
self.plan = self.store.create(
"Plan Name", self.template.uuid, self.env.uuid)
def test_create(self):
name = "deployment_plan name"
result = self.store.create(name, self.template.uuid, self.env.uuid)
self.assertEqual(result.name, name)
def test_create_duplicate(self):
# setup
name = "deployment_plan name"
self.store.create(name, self.template.uuid, self.env.uuid)
create_call = partial(self.store.create, name)
# test & verify
self.assertRaises(NameAlreadyUsed, create_call)
def test_create_no_template(self):
name = "deployment_plan name"
result = self.store.create(name, environment_uuid=self.env.uuid)
self.assertEqual(result.name, name)
def test_create_no_environment(self):
name = "deployment_plan name"
template_uuid = self.template.uuid
result = self.store.create(name, master_template_uuid=template_uuid)
self.assertEqual(result.name, name)
def test_retrieve(self):
# test
retrieved = self.store.retrieve(self.plan.uuid)
# verify
self.assertEqual(self.plan.uuid, retrieved.uuid)
self.assertEqual(self.template.uuid, retrieved.master_template.uuid)
self.assertEqual(self.env.uuid, retrieved.environment_file.uuid)
def test_update_template(self):
# setup
plan = self.store.create("plan")
new_template = self.store._template_store.update(
plan.master_template.uuid, "NEW CONTENT")
# test
updated = self.store.update(
plan.uuid, master_template_uuid=new_template.uuid)
# verify
retrieved = self.store.retrieve(plan.uuid)
self.assertEqual(plan.uuid, retrieved.uuid)
self.assertEqual(updated.master_template.uuid, new_template.uuid)
def test_update_environment(self):
# setup
plan = self.store.create("plan")
new_env = self.store._env_file_store.update(
plan.environment_file.uuid, "NEW CONTENT")
# test
updated = self.store.update(
plan.uuid, environment_uuid=new_env.uuid)
# verify
retrieved = self.store.retrieve(plan.uuid)
self.assertEqual(plan.uuid, retrieved.uuid)
self.assertEqual(updated.environment_file.uuid, new_env.uuid)
def test_update_nothing(self):
# setup
update_call = partial(self.store.update, self.plan.uuid)
# test & verify
self.assertRaises(ValueError, update_call)
def test_list(self):
plans = self.store.list()
self.assertEqual(1, len(plans))
plan, = plans
self.assertEqual(plan.uuid, self.plan.uuid)
def test_retrieve_by_name(self):
plan = self.store.retrieve_by_name("Plan Name")
self.assertEqual(plan.uuid, self.plan.uuid)