Added Mixin-related filters on DB queries
As a pre-requisite for being able to query the database for objects that are expired, I need a way to express date comparison on the 'deleted_at' field which is common for every Watcher object. As they are coming from mixins, I decided to implement these filters with a syntax borrowed from the Django ORM where the field is suffixed by the comparison operator you want to apply: - The '__lt' suffix stands for 'less than' - The '__lte' suffix stands for 'less than or equal to' - The '__gt' suffix stands for 'greater than' - The '__gte' suffix stands for 'greater than or equal to' - The '__eq' suffix stands for 'equal to' I also added a 'uuid' filter to later on be able to filter by uuid. Partially Implements: blueprint db-purge-engine Change-Id: I763f330c1b8ea8395990d2276b71e87f5b3f3ddc
This commit is contained in:
Vincent Françoise
parent
5d0fe553c4
commit
c3db66ca09
@ -5,6 +5,7 @@
|
||||
coverage>=3.6 # Apache-2.0
|
||||
discover # BSD
|
||||
doc8 # Apache-2.0
|
||||
freezegun # Apache-2.0
|
||||
hacking<0.11,>=0.10.2
|
||||
mock>=1.2 # BSD
|
||||
oslotest>=1.10.0 # Apache-2.0
|
||||
|
||||
@ -29,7 +29,10 @@ from watcher.common import exception
|
||||
from watcher.common import utils
|
||||
from watcher.db import api
|
||||
from watcher.db.sqlalchemy import models
|
||||
from watcher.objects import action as action_objects
|
||||
from watcher.objects import action_plan as ap_objects
|
||||
from watcher.objects import audit as audit_objects
|
||||
from watcher.objects import utils as objutils
|
||||
|
||||
CONF = cfg.CONF
|
||||
LOG = log.getLogger(__name__)
|
||||
@ -105,12 +108,88 @@ class Connection(api.BaseConnection):
|
||||
"""SqlAlchemy connection."""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
super(Connection, self).__init__()
|
||||
|
||||
def __add_soft_delete_mixin_filters(self, query, filters, model):
|
||||
if 'deleted' in filters:
|
||||
if bool(filters['deleted']):
|
||||
query = query.filter(model.deleted != 0)
|
||||
else:
|
||||
query = query.filter(model.deleted == 0)
|
||||
if 'deleted_at__eq' in filters:
|
||||
query = query.filter(
|
||||
model.deleted_at == objutils.datetime_or_str_or_none(
|
||||
filters['deleted_at__eq']))
|
||||
if 'deleted_at__gt' in filters:
|
||||
query = query.filter(
|
||||
model.deleted_at > objutils.datetime_or_str_or_none(
|
||||
filters['deleted_at__gt']))
|
||||
if 'deleted_at__gte' in filters:
|
||||
query = query.filter(
|
||||
model.deleted_at >= objutils.datetime_or_str_or_none(
|
||||
filters['deleted_at__gte']))
|
||||
if 'deleted_at__lt' in filters:
|
||||
query = query.filter(
|
||||
model.deleted_at < objutils.datetime_or_str_or_none(
|
||||
filters['deleted_at__lt']))
|
||||
if 'deleted_at__lte' in filters:
|
||||
query = query.filter(
|
||||
model.deleted_at <= objutils.datetime_or_str_or_none(
|
||||
filters['deleted_at__lte']))
|
||||
|
||||
return query
|
||||
|
||||
def __add_timestamp_mixin_filters(self, query, filters, model):
|
||||
if 'created_at__eq' in filters:
|
||||
query = query.filter(
|
||||
model.created_at == objutils.datetime_or_str_or_none(
|
||||
filters['created_at__eq']))
|
||||
if 'created_at__gt' in filters:
|
||||
query = query.filter(
|
||||
model.created_at > objutils.datetime_or_str_or_none(
|
||||
filters['created_at__gt']))
|
||||
if 'created_at__gte' in filters:
|
||||
query = query.filter(
|
||||
model.created_at >= objutils.datetime_or_str_or_none(
|
||||
filters['created_at__gte']))
|
||||
if 'created_at__lt' in filters:
|
||||
query = query.filter(
|
||||
model.created_at < objutils.datetime_or_str_or_none(
|
||||
filters['created_at__lt']))
|
||||
if 'created_at__lte' in filters:
|
||||
query = query.filter(
|
||||
model.created_at <= objutils.datetime_or_str_or_none(
|
||||
filters['created_at__lte']))
|
||||
|
||||
if 'updated_at__eq' in filters:
|
||||
query = query.filter(
|
||||
model.updated_at == objutils.datetime_or_str_or_none(
|
||||
filters['updated_at__eq']))
|
||||
if 'updated_at__gt' in filters:
|
||||
query = query.filter(
|
||||
model.updated_at > objutils.datetime_or_str_or_none(
|
||||
filters['updated_at__gt']))
|
||||
if 'updated_at__gte' in filters:
|
||||
query = query.filter(
|
||||
model.updated_at >= objutils.datetime_or_str_or_none(
|
||||
filters['updated_at__gte']))
|
||||
if 'updated_at__lt' in filters:
|
||||
query = query.filter(
|
||||
model.updated_at < objutils.datetime_or_str_or_none(
|
||||
filters['updated_at__lt']))
|
||||
if 'updated_at__lte' in filters:
|
||||
query = query.filter(
|
||||
model.updated_at <= objutils.datetime_or_str_or_none(
|
||||
filters['updated_at__lte']))
|
||||
|
||||
return query
|
||||
|
||||
def _add_audit_templates_filters(self, query, filters):
|
||||
if filters is None:
|
||||
filters = []
|
||||
|
||||
if 'uuid' in filters:
|
||||
query = query.filter_by(uuid=filters['uuid'])
|
||||
if 'name' in filters:
|
||||
query = query.filter_by(name=filters['name'])
|
||||
if 'host_aggregate' in filters:
|
||||
@ -118,12 +197,19 @@ class Connection(api.BaseConnection):
|
||||
if 'goal' in filters:
|
||||
query = query.filter_by(goal=filters['goal'])
|
||||
|
||||
query = self.__add_soft_delete_mixin_filters(
|
||||
query, filters, models.AuditTemplate)
|
||||
query = self.__add_timestamp_mixin_filters(
|
||||
query, filters, models.AuditTemplate)
|
||||
|
||||
return query
|
||||
|
||||
def _add_audits_filters(self, query, filters):
|
||||
if filters is None:
|
||||
filters = []
|
||||
|
||||
if 'uuid' in filters:
|
||||
query = query.filter_by(uuid=filters['uuid'])
|
||||
if 'type' in filters:
|
||||
query = query.filter_by(type=filters['type'])
|
||||
if 'state' in filters:
|
||||
@ -144,12 +230,20 @@ class Connection(api.BaseConnection):
|
||||
query = query.filter(
|
||||
models.AuditTemplate.name ==
|
||||
filters['audit_template_name'])
|
||||
|
||||
query = self.__add_soft_delete_mixin_filters(
|
||||
query, filters, models.Audit)
|
||||
query = self.__add_timestamp_mixin_filters(
|
||||
query, filters, models.Audit)
|
||||
|
||||
return query
|
||||
|
||||
def _add_action_plans_filters(self, query, filters):
|
||||
if filters is None:
|
||||
filters = []
|
||||
|
||||
if 'uuid' in filters:
|
||||
query = query.filter_by(uuid=filters['uuid'])
|
||||
if 'state' in filters:
|
||||
query = query.filter_by(state=filters['state'])
|
||||
if 'audit_id' in filters:
|
||||
@ -158,12 +252,20 @@ class Connection(api.BaseConnection):
|
||||
query = query.join(models.Audit,
|
||||
models.ActionPlan.audit_id == models.Audit.id)
|
||||
query = query.filter(models.Audit.uuid == filters['audit_uuid'])
|
||||
|
||||
query = self.__add_soft_delete_mixin_filters(
|
||||
query, filters, models.ActionPlan)
|
||||
query = self.__add_timestamp_mixin_filters(
|
||||
query, filters, models.ActionPlan)
|
||||
|
||||
return query
|
||||
|
||||
def _add_actions_filters(self, query, filters):
|
||||
if filters is None:
|
||||
filters = []
|
||||
|
||||
if 'uuid' in filters:
|
||||
query = query.filter_by(uuid=filters['uuid'])
|
||||
if 'action_plan_id' in filters:
|
||||
query = query.filter_by(action_plan_id=filters['action_plan_id'])
|
||||
if 'action_plan_uuid' in filters:
|
||||
@ -184,6 +286,11 @@ class Connection(api.BaseConnection):
|
||||
if 'alarm' in filters:
|
||||
query = query.filter_by(alarm=filters['alarm'])
|
||||
|
||||
query = self.__add_soft_delete_mixin_filters(
|
||||
query, filters, models.Action)
|
||||
query = self.__add_timestamp_mixin_filters(
|
||||
query, filters, models.Action)
|
||||
|
||||
return query
|
||||
|
||||
def get_audit_template_list(self, context, filters=None, limit=None,
|
||||
@ -193,7 +300,6 @@ class Connection(api.BaseConnection):
|
||||
query = self._add_audit_templates_filters(query, filters)
|
||||
if not context.show_deleted:
|
||||
query = query.filter_by(deleted_at=None)
|
||||
|
||||
return _paginate_query(models.AuditTemplate, limit, marker,
|
||||
sort_key, sort_dir, query)
|
||||
|
||||
@ -312,7 +418,8 @@ class Connection(api.BaseConnection):
|
||||
query = model_query(models.Audit)
|
||||
query = self._add_audits_filters(query, filters)
|
||||
if not context.show_deleted:
|
||||
query = query.filter(~(models.Audit.state == 'DELETED'))
|
||||
query = query.filter(
|
||||
~(models.Audit.state == audit_objects.State.DELETED))
|
||||
|
||||
return _paginate_query(models.Audit, limit, marker,
|
||||
sort_key, sort_dir, query)
|
||||
@ -340,7 +447,7 @@ class Connection(api.BaseConnection):
|
||||
try:
|
||||
audit = query.one()
|
||||
if not context.show_deleted:
|
||||
if audit.state == 'DELETED':
|
||||
if audit.state == audit_objects.State.DELETED:
|
||||
raise exception.AuditNotFound(audit=audit_id)
|
||||
return audit
|
||||
except exc.NoResultFound:
|
||||
@ -353,7 +460,7 @@ class Connection(api.BaseConnection):
|
||||
try:
|
||||
audit = query.one()
|
||||
if not context.show_deleted:
|
||||
if audit.state == 'DELETED':
|
||||
if audit.state == audit_objects.State.DELETED:
|
||||
raise exception.AuditNotFound(audit=audit_uuid)
|
||||
return audit
|
||||
except exc.NoResultFound:
|
||||
@ -421,7 +528,8 @@ class Connection(api.BaseConnection):
|
||||
query = model_query(models.Action)
|
||||
query = self._add_actions_filters(query, filters)
|
||||
if not context.show_deleted:
|
||||
query = query.filter(~(models.Action.state == 'DELETED'))
|
||||
query = query.filter(
|
||||
~(models.Action.state == action_objects.State.DELETED))
|
||||
return _paginate_query(models.Action, limit, marker,
|
||||
sort_key, sort_dir, query)
|
||||
|
||||
@ -444,7 +552,7 @@ class Connection(api.BaseConnection):
|
||||
try:
|
||||
action = query.one()
|
||||
if not context.show_deleted:
|
||||
if action.state == 'DELETED':
|
||||
if action.state == action_objects.State.DELETED:
|
||||
raise exception.ActionNotFound(
|
||||
action=action_id)
|
||||
return action
|
||||
@ -457,7 +565,7 @@ class Connection(api.BaseConnection):
|
||||
try:
|
||||
action = query.one()
|
||||
if not context.show_deleted:
|
||||
if action.state == 'DELETED':
|
||||
if action.state == action_objects.State.DELETED:
|
||||
raise exception.ActionNotFound(
|
||||
action=action_uuid)
|
||||
return action
|
||||
@ -514,7 +622,8 @@ class Connection(api.BaseConnection):
|
||||
query = model_query(models.ActionPlan)
|
||||
query = self._add_action_plans_filters(query, filters)
|
||||
if not context.show_deleted:
|
||||
query = query.filter(~(models.ActionPlan.state == 'DELETED'))
|
||||
query = query.filter(
|
||||
~(models.ActionPlan.state == ap_objects.State.DELETED))
|
||||
|
||||
return _paginate_query(models.ActionPlan, limit, marker,
|
||||
sort_key, sort_dir, query)
|
||||
@ -539,7 +648,7 @@ class Connection(api.BaseConnection):
|
||||
try:
|
||||
action_plan = query.one()
|
||||
if not context.show_deleted:
|
||||
if action_plan.state == 'DELETED':
|
||||
if action_plan.state == ap_objects.State.DELETED:
|
||||
raise exception.ActionPlanNotFound(
|
||||
action_plan=action_plan_id)
|
||||
return action_plan
|
||||
@ -553,7 +662,7 @@ class Connection(api.BaseConnection):
|
||||
try:
|
||||
action_plan = query.one()
|
||||
if not context.show_deleted:
|
||||
if action_plan.state == 'DELETED':
|
||||
if action_plan.state == ap_objects.State.DELETED:
|
||||
raise exception.ActionPlanNotFound(
|
||||
action_plan=action_plan__uuid)
|
||||
return action_plan
|
||||
|
||||
@ -92,6 +92,12 @@ class DbTestCase(base.TestCase):
|
||||
def setUp(self):
|
||||
cfg.CONF.set_override("enable_authentication", False,
|
||||
enforce_type=True)
|
||||
# To use in-memory SQLite DB
|
||||
cfg.CONF.set_override("connection", "sqlite://", group="database",
|
||||
enforce_type=True)
|
||||
cfg.CONF.set_override("sqlite_db", "", group="database",
|
||||
enforce_type=True)
|
||||
|
||||
super(DbTestCase, self).setUp()
|
||||
|
||||
self.dbapi = dbapi.get_instance()
|
||||
|
||||
@ -15,13 +15,222 @@
|
||||
|
||||
"""Tests for manipulating Action via the DB API"""
|
||||
|
||||
import freezegun
|
||||
import six
|
||||
|
||||
from watcher.common import exception
|
||||
from watcher.common import utils as w_utils
|
||||
from watcher.objects import action as act_objects
|
||||
from watcher.tests.db import base
|
||||
from watcher.tests.db import utils
|
||||
|
||||
|
||||
class TestDbActionFilters(base.DbTestCase):
|
||||
|
||||
FAKE_OLDER_DATE = '2014-01-01T09:52:05.219414'
|
||||
FAKE_OLD_DATE = '2015-01-01T09:52:05.219414'
|
||||
FAKE_TODAY = '2016-02-24T09:52:05.219414'
|
||||
|
||||
def setUp(self):
|
||||
super(TestDbActionFilters, self).setUp()
|
||||
self.context.show_deleted = True
|
||||
self._data_setup()
|
||||
|
||||
def _data_setup(self):
|
||||
self.audit_template_name = "Audit Template"
|
||||
|
||||
self.audit_template = utils.create_test_audit_template(
|
||||
name=self.audit_template_name, id=1, uuid=None)
|
||||
self.audit = utils.create_test_audit(
|
||||
audit_template_id=self.audit_template.id, id=1, uuid=None)
|
||||
self.action_plan = utils.create_test_action_plan(
|
||||
audit_id=self.audit.id, id=1, uuid=None)
|
||||
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.action1 = utils.create_test_action(
|
||||
action_plan_id=self.action_plan.id, id=1, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.action2 = utils.create_test_action(
|
||||
action_plan_id=self.action_plan.id, id=2, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.action3 = utils.create_test_action(
|
||||
action_plan_id=self.action_plan.id, id=3, uuid=None)
|
||||
|
||||
def _soft_delete_actions(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_action(self.action1.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.soft_delete_action(self.action2.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.soft_delete_action(self.action3.uuid)
|
||||
|
||||
def _update_actions(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.update_action(
|
||||
self.action1.uuid,
|
||||
values={"state": act_objects.State.SUCCEEDED})
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.update_action(
|
||||
self.action2.uuid,
|
||||
values={"state": act_objects.State.SUCCEEDED})
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.update_action(
|
||||
self.action3.uuid,
|
||||
values={"state": act_objects.State.SUCCEEDED})
|
||||
|
||||
def test_get_action_filter_deleted_true(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_action(self.action1.uuid)
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'deleted': True})
|
||||
|
||||
self.assertEqual([self.action1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_filter_deleted_false(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_action(self.action1.uuid)
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'deleted': False})
|
||||
|
||||
self.assertEqual([self.action2['id'], self.action3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_filter_deleted_at_eq(self):
|
||||
self._soft_delete_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'deleted_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.action1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_filter_deleted_at_lt(self):
|
||||
self._soft_delete_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'deleted_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action2['id'], self.action3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_filter_deleted_at_lte(self):
|
||||
self._soft_delete_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'deleted_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action2['id'], self.action3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_filter_deleted_at_gt(self):
|
||||
self._soft_delete_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'deleted_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.action1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_filter_deleted_at_gte(self):
|
||||
self._soft_delete_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'deleted_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action1['id'], self.action2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# created_at #
|
||||
|
||||
def test_get_action_filter_created_at_eq(self):
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'created_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.action1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_filter_created_at_lt(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'created_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action2['id'], self.action3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_filter_created_at_lte(self):
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'created_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action2['id'], self.action3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_filter_created_at_gt(self):
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'created_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.action1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_filter_created_at_gte(self):
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'created_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action1['id'], self.action2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# updated_at #
|
||||
|
||||
def test_get_action_filter_updated_at_eq(self):
|
||||
self._update_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'updated_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.action1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_filter_updated_at_lt(self):
|
||||
self._update_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'updated_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action2['id'], self.action3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_filter_updated_at_lte(self):
|
||||
self._update_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'updated_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action2['id'], self.action3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_filter_updated_at_gt(self):
|
||||
self._update_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'updated_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.action1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_filter_updated_at_gte(self):
|
||||
self._update_actions()
|
||||
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'updated_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action1['id'], self.action2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
|
||||
class DbActionTestCase(base.DbTestCase):
|
||||
|
||||
def _create_test_action(self, **kwargs):
|
||||
@ -36,7 +245,7 @@ class DbActionTestCase(base.DbTestCase):
|
||||
|
||||
def test_get_action_list(self):
|
||||
uuids = []
|
||||
for i in range(1, 6):
|
||||
for _ in range(1, 6):
|
||||
action = utils.create_test_action(uuid=w_utils.generate_uuid())
|
||||
uuids.append(six.text_type(action['uuid']))
|
||||
res = self.dbapi.get_action_list(self.context)
|
||||
@ -101,6 +310,14 @@ class DbActionTestCase(base.DbTestCase):
|
||||
for action in res:
|
||||
self.assertEqual(action_plan['id'], action.action_plan_id)
|
||||
|
||||
def test_get_action_list_with_filter_by_uuid(self):
|
||||
action = self._create_test_action()
|
||||
res = self.dbapi.get_action_list(
|
||||
self.context, filters={'uuid': action["uuid"]})
|
||||
|
||||
self.assertEqual(len(res), 1)
|
||||
self.assertEqual(action['uuid'], res[0].uuid)
|
||||
|
||||
def test_get_action_by_id(self):
|
||||
action = self._create_test_action()
|
||||
action = self.dbapi.get_action_by_id(self.context, action['id'])
|
||||
|
||||
@ -15,13 +15,219 @@
|
||||
|
||||
"""Tests for manipulating ActionPlan via the DB API"""
|
||||
|
||||
import freezegun
|
||||
import six
|
||||
|
||||
from watcher.common import exception
|
||||
from watcher.common import utils as w_utils
|
||||
from watcher.objects import action_plan as ap_objects
|
||||
from watcher.tests.db import base
|
||||
from watcher.tests.db import utils
|
||||
|
||||
|
||||
class TestDbActionPlanFilters(base.DbTestCase):
|
||||
|
||||
FAKE_OLDER_DATE = '2014-01-01T09:52:05.219414'
|
||||
FAKE_OLD_DATE = '2015-01-01T09:52:05.219414'
|
||||
FAKE_TODAY = '2016-02-24T09:52:05.219414'
|
||||
|
||||
def setUp(self):
|
||||
super(TestDbActionPlanFilters, self).setUp()
|
||||
self.context.show_deleted = True
|
||||
self._data_setup()
|
||||
|
||||
def _data_setup(self):
|
||||
self.audit_template_name = "Audit Template"
|
||||
|
||||
self.audit_template = utils.create_test_audit_template(
|
||||
name=self.audit_template_name, id=1, uuid=None)
|
||||
self.audit = utils.create_test_audit(
|
||||
audit_template_id=self.audit_template.id, id=1, uuid=None)
|
||||
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.action_plan1 = utils.create_test_action_plan(
|
||||
audit_id=self.audit.id, id=1, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.action_plan2 = utils.create_test_action_plan(
|
||||
audit_id=self.audit.id, id=2, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.action_plan3 = utils.create_test_action_plan(
|
||||
audit_id=self.audit.id, id=3, uuid=None)
|
||||
|
||||
def _soft_delete_action_plans(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_action_plan(self.action_plan1.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.soft_delete_action_plan(self.action_plan2.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.soft_delete_action_plan(self.action_plan3.uuid)
|
||||
|
||||
def _update_action_plans(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.update_action_plan(
|
||||
self.action_plan1.uuid,
|
||||
values={"state": ap_objects.State.SUCCEEDED})
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.update_action_plan(
|
||||
self.action_plan2.uuid,
|
||||
values={"state": ap_objects.State.SUCCEEDED})
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.update_action_plan(
|
||||
self.action_plan3.uuid,
|
||||
values={"state": ap_objects.State.SUCCEEDED})
|
||||
|
||||
def test_get_action_plan_list_filter_deleted_true(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_action_plan(self.action_plan1.uuid)
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'deleted': True})
|
||||
|
||||
self.assertEqual([self.action_plan1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_deleted_false(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_action_plan(self.action_plan1.uuid)
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'deleted': False})
|
||||
|
||||
self.assertEqual([self.action_plan2['id'], self.action_plan3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_deleted_at_eq(self):
|
||||
self._soft_delete_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'deleted_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.action_plan1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_deleted_at_lt(self):
|
||||
self._soft_delete_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'deleted_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan2['id'], self.action_plan3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_deleted_at_lte(self):
|
||||
self._soft_delete_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'deleted_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan2['id'], self.action_plan3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_deleted_at_gt(self):
|
||||
self._soft_delete_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'deleted_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.action_plan1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_deleted_at_gte(self):
|
||||
self._soft_delete_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'deleted_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan1['id'], self.action_plan2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# created_at #
|
||||
|
||||
def test_get_action_plan_list_filter_created_at_eq(self):
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'created_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.action_plan1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_created_at_lt(self):
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'created_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan2['id'], self.action_plan3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_created_at_lte(self):
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'created_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan2['id'], self.action_plan3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_created_at_gt(self):
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'created_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.action_plan1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_created_at_gte(self):
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'created_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan1['id'], self.action_plan2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# updated_at #
|
||||
|
||||
def test_get_action_plan_list_filter_updated_at_eq(self):
|
||||
self._update_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'updated_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.action_plan1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_updated_at_lt(self):
|
||||
self._update_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'updated_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan2['id'], self.action_plan3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_updated_at_lte(self):
|
||||
self._update_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'updated_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan2['id'], self.action_plan3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_updated_at_gt(self):
|
||||
self._update_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'updated_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.action_plan1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_action_plan_list_filter_updated_at_gte(self):
|
||||
self._update_action_plans()
|
||||
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'updated_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.action_plan1['id'], self.action_plan2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
|
||||
class DbActionPlanTestCase(base.DbTestCase):
|
||||
|
||||
def _create_test_audit(self, **kwargs):
|
||||
@ -80,6 +286,14 @@ class DbActionPlanTestCase(base.DbTestCase):
|
||||
for r in res:
|
||||
self.assertEqual(audit['id'], r.audit_id)
|
||||
|
||||
def test_get_action_plan_list_with_filter_by_uuid(self):
|
||||
action_plan = self._create_test_action_plan()
|
||||
res = self.dbapi.get_action_plan_list(
|
||||
self.context, filters={'uuid': action_plan["uuid"]})
|
||||
|
||||
self.assertEqual(len(res), 1)
|
||||
self.assertEqual(action_plan['uuid'], res[0].uuid)
|
||||
|
||||
def test_get_action_plan_by_id(self):
|
||||
action_plan = self._create_test_action_plan()
|
||||
action_plan = self.dbapi.get_action_plan_by_id(
|
||||
|
||||
@ -15,13 +15,217 @@
|
||||
|
||||
"""Tests for manipulating Audit via the DB API"""
|
||||
|
||||
import freezegun
|
||||
import six
|
||||
|
||||
from watcher.common import exception
|
||||
from watcher.common import utils as w_utils
|
||||
from watcher.objects import audit as audit_objects
|
||||
from watcher.tests.db import base
|
||||
from watcher.tests.db import utils
|
||||
|
||||
|
||||
class TestDbAuditFilters(base.DbTestCase):
|
||||
|
||||
FAKE_OLDER_DATE = '2014-01-01T09:52:05.219414'
|
||||
FAKE_OLD_DATE = '2015-01-01T09:52:05.219414'
|
||||
FAKE_TODAY = '2016-02-24T09:52:05.219414'
|
||||
|
||||
def setUp(self):
|
||||
super(TestDbAuditFilters, self).setUp()
|
||||
self.context.show_deleted = True
|
||||
self._data_setup()
|
||||
|
||||
def _data_setup(self):
|
||||
self.audit_template_name = "Audit Template"
|
||||
|
||||
self.audit_template = utils.create_test_audit_template(
|
||||
name=self.audit_template_name, id=1, uuid=None)
|
||||
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.audit1 = utils.create_test_audit(
|
||||
audit_template_id=self.audit_template.id, id=1, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.audit2 = utils.create_test_audit(
|
||||
audit_template_id=self.audit_template.id, id=2, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.audit3 = utils.create_test_audit(
|
||||
audit_template_id=self.audit_template.id, id=3, uuid=None)
|
||||
|
||||
def _soft_delete_audits(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_audit(self.audit1.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.soft_delete_audit(self.audit2.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.soft_delete_audit(self.audit3.uuid)
|
||||
|
||||
def _update_audits(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.update_audit(
|
||||
self.audit1.uuid,
|
||||
values={"state": audit_objects.State.SUCCEEDED})
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.update_audit(
|
||||
self.audit2.uuid,
|
||||
values={"state": audit_objects.State.SUCCEEDED})
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.update_audit(
|
||||
self.audit3.uuid,
|
||||
values={"state": audit_objects.State.SUCCEEDED})
|
||||
|
||||
def test_get_audit_list_filter_deleted_true(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_audit(self.audit1.uuid)
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'deleted': True})
|
||||
|
||||
self.assertEqual([self.audit1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_deleted_false(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_audit(self.audit1.uuid)
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'deleted': False})
|
||||
|
||||
self.assertEqual([self.audit2['id'], self.audit3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_deleted_at_eq(self):
|
||||
self._soft_delete_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'deleted_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.audit1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_deleted_at_lt(self):
|
||||
self._soft_delete_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'deleted_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit2['id'], self.audit3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_deleted_at_lte(self):
|
||||
self._soft_delete_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'deleted_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit2['id'], self.audit3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_deleted_at_gt(self):
|
||||
self._soft_delete_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'deleted_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.audit1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_deleted_at_gte(self):
|
||||
self._soft_delete_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'deleted_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit1['id'], self.audit2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# created_at #
|
||||
|
||||
def test_get_audit_list_filter_created_at_eq(self):
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'created_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.audit1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_created_at_lt(self):
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'created_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit2['id'], self.audit3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_created_at_lte(self):
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'created_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit2['id'], self.audit3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_created_at_gt(self):
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'created_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.audit1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_created_at_gte(self):
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'created_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit1['id'], self.audit2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# updated_at #
|
||||
|
||||
def test_get_audit_list_filter_updated_at_eq(self):
|
||||
self._update_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'updated_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.audit1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_updated_at_lt(self):
|
||||
self._update_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'updated_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit2['id'], self.audit3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_updated_at_lte(self):
|
||||
self._update_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'updated_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit2['id'], self.audit3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_updated_at_gt(self):
|
||||
self._update_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'updated_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.audit1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_filter_updated_at_gte(self):
|
||||
self._update_audits()
|
||||
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'updated_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit1['id'], self.audit2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
|
||||
class DbAuditTestCase(base.DbTestCase):
|
||||
|
||||
def _create_test_audit(self, **kwargs):
|
||||
@ -31,7 +235,7 @@ class DbAuditTestCase(base.DbTestCase):
|
||||
|
||||
def test_get_audit_list(self):
|
||||
uuids = []
|
||||
for i in range(1, 6):
|
||||
for _ in range(1, 6):
|
||||
audit = utils.create_test_audit(uuid=w_utils.generate_uuid())
|
||||
uuids.append(six.text_type(audit['uuid']))
|
||||
res = self.dbapi.get_audit_list(self.context)
|
||||
@ -70,6 +274,14 @@ class DbAuditTestCase(base.DbTestCase):
|
||||
filters={'state': 'PENDING'})
|
||||
self.assertEqual([audit2['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_list_with_filter_by_uuid(self):
|
||||
audit = self._create_test_audit()
|
||||
res = self.dbapi.get_audit_list(
|
||||
self.context, filters={'uuid': audit["uuid"]})
|
||||
|
||||
self.assertEqual(len(res), 1)
|
||||
self.assertEqual(audit['uuid'], res[0].uuid)
|
||||
|
||||
def test_get_audit_by_id(self):
|
||||
audit = self._create_test_audit()
|
||||
audit = self.dbapi.get_audit_by_id(self.context, audit['id'])
|
||||
|
||||
@ -15,13 +15,214 @@
|
||||
|
||||
"""Tests for manipulating AuditTemplate via the DB API"""
|
||||
|
||||
import freezegun
|
||||
import six
|
||||
|
||||
from watcher.common import exception
|
||||
from watcher.common import utils as w_utils
|
||||
from watcher.tests.db import base
|
||||
from watcher.tests.db import utils
|
||||
|
||||
|
||||
class TestDbAuditTemplateFilters(base.DbTestCase):
|
||||
|
||||
FAKE_OLDER_DATE = '2014-01-01T09:52:05.219414'
|
||||
FAKE_OLD_DATE = '2015-01-01T09:52:05.219414'
|
||||
FAKE_TODAY = '2016-02-24T09:52:05.219414'
|
||||
|
||||
def setUp(self):
|
||||
super(TestDbAuditTemplateFilters, self).setUp()
|
||||
self.context.show_deleted = True
|
||||
self._data_setup()
|
||||
|
||||
def _data_setup(self):
|
||||
gen_name = lambda: "Audit Template %s" % w_utils.generate_uuid()
|
||||
self.audit_template1_name = gen_name()
|
||||
self.audit_template2_name = gen_name()
|
||||
self.audit_template3_name = gen_name()
|
||||
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.audit_template1 = utils.create_test_audit_template(
|
||||
name=self.audit_template1_name, id=1, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.audit_template2 = utils.create_test_audit_template(
|
||||
name=self.audit_template2_name, id=2, uuid=None)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.audit_template3 = utils.create_test_audit_template(
|
||||
name=self.audit_template3_name, id=3, uuid=None)
|
||||
|
||||
def _soft_delete_audit_templates(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_audit_template(self.audit_template1.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.soft_delete_audit_template(self.audit_template2.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.soft_delete_audit_template(self.audit_template3.uuid)
|
||||
|
||||
def _update_audit_templates(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.update_audit_template(
|
||||
self.audit_template1.uuid, values={"name": "audit_template1"})
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.update_audit_template(
|
||||
self.audit_template2.uuid, values={"name": "audit_template2"})
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.update_audit_template(
|
||||
self.audit_template3.uuid, values={"name": "audit_template3"})
|
||||
|
||||
def test_get_audit_template_list_filter_deleted_true(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_audit_template(self.audit_template1.uuid)
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'deleted': True})
|
||||
|
||||
self.assertEqual([self.audit_template1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_deleted_false(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_audit_template(self.audit_template1.uuid)
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'deleted': False})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template2['id'], self.audit_template3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_deleted_at_eq(self):
|
||||
self._soft_delete_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'deleted_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.audit_template1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_deleted_at_lt(self):
|
||||
self._soft_delete_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'deleted_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template2['id'], self.audit_template3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_deleted_at_lte(self):
|
||||
self._soft_delete_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'deleted_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template2['id'], self.audit_template3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_deleted_at_gt(self):
|
||||
self._soft_delete_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'deleted_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.audit_template1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_deleted_at_gte(self):
|
||||
self._soft_delete_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'deleted_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template1['id'], self.audit_template2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# created_at #
|
||||
|
||||
def test_get_audit_template_list_filter_created_at_eq(self):
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'created_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.audit_template1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_created_at_lt(self):
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'created_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template2['id'], self.audit_template3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_created_at_lte(self):
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'created_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template2['id'], self.audit_template3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_created_at_gt(self):
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'created_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.audit_template1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_created_at_gte(self):
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'created_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template1['id'], self.audit_template2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# updated_at #
|
||||
|
||||
def test_get_audit_template_list_filter_updated_at_eq(self):
|
||||
self._update_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'updated_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.audit_template1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_updated_at_lt(self):
|
||||
self._update_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'updated_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template2['id'], self.audit_template3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_updated_at_lte(self):
|
||||
self._update_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'updated_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template2['id'], self.audit_template3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_updated_at_gt(self):
|
||||
self._update_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'updated_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.audit_template1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_filter_updated_at_gte(self):
|
||||
self._update_audit_templates()
|
||||
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'updated_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.audit_template1['id'], self.audit_template2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
|
||||
class DbAuditTemplateTestCase(base.DbTestCase):
|
||||
|
||||
def _create_test_audit_template(self, **kwargs):
|
||||
@ -77,6 +278,14 @@ class DbAuditTemplateTestCase(base.DbTestCase):
|
||||
filters={'name': 'My Audit Template 2'})
|
||||
self.assertEqual([audit_template2['id']], [r.id for r in res])
|
||||
|
||||
def test_get_audit_template_list_with_filter_by_uuid(self):
|
||||
audit_template = self._create_test_audit_template()
|
||||
res = self.dbapi.get_audit_template_list(
|
||||
self.context, filters={'uuid': audit_template["uuid"]})
|
||||
|
||||
self.assertEqual(len(res), 1)
|
||||
self.assertEqual(audit_template['uuid'], res[0].uuid)
|
||||
|
||||
def test_get_audit_template_by_id(self):
|
||||
audit_template = self._create_test_audit_template()
|
||||
audit_template = self.dbapi.get_audit_template_by_id(
|
||||
@ -132,14 +341,6 @@ class DbAuditTemplateTestCase(base.DbTestCase):
|
||||
self.assertRaises(exception.AuditTemplateNotFound,
|
||||
self.dbapi.destroy_audit_template, 1234)
|
||||
|
||||
# def test_destroy_audit_template_that_referenced_by_goals(self):
|
||||
# audit_template = self._create_test_audit_template()
|
||||
# goal = utils.create_test_goal(audit_template=audit_template['uuid'])
|
||||
# self.assertEqual(audit_template['uuid'], goal.audit_template)
|
||||
# self.assertRaises(exception.AuditTemplateReferenced,
|
||||
# self.dbapi.destroy_audit_template,
|
||||
# audit_template['id'])
|
||||
|
||||
def test_create_audit_template_already_exists(self):
|
||||
uuid = w_utils.generate_uuid()
|
||||
self._create_test_audit_template(id=1, uuid=uuid)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user