openstack/blazar
Code Issues Proposed changes
822e568921
blazar/climate/tests/db/sqlalchemy/test_sqlalchemy_api.py
Cristian A Sanchez 6b26ecaf75 Change references of tenant to project 
Refernces from tenant to project were made in several
places, including DB, HTTP Headers, configuration items,
json in responses and devstack integration. Additionally,
a migration script using Alembic was included.

Implements: blueprint tenant-to-project-ref
Change-Id: I4b21182f555ccd412a4ca4e0ce753f07edcd07f8
2014-04-17 11:16:09 -03:00

945 lines
36 KiB
Python
Raw Blame History

# Copyright (c) 2013 Bull.
#
# 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
import operator
import uuid
from climate.db import exceptions as db_exceptions
from climate.db.sqlalchemy import api as db_api
from climate.db.sqlalchemy import models
from climate import tests
from climate.plugins import instances as vm_plugin
from climate.plugins import oshosts as host_plugin
def _get_fake_random_uuid():
return unicode(str(uuid.uuid4()))
def _get_fake_lease_uuid():
"""Returns a fake uuid."""
return 'aaaaaaaa-1111-bbbb-2222-cccccccccccc'
def _get_fake_phys_reservation_values(id=_get_fake_random_uuid(),
lease_id=_get_fake_lease_uuid(),
resource_id=None):
return {'id': id,
'lease_id': lease_id,
'resource_id': '1234' if not resource_id else resource_id,
'resource_type': host_plugin.RESOURCE_TYPE,
'hypervisor_properties': '[\"=\", \"$hypervisor_type\", \"QEMU\"]',
'resource_properties': '',
'min': 1, 'max': 1}
def _get_fake_virt_reservation_values(lease_id=_get_fake_lease_uuid(),
resource_id=None):
return {'lease_id': lease_id,
'resource_id': '5678' if not resource_id else resource_id,
'resource_type': vm_plugin.RESOURCE_TYPE}
def _get_fake_event_values(id=_get_fake_random_uuid(),
lease_id=_get_fake_lease_uuid(),
event_type='fake_event_type',
time=None,
status='fake_event_status'):
return {'id': id,
'lease_id': lease_id,
'event_type': event_type,
'time': _get_datetime('2030-03-01 00:00') if not time else time,
'status': status}
def _get_datetime(value='2030-01-01 00:00'):
return datetime.datetime.strptime(value, '%Y-%m-%d %H:%M')
def _get_fake_virt_lease_values(id=_get_fake_lease_uuid(),
name='fake_virt_lease',
start_date=_get_datetime('2030-01-01 00:00'),
end_date=_get_datetime('2030-01-02 00:00'),
resource_id=None):
return {'id': id,
'name': name,
'user_id': _get_fake_random_uuid(),
'project_id': _get_fake_random_uuid(),
'start_date': start_date,
'end_date': end_date,
'trust': 'trust',
'reservations': [_get_fake_virt_reservation_values(
lease_id=id,
resource_id=resource_id)],
'events': []
}
def _get_fake_phys_lease_values(id=_get_fake_lease_uuid(),
name='fake_phys_lease',
start_date=_get_datetime('2030-01-01 00:00'),
end_date=_get_datetime('2030-01-02 00:00'),
resource_id=None):
return {'id': id,
'name': name,
'user_id': 'fake',
'project_id': 'fake',
'start_date': start_date,
'end_date': end_date,
'trust': 'trust',
'reservations': [_get_fake_phys_reservation_values(
id=_get_fake_random_uuid(),
lease_id=id,
resource_id=resource_id)],
'events': []
}
def _get_fake_host_allocation_values(
id=None,
compute_host_id=_get_fake_random_uuid(),
reservation_id=_get_fake_random_uuid()):
values = {'compute_host_id': compute_host_id,
'reservation_id': reservation_id}
if id is not None:
values.update({'id': id})
return values
def _create_virtual_lease(values=_get_fake_virt_lease_values(),
random=False):
"""Creating fake lease having a single virtual resource."""
return db_api.lease_create(values)
def _create_physical_lease(values=_get_fake_phys_lease_values(),
random=False):
"""Creating fake lease having a single physical resource."""
if random is True:
values = _get_fake_phys_lease_values(id=_get_fake_random_uuid(),
name=_get_fake_random_uuid())
lease = db_api.lease_create(values)
phys_res = _get_fake_phys_reservation_values()
for reservation in db_api.reservation_get_all_by_lease_id(lease['id']):
allocation_values = {
'id': _get_fake_random_uuid(),
'compute_host_id': values['reservations'][0]['resource_id'],
'reservation_id': reservation['id']
}
db_api.host_allocation_create(allocation_values)
computehost_reservations = {
'id': _get_fake_random_uuid(),
'reservation_id': values['reservations'][0]['id'],
'resource_properties': phys_res['resource_properties'],
'hypervisor_properties': phys_res['hypervisor_properties'],
'count_range': "{0} - {1}".format(phys_res['min'],
phys_res['max'])
}
db_api.host_reservation_create(computehost_reservations)
return lease
def _get_fake_host_reservation_values(id=_get_fake_random_uuid(),
reservation_id=_get_fake_random_uuid()):
return {'id': id,
'reservation_id': reservation_id,
'resource_properties': "fake",
'hypervisor_properties': "fake",
'min': 1, 'max': 1}
def _get_fake_cpu_info():
return str({'vendor': 'Intel',
'model': 'Westmere',
'arch': 'x86_64',
'features': ['rdtscp', 'pdpe1gb', 'hypervisor', 'vmx', 'ss',
'vme'],
'topology': {'cores': 1, 'threads': 1, 'sockets': 2}})
def _get_fake_host_values(id=_get_fake_random_uuid(), mem=8192, disk=10):
return {'id': id,
'vcpus': 1,
'cpu_info': _get_fake_cpu_info(),
'hypervisor_type': 'QEMU',
'hypervisor_version': 1000,
'memory_mb': mem,
'local_gb': disk,
'status': 'free'
}
def _get_fake_host_extra_capabilities(id=_get_fake_random_uuid(),
computehost_id=_get_fake_random_uuid()):
return {'id': id,
'computehost_id': computehost_id,
'capability_name': 'vgpu',
'capability_value': '2'}
def is_result_sorted_correctly(results, sort_key, sort_dir='asc'):
sorted_list = sorted(results,
key=operator.itemgetter(sort_key),
reverse=False if sort_dir == 'asc' else True)
return sorted_list == results
class SQLAlchemyDBApiTestCase(tests.DBTestCase):
"""Test case for SQLAlchemy DB API."""
def setUp(self):
super(SQLAlchemyDBApiTestCase, self).setUp()
def test_model_query(self):
lease = db_api.lease_create(_get_fake_virt_lease_values())
query = db_api.model_query(models.Lease)
self.assertEqual([lease.to_dict()], [l.to_dict() for l in query.all()])
def test_create_virt_lease(self):
"""Create a virtual lease and verify that all tables have been
populated.
"""
result = db_api.lease_create(_get_fake_virt_lease_values())
self.assertEqual(result['name'],
_get_fake_virt_lease_values()['name'])
self.assertEqual(0, len(db_api.event_get_all()))
self.assertEqual(1, len(db_api.reservation_get_all()))
def test_create_phys_lease(self):
"""Create a physical lease and verify that all tables have been
populated.
"""
result = db_api.lease_create(_get_fake_phys_lease_values())
self.assertEqual(result['name'],
_get_fake_phys_lease_values()['name'])
self.assertEqual(0, len(db_api.event_get_all()))
self.assertEqual(1, len(db_api.reservation_get_all()))
def test_create_duplicate_leases(self):
"""Create two leases with same names, and checks it raises an error.
"""
db_api.lease_create(_get_fake_phys_lease_values())
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.lease_create,
_get_fake_phys_lease_values())
def test_create_leases_with_duplicated_reservation(self):
"""Create two leases with a duplicated reservation,
and checks it raises an error.
"""
lease_values = _get_fake_phys_lease_values()
db_api.lease_create(lease_values)
lease_values['id'] = _get_fake_random_uuid()
lease_values['name'] = 'duplicated_reservation'
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.lease_create,
lease_values)
def test_create_leases_with_duplicated_event(self):
"""Create two leases with a duplicated event,
and checks it raises an error.
"""
lease_values = _get_fake_phys_lease_values()
lease_values['events'] = [_get_fake_event_values()]
db_api.lease_create(lease_values)
lease_values['id'] = _get_fake_random_uuid()
lease_values['name'] = 'duplicated_event'
lease_values['reservations'][0]['id'] = _get_fake_random_uuid()
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.lease_create,
lease_values)
def test_create_lease_with_event(self):
"""Create a lease including a fake event and check all tables."""
lease = _get_fake_phys_lease_values()
lease['events'].append(_get_fake_event_values(lease_id=lease['id']))
result = db_api.lease_create(lease)
self.assertEqual(result['name'],
_get_fake_phys_lease_values()['name'])
self.assertEqual(1, len(db_api.event_get_all()))
def test_delete_wrong_lease(self):
"""Delete a lease that doesn't exist and check that raises an error."""
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.lease_destroy, 'fake_id')
def test_get_physical_lease(self):
"""Test if physical host reservation contains data of reservation."""
lease = _get_fake_phys_lease_values()
lease['events'].append(_get_fake_event_values(lease_id=lease['id']))
result = _create_physical_lease(values=lease)
result = db_api.lease_get(result['id'])
res = result.to_dict()
self.assertEqual(res['reservations'][0]['hypervisor_properties'],
lease['reservations'][0]['hypervisor_properties'])
self.assertEqual(res['reservations'][0]['resource_properties'],
lease['reservations'][0]['resource_properties'])
self.assertEqual(res['reservations'][0]['min'],
lease['reservations'][0]['min'])
self.assertEqual(res['reservations'][0]['max'],
lease['reservations'][0]['max'])
def test_delete_correct_lease(self):
"""Delete a lease and check that deletion has been cascaded to FKs."""
lease = _get_fake_phys_lease_values()
lease['events'].append(_get_fake_event_values(lease_id=lease['id']))
result = _create_physical_lease(values=lease)
db_api.lease_destroy(result['id'])
self.assertIsNone(db_api.lease_get(result['id']))
self.assertEqual(0, len(db_api.reservation_get_all()))
self.assertEqual(0, len(db_api.event_get_all()))
def test_lease_get_all(self):
"""Check the number of leases we get."""
_create_physical_lease(random=True)
self.assertEqual(1, len(db_api.lease_get_all()))
_create_physical_lease(random=True)
self.assertEqual(2, len(db_api.lease_get_all()))
def test_lease_list(self):
"""Not implemented yet until lease_list returns list of IDs."""
# TODO(sbauza): Enable this test when lease_list will return only IDs
self.assertTrue(True)
return
_create_physical_lease(
values=_get_fake_phys_lease_values(id='1', name='fake1'))
_create_physical_lease(
values=_get_fake_phys_lease_values(id='2', name='fake2'))
self.assertEqual(['1', '2'], db_api.lease_list())
def test_lease_update(self):
"""Update both start_time and name and check lease has been updated."""
result = _create_physical_lease()
result = db_api.lease_update(result['id'],
values={'name': 'lease_renamed'})
self.assertEqual('lease_renamed', result['name'])
result = db_api.lease_update(
result['id'],
values={'start_date': _get_datetime('2014-02-01 00:00')})
self.assertEqual(_get_datetime('2014-02-01 00:00'),
result['start_date'])
# Reservations
def test_create_reservation(self):
"""Create a reservation and verify that all tables have been
populated.
"""
result = db_api.reservation_create(_get_fake_phys_reservation_values())
self.assertEqual(result['lease_id'],
_get_fake_phys_reservation_values()
['lease_id'])
def test_reservation_get_all_by_values(self):
"""Create two reservations and verify that we can find reservation per
resource_id or resource_type.
"""
db_api.reservation_create(_get_fake_phys_reservation_values())
db_api.reservation_create(_get_fake_virt_reservation_values())
self.assertEqual(2, len(db_api.reservation_get_all_by_values()))
self.assertEqual(1, len(db_api.reservation_get_all_by_values(
resource_id='5678')))
self.assertEqual(1, len(db_api.reservation_get_all_by_values(
resource_type=host_plugin.RESOURCE_TYPE)))
def test_reservation_update(self):
result = db_api.reservation_create(_get_fake_phys_reservation_values())
self.assertNotEqual('fake', result.resource_type)
result = db_api.reservation_update(result.id,
{"resource_type": 'fake'})
self.assertEqual('fake', result.resource_type)
def test_reservation_destroy_for_reservation_not_found(self):
self.assertFalse(db_api.reservation_get('1'))
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.reservation_destroy, '1')
def test_create_duplicate_reservation(self):
"""Create a reservation and verify that an exception is raised if a
duplicated reservation is created.
"""
uuid = _get_fake_random_uuid()
db_api.reservation_create(_get_fake_phys_reservation_values(id=uuid))
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.reservation_create,
_get_fake_phys_reservation_values(id=uuid))
# Host reservations
def test_create_host_reservation(self):
"""Create a host reservation and verify that all tables
have been populated.
"""
result = db_api.host_reservation_create(
_get_fake_host_reservation_values(id='1'))
self.assertEqual(result['id'],
_get_fake_host_reservation_values(id='1')
['id'])
def test_create_duplicate_host_reservation(self):
"""Create a duplicated host reservation and verify that an exception is
raised.
"""
db_api.host_reservation_create(
_get_fake_host_reservation_values(id='1'))
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.host_reservation_create,
_get_fake_host_reservation_values(id='1'))
def test_delete_host_reservation(self):
"""Check all deletion cases for host reservation,
including cascade deletion from reservations table.
"""
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.host_reservation_destroy, 'fake_id')
result = db_api.host_reservation_create(
_get_fake_host_reservation_values())
db_api.host_reservation_destroy(result['id'])
self.assertIsNone(db_api.host_reservation_get(result['id']))
reserv = db_api.reservation_create(_get_fake_phys_reservation_values())
result = db_api.host_reservation_create(
_get_fake_host_reservation_values(reservation_id=reserv['id']))
db_api.reservation_destroy(reserv['id'])
self.assertIsNone(db_api.host_reservation_get(result['id']))
def test_host_reservation_get_all(self):
"""Check that we return 2 hosts."""
db_api.host_reservation_create(_get_fake_host_reservation_values(id=1))
db_api.host_reservation_create(_get_fake_host_reservation_values(id=2))
hosts_reservations = db_api.host_reservation_get_all()
self.assertEqual(['1', '2'], [x['id'] for x in hosts_reservations])
def test_host_reservation_get_by_reservation_id(self):
"""Check that we return 2 hosts."""
db_api.host_reservation_create(
_get_fake_host_reservation_values(id=1, reservation_id=1))
db_api.host_reservation_create(
_get_fake_host_reservation_values(id=2, reservation_id=2))
res = db_api.host_reservation_get_by_reservation_id(2)
self.assertEqual('2', res['id'])
def test_update_host_reservation(self):
db_api.host_reservation_create(_get_fake_host_reservation_values(id=1))
db_api.host_reservation_update(1, {'resource_properties': 'updated'})
res = db_api.host_reservation_get(1)
self.assertEqual('updated', res['resource_properties'])
def test_create_host(self):
"""Create a host and verify that all tables
have been populated.
"""
result = db_api.host_create(_get_fake_host_values(id='1'))
self.assertEqual(result['id'], _get_fake_host_values(id='1')['id'])
def test_create_duplicated_host(self):
"""Create a duplicated host and verify that an exception is raised.
"""
db_api.host_create(_get_fake_host_values(id='1'))
# Making sure we still raise a DuplicateDBEntry
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.host_create,
_get_fake_host_values(id='1'))
def test_search_for_hosts_by_ram(self):
"""Create two hosts and check that we can find a host per its RAM info.
"""
db_api.host_create(_get_fake_host_values(id=1, mem=2048))
db_api.host_create(_get_fake_host_values(id=2, mem=4096))
self.assertEqual(2, len(
db_api.host_get_all_by_queries(['memory_mb >= 2048'])))
self.assertEqual(0, len(
db_api.host_get_all_by_queries(['memory_mb lt 2048'])))
def test_search_for_hosts_by_cpu_info(self):
"""Create one host and search within cpu_info."""
db_api.host_create(_get_fake_host_values())
self.assertEqual(1, len(
db_api.host_get_all_by_queries(['cpu_info like %Westmere%'])))
def test_search_for_hosts_by_extra_capability(self):
"""Create one host and test extra capability queries."""
db_api.host_create(_get_fake_host_values(id=1))
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(computehost_id=1))
self.assertEqual(1, len(
db_api.host_get_all_by_queries(['vgpu == 2'])))
self.assertEqual(0, len(
db_api.host_get_all_by_queries(['vgpu != 2'])))
self.assertEqual(1, len(
db_api.host_get_all_by_queries(['cpu_info like %Westmere%',
'vgpu == 2'])))
self.assertEqual(0, len(
db_api.host_get_all_by_queries(['cpu_info like %wrongcpu%',
'vgpu == 2'])))
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.host_get_all_by_queries, ['apples < 2048'])
def test_search_for_hosts_by_composed_queries(self):
"""Create one host and test composed queries."""
db_api.host_create(_get_fake_host_values(mem=8192))
self.assertEqual(1, len(
db_api.host_get_all_by_queries(['memory_mb > 2048',
'cpu_info like %Westmere%'])))
self.assertEqual(0, len(
db_api.host_get_all_by_queries(['memory_mb < 2048',
'cpu_info like %Westmere%'])))
self.assertRaises(db_exceptions.ClimateDBInvalidFilter,
db_api.host_get_all_by_queries, ['memory_mb <'])
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.host_get_all_by_queries, ['apples < 2048'])
self.assertRaises(db_exceptions.ClimateDBInvalidFilterOperator,
db_api.host_get_all_by_queries,
['memory_mb wrongop 2048'])
self.assertEqual(1, len(
db_api.host_get_all_by_queries(['memory_mb in 4096,8192'])))
self.assertEqual(1, len(
db_api.host_get_all_by_queries(['memory_mb != null'])))
def test_list_hosts(self):
db_api.host_create(_get_fake_host_values(id=1))
db_api.host_create(_get_fake_host_values(id=2))
self.assertEqual(2, len(db_api.host_list()))
def test_get_hosts_per_filter(self):
db_api.host_create(_get_fake_host_values(id=1))
db_api.host_create(_get_fake_host_values(id=2))
filters = {'status': 'free'}
self.assertEqual(2, len(
db_api.host_get_all_by_filters(filters)))
def test_update_host(self):
db_api.host_create(_get_fake_host_values(id=1))
db_api.host_update(1, {'status': 'updated'})
self.assertEqual('updated', db_api.host_get(1)['status'])
def test_delete_host(self):
db_api.host_create(_get_fake_host_values(id=1))
db_api.host_destroy(1)
self.assertEqual(None, db_api.host_get(1))
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.host_destroy, 2)
def test_create_host_extra_capability(self):
result = db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id=1))
self.assertEqual(result['id'], _get_fake_host_values(id='1')['id'])
def test_create_duplicated_host_extra_capability(self):
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id=1))
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.host_extra_capability_create,
_get_fake_host_extra_capabilities(id='1'))
def test_get_host_extra_capability_per_id(self):
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id='1'))
result = db_api.host_extra_capability_get('1')
self.assertEqual('1', result['id'])
def test_host_extra_capability_get_all_per_host(self):
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id='1', computehost_id='1'))
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id='2', computehost_id='1'))
res = db_api.host_extra_capability_get_all_per_host('1')
self.assertEqual(2, len(res))
def test_update_host_extra_capability(self):
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id='1'))
db_api.host_extra_capability_update('1', {'capability_value': '2'})
res = db_api.host_extra_capability_get('1')
self.assertEqual('2', res['capability_value'])
def test_delete_host_extra_capability(self):
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id='1'))
db_api.host_extra_capability_destroy('1')
self.assertEqual(None, db_api.host_extra_capability_get('1'))
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.host_extra_capability_destroy, '1')
def test_host_extra_capability_get_all_per_name(self):
db_api.host_extra_capability_create(
_get_fake_host_extra_capabilities(id='1', computehost_id='1'))
res = db_api.host_extra_capability_get_all_per_name('1', 'vgpu')
self.assertEqual(1, len(res))
self.assertEqual([],
db_api.host_extra_capability_get_all_per_name('1',
'bad'))
# Host allocations
def test_host_allocation_get_all(self):
self.assertFalse(db_api.host_allocation_get_all())
db_api.host_allocation_create(_get_fake_host_allocation_values(id='1'))
db_api.host_allocation_create(_get_fake_host_allocation_values(id='2'))
self.assertEqual(2, len(db_api.host_allocation_get_all()))
def test_host_allocation_create_for_duplicated_hosts(self):
db_api.host_allocation_create(
_get_fake_host_allocation_values(id='1')
)
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.host_allocation_create,
_get_fake_host_allocation_values(id='1'))
def test_host_allocation_update_for_host(self):
host_allocation = db_api.host_allocation_create(
_get_fake_host_allocation_values(
compute_host_id="1",
reservation_id="1"
))
new_host_allocation = db_api.host_allocation_update(
host_allocation.id,
_get_fake_host_allocation_values(
compute_host_id="2",
reservation_id="2"
))
self.assertEqual('2', new_host_allocation.compute_host_id)
self.assertEqual('2', new_host_allocation.reservation_id)
self.assertNotEqual(host_allocation.compute_host_id,
new_host_allocation.compute_host_id)
def test_host_allocation_destroy_for_host(self):
host_allocation = db_api.host_allocation_create(
_get_fake_host_allocation_values()
)
db_api.host_allocation_destroy(host_allocation.id)
self.assertIsNone(db_api.host_allocation_get(host_allocation.id))
def test_host_allocation_destroy_for_host_not_found(self):
host_allocation_id = _get_fake_random_uuid()
self.assertIsNone(db_api.host_allocation_get(host_allocation_id))
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.host_allocation_destroy,
host_allocation_id)
def test_host_allocation_get_all_by_values(self):
db_api.host_allocation_create(_get_fake_host_allocation_values(
compute_host_id="1", reservation_id="1"))
db_api.host_allocation_create(_get_fake_host_allocation_values(
compute_host_id="1", reservation_id="1234"))
self.assertEqual(2, len(db_api.host_allocation_get_all_by_values()))
self.assertEqual(1, len(db_api.host_allocation_get_all_by_values(
reservation_id='1234')))
# Event
def test_event_create(self):
fake_event_type = 'test_event'
test_event = db_api.event_create(_get_fake_event_values(
event_type=fake_event_type))
self.assertTrue(test_event)
self.assertEqual(fake_event_type, test_event.event_type)
def test_create_duplicated_event(self):
self.assertFalse(db_api.event_get('1'))
fake_values = _get_fake_event_values(id='1')
test_event = db_api.event_create(fake_values)
self.assertTrue(test_event)
self.assertRaises(db_exceptions.ClimateDBDuplicateEntry,
db_api.event_create, fake_values)
def test_event_update(self):
self.assertFalse(db_api.event_get('1'))
test_event = db_api.event_create(_get_fake_event_values(id='1'))
self.assertTrue(test_event)
test_event = db_api.event_update(test_event.id, {'status': 'changed'})
self.assertEqual('changed', test_event.status)
def test_event_destroy(self):
self.assertFalse(db_api.event_get('1'))
db_api.event_create(_get_fake_event_values(
id='1'))
self.assertTrue(db_api.event_get('1'))
db_api.event_destroy('1')
self.assertFalse(db_api.event_get('1'))
def test_destroy_for_event_not_found(self):
self.assertFalse(db_api.event_get('1'))
self.assertRaises(db_exceptions.ClimateDBNotFound,
db_api.event_destroy, '1')
def test_event_get_first_sorted_by_event_type_filter(self):
fake_event_type = 'test_event'
db_api.event_create(_get_fake_event_values(
id='1'
))
db_api.event_create(_get_fake_event_values(
id='2',
event_type=fake_event_type
))
db_api.event_create(_get_fake_event_values(
id='3',
event_type=fake_event_type
))
filtered_events = db_api.event_get_first_sorted_by_filters(
sort_key='time',
sort_dir='asc',
filters={'event_type': fake_event_type}
)
self.assertEqual(fake_event_type, filtered_events.event_type)
self.assertEqual('2', filtered_events.id)
def test_event_get_first_sorted_by_status_filter(self):
fake_status = 'test_status'
db_api.event_create(_get_fake_event_values(
id='1'
))
db_api.event_create(_get_fake_event_values(
id='2',
status=fake_status
))
db_api.event_create(_get_fake_event_values(
id='3',
status=fake_status
))
filtered_events = db_api.event_get_first_sorted_by_filters(
sort_key='time',
sort_dir='asc',
filters={'status': fake_status}
)
self.assertEqual(fake_status, filtered_events.status)
self.assertEqual('2', filtered_events.id)
def test_event_get_first_sorted_by_lease_id_filter(self):
fake_lease_id = '1234'
db_api.event_create(_get_fake_event_values(
id='1'
))
db_api.event_create(_get_fake_event_values(
id='2',
lease_id=fake_lease_id
))
db_api.event_create(_get_fake_event_values(
id='3',
lease_id=fake_lease_id
))
filtered_events = db_api.event_get_first_sorted_by_filters(
sort_key='time',
sort_dir='asc',
filters={'lease_id': fake_lease_id}
)
self.assertEqual(fake_lease_id, filtered_events.lease_id)
self.assertEqual('2', filtered_events.id)
def test_event_get_sorted_asc_by_event_type_filter(self):
fake_event_type = 'test_event'
sort_dir = 'asc'
sort_key = 'time'
db_api.event_create(_get_fake_event_values(
id='1',
event_type=fake_event_type,
time=datetime.datetime.utcnow()
))
db_api.event_create(_get_fake_event_values(
id='2',
event_type=fake_event_type,
time=datetime.datetime.utcnow()
))
filtered_events = db_api.event_get_all_sorted_by_filters(
sort_key=sort_key,
sort_dir=sort_dir,
filters={'event_type': fake_event_type}
)
self.assertEqual(2, len(filtered_events))
self.assertEqual(fake_event_type, filtered_events[0].event_type)
#testing sort
self.assertTrue(is_result_sorted_correctly(filtered_events,
sort_key=sort_key,
sort_dir=sort_dir))
def test_event_get_sorted_asc_by_status_filter(self):
fake_status = 'test_status'
sort_dir = 'asc'
sort_key = 'time'
db_api.event_create(_get_fake_event_values(
id='1',
status=fake_status
))
db_api.event_create(_get_fake_event_values(
id='2'
))
filtered_events = db_api.event_get_all_sorted_by_filters(
sort_key=sort_key,
sort_dir=sort_dir,
filters={'status': fake_status}
)
self.assertEqual(1, len(filtered_events))
self.assertEqual(fake_status, filtered_events[0].status)
#testing sort
self.assertTrue(is_result_sorted_correctly(filtered_events,
sort_key=sort_key,
sort_dir=sort_dir))
def test_event_get_sorted_asc_by_lease_id_filter(self):
fake_lease_id = '1234'
sort_dir = 'asc'
sort_key = 'time'
db_api.event_create(_get_fake_event_values(
id='1',
lease_id=fake_lease_id
))
db_api.event_create(_get_fake_event_values(
id='2'
))
filtered_events = db_api.event_get_all_sorted_by_filters(
sort_key=sort_key,
sort_dir=sort_dir,
filters={'lease_id': fake_lease_id}
)
self.assertEqual(1, len(filtered_events))
self.assertEqual(fake_lease_id, filtered_events[0].lease_id)
#testing sort
self.assertTrue(is_result_sorted_correctly(filtered_events,
sort_key=sort_key,
sort_dir=sort_dir))
def test_event_get_sorted_desc_by_event_type_filter(self):
fake_event_type = 'test_event'
sort_dir = 'desc'
sort_key = 'time'
db_api.event_create(_get_fake_event_values(
id='1',
event_type=fake_event_type
))
db_api.event_create(_get_fake_event_values(
id='2',
event_type=fake_event_type
))
filtered_events = db_api.event_get_all_sorted_by_filters(
sort_key=sort_key,
sort_dir=sort_dir,
filters={'event_type': fake_event_type}
)
self.assertEqual(2, len(filtered_events))
self.assertEqual(fake_event_type, filtered_events[0].event_type)
#testing sort
self.assertTrue(is_result_sorted_correctly(filtered_events,
sort_key=sort_key,
sort_dir=sort_dir))
def test_event_get_sorted_desc_by_status_filter(self):
fake_status = 'test_status'
sort_dir = 'desc'
sort_key = 'time'
db_api.event_create(_get_fake_event_values(
id='1',
status=fake_status
))
db_api.event_create(_get_fake_event_values(
id='2'
))
filtered_events = db_api.event_get_all_sorted_by_filters(
sort_key=sort_key,
sort_dir=sort_dir,
filters={'status': fake_status}
)
self.assertEqual(1, len(filtered_events))
self.assertEqual(fake_status, filtered_events[0].status)
#testing sort
self.assertTrue(is_result_sorted_correctly(filtered_events,
sort_key=sort_key,
sort_dir=sort_dir))
def test_event_get_sorted_desc_by_lease_id_filter(self):
fake_lease_id = '1234'
sort_dir = 'desc'
sort_key = 'time'
db_api.event_create(_get_fake_event_values(
id='1',
lease_id=fake_lease_id
))
db_api.event_create(_get_fake_event_values(
id='2'
))
filtered_events = db_api.event_get_all_sorted_by_filters(
sort_key=sort_key,
sort_dir=sort_dir,
filters={'lease_id': fake_lease_id}
)
self.assertEqual(1, len(filtered_events))
self.assertEqual(fake_lease_id, filtered_events[0].lease_id)
#testing sort
self.assertTrue(is_result_sorted_correctly(filtered_events,
sort_key=sort_key,
sort_dir=sort_dir))
View Git Blame Copy Permalink