Arnike/watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
46d5094addfa3963a2eda79a224b151b28060a7a
watcher/watcher/tests/db/test_audit_template.py
Vincent Françoise c3db66ca09 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
2016-03-14 15:46:58 +01:00

373 lines
15 KiB
Python
Raw Blame History

# Copyright 2015 OpenStack Foundation
# All Rights Reserved.
#
# 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.
"""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):
audit_template = utils.get_test_audit_template(**kwargs)
self.dbapi.create_audit_template(audit_template)
return audit_template
def test_get_audit_template_list(self):
uuids = []
for i in range(1, 6):
audit_template = utils.create_test_audit_template(
uuid=w_utils.generate_uuid(),
name='My Audit Template {0}'.format(i))
uuids.append(six.text_type(audit_template['uuid']))
res = self.dbapi.get_audit_template_list(self.context)
res_uuids = [r.uuid for r in res]
self.assertEqual(uuids.sort(), res_uuids.sort())
def test_get_audit_template_list_with_filters(self):
audit_template1 = self._create_test_audit_template(
id=1,
uuid=w_utils.generate_uuid(),
name='My Audit Template 1',
description='Description of my audit template 1',
host_aggregate=5,
goal='DUMMY',
extra={'automatic': True})
audit_template2 = self._create_test_audit_template(
id=2,
uuid=w_utils.generate_uuid(),
name='My Audit Template 2',
description='Description of my audit template 2',
host_aggregate=3,
goal='DUMMY',
extra={'automatic': True})
res = self.dbapi.get_audit_template_list(self.context,
filters={'host_aggregate': 5})
self.assertEqual([audit_template1['id']], [r.id for r in res])
res = self.dbapi.get_audit_template_list(self.context,
filters={'host_aggregate': 1})
self.assertEqual([], [r.id for r in res])
res = self.dbapi.get_audit_template_list(
self.context,
filters={'goal': 'DUMMY'})
self.assertEqual([audit_template1['id'], audit_template2['id']],
[r.id for r in res])
res = self.dbapi.get_audit_template_list(
self.context,
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(
self.context, audit_template['id'])
self.assertEqual(audit_template['uuid'], audit_template.uuid)
def test_get_audit_template_by_uuid(self):
audit_template = self._create_test_audit_template()
audit_template = self.dbapi.get_audit_template_by_uuid(
self.context, audit_template['uuid'])
self.assertEqual(audit_template['id'], audit_template.id)
def test_get_audit_template_that_does_not_exist(self):
self.assertRaises(exception.AuditTemplateNotFound,
self.dbapi.get_audit_template_by_id,
self.context, 1234)
def test_update_audit_template(self):
audit_template = self._create_test_audit_template()
res = self.dbapi.update_audit_template(audit_template['id'],
{'name': 'updated-model'})
self.assertEqual('updated-model', res.name)
def test_update_audit_template_that_does_not_exist(self):
self.assertRaises(exception.AuditTemplateNotFound,
self.dbapi.update_audit_template, 1234, {'name': ''})
def test_update_audit_template_uuid(self):
audit_template = self._create_test_audit_template()
self.assertRaises(exception.Invalid,
self.dbapi.update_audit_template,
audit_template['id'],
{'uuid': 'hello'})
def test_destroy_audit_template(self):
audit_template = self._create_test_audit_template()
self.dbapi.destroy_audit_template(audit_template['id'])
self.assertRaises(exception.AuditTemplateNotFound,
self.dbapi.get_audit_template_by_id,
self.context, audit_template['id'])
def test_destroy_audit_template_by_uuid(self):
uuid = w_utils.generate_uuid()
self._create_test_audit_template(uuid=uuid)
self.assertIsNotNone(self.dbapi.get_audit_template_by_uuid(
self.context, uuid))
self.dbapi.destroy_audit_template(uuid)
self.assertRaises(exception.AuditTemplateNotFound,
self.dbapi.get_audit_template_by_uuid,
self.context, uuid)
def test_destroy_audit_template_that_does_not_exist(self):
self.assertRaises(exception.AuditTemplateNotFound,
self.dbapi.destroy_audit_template, 1234)
def test_create_audit_template_already_exists(self):
uuid = w_utils.generate_uuid()
self._create_test_audit_template(id=1, uuid=uuid)
self.assertRaises(exception.AuditTemplateAlreadyExists,
self._create_test_audit_template,
id=2, uuid=uuid)
def test_audit_template_create_same_name(self):
audit_template1 = utils.create_test_audit_template(
uuid=w_utils.generate_uuid(),
name='audit_template_name')
self.assertEqual(audit_template1['uuid'], audit_template1.uuid)
self.assertRaises(
exception.AuditTemplateAlreadyExists,
utils.create_test_audit_template,
uuid=w_utils.generate_uuid(),
name='audit_template_name')
def test_audit_template_create_same_uuid(self):
uuid = w_utils.generate_uuid()
audit_template1 = utils.create_test_audit_template(
uuid=uuid,
name='audit_template_name_1')
self.assertEqual(audit_template1['uuid'], audit_template1.uuid)
self.assertRaises(
exception.AuditTemplateAlreadyExists,
utils.create_test_audit_template,
uuid=uuid,
name='audit_template_name_2')
Reference in New Issue View Git Blame Copy Permalink