Added EfficacyIndicator model in DB
In this changeset, I created a new model named EfficacyIndicator which is responsible for storing information regarding the efficacy indicators that were computed by the strategy in its solution. Every efficacy indicator should relate to a single Action Plan. Partially Implements: blueprint efficacy-indicator Change-Id: Ifc14ea5e16e92f032d7912c9b3fdc270af79cab6
This commit is contained in:
Vincent Françoise
385
watcher/tests/db/test_efficacy_indicator.py
Normal file
385
watcher/tests/db/test_efficacy_indicator.py
Normal file
@@ -0,0 +1,385 @@
|
||||
# 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 EfficacyIndicator 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 TestDbEfficacyIndicatorFilters(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(TestDbEfficacyIndicatorFilters, 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.efficacy_indicator1 = utils.create_test_efficacy_indicator(
|
||||
action_plan_id=self.action_plan.id, id=1, uuid=None,
|
||||
name="efficacy_indicator1", description="Test Indicator 1")
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.efficacy_indicator2 = utils.create_test_efficacy_indicator(
|
||||
action_plan_id=self.action_plan.id, id=2, uuid=None,
|
||||
name="efficacy_indicator2", description="Test Indicator 2")
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.efficacy_indicator3 = utils.create_test_efficacy_indicator(
|
||||
action_plan_id=self.action_plan.id, id=3, uuid=None,
|
||||
name="efficacy_indicator3", description="Test Indicator 3")
|
||||
|
||||
def _soft_delete_efficacy_indicators(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_efficacy_indicator(
|
||||
self.efficacy_indicator1.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.soft_delete_efficacy_indicator(
|
||||
self.efficacy_indicator2.uuid)
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.soft_delete_efficacy_indicator(
|
||||
self.efficacy_indicator3.uuid)
|
||||
|
||||
def _update_efficacy_indicators(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.update_efficacy_indicator(
|
||||
self.efficacy_indicator1.uuid,
|
||||
values={"description": "New decription 1"})
|
||||
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
||||
self.dbapi.update_efficacy_indicator(
|
||||
self.efficacy_indicator2.uuid,
|
||||
values={"description": "New decription 2"})
|
||||
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
||||
self.dbapi.update_efficacy_indicator(
|
||||
self.efficacy_indicator3.uuid,
|
||||
values={"description": "New decription 3"})
|
||||
|
||||
def test_get_efficacy_indicator_filter_deleted_true(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_efficacy_indicator(
|
||||
self.efficacy_indicator1.uuid)
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'deleted': True})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_deleted_false(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
self.dbapi.soft_delete_efficacy_indicator(
|
||||
self.efficacy_indicator1.uuid)
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'deleted': False})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator2['id'],
|
||||
self.efficacy_indicator3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_deleted_at_eq(self):
|
||||
self._soft_delete_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'deleted_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_deleted_at_lt(self):
|
||||
self._soft_delete_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'deleted_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator2['id'], self.efficacy_indicator3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_deleted_at_lte(self):
|
||||
self._soft_delete_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'deleted_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator2['id'], self.efficacy_indicator3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_deleted_at_gt(self):
|
||||
self._soft_delete_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'deleted_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_deleted_at_gte(self):
|
||||
self._soft_delete_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'deleted_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator1['id'], self.efficacy_indicator2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# created_at #
|
||||
|
||||
def test_get_efficacy_indicator_filter_created_at_eq(self):
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'created_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_created_at_lt(self):
|
||||
with freezegun.freeze_time(self.FAKE_TODAY):
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'created_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator2['id'], self.efficacy_indicator3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_created_at_lte(self):
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'created_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator2['id'], self.efficacy_indicator3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_created_at_gt(self):
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'created_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_created_at_gte(self):
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'created_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator1['id'], self.efficacy_indicator2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
# updated_at #
|
||||
|
||||
def test_get_efficacy_indicator_filter_updated_at_eq(self):
|
||||
self._update_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'updated_at__eq': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_updated_at_lt(self):
|
||||
self._update_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'updated_at__lt': self.FAKE_TODAY})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator2['id'], self.efficacy_indicator3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_updated_at_lte(self):
|
||||
self._update_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'updated_at__lte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator2['id'], self.efficacy_indicator3['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_updated_at_gt(self):
|
||||
self._update_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'updated_at__gt': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual([self.efficacy_indicator1['id']], [r.id for r in res])
|
||||
|
||||
def test_get_efficacy_indicator_filter_updated_at_gte(self):
|
||||
self._update_efficacy_indicators()
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'updated_at__gte': self.FAKE_OLD_DATE})
|
||||
|
||||
self.assertEqual(
|
||||
[self.efficacy_indicator1['id'], self.efficacy_indicator2['id']],
|
||||
[r.id for r in res])
|
||||
|
||||
|
||||
class DbEfficacyIndicatorTestCase(base.DbTestCase):
|
||||
|
||||
def _create_test_efficacy_indicator(self, **kwargs):
|
||||
efficacy_indicator_dict = utils.get_test_efficacy_indicator(**kwargs)
|
||||
efficacy_indicator = self.dbapi.create_efficacy_indicator(
|
||||
efficacy_indicator_dict)
|
||||
return efficacy_indicator
|
||||
|
||||
def _create_test_action_plan(self, **kwargs):
|
||||
action_plan_dict = utils.get_test_action_plan(**kwargs)
|
||||
action_plan = self.dbapi.create_action_plan(action_plan_dict)
|
||||
return action_plan
|
||||
|
||||
def test_get_efficacy_indicator_list(self):
|
||||
uuids = []
|
||||
action_plan = self._create_test_action_plan()
|
||||
for id_ in range(1, 6):
|
||||
efficacy_indicator = utils.create_test_efficacy_indicator(
|
||||
action_plan_id=action_plan.id, id=id_, uuid=None,
|
||||
name="efficacy_indicator", description="Test Indicator ")
|
||||
uuids.append(six.text_type(efficacy_indicator['uuid']))
|
||||
res = self.dbapi.get_efficacy_indicator_list(self.context)
|
||||
res_uuids = [r.uuid for r in res]
|
||||
self.assertEqual(uuids.sort(), res_uuids.sort())
|
||||
|
||||
def test_get_efficacy_indicator_list_with_filters(self):
|
||||
audit = utils.create_test_audit(uuid=w_utils.generate_uuid())
|
||||
action_plan = self._create_test_action_plan(
|
||||
id=1,
|
||||
uuid=w_utils.generate_uuid(),
|
||||
audit_id=audit.id,
|
||||
first_efficacy_indicator_id=None,
|
||||
state='RECOMMENDED')
|
||||
efficacy_indicator1 = self._create_test_efficacy_indicator(
|
||||
id=1,
|
||||
name='indicator_1',
|
||||
uuid=w_utils.generate_uuid(),
|
||||
action_plan_id=1,
|
||||
description='Description efficacy indicator 1',
|
||||
unit='%')
|
||||
efficacy_indicator2 = self._create_test_efficacy_indicator(
|
||||
id=2,
|
||||
name='indicator_2',
|
||||
uuid=w_utils.generate_uuid(),
|
||||
action_plan_id=2,
|
||||
description='Description efficacy indicator 2',
|
||||
unit='%')
|
||||
efficacy_indicator3 = self._create_test_efficacy_indicator(
|
||||
id=3,
|
||||
name='indicator_3',
|
||||
uuid=w_utils.generate_uuid(),
|
||||
action_plan_id=1,
|
||||
description='Description efficacy indicator 3',
|
||||
unit='%')
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'name': 'indicator_3'})
|
||||
self.assertEqual([efficacy_indicator3['id']], [r.id for r in res])
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'unit': 'kWh'})
|
||||
self.assertEqual([], [r.id for r in res])
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context,
|
||||
filters={'action_plan_id': 2})
|
||||
self.assertEqual([efficacy_indicator2['id']], [r.id for r in res])
|
||||
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context,
|
||||
filters={'action_plan_uuid': action_plan['uuid']})
|
||||
self.assertEqual(
|
||||
[efficacy_indicator1['id'], efficacy_indicator3['id']].sort(),
|
||||
[r.id for r in res].sort())
|
||||
|
||||
def test_get_efficacy_indicator_list_with_filter_by_uuid(self):
|
||||
efficacy_indicator = self._create_test_efficacy_indicator()
|
||||
res = self.dbapi.get_efficacy_indicator_list(
|
||||
self.context, filters={'uuid': efficacy_indicator.uuid})
|
||||
|
||||
self.assertEqual(len(res), 1)
|
||||
self.assertEqual(efficacy_indicator.uuid, res[0].uuid)
|
||||
|
||||
def test_get_efficacy_indicator_by_id(self):
|
||||
efficacy_indicator = self._create_test_efficacy_indicator()
|
||||
efficacy_indicator = self.dbapi.get_efficacy_indicator_by_id(
|
||||
self.context, efficacy_indicator.id)
|
||||
self.assertEqual(efficacy_indicator.uuid, efficacy_indicator.uuid)
|
||||
|
||||
def test_get_efficacy_indicator_by_uuid(self):
|
||||
efficacy_indicator = self._create_test_efficacy_indicator()
|
||||
efficacy_indicator = self.dbapi.get_efficacy_indicator_by_uuid(
|
||||
self.context, efficacy_indicator.uuid)
|
||||
self.assertEqual(efficacy_indicator['id'], efficacy_indicator.id)
|
||||
|
||||
def test_get_efficacy_indicator_that_does_not_exist(self):
|
||||
self.assertRaises(
|
||||
exception.EfficacyIndicatorNotFound,
|
||||
self.dbapi.get_efficacy_indicator_by_id, self.context, 1234)
|
||||
|
||||
def test_update_efficacy_indicator(self):
|
||||
efficacy_indicator = self._create_test_efficacy_indicator()
|
||||
res = self.dbapi.update_efficacy_indicator(
|
||||
efficacy_indicator.id, {'state': 'CANCELLED'})
|
||||
self.assertEqual('CANCELLED', res.state)
|
||||
|
||||
def test_update_efficacy_indicator_that_does_not_exist(self):
|
||||
self.assertRaises(
|
||||
exception.EfficacyIndicatorNotFound,
|
||||
self.dbapi.update_efficacy_indicator, 1234, {'state': ''})
|
||||
|
||||
def test_update_efficacy_indicator_uuid(self):
|
||||
efficacy_indicator = self._create_test_efficacy_indicator()
|
||||
self.assertRaises(
|
||||
exception.Invalid,
|
||||
self.dbapi.update_efficacy_indicator, efficacy_indicator.id,
|
||||
{'uuid': 'hello'})
|
||||
|
||||
def test_destroy_efficacy_indicator(self):
|
||||
efficacy_indicator = self._create_test_efficacy_indicator()
|
||||
self.dbapi.destroy_efficacy_indicator(efficacy_indicator['id'])
|
||||
self.assertRaises(exception.EfficacyIndicatorNotFound,
|
||||
self.dbapi.get_efficacy_indicator_by_id,
|
||||
self.context, efficacy_indicator['id'])
|
||||
|
||||
def test_destroy_efficacy_indicator_by_uuid(self):
|
||||
uuid = w_utils.generate_uuid()
|
||||
self._create_test_efficacy_indicator(uuid=uuid)
|
||||
self.assertIsNotNone(self.dbapi.get_efficacy_indicator_by_uuid(
|
||||
self.context, uuid))
|
||||
self.dbapi.destroy_efficacy_indicator(uuid)
|
||||
self.assertRaises(
|
||||
exception.EfficacyIndicatorNotFound,
|
||||
self.dbapi.get_efficacy_indicator_by_uuid, self.context, uuid)
|
||||
|
||||
def test_destroy_efficacy_indicator_that_does_not_exist(self):
|
||||
self.assertRaises(exception.EfficacyIndicatorNotFound,
|
||||
self.dbapi.destroy_efficacy_indicator, 1234)
|
||||
|
||||
def test_create_efficacy_indicator_already_exists(self):
|
||||
uuid = w_utils.generate_uuid()
|
||||
self._create_test_efficacy_indicator(id=1, uuid=uuid)
|
||||
self.assertRaises(exception.EfficacyIndicatorAlreadyExists,
|
||||
self._create_test_efficacy_indicator,
|
||||
id=2, uuid=uuid)
|
||||
@@ -273,7 +273,12 @@ class DbGoalTestCase(base.DbTestCase):
|
||||
self.assertEqual([goal2['uuid']], [r.uuid for r in res])
|
||||
|
||||
def test_get_goal_by_uuid(self):
|
||||
created_goal = self._create_test_goal()
|
||||
efficacy_spec = [{"unit": "%", "name": "dummy",
|
||||
"schema": "Range(min=0, max=100, min_included=True, "
|
||||
"max_included=True, msg=None)",
|
||||
"description": "Dummy indicator"}]
|
||||
created_goal = self._create_test_goal(
|
||||
efficacy_specification=efficacy_spec)
|
||||
goal = self.dbapi.get_goal_by_uuid(self.context, created_goal['uuid'])
|
||||
self.assertEqual(goal.uuid, created_goal['uuid'])
|
||||
|
||||
|
||||
@@ -189,3 +189,33 @@ def create_test_strategy(**kwargs):
|
||||
strategy = get_test_strategy(**kwargs)
|
||||
dbapi = db_api.get_instance()
|
||||
return dbapi.create_strategy(strategy)
|
||||
|
||||
|
||||
def get_test_efficacy_indicator(**kwargs):
|
||||
return {
|
||||
'id': kwargs.get('id', 1),
|
||||
'uuid': kwargs.get('uuid', '202cfcf9-811c-411a-8a35-d8351f64eb24'),
|
||||
'name': kwargs.get('name', 'test_indicator'),
|
||||
'description': kwargs.get('description', 'Test indicator'),
|
||||
'unit': kwargs.get('unit', '%'),
|
||||
'value': kwargs.get('value', 0),
|
||||
'action_plan_id': kwargs.get('action_plan_id', 1),
|
||||
'created_at': kwargs.get('created_at'),
|
||||
'updated_at': kwargs.get('updated_at'),
|
||||
'deleted_at': kwargs.get('deleted_at'),
|
||||
}
|
||||
|
||||
|
||||
def create_test_efficacy_indicator(**kwargs):
|
||||
"""Create and return a test efficacy indicator entry in DB.
|
||||
|
||||
Function to be used to create test EfficacyIndicator objects in the DB.
|
||||
:param kwargs: kwargs for overriding the values of the attributes
|
||||
:returns: Test EfficacyIndicator DB object.
|
||||
"""
|
||||
efficacy_indicator = get_test_efficacy_indicator(**kwargs)
|
||||
# Let DB generate ID if it isn't specified explicitly
|
||||
if 'id' not in kwargs:
|
||||
del efficacy_indicator['id']
|
||||
dbapi = db_api.get_instance()
|
||||
return dbapi.create_efficacy_indicator(efficacy_indicator)
|
||||
|
||||
Reference in New Issue
Block a user