x/tuskar
Code Issues Proposed changes

Merge "Fix delete cascade for attribute and counts"

Browse Source
This commit is contained in:
Jenkins 2014-02-14 07:50:58 +00:00 committed by Gerrit Code Review
commit dad9a3e0df
4 changed files with 47 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tuskar/common/exception.py
View File

@ -151,6 +151,10 @@ class OvercloudRoleExists(DuplicateEntry):
message = _("Overcloud role with name %(name)s already exists.")
class OvercloudRoleInUse(TuskarException):
message = _('Role %(name)s is in use by an overcloud.')
class OvercloudRoleCountExists(DuplicateEntry):
message = _("Count for overcloud %(cloud)s and "
"role %(role)s already exists.")

44
tuskar/db/sqlalchemy/api.py
View File

@ -19,6 +19,7 @@
from oslo.config import cfg
# TODO(deva): import MultipleResultsFound and handle it appropriately
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import NoResultFound
from sqlalchemy.orm import subqueryload
@ -53,6 +54,10 @@ def model_query(model, *args, **kwargs):
return query
def get_session():
return db_session.get_session(sqlite_fk=True)
class Connection(api.Connection):
"""SqlAlchemy connection."""
@ -63,21 +68,19 @@ class Connection(api.Connection):
# here.
pass
@staticmethod
def get_overcloud_roles():
def get_overcloud_roles(self):
"""Returns all overcloud roles known to Tuskar.
:return: list of roles; empty list if none are found
:rtype: list of tuskar.db.sqlalchemy.models.OvercloudRole
"""
session = db_session.get_session()
session = get_session()
roles = session.query(models.OvercloudRole).all()
session.close()
return roles
@staticmethod
def get_overcloud_role_by_id(role_id):
def get_overcloud_role_by_id(self, role_id):
"""Single overcloud role query.
:return: role if one exists with the given ID
@ -87,7 +90,7 @@ class Connection(api.Connection):
role with the given ID exists
"""
session = db_session.get_session()
session = get_session()
try:
query = session.query(models.OvercloudRole).filter_by(
id=role_id)
@ -101,8 +104,7 @@ class Connection(api.Connection):
return result
@staticmethod
def create_overcloud_role(overcloud_role):
def create_overcloud_role(self, overcloud_role):
"""Creates a new overcloud role in the database.
:param overcloud_role: role instance to save
@ -115,7 +117,7 @@ class Connection(api.Connection):
:raises: tuskar.common.exception.OvercloudRoleExists: if a role
with the given name exists
"""
session = db_session.get_session()
session = get_session()
session.begin()
try:
@ -160,25 +162,30 @@ class Connection(api.Connection):
"""
role = self.get_overcloud_role_by_id(role_id)
session = db_session.get_session()
session = get_session()
session.begin()
try:
session.delete(role)
session.commit()
except db_exception.DBError as e:
if isinstance(e.inner_exception, IntegrityError):
raise exception.OvercloudRoleInUse(name=role.name)
else:
raise
finally:
session.close()
@staticmethod
def get_overclouds():
def get_overclouds(self):
"""Returns all overcloud instances from the database.
:return: list of overcloud instances; empty list if none are found
:rtype: list of tuskar.db.sqlalchemy.models.Overcloud
"""
session = db_session.get_session()
session = get_session()
overclouds = session.query(models.Overcloud).\
options(subqueryload(models.Overcloud.attributes)).\
options(subqueryload(models.Overcloud.counts)).\
@ -186,8 +193,7 @@ class Connection(api.Connection):
session.close()
return overclouds
@staticmethod
def get_overcloud_by_id(overcloud_id):
def get_overcloud_by_id(self, overcloud_id):
"""Returns a specific overcloud instance.
:return: overcloud if one exists with the given ID
@ -197,7 +203,7 @@ class Connection(api.Connection):
overcloud with the given ID exists
"""
session = db_session.get_session()
session = get_session()
try:
query = session.query(models.Overcloud).\
options(subqueryload(models.Overcloud.attributes)).\
@ -226,7 +232,7 @@ class Connection(api.Connection):
:raises: tuskar.common.exception.OvercloudExists: if an overcloud
role with the given name exists
"""
session = db_session.get_session()
session = get_session()
session.begin()
try:
@ -276,7 +282,7 @@ class Connection(api.Connection):
existing = self.get_overcloud_by_id(updated.id)
session = db_session.get_session()
session = get_session()
session.begin()
try:
@ -383,7 +389,7 @@ class Connection(api.Connection):
"""
overcloud = self.get_overcloud_by_id(overcloud_id)
session = db_session.get_session()
session = get_session()
session.begin()
try:

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

@ -122,7 +122,7 @@ class OvercloudRoleCount(Base):
# Overcloud in which the role is being deployed
overcloud_id = \
Column(Integer,
ForeignKey('%s.id' % TABLE_OVERCLOUD),
ForeignKey('%s.id' % TABLE_OVERCLOUD, ondelete='CASCADE'),
nullable=False)
# Number of nodes of this configuration that should be deployed
@ -148,7 +148,8 @@ class OvercloudAttribute(Base):
# Reference back to the overcloud being configured
overcloud_id = Column(Integer,
ForeignKey('%s.id' % TABLE_OVERCLOUD),
ForeignKey('%s.id' % TABLE_OVERCLOUD,
ondelete='CASCADE'),
nullable=False)
# Identifier and value of the configuration attribute
@ -182,10 +183,12 @@ class Overcloud(Base):
description = Column(String(length=LENGTH_DESCRIPTION))
# List of configuration attributes for the overcloud
attributes = relationship(OvercloudAttribute.__name__)
attributes = relationship(OvercloudAttribute.__name__,
cascade='all,delete')
# List of counts of overcloud roles to deploy
counts = relationship(OvercloudRoleCount.__name__)
counts = relationship(OvercloudRoleCount.__name__,
cascade='all,delete')
def __eq__(self, other):
return self.name == other.name

11
tuskar/tests/db/test_api.py
View File

@ -363,6 +363,17 @@ class OvercloudTests(db_base.DbTestCase):
found = self.connection.get_overclouds()
self.assertEqual(0, len(found))
# Ensure the joined tables are clear too
session = dbapi.get_session()
all_counts = session.query(models.OvercloudAttribute).all()
session.close()
self.assertEqual(0, len(all_counts))
session = dbapi.get_session()
all_counts = session.query(models.OvercloudRoleCount).all()
session.close()
self.assertEqual(0, len(all_counts))
def test_delete_nonexistent_overcloud(self):
self.assertRaises(exception.OvercloudNotFound,
self.connection.delete_overcloud_by_id,