440 lines
16 KiB
Python
440 lines
16 KiB
Python
# Copyright (c) 2020 Nokia Corporation.
|
|
#
|
|
# 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.
|
|
|
|
import datetime
|
|
|
|
from oslo_utils import uuidutils
|
|
|
|
from fenix.db import exceptions as db_exceptions
|
|
from fenix.db.sqlalchemy import api as db_api
|
|
from fenix import tests
|
|
|
|
|
|
def _get_fake_random_uuid():
|
|
return uuidutils.generate_uuid()
|
|
|
|
|
|
def _get_fake_uuid():
|
|
"""Returns a fake uuid."""
|
|
return 'aaaaaaaa-1111-bbbb-2222-cccccccccccc'
|
|
|
|
|
|
def _get_fake_uuid1():
|
|
return 'dddddddd-3333-eeee-4444-ffffffffffff'
|
|
|
|
|
|
def _get_datetime(value='2020-03-19 00:00'):
|
|
return datetime.datetime.strptime(value, '%Y-%m-%d %H:%M')
|
|
|
|
|
|
def _get_fake_session_values():
|
|
session = {
|
|
'session_id': _get_fake_uuid(),
|
|
'prev_state': None,
|
|
'state': 'MAINTENANCE',
|
|
'maintenance_at': _get_datetime(),
|
|
'meta': "{'openstack_version': 'Train'}",
|
|
'workflow': "default"}
|
|
return session
|
|
|
|
|
|
def _get_fake_session_db_values(db_object_dict):
|
|
db_session = _get_fake_session_values()
|
|
db_session['created_at'] = db_object_dict['created_at']
|
|
db_session['created_at'] = db_object_dict['created_at']
|
|
db_session['updated_at'] = None
|
|
return db_session
|
|
|
|
|
|
def _get_fake_action_plugin_values(uuid=_get_fake_uuid(),
|
|
plugin='compute_upgrade'):
|
|
adict = {'session_id': uuid,
|
|
'plugin': plugin,
|
|
'type': 'compute',
|
|
'meta': "{'os': 'Linux123', 'esw': 'OS_Train'}"}
|
|
return adict
|
|
|
|
|
|
def _get_fake_action_plugin_db_values(db_object_dict,
|
|
uuid=_get_fake_uuid(),
|
|
plugin='compute_upgrade'):
|
|
db_adict = _get_fake_action_plugin_values(uuid, plugin)
|
|
db_adict['id'] = db_object_dict['id']
|
|
db_adict['created_at'] = db_object_dict['created_at']
|
|
db_adict['updated_at'] = None
|
|
return db_adict
|
|
|
|
|
|
def _get_fake_action_plugin_instance_values(uuid=_get_fake_uuid()):
|
|
aidict = {'session_id': uuid,
|
|
'plugin': 'compute_upgrade',
|
|
'hostname': 'compute-1',
|
|
'state': 'DONE'}
|
|
return aidict
|
|
|
|
|
|
def _get_fake_action_plugin_instance_db_values(db_object_dict):
|
|
db_aidict = _get_fake_action_plugin_instance_values()
|
|
db_aidict['id'] = db_object_dict['id']
|
|
db_aidict['created_at'] = db_object_dict['created_at']
|
|
db_aidict['updated_at'] = None
|
|
return db_aidict
|
|
|
|
|
|
def _get_fake_download_values(uuid=_get_fake_uuid(),
|
|
local_file="/tmp/compute.tar.gz"):
|
|
ddict = {'session_id': uuid,
|
|
'local_file': local_file}
|
|
return ddict
|
|
|
|
|
|
def _get_fake_download_db_values(db_object_dict,
|
|
uuid=_get_fake_uuid(),
|
|
local_file="/tmp/compute.tar.gz"):
|
|
db_ddict = _get_fake_download_values(uuid, local_file)
|
|
db_ddict['id'] = db_object_dict['id']
|
|
db_ddict['created_at'] = db_object_dict['created_at']
|
|
db_ddict['updated_at'] = None
|
|
return db_ddict
|
|
|
|
|
|
def _get_fake_host_values(uuid=_get_fake_uuid(),
|
|
hostname='compute-1'):
|
|
hdict = {'session_id': uuid,
|
|
'hostname': hostname,
|
|
'type': 'compute',
|
|
'maintained': False,
|
|
'disabled': False,
|
|
'details': None}
|
|
return hdict
|
|
|
|
|
|
def _get_fake_host_db_values(db_object_dict, hostname='compute-1'):
|
|
db_hdict = _get_fake_host_values(hostname=hostname)
|
|
db_hdict['id'] = db_object_dict['id']
|
|
db_hdict['created_at'] = db_object_dict['created_at']
|
|
db_hdict['updated_at'] = None
|
|
return db_hdict
|
|
|
|
|
|
def _get_fake_project_values(uuid=_get_fake_uuid(),
|
|
project_id=_get_fake_uuid1()):
|
|
pdict = {'session_id': uuid,
|
|
'project_id': project_id,
|
|
'state': None}
|
|
return pdict
|
|
|
|
|
|
def _get_fake_project_db_values(db_object_dict, project_id=_get_fake_uuid1()):
|
|
db_pdict = _get_fake_project_values(project_id=project_id)
|
|
db_pdict['id'] = db_object_dict['id']
|
|
db_pdict['created_at'] = db_object_dict['created_at']
|
|
db_pdict['updated_at'] = None
|
|
return db_pdict
|
|
|
|
|
|
def _get_fake_instance_values(uuid=_get_fake_uuid(),
|
|
instance_id=_get_fake_uuid1()):
|
|
idict = {'session_id': uuid,
|
|
'instance_id': instance_id,
|
|
'state': 'Running',
|
|
'action': None,
|
|
'project_id': _get_fake_uuid1(),
|
|
'project_state': None,
|
|
'instance_name': 'Instance1',
|
|
'action_done': False,
|
|
'host': 'compute-1',
|
|
'details': None}
|
|
return idict
|
|
|
|
|
|
def _get_fake_instance_db_values(db_object_dict,
|
|
instance_id=_get_fake_uuid1()):
|
|
db_idict = _get_fake_instance_values(instance_id=instance_id)
|
|
db_idict['id'] = db_object_dict['id']
|
|
db_idict['created_at'] = db_object_dict['created_at']
|
|
db_idict['updated_at'] = None
|
|
return db_idict
|
|
|
|
|
|
def _get_fake_project_instance_values(uuid=_get_fake_uuid()):
|
|
idict = {'instance_id': uuid,
|
|
'project_id': _get_fake_uuid1(),
|
|
'group_id': _get_fake_uuid1(),
|
|
'instance_name': 'Instance1',
|
|
'max_interruption_time': 10,
|
|
'migration_type': 'LIVE_MIGRATION',
|
|
'resource_mitigation': False,
|
|
'lead_time': 30}
|
|
return idict
|
|
|
|
|
|
def _get_fake_project_instance_db_values(db_object_dict):
|
|
db_idict = _get_fake_project_instance_values()
|
|
db_idict['created_at'] = db_object_dict['created_at']
|
|
db_idict['updated_at'] = None
|
|
return db_idict
|
|
|
|
|
|
def _get_fake_instance_group_values(uuid=_get_fake_uuid()):
|
|
idict = {'group_id': uuid,
|
|
'project_id': _get_fake_uuid1(),
|
|
'group_name': 'ha_group',
|
|
'anti_affinity_group': True,
|
|
'max_instances_per_host': 1,
|
|
'max_impacted_members': 1,
|
|
'recovery_time': 10,
|
|
'resource_mitigation': False}
|
|
return idict
|
|
|
|
|
|
def _get_fake_instance_group_db_values(db_object_dict):
|
|
db_idict = _get_fake_instance_group_values()
|
|
db_idict['created_at'] = db_object_dict['created_at']
|
|
db_idict['updated_at'] = None
|
|
return db_idict
|
|
|
|
|
|
class SQLAlchemyDBApiTestCase(tests.DBTestCase):
|
|
"""Test case for SQLAlchemy DB API."""
|
|
|
|
def setUp(self):
|
|
super(SQLAlchemyDBApiTestCase, self).setUp()
|
|
|
|
# Maintenance session
|
|
|
|
def test_create_session(self):
|
|
"""Test maintenance session create
|
|
|
|
Create session and check results equals to given values.
|
|
"""
|
|
result = db_api.create_session(_get_fake_session_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_session_db_values(result.to_dict()))
|
|
|
|
# We cannot create duplicate, so no need to test
|
|
|
|
def test_remove_session(self):
|
|
"""Test maintenance session removal
|
|
|
|
Check session does not exist after removal
|
|
"""
|
|
self.assertRaises(db_exceptions.FenixDBNotFound,
|
|
db_api.remove_session, _get_fake_uuid())
|
|
|
|
db_api.create_session(_get_fake_session_values())
|
|
db_api.remove_session(_get_fake_uuid())
|
|
self.assertIsNone(
|
|
db_api.maintenance_session_get(_get_fake_uuid()))
|
|
|
|
def test_remove_session_all(self):
|
|
"""Test maintenance session removal with all tables
|
|
|
|
Remove maintenance session that includes all other
|
|
maintenance session tables.
|
|
"""
|
|
# TBD make other db cases first to make this one
|
|
pass
|
|
|
|
# Action plug-in
|
|
|
|
def test_create_action_plugin(self):
|
|
result = db_api.create_action_plugin(_get_fake_action_plugin_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_action_plugin_db_values(result.to_dict()))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_action_plugin,
|
|
_get_fake_action_plugin_values())
|
|
|
|
def test_create_action_plugins(self):
|
|
ap_list = [_get_fake_action_plugin_values(plugin='compute_upgrade'),
|
|
_get_fake_action_plugin_values(plugin='compute_os')]
|
|
results = db_api.create_action_plugins(ap_list)
|
|
for result in results:
|
|
self.assertEqual(
|
|
result.to_dict(),
|
|
_get_fake_action_plugin_db_values(result.to_dict(),
|
|
plugin=result.plugin))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_action_plugins,
|
|
[_get_fake_action_plugin_values(),
|
|
_get_fake_action_plugin_values()])
|
|
|
|
# Action plug-in instance
|
|
|
|
def test_create_action_plugin_instance(self):
|
|
result = db_api.create_action_plugin_instance(
|
|
_get_fake_action_plugin_instance_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_action_plugin_instance_db_values(
|
|
result.to_dict()))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_action_plugin_instance,
|
|
_get_fake_action_plugin_instance_values())
|
|
|
|
def test_remove_action_plugin_instance(self):
|
|
result = db_api.create_action_plugin_instance(
|
|
_get_fake_action_plugin_instance_values())
|
|
api = _get_fake_action_plugin_instance_values()
|
|
db_api.remove_action_plugin_instance(result)
|
|
self.assertIsNone(
|
|
db_api.action_plugin_instance_get(api['session_id'],
|
|
api['plugin'],
|
|
api['hostname']))
|
|
self.assertRaises(db_exceptions.FenixDBNotFound,
|
|
db_api.remove_action_plugin_instance, result)
|
|
|
|
# Download
|
|
|
|
def test_create_downloads(self):
|
|
downloads = [_get_fake_download_values(),
|
|
_get_fake_download_values(local_file="foo")]
|
|
results = db_api.create_downloads(downloads)
|
|
for result in results:
|
|
self.assertEqual(
|
|
result.to_dict(),
|
|
_get_fake_download_db_values(result.to_dict(),
|
|
local_file=result.local_file))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_downloads,
|
|
[_get_fake_download_values(),
|
|
_get_fake_download_values()])
|
|
|
|
# Host
|
|
|
|
def test_create_host(self):
|
|
result = db_api.create_host(_get_fake_host_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_host_db_values(
|
|
result.to_dict()))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_host,
|
|
_get_fake_host_values())
|
|
|
|
def test_create_hosts(self):
|
|
hosts = [_get_fake_host_values(),
|
|
_get_fake_host_values(hostname='compute-2')]
|
|
results = db_api.create_hosts(hosts)
|
|
for result in results:
|
|
self.assertEqual(
|
|
result.to_dict(),
|
|
_get_fake_host_db_values(result.to_dict(),
|
|
hostname=result.hostname))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_hosts,
|
|
[_get_fake_host_values(),
|
|
_get_fake_host_values()])
|
|
|
|
# Project
|
|
|
|
def test_create_project(self):
|
|
result = db_api.create_project(_get_fake_project_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_project_db_values(
|
|
result.to_dict()))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_project,
|
|
_get_fake_project_values())
|
|
|
|
def test_create_projects(self):
|
|
projects = [_get_fake_project_values(),
|
|
_get_fake_project_values(project_id="1234567890")]
|
|
results = db_api.create_projects(projects)
|
|
for result in results:
|
|
self.assertEqual(
|
|
result.to_dict(),
|
|
_get_fake_project_db_values(result.to_dict(),
|
|
project_id=result.project_id))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_projects,
|
|
[_get_fake_project_values(),
|
|
_get_fake_project_values()])
|
|
|
|
# Instance
|
|
|
|
def test_create_instance(self):
|
|
result = db_api.create_instance(_get_fake_instance_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_instance_db_values(
|
|
result.to_dict()))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_instance,
|
|
_get_fake_instance_values())
|
|
|
|
def test_create_instances(self):
|
|
instances = [_get_fake_instance_values(),
|
|
_get_fake_instance_values(instance_id="123456789")]
|
|
results = db_api.create_instances(instances)
|
|
for result in results:
|
|
self.assertEqual(
|
|
result.to_dict(),
|
|
_get_fake_instance_db_values(result.to_dict(),
|
|
instance_id=result.instance_id))
|
|
self.assertRaises(db_exceptions.FenixDBDuplicateEntry,
|
|
db_api.create_instances,
|
|
[_get_fake_instance_values(),
|
|
_get_fake_instance_values()])
|
|
|
|
def test_remove_instance(self):
|
|
db_api.create_instance(_get_fake_instance_values())
|
|
iv = _get_fake_instance_values()
|
|
db_api.remove_instance(iv['session_id'], iv['instance_id'])
|
|
self.assertIsNone(
|
|
db_api.instance_get(iv['session_id'], iv['instance_id']))
|
|
self.assertRaises(db_exceptions.FenixDBNotFound,
|
|
db_api.remove_instance,
|
|
iv['session_id'],
|
|
iv['instance_id'])
|
|
|
|
# Project instances
|
|
|
|
def test_update_project_instance(self):
|
|
result = db_api.update_project_instance(
|
|
_get_fake_project_instance_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_project_instance_db_values(
|
|
result.to_dict()))
|
|
# No need to test duplicate as always overwritten
|
|
|
|
def test_remove_project_instance(self):
|
|
db_api.update_project_instance(_get_fake_project_instance_values())
|
|
pi = _get_fake_project_instance_values()
|
|
db_api.remove_project_instance(pi['instance_id'])
|
|
self.assertIsNone(
|
|
db_api.project_instance_get(pi['instance_id']))
|
|
self.assertRaises(db_exceptions.FenixDBNotFound,
|
|
db_api.remove_project_instance,
|
|
pi['instance_id'])
|
|
|
|
# Instances groups
|
|
|
|
def test_update_instance_group(self):
|
|
result = db_api.update_instance_group(
|
|
_get_fake_instance_group_values())
|
|
self.assertEqual(result.to_dict(),
|
|
_get_fake_instance_group_db_values(
|
|
result.to_dict()))
|
|
# No need to test duplicate as always overwritten
|
|
|
|
def test_remove_instance_group(self):
|
|
db_api.update_instance_group(_get_fake_instance_group_values())
|
|
gi = _get_fake_instance_group_values()
|
|
db_api.remove_instance_group(gi['group_id'])
|
|
self.assertIsNone(
|
|
db_api.instance_group_get(gi['group_id']))
|
|
self.assertRaises(db_exceptions.FenixDBNotFound,
|
|
db_api.remove_instance_group,
|
|
gi['group_id'])
|