368 lines
14 KiB
Python
368 lines
14 KiB
Python
# -*- encoding: utf-8 -*-
|
|
# Copyright (c) 2016 b<>com
|
|
#
|
|
# 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 Strategy via the DB API"""
|
|
|
|
import freezegun
|
|
|
|
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 TestDbStrategyFilters(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(TestDbStrategyFilters, self).setUp()
|
|
self.context.show_deleted = True
|
|
self._data_setup()
|
|
|
|
def _data_setup(self):
|
|
strategy1_name = "STRATEGY_ID_1"
|
|
strategy2_name = "STRATEGY_ID_2"
|
|
strategy3_name = "STRATEGY_ID_3"
|
|
|
|
self.goal1 = utils.create_test_goal(
|
|
id=1, uuid=w_utils.generate_uuid(),
|
|
name="GOAL_ID", display_name="Goal")
|
|
self.goal2 = utils.create_test_goal(
|
|
id=2, uuid=w_utils.generate_uuid(),
|
|
name="DUMMY", display_name="Dummy")
|
|
|
|
with freezegun.freeze_time(self.FAKE_TODAY):
|
|
self.strategy1 = utils.create_test_strategy(
|
|
id=1, uuid=w_utils.generate_uuid(),
|
|
name=strategy1_name, display_name="Strategy 1",
|
|
goal_id=self.goal1.id)
|
|
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
|
self.strategy2 = utils.create_test_strategy(
|
|
id=2, uuid=w_utils.generate_uuid(),
|
|
name=strategy2_name, display_name="Strategy 2",
|
|
goal_id=self.goal1.id)
|
|
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
|
self.strategy3 = utils.create_test_strategy(
|
|
id=3, uuid=w_utils.generate_uuid(),
|
|
name=strategy3_name, display_name="Strategy 3",
|
|
goal_id=self.goal2.id)
|
|
|
|
def _soft_delete_strategys(self):
|
|
with freezegun.freeze_time(self.FAKE_TODAY):
|
|
self.dbapi.soft_delete_strategy(self.strategy1.id)
|
|
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
|
self.dbapi.soft_delete_strategy(self.strategy2.id)
|
|
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
|
self.dbapi.soft_delete_strategy(self.strategy3.id)
|
|
|
|
def _update_strategies(self):
|
|
with freezegun.freeze_time(self.FAKE_TODAY):
|
|
self.dbapi.update_strategy(
|
|
self.strategy1.id, values={"display_name": "strategy1"})
|
|
with freezegun.freeze_time(self.FAKE_OLD_DATE):
|
|
self.dbapi.update_strategy(
|
|
self.strategy2.id, values={"display_name": "strategy2"})
|
|
with freezegun.freeze_time(self.FAKE_OLDER_DATE):
|
|
self.dbapi.update_strategy(
|
|
self.strategy3.id, values={"display_name": "strategy3"})
|
|
|
|
def test_get_strategy_list_filter_deleted_true(self):
|
|
with freezegun.freeze_time(self.FAKE_TODAY):
|
|
self.dbapi.soft_delete_strategy(self.strategy1.id)
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'deleted': True})
|
|
|
|
self.assertEqual([self.strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_list_filter_deleted_false(self):
|
|
with freezegun.freeze_time(self.FAKE_TODAY):
|
|
self.dbapi.soft_delete_strategy(self.strategy1.id)
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'deleted': False})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy2['uuid'], self.strategy3['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
def test_get_strategy_list_filter_deleted_at_eq(self):
|
|
self._soft_delete_strategys()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'deleted_at__eq': self.FAKE_TODAY})
|
|
|
|
self.assertEqual([self.strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_list_filter_deleted_at_lt(self):
|
|
self._soft_delete_strategys()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'deleted_at__lt': self.FAKE_TODAY})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy2['uuid'], self.strategy3['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
def test_get_strategy_list_filter_deleted_at_lte(self):
|
|
self._soft_delete_strategys()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'deleted_at__lte': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy2['uuid'], self.strategy3['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
def test_get_strategy_list_filter_deleted_at_gt(self):
|
|
self._soft_delete_strategys()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'deleted_at__gt': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual([self.strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_list_filter_deleted_at_gte(self):
|
|
self._soft_delete_strategys()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'deleted_at__gte': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy1['uuid'], self.strategy2['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
# created_at #
|
|
|
|
def test_get_strategy_list_filter_created_at_eq(self):
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'created_at__eq': self.FAKE_TODAY})
|
|
|
|
self.assertEqual([self.strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_list_filter_created_at_lt(self):
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'created_at__lt': self.FAKE_TODAY})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy2['uuid'], self.strategy3['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
def test_get_strategy_list_filter_created_at_lte(self):
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'created_at__lte': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy2['uuid'], self.strategy3['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
def test_get_strategy_list_filter_created_at_gt(self):
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'created_at__gt': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual([self.strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_list_filter_created_at_gte(self):
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'created_at__gte': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy1['uuid'], self.strategy2['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
# updated_at #
|
|
|
|
def test_get_strategy_list_filter_updated_at_eq(self):
|
|
self._update_strategies()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'updated_at__eq': self.FAKE_TODAY})
|
|
|
|
self.assertEqual([self.strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_list_filter_updated_at_lt(self):
|
|
self._update_strategies()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'updated_at__lt': self.FAKE_TODAY})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy2['uuid'], self.strategy3['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
def test_get_strategy_list_filter_updated_at_lte(self):
|
|
self._update_strategies()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'updated_at__lte': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy2['uuid'], self.strategy3['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
def test_get_strategy_list_filter_updated_at_gt(self):
|
|
self._update_strategies()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'updated_at__gt': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual([self.strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_list_filter_updated_at_gte(self):
|
|
self._update_strategies()
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'updated_at__gte': self.FAKE_OLD_DATE})
|
|
|
|
self.assertEqual(
|
|
set([self.strategy1['uuid'], self.strategy2['uuid']]),
|
|
set([r.uuid for r in res]))
|
|
|
|
|
|
class DbStrategyTestCase(base.DbTestCase):
|
|
|
|
def test_get_strategy_list(self):
|
|
uuids = []
|
|
for i in range(1, 4):
|
|
strategy = utils.create_test_strategy(
|
|
id=i,
|
|
uuid=w_utils.generate_uuid(),
|
|
name="STRATEGY_ID_%s" % i,
|
|
display_name='My Strategy {0}'.format(i))
|
|
uuids.append(str(strategy['uuid']))
|
|
strategies = self.dbapi.get_strategy_list(self.context)
|
|
strategy_uuids = [s.uuid for s in strategies]
|
|
self.assertEqual(sorted(uuids), sorted(strategy_uuids))
|
|
for strategy in strategies:
|
|
self.assertIsNone(strategy.goal)
|
|
|
|
def test_get_strategy_list_eager(self):
|
|
_goal = utils.get_test_goal()
|
|
goal = self.dbapi.create_goal(_goal)
|
|
uuids = []
|
|
for i in range(1, 4):
|
|
strategy = utils.create_test_strategy(
|
|
id=i,
|
|
uuid=w_utils.generate_uuid(),
|
|
name="STRATEGY_ID_%s" % i,
|
|
display_name='My Strategy {0}'.format(i),
|
|
goal_id=goal.id)
|
|
uuids.append(str(strategy['uuid']))
|
|
strategys = self.dbapi.get_strategy_list(self.context, eager=True)
|
|
strategy_map = {a.uuid: a for a in strategys}
|
|
self.assertEqual(sorted(uuids), sorted(strategy_map.keys()))
|
|
eager_strategy = strategy_map[strategy.uuid]
|
|
self.assertEqual(goal.as_dict(), eager_strategy.goal.as_dict())
|
|
|
|
def test_get_strategy_list_with_filters(self):
|
|
# NOTE(erakli): we don't create goal in database but links to
|
|
# goal_id = 1. There is no error in dbapi.create_strategy() method.
|
|
# Is it right behaviour?
|
|
strategy1 = utils.create_test_strategy(
|
|
id=1,
|
|
uuid=w_utils.generate_uuid(),
|
|
name="STRATEGY_ID_1",
|
|
display_name='Strategy 1',
|
|
)
|
|
strategy2 = utils.create_test_strategy(
|
|
id=2,
|
|
uuid=w_utils.generate_uuid(),
|
|
name="STRATEGY_ID_2",
|
|
display_name='Strategy 2',
|
|
)
|
|
strategy3 = utils.create_test_strategy(
|
|
id=3,
|
|
uuid=w_utils.generate_uuid(),
|
|
name="STRATEGY_ID_3",
|
|
display_name='Strategy 3',
|
|
)
|
|
|
|
self.dbapi.soft_delete_strategy(strategy3['uuid'])
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'display_name': 'Strategy 1'})
|
|
self.assertEqual([strategy1['uuid']], [r.uuid for r in res])
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'display_name': 'Strategy 3'})
|
|
self.assertEqual([], [r.uuid for r in res])
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'goal_id': 1})
|
|
self.assertEqual([strategy1['uuid'], strategy2['uuid']],
|
|
[r.uuid for r in res])
|
|
|
|
res = self.dbapi.get_strategy_list(
|
|
self.context, filters={'display_name': 'Strategy 2'})
|
|
self.assertEqual([strategy2['uuid']], [r.uuid for r in res])
|
|
|
|
def test_get_strategy_by_uuid(self):
|
|
created_strategy = utils.create_test_strategy()
|
|
strategy = self.dbapi.get_strategy_by_uuid(
|
|
self.context, created_strategy['uuid'])
|
|
self.assertEqual(strategy.uuid, created_strategy['uuid'])
|
|
|
|
def test_get_strategy_by_name(self):
|
|
created_strategy = utils.create_test_strategy()
|
|
strategy = self.dbapi.get_strategy_by_name(
|
|
self.context, created_strategy['name'])
|
|
self.assertEqual(strategy.name, created_strategy['name'])
|
|
|
|
def test_get_strategy_that_does_not_exist(self):
|
|
self.assertRaises(exception.StrategyNotFound,
|
|
self.dbapi.get_strategy_by_id,
|
|
self.context, 404)
|
|
|
|
def test_update_strategy(self):
|
|
strategy = utils.create_test_strategy()
|
|
res = self.dbapi.update_strategy(
|
|
strategy['uuid'], {'display_name': 'updated-model'})
|
|
self.assertEqual('updated-model', res.display_name)
|
|
|
|
def test_update_goal_id(self):
|
|
strategy = utils.create_test_strategy()
|
|
self.assertRaises(exception.Invalid,
|
|
self.dbapi.update_strategy, strategy['uuid'],
|
|
{'uuid': 'new_strategy_id'})
|
|
|
|
def test_update_strategy_that_does_not_exist(self):
|
|
self.assertRaises(exception.StrategyNotFound,
|
|
self.dbapi.update_strategy,
|
|
404,
|
|
{'display_name': ''})
|
|
|
|
def test_destroy_strategy(self):
|
|
strategy = utils.create_test_strategy()
|
|
self.dbapi.destroy_strategy(strategy['uuid'])
|
|
self.assertRaises(exception.StrategyNotFound,
|
|
self.dbapi.get_strategy_by_id,
|
|
self.context, strategy['uuid'])
|
|
|
|
def test_destroy_strategy_that_does_not_exist(self):
|
|
self.assertRaises(exception.StrategyNotFound,
|
|
self.dbapi.destroy_strategy, 404)
|
|
|
|
def test_create_strategy_already_exists(self):
|
|
strategy_id = "STRATEGY_ID"
|
|
utils.create_test_strategy(name=strategy_id)
|
|
self.assertRaises(exception.StrategyAlreadyExists,
|
|
utils.create_test_strategy,
|
|
name=strategy_id)
|