#!/usr/bin/python
# Copyright (C) 2012-2015 ASTRON (Netherlands Institute for Radio Astronomy)
# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# The LOFAR software suite is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
# $Id: $
import unittest
import testing.postgresql
import psycopg2
import os
from datetime import datetime, timedelta
from dateutil import parser
import logging
from pprint import pformat
logger = logging.getLogger(__name__)
try:
import mock
except ImportError as e:
print str(e)
print 'Please install python package mock: sudo pip install mock'
exit(3) # special lofar test exit code: skipped test
try:
import testing.postgresql
except ImportError as e:
print str(e)
print 'Please install python package testing.postgresql: sudo pip install testing.postgresql'
exit(3) # special lofar test exit code: skipped test
from lofar.common.dbcredentials import Credentials
from lofar.sas.resourceassignment.database.radb import RADatabase
from lofar.common.postgres import PostgresListener
# Create shared test database for better performance
database_credentials = Credentials()
Postgresql = testing.postgresql.PostgresqlFactory(cache_initialized_db=True)
def tearDownModule():
# clear cached database at end of tests
logger.info('tearDownModule')
Postgresql.clear_cache()
class ResourceAssignmentDatabaseTest(unittest.TestCase):
class test_task:
""" A lot of tests involve manipulation of a task (and its corresponding specification) in the RADB. A test task
that can be written to the RADB in preparation of the actual test makes the test-code more clean. """
task_status = "prescheduled"
task_type = "observation"
starttime = '2017-05-10 10:00:00'
endtime = '2017-05-10 12:00:00'
content = ""
cluster = "CEP4"
def setUp(self):
logger.info('setting up test RA database...')
# connect to shared test db
self.postgresql = Postgresql() # fresh db instead of shared one: self.postgresql = testing.postgresql.Postgresql()
# set up fixtures
# Note: In theory, this can be moved to the PostgresqlFactory call as kwarg 'on_initialized=populatedb'
# ...but for some reason that was much slower than keeping it here.
self.populate_db()
# update credentials (e.g. port changes for each test)
database_credentials.host = self.postgresql.dsn()['host']
database_credentials.database = self.postgresql.dsn()['database']
database_credentials.port = self.postgresql.dsn()['port']
# connect with useradministration role for tests
self.connection = psycopg2.connect(host=database_credentials.host,
user=database_credentials.user,
password=database_credentials.password,
dbname=database_credentials.database,
port=database_credentials.port)
# set up PostgresListener for notifications:
self.listener = PostgresListener(host=database_credentials.host,
username=database_credentials.user,
password=database_credentials.password,
database=database_credentials.database,
port=database_credentials.port)
# set up radb python module
self.radb = RADatabase(database_credentials, log_queries=True)
logger.info('...finished setting up test RA database')
def tearDown(self):
logger.info('removing test RA database...')
self.connection.close()
# self.Postgresql.clear_cache() # for fresh db during setUp, do instead:
self.postgresql.stop()
def populate_db(self):
# connect to db as root
conn = psycopg2.connect(**self.postgresql.dsn())
cursor = conn.cursor()
# set credentials to be used during tests
database_credentials.user = 'resourceassignment'
database_credentials.password = 'blabla' # cannot be empty...
# create user role
# Note: NOSUPERUSER currently raises "permission denied for schema virtual_instrument"
# Maybe we want to sort out user creation and proper permissions in the sql scripts?
query = "CREATE USER %s WITH SUPERUSER PASSWORD '%s'" % (
database_credentials.user,
database_credentials.password)
cursor.execute(query)
# populate db tables
# These are applied in given order to set up test db
# Note: cannot use create_and_populate_database.sql since '\i' is not understood by cursor.execute()
sql_basepath = os.environ['LOFARROOT'] + "/share/radb/sql/"
sql_createdb_paths = [sql_basepath + "create_database.sql",
sql_basepath + "/add_resource_allocation_statics.sql",
sql_basepath + "/add_virtual_instrument.sql",
sql_basepath + "/add_notifications.sql",
sql_basepath + "/add_functions_and_triggers.sql"
]
for sql_path in sql_createdb_paths:
with open(sql_path) as sql:
cursor.execute(sql.read())
cursor.close()
conn.commit()
conn.close()
def _execute_query(self, query, fetch=False):
cursor = self.connection.cursor()
cursor.execute(query)
ret = None
if fetch:
ret = cursor.fetchall()
cursor.close()
self.connection.commit()
return ret
# --- tests start here
# integrity tests of postgres database itself
#
# Note: These are meant to make sure the setup generally works and all sql scripts were applied.
# I don't see much benefit in full coverage here since it should be all be tested through RADataBase functionality.
# Of course new tests can be added here where db functionality like triggers should be tested separately from the
# Python part of the job.
# database created?
def test_select_tables_contains_tables_for_each_schema(self):
query = "SELECT table_schema,table_name FROM information_schema.tables"
fetch = self._execute_query(query, fetch=True)
self.assertTrue('resource_allocation' in str(fetch))
self.assertTrue('resource_monitoring' in str(fetch))
self.assertTrue('virtual_instrument' in str(fetch))
# resource allocation_statics there?
def test_select_task_types_contains_obervation(self):
query = "SELECT * FROM resource_allocation.task_type"
fetch = self._execute_query(query, fetch=True)
self.assertTrue('observation' in str(fetch))
# virtual instrument there?
def test_select_virtualinstrument_units_contain_rcuboard(self):
query = "SELECT * FROM virtual_instrument.unit"
fetch = self._execute_query(query, fetch=True)
self.assertTrue('rcu_board' in str(fetch))
def _insert_test_spec(self,
starttime='2017-05-10 13:00:00',
endtime='2017-05-10 14:00:00',
content='testcontent',
cluster='CEP4'):
query = "INSERT INTO resource_allocation.specification (starttime, endtime, content, cluster) " \
"VALUES ('%s', '%s', '%s', '%s') RETURNING id" % (starttime, endtime, content, cluster)
res = self._execute_query(query, fetch=True)
return res[0][0]
#
def test_insert_specification_creates_new_entry(self):
# insert spec
content = 'testcontent'
ident = self._insert_test_spec(content=content)
# check it is there
query = "SELECT content FROM resource_allocation.specification WHERE id=%s" % ident
fetch = self._execute_query(query, fetch=True)
self.assertTrue(content in str(fetch))
def test_update_specification_changes_entry(self):
# insert spec
ident = self._insert_test_spec()
# update existing spec content
newcontent = 'testcontent_new'
query = "UPDATE resource_allocation.specification SET (content) = ('%s')" % newcontent
self._execute_query(query)
# check updated content
query = "SELECT content FROM resource_allocation.specification WHERE id=%s" % ident
fetch = self._execute_query(query, fetch=True)
self.assertTrue(newcontent in str(fetch))
def test_delete_specification(self):
# insert spec
content = 'deletecontent'
ident = self._insert_test_spec(content=content)
# make sure it's there
query = "SELECT content FROM resource_allocation.specification WHERE id=%s" % ident
fetch = self._execute_query(query, fetch=True)
self.assertTrue(content in str(fetch))
# delete testspec again
query = "DELETE FROM resource_allocation.specification WHERE id = %s" % ident
self._execute_query(query)
# make sure it's gone
query = "SELECT content FROM resource_allocation.specification WHERE id=%s" % ident
fetch = self._execute_query(query, fetch=True)
self.assertFalse(content in str(fetch))
# triggers in place?
def test_insert_specification_swaps_startendtimes_if_needed(self):
#when inserting spec with start>endtime, should raise error
with self.assertRaises(psycopg2.InternalError) as context:
# insert spec
starttime = '2017-05-10 12:00:00'
endtime = '2017-05-10 10:00:00'
ident = self._insert_test_spec(starttime=starttime, endtime=endtime)
# notifications in place?
def test_insert_task_triggers_notification(self):
# insert specification to not raise INtegrityError
ident = self._insert_test_spec()
# listen on notification
cursor = self.connection.cursor()
cursor.execute("LISTEN %s;", (psycopg2.extensions.AsIs('task_insert'),))
# todo: fix this and use this instead to listen for notifications.
# todo: ...Problem: For some reason callback function is not called.
# set up listener in a way we can check it was called
# callback = mock.Mock()
# callback.listen.return_value = 42
# self.listener.subscribe('task_insert', callback.listen)
# trigger notification
query = "INSERT INTO resource_allocation.task (mom_id, otdb_id, status_id, type_id, specification_id)" \
"VALUES (%s, %s, %s, %s, %s)" % (1, 1, 200, 0, ident)
self._execute_query(query)
# wait for notification
notification = ''
self.connection.poll()
while self.connection.notifies:
try:
notification = self.connection.notifies.pop(0)
break
except Exception:
pass
self.assertTrue('task_insert' in str(notification))
# todo: fix listener and use instead of polling:
# callback.listen.assert_called()
#
# radb functionality tests
#
#
def _insert_test_task_and_specification(self, mom_id=1, otdb_id=2):
""" Inserts a sample task and specification (see self.test_class) to the RADB
:param mom_id: optional alternative MoM ID. Should be set uniquely if within a test multiple tasks are to be
inserted
:param otdb_id: optional alternative OTDB OD. Should be set uniquely if within a test multiple tasks are to be
inserted.
:returns 2-tuple (task_id, spec_id) or None if task wasn't inserted
"""
task = self.radb.insertSpecificationAndTask(mom_id=mom_id,
otdb_id=otdb_id,
task_status=self.test_task.task_status,
task_type=self.test_task.task_type,
starttime=self.test_task.starttime,
endtime=self.test_task.endtime,
content=self.test_task.content,
cluster=self.test_task.cluster)
if task['inserted']:
return task['task_id'], task['specification_id']
def test_getTaskStatuses_succeeds(self):
""" Verifies if radb.getTaskStatuses() successfully fetches all expected task statuses """
expected_statuses = [
{'id': 200, 'name': 'prepared'},
{'id': 300, 'name': 'approved'},
{'id': 320, 'name': 'on_hold'},
{'id': 335, 'name': 'conflict'},
{'id': 350, 'name': 'prescheduled'},
{'id': 400, 'name': 'scheduled'},
{'id': 500, 'name': 'queued'},
{'id': 600, 'name': 'active'},
{'id': 900, 'name': 'completing'},
{'id': 1000, 'name': 'finished'},
{'id': 1100, 'name': 'aborted'},
{'id': 1150, 'name': 'error'},
{'id': 1200, 'name': 'obsolete'}]
statuses = self.radb.getTaskStatuses()
self.assertEquals(statuses, expected_statuses)
def test_getTaskStatusNames_succeeds(self):
""" Verifies if radb.getTaskStatusNames() successfully fetches all expected task status names """
expected_names = ['prepared', 'approved', 'on_hold', 'conflict', 'prescheduled', 'scheduled', 'queued',
'active', 'completing', 'finished', 'aborted', 'error', 'obsolete']
names = self.radb.getTaskStatusNames()
self.assertEqual(sorted(expected_names), sorted(names))
def test_getTaskStatusId_wrong_status_fails(self):
""" Verifies if radb.getTaskStatusId() raises an Exception if the idea of an unknown status is requested """
wrong_status = 'willywonka'
self.assertRaises(KeyError, self.radb.getTaskStatusId, wrong_status)
def test_getTaskStatusId_right_status_succeeds(self):
""" Verifies if radb.getTaskStatusId() successfully fetches the expected status id for a given status. """
status = 'scheduled'
expected_status_id = 400
status_id = self.radb.getTaskStatusId(status)
self.assertEqual(status_id, expected_status_id)
def test_getTaskTypes_succeeds(self):
""" Verifies if radb.getTaskTypes() successfully fetches all expected task types """
expected_task_types = [
{'id': 0, 'name': 'observation'},
{'id': 1, 'name': 'pipeline'},
{'id': 2, 'name': 'reservation'}]
task_types = self.radb.getTaskTypes()
self.assertEqual(task_types, expected_task_types)
def test_getTaskTypeNames_succeeds(self):
""" Verifies if radb.getTaskTypeNames() successfully fetches all expected task type names """
expected_task_type_names = ['observation', 'pipeline', 'reservation']
task_type_names = self.radb.getTaskTypeNames()
self.assertEqual(task_type_names, expected_task_type_names)
def test_getTaskTypeId_wrong_type_name_fails(self):
""" Verifies if radb.getTaskTypeId() raises an exception if a type id is requested for a wrong type name """
wrong_type_name = 'willywonka'
self.assertRaises(KeyError, self.radb.getTaskTypeId, wrong_type_name)
def test_getTaskTypeId_right_type_name_succeeds(self):
""" Verifies if radb.getTaskTypeId() successfully fetches the type id for a given type name. """
type_name = 'reservation'
expected_type_id = 2
type_id = self.radb.getTaskTypeId(type_name)
self.assertEqual(type_id, expected_type_id)
def test_getResourceClaimStatuses_succeeds(self):
""" Verifies if radb.getResourceClaimStatuses() successfully fetches all expected claim statuses. """
expected_claim_statuses = [
{'id': 0, 'name': 'tentative'},
{'id': 1, 'name': 'claimed'},
{'id': 2, 'name': 'conflict'}]
claim_statuses = self.radb.getResourceClaimStatuses()
self.assertEqual(claim_statuses, expected_claim_statuses)
def test_getResourceClaimStatusNames_succeeds(self):
""" Verifies if radb.getResourceClaimStatusNames() successfully fetches all expected claim status names. """
expected_claim_status_names = ['tentative', 'claimed', 'conflict']
claim_status_names = self.radb.getResourceClaimStatusNames()
self.assertEqual(claim_status_names, expected_claim_status_names)
def test_getResourceClaimStatusId_wrong_claim_name_fails(self):
""" Verifies if radb.getResourceClaimStatusId() raises an exception if a claim status id is requested for wrong
claim name. """
wrong_claim_name = 'willywonka'
self.assertRaises(KeyError, self.radb.getResourceClaimStatusId, wrong_claim_name)
def test_getResourceClaimStatusId_right_claim_name_succeeds(self):
""" Verifies if radb.getResourceClaimStatusId() successfully fetches the expected claim ID for a given claim
name. """
claim_name = 'conflict'
expected_claim_id = 2
claim_id = self.radb.getResourceClaimStatusId(claim_name)
self.assertEqual(claim_id, expected_claim_id)
def test_getTasksTimeWindow_no_ids_fails(self):
""" Verify if radb.getTasksTimeWindow() raises an exception when called with an empty ID lists for every ID
type. """
self.assertRaises(KeyError, self.radb.getTasksTimeWindow, task_ids=[], mom_ids=[], otdb_ids=[])
def test_getTasksTimeWindow_multiple_kinds_of_ids_fails(self):
""" Verify if radb.getTasksTimeWindow() raises an exception when called with IDs of more than one type. """
task_ids = [0, 1, 2, 3]
mom_ids = [4, 5, 6, 7]
otdb_ids = [8, 9, 10, 11]
self.assertRaises(KeyError, self.radb.getTasksTimeWindow, task_ids, mom_ids, otdb_ids)
def test_getTasksTimeWindow_empty_ids_list_succeeds(self):
""" Verify if radb.getTasksTimeWindow() returns an empty list when requesting a time window for an empty list
of IDs. """
time_windows = [self.radb.getTasksTimeWindow([], None, None),
self.radb.getTasksTimeWindow(None, [], None),
self.radb.getTasksTimeWindow(None, None, [])]
expected_time_windows = [[], [], []]
self.assertItemsEqual(time_windows, expected_time_windows)
def test_getTasksTimeWindow_empty_db_returns_no_time_window_succeeds(self):
""" Verify if radb.getTasksTimeWindow() returns an invalid time window when requesting a time window for a
non-existing task. """
# Ask time window for a non-existing task id
time_window = self.radb.getTasksTimeWindow([0], None, None)
time_window = [time_window['min_starttime'], time_window['max_endtime']]
expected_time_window = [None, None]
self.assertItemsEqual(time_window, expected_time_window)
def test_getTasksTimeWindow_normal_use_succeeds(self):
""" Verify if radb.getTasksTimeWindow() successfully return the expected time window when requesting a time
window for an existing task. """
# Shoot a task into RADB which time window can later be queried
starttime = '2017-05-10 10:00:00'
endtime = '2017-05-10 12:00:00'
mom_id = 1
otdb_id = 2
task_id, _ = self._insert_test_task_and_specification(mom_id=mom_id, otdb_id=otdb_id)
# Now query RADB for time_window based on task_id, mom_id, and otdb_id
time_windows = [self.radb.getTasksTimeWindow([task_id], None, None),
self.radb.getTasksTimeWindow(None, [mom_id], None),
self.radb.getTasksTimeWindow(None, None, [otdb_id])]
# The time_window based on task_id, mom_id, and otdb_id should be the same
expected_time_windows = 3*[{'max_endtime': parser.parse(endtime), 'min_starttime': parser.parse(starttime)}]
self.assertItemsEqual(time_windows, expected_time_windows)
def test_getTasks_no_ids_fails(self):
""" Verify if radb.getTasks() raises an exception when called with an empty ID lists for every ID type. """
self.assertRaises(KeyError, self.radb.getTasks, task_ids=[], mom_ids=[], otdb_ids=[])
def test_getTasks_multiple_kinds_of_ids_fails(self):
""" Verify if radb.getTasks() raises an exception when called with filled ID lists for multiple ID types. """
task_ids = [0, 1, 2, 3]
mom_ids = [4, 5, 6, 7]
otdb_ids = [8, 9, 10, 11]
self.assertRaises(KeyError, self.radb.getTasks, task_ids=task_ids, mom_ids=mom_ids, otdb_ids=otdb_ids)
def test_getTasks_empty_ids_list_succeeds(self):
tasks = [self.radb.getTasks(task_ids=[], mom_ids=None, otdb_ids=None),
self.radb.getTasks(task_ids=None, mom_ids=[], otdb_ids=None),
self.radb.getTasks(task_ids=None, mom_ids=None, otdb_ids=[])]
expected_tasks = [[], [], []]
self.assertItemsEqual(tasks, expected_tasks)
def test_getTasks_empty_db_returns_empty_list_succeeds(self):
""" Verify if radb.getTasks() successfully returns an empty list when called with a task ID that is non-existing
in RADB. """
tasks = self.radb.getTasks(task_ids=[0])
self.assertEqual(tasks, [])
def test_getTasks_normal_use_succeeds(self):
""" Verify if radb.getTasks() successfully returns the expected tasks when requesting tasks related to an
existing task. """
# Shoot a task into RADB which can later be fetched
task_id, _ = self._insert_test_task_and_specification()
# Now query RADB for the task based on task_id
task = self.radb.getTasks(task_ids=[task_id])[0]
# The task's task ID should be the same to pass this test
self.assertEqual(task['id'], task_id)
def test_getTask_no_ids_fails(self):
""" Verify if radb.getTask() raises an exception when called without arguments. """
self.assertRaises(KeyError, self.radb.getTask)
def test_getTask_multiple_kinds_of_ids_fails(self):
""" Verify if radb.getTask() raises an exception when called with multiple ID types defined. """
self.assertRaises(KeyError, self.radb.getTask, 1, 2, 3, 4)
def test_getTask_empty_db_returns_none_succeeds(self):
""" Verify if radb.getTask() successfully returns an None when called with a task ID that doesn't exist in
RADB. """
task = self.radb.getTask(id=0)
self.assertIsNone(task)
def test_getTask_normal_use_succeeds(self):
""" Verify if radb.getTask() successfully returns the expected task when requested to. """
# Shoot a task into RADB which fetched
task_id, _ = self._insert_test_task_and_specification()
task = self.radb.getTask(id=task_id)
self.assertEqual(task['id'], task_id)
def test_insertTask_with_invalid_specification_id_raises_exception(self):
""" Verify if radb.insertTask() raises an exception when called with non-existing specification ID """
with self.assertRaises(Exception):
self.radb.insertTask(0, 0, 'conflict', 'observation', 1)
def test_insertTask_with_invalid_id_type_raises_exception(self):
""" Verify if radb.insertTask() raises an exception when called with illegal mom_id and otdb_id types """
# Insert a specification in order to be sure we use a valid specification_id
spec_id = self.radb.insertSpecification(starttime='2017-05-10 10:00:00', endtime='2017-05-10 12:00:00',
content="", cluster="CEP4")
with self.assertRaises(Exception):
self.radb.insertTask('monkey see', 'is monkey do', 'conflict', 'observation', spec_id)
def test_insertTask_allows_nonexisting_mom_and_otdb_ids(self):
""" Verify if radb.insertTask() allows the insertion of a task with non-exising mom_id and otdb_id values """
# Insert a specification in order to be sure we use a valid specification_id
spec_id = self.radb.insertSpecification(starttime='2017-05-10 10:00:00', endtime='2017-05-10 12:00:00',
content="", cluster="CEP4")
mom_id = otdb_id = -1
task_id = self.radb.insertTask(mom_id, otdb_id, 'conflict', 'observation', spec_id)
self.assertEqual(task_id, 1)
def test_insertTask_duplicate_mom_ids_fails(self):
""" Verify if radb.insertTask() raises exception when called with already occupied mom_id """
# Insert a specification in order to be sure we use a valid specification_id
spec_id = self.radb.insertSpecification(starttime='2017-05-10 10:00:00', endtime='2017-05-10 12:00:00',
content="", cluster="CEP4")
with self.assertRaises(Exception):
self.radb.insertTask(1, 1, 'conflict', 'observation', spec_id)
self.radb.insertTask(1, 2, 'conflict', 'observation', spec_id)
def test_insertTask_duplicate_otdb_ids_fails(self):
""" Verify if radb.insertTask() raises exception when called with already occupied otdb_id """
# Insert a specification in order to be sure we use a valid specification_id
spec_id = self.radb.insertSpecification(starttime='2017-05-10 10:00:00', endtime='2017-05-10 12:00:00',
content="", cluster="CEP4")
with self.assertRaises(Exception):
self.radb.insertTask(1, 1, 'conflict', 'observation', spec_id)
self.radb.insertTask(2, 1, 'conflict', 'observation', spec_id)
def test_insertTask_with_invalid_task_status_raises_exception(self):
""" Verify if radb.insertTask() raises an exception when called with invalid task status """
# Insert a specification in order to be sure we use a valid specification_id
specification_id = self.radb.insertSpecification(starttime='2017-05-10 10:00:00',
endtime='2017-05-10 12:00:00',
content="", cluster="CEP4")
with self.assertRaises(Exception):
self.radb.insertTask(0, 0, 'willywonka', 'observation', specification_id)
def test_insertTask_with_invalid_task_type_raises_exception(self):
""" Verify if radb.insertTask() raises an exception when called with invalid task type """
# Insert a specification in order to be sure we use a valid specification_id
specification_id = self.radb.insertSpecification(starttime='2017-05-10 10:00:00',
endtime='2017-05-10 12:00:00',
content="", cluster="CEP4")
with self.assertRaises(Exception):
self.radb.insertTask(0, 0, 'conflict', 'willywonka', specification_id)
def test_insertTask_normal_use_succeeds(self):
""" Verify if radb.insertTask() successfully inserts a task when called with valid arguments. """
sample_starttime = '2017-05-10 10:00:00'
sample_endtime = '2017-05-10 12:00:00'
sample_task = {
'id': 1,
'starttime': parser.parse(sample_starttime),
'endtime': parser.parse(sample_endtime),
'cluster': 'CEP4',
'status': 'conflict',
'status_id': 335,
'type': 'observation',
'type_id': 0,
'mom_id': 0,
'otdb_id': 0,
'blocked_by_ids': [],
'predecessor_ids': [],
'successor_ids': [],
'duration': (parser.parse(sample_endtime) - parser.parse(sample_starttime)).seconds,
}
# Insert a specification in order to be sure we use a valid specification_id
sample_task['specification_id'] = self.radb.insertSpecification(starttime=sample_starttime,
endtime=sample_endtime,
cluster=sample_task['cluster'],
content='',)
task_id = self.radb.insertTask(sample_task['mom_id'], sample_task['otdb_id'], sample_task['status'],
sample_task['type'], sample_task['specification_id'])
task = self.radb.getTask(id=task_id)
self.assertEqual(task, sample_task)
def test_deleteTask_with_non_excisting_task_id_fails(self):
""" Verify if radb.deleteTask() fails when called with a non-excisting task ID. """
successfully_deleted = self.radb.deleteTask(0)
self.assertFalse(successfully_deleted)
def test_deleteTask_removes_task_successfully(self):
""" Verify if radb.deleteTask() successfully deletes the expected task """
# Shoot a task and corresponding specification into RADB which can later be deleted
task_id, spec_id = self._insert_test_task_and_specification()
successfully_deleted = self.radb.deleteTask(task_id)
self.assertTrue(successfully_deleted)
self.assertIsNone(self.radb.getTask(id=task_id))
def test_deleteTask_leaves_specification_untouched(self):
""" Verify if radb.deleteTask() leaves a task's specification untouched when deleting the task """
# Shoot a task and corresponding specification into RADB which can later be deleted
task_id, spec_id = self._insert_test_task_and_specification()
self.radb.deleteTask(task_id)
self.assertNotEqual(self.radb.getSpecification(spec_id), [])
def test_updateTask_nonexisting_task_id_fails(self):
""" Verify if radb.updateTask() fails when called with a non-existing task ID """
task_id = -1
task_updated = self.radb.updateTask(task_id)
self.assertFalse(task_updated)
def test_updateTask_invalid_task_status_raises_exception(self):
""" Verify if radb.updateTask() raises an exception when called with illegal task_status """
# Shoot a task and corresponding specification into RADB which can later be updated
task_id, spec_id = self._insert_test_task_and_specification(mom_id=1, otdb_id=2)
with self.assertRaises(Exception):
self.radb.updateTask(task_id, task_status="willywonka")
def test_updateTask_invalid_task_type_raises_exception(self):
""" Verify if radb.updateTask() raises an exception when called with illegal task_type """
# Shoot a task and corresponding specification into RADB which can later be updated
task_id, spec_id = self._insert_test_task_and_specification(mom_id=1, otdb_id=2)
with self.assertRaises(Exception):
self.radb.updateTask(task_id, task_type="willywonka")
def test_updateTask_invalid_specification_id_raises_exception(self):
""" Verify if radb.updateTask() raises an exception when called with illegal specification ID """
# Shoot a task and corresponding specification into RADB which can later be updated
task_id, spec_id = self._insert_test_task_and_specification(mom_id=1, otdb_id=2)
with self.assertRaises(Exception):
self.radb.updateTask(task_id, spec_id=-1) # Illegal spec_id
def test_updateTask_duplicate_mom_id_fail(self):
""" Verify if radb.updateTask() raises an exception when called with an already occupied mom_id """
# Shoot in two tasks and corresponding specifications into RADB with different mom_id and otdb_ids which can
# later be updated
task_id, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=11)
self._insert_test_task_and_specification(mom_id=2, otdb_id=12)
task_updated = self.radb.updateTask(task_id, mom_id=2)
self.assertFalse(task_updated)
def test_updateTask_duplicate_otdb_id_fail(self):
""" Verify if radb.updateTask() raises an exception when called with already existing otdb_id """
# Shoot in two tasks and corresponding specifications into RADB with different mom_id and otdb_ids which can
# later be updated
task_id, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=11)
self._insert_test_task_and_specification(mom_id=2, otdb_id=12)
task_updated = self.radb.updateTask(task_id, otdb_id=12)
self.assertFalse(task_updated)
def test_updateTask_normal_use_succeeds(self):
""" Verify if radb.updateTask() successfully updates a task in RADB """
# Shoot a task and corresponding specification into RADB which can later be updated
mom_id = 1
otdb_id = 2
task_id, spec_id = self._insert_test_task_and_specification(mom_id=mom_id, otdb_id=otdb_id)
new_task_status = "approved"
new_task_type = "reservation"
task_is_updated = self.radb.updateTask(task_id=task_id,
mom_id=mom_id,
otdb_id=otdb_id,
task_status=new_task_status,
task_type=new_task_type,
specification_id=spec_id)
task = self.radb.getTask(id=task_id)
self.assertTrue(task_is_updated)
self.assertEqual(task['status'], new_task_status)
self.assertEqual(task['type'], new_task_type)
def test_getTaskPredecessorIds_invalid_id_returns_empty_dict(self):
""" Verify if radb.getTaskPredecessorIds() returns an empty dict when called with an invalid ID """
id = -1
task_and_predecessors = self.radb.getTaskPredecessorIds(id)
self.assertEqual(task_and_predecessors, {})
def test_getTaskPredecessorIds_valid_nonexisting_id_returns_empty_dict(self):
""" Verify if radb.getTaskPredecessorIds() returns an empty dict when called with a valid ID that doesn't exist
in RADB """
id = 1
task_and_predecessors = self.radb.getTaskPredecessorIds(id)
self.assertEqual(task_and_predecessors, {})
def test_getTaskPredecessorIds_normal_use_with_predecessor_succeeds(self):
""" Verify if radb.getTaskPredecessorIds() returns an empty dict when called with a valid ID that exists in RADB
and has a predecessor """
# Shoot 2 unique tasks and corresponding specifications into RADB
task_id, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=10)
task_id_pre1, _ = self._insert_test_task_and_specification(mom_id=2, otdb_id=11)
# Add predecessors to task relation
id = self.radb.insertTaskPredecessor(task_id, task_id_pre1)
task_and_predecessors = self.radb.getTaskPredecessorIds(id)
self.assertEqual(task_and_predecessors, {task_id: [task_id_pre1]})
def test_getTaskSuccessorIds_invalid_id_returns_empty_dict(self):
""" Verify if radb.getTaskSuccessorIds() returns an empty dict when called with an invalid ID """
id = -1
task_and_successors = self.radb.getTaskSuccessorIds(id)
self.assertEqual(task_and_successors, {})
def test_getTaskSuccessorIds_valid_nonexisting_id_returns_empty_dict(self):
""" Verify if radb.getTaskSuccessorIds() returns an empty dict when called with a valid ID that doesn't exist in
RADB """
id = 1
task_and_successors = self.radb.getTaskSuccessorIds(id)
self.assertEqual(task_and_successors, {})
def test_getTaskSuccessorIds_normal_use_with_successor_succeeds(self):
""" Verify if radb.getTaskSuccessorIds() returns an empty dict when called with a valid ID that exists in RADB
and has a successor """
# Shoot 2 unique tasks and corresponding specifications into RADB
task_id, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=10)
task_id_suc1, _ = self._insert_test_task_and_specification(mom_id=2, otdb_id=11)
# Add predecessors to task relation
id = self.radb.insertTaskPredecessor(task_id_suc1, task_id)
task_and_successors = self.radb.getTaskSuccessorIds(id)
self.assertEqual(task_and_successors, {task_id: [task_id_suc1]})
def test_getTaskPredecessorIdsForTask_invalid_task_id_returns_empty_dict(self):
""" Verify if radb.getTaskPredecessorIdsForTask() returns an empty dict when called with an invalid task ID """
task_id = -1
predecessors = self.radb.getTaskPredecessorIdsForTask(task_id)
self.assertEqual(predecessors, [])
def test_getTaskPredecessorIdsForTask_valid_nonexisting_task_id_returns_empty_dict(self):
""" Verify if radb.getTaskPredecessorIdsForTask() returns an empty dict when called with a valid task ID that
doesn't exist in RADB """
task_id = 1
predecessors = self.radb.getTaskPredecessorIdsForTask(task_id)
self.assertEqual(predecessors, [])
def test_getTaskPredecessorIdsForTask_normal_use_with_successor_succeeds(self):
""" Verify if radb.getTaskPredecessorIdsForTask() returns an empty dict when called with a valid task ID that
exists in RADB and has a predecessor """
# Shoot 2 unique tasks and corresponding specifications into RADB
task_id, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=10)
task_id_pre1, _ = self._insert_test_task_and_specification(mom_id=2, otdb_id=11)
# Add predecessors to task relation
self.radb.insertTaskPredecessor(task_id, task_id_pre1)
predecessors = self.radb.getTaskPredecessorIdsForTask(task_id)
self.assertEqual(predecessors, [task_id_pre1])
def test_getTaskSuccessorIdsForTask_invalid_task_id_returns_empty_dict(self):
""" Verify if radb.getTaskSuccessorIdsForTask() returns an empty dict when called with an invalid task ID """
task_id = -1
successors = self.radb.getTaskSuccessorIdsForTask(task_id)
self.assertEqual(successors, [])
def test_getTaskSuccessorIdsForTask_valid_nonexisting_task_id_returns_empty_dict(self):
""" Verify if radb.getTaskSuccessorIdsForTask() returns an empty dict when called with a valid task ID that
doesn't exist in RADB """
task_id = 1
successors = self.radb.getTaskSuccessorIdsForTask(task_id)
self.assertEqual(successors, [])
def test_getTaskSuccessorIdsForTask_normal_use_with_successor_succeeds(self):
""" Verify if radb.getTaskSuccessorIdsForTask() returns an empty dict when called with a valid task ID that
exists in RADB and has a successor """
# Shoot 2 unique tasks and corresponding specifications into RADB
task_id, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=10)
task_id_suc1, _ = self._insert_test_task_and_specification(mom_id=2, otdb_id=11)
# Add predecessors to task relation
self.radb.insertTaskPredecessor(task_id_suc1, task_id)
successors = self.radb.getTaskSuccessorIdsForTask(task_id)
self.assertEqual(successors, [task_id_suc1])
def test_insertTaskPredecessor_invalid_ids_returns_none(self):
""" Verify if radb.insertTaskPredecessor() returns None when called with invalid task ID and/or predecessor ID
"""
invalid_id = -1
ids = [self.radb.insertTaskPredecessor(invalid_id, 1),
self.radb.insertTaskPredecessor(1, invalid_id),
self.radb.insertTaskPredecessor(invalid_id, invalid_id)]
self.assertItemsEqual(ids, [None, None, None])
def test_insertTaskPredecessor_valid_nonexisting_ids_return_none(self):
""" Verify if radb.insertTaskPredecessor() returns None when called with valid but non-existing task ID and
predecessor ID """
task_id = 1
predecessor_id = 1
_id = self.radb.insertTaskPredecessor(task_id, predecessor_id)
self.assertIsNone(_id)
def test_insertTaskPredecessor_normal_use_succeeds(self):
""" Verify if radb.insertTaskPredecessor() returns an ID when called with valid and existing task and
predecessor IDs. """
# Shoot 2 unique tasks and corresponding specifications into RADB
task_id_a, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=10)
task_id_b, _ = self._insert_test_task_and_specification(mom_id=2, otdb_id=11)
_id = self.radb.insertTaskPredecessor(task_id_a, task_id_b)
self.assertIsNotNone(_id)
def test_insertTaskPredecessors_normal_use_succeeds(self):
""" Verify if radb.insertTaskPredecessors() returns a list of IDs when called with valid and existing task and
predecessor IDs. """
# Shoot 2 unique tasks and corresponding specifications into RADB
task_id_a, _ = self._insert_test_task_and_specification(mom_id=1, otdb_id=10)
task_id_b, _ = self._insert_test_task_and_specification(mom_id=2, otdb_id=11)
task_id_c, _ = self._insert_test_task_and_specification(mom_id=3, otdb_id=12)
ids = self.radb.insertTaskPredecessors(task_id_a, [task_id_b, task_id_c])
self.assertIs(len(ids), 2)
def test_getSpecifications_select_all_on_empty_db_succeeds(self):
""" Verify if radb.getSpecifications() returns an empty list on an empty RADB """
self.assertEqual(self.radb.getSpecifications(), [])
def test_getSpecifications_normal_use_no_filter_succeeds(self):
""" Verify if radb.getSpecifications() returns a list containing all specifications that exist in the RADB """
spec_ids = [self._insert_test_spec(), self._insert_test_spec(), self._insert_test_spec()]
specifications = self.radb.getSpecifications()
self.assertEqual(len(spec_ids), len(specifications))
def test_getSpecifications_normal_use_select_one_succeeds(self):
""" Verify if radb.getSpecifications() returns a list containing one of the three specifications that exist in
the RADB """
spec_ids = [self._insert_test_spec(), self._insert_test_spec(), self._insert_test_spec()]
specifications = self.radb.getSpecifications(spec_ids[1])
self.assertEqual(len(specifications), 1)
def test_getSpecifications_normal_use_select_multiple_succeeds(self):
""" Verify if radb.getSpecifications() returns a list containing two of the three specifications that exist in
the RADB """
spec_ids = [self._insert_test_spec(), self._insert_test_spec(), self._insert_test_spec()]
specifications = self.radb.getSpecifications([spec_ids[1], spec_ids[2]])
self.assertEqual(len(specifications), 2)
def test_getSpecification_normal_use_select_one_succeeds(self):
""" Verify if radb.getSpecification() returns a list containing a single specification """
spec_ids = [self._insert_test_spec(), self._insert_test_spec(), self._insert_test_spec()]
specifications = self.radb.getSpecification(spec_ids[1])
self.assertEqual(len(specifications), 1)
def test_task_and_claim_conflicts(self):
# TODO: split up once the test setup is faster (not creating a new db for each test method)
# for testing purposous let's give CEP4 storage a total size of 100
self.assertTrue(self.radb.updateResourceAvailability(117, available_capacity=100, total_capacity=100))
self.assertEqual(100, self.radb.getResources(117, include_availability=True)[0]['total_capacity'])
now = datetime.utcnow()
now -= timedelta(seconds=now.second, microseconds=now.microsecond)
result = self.radb.insertSpecificationAndTask(0, 0, 'approved', 'observation', now, now+timedelta(hours=1), 'foo', 'CEP4')
self.assertTrue(result['inserted'])
spec_id1 = result['specification_id']
task_id1 = result['task_id']
task1 = self.radb.getTask(task_id1)
self.assertTrue(task1)
self.assertEqual(task_id1, task1['id'])
t1_claim1 = { 'resource_id': 117,
'starttime': task1['starttime'],
'endtime': task1['endtime'],
'status': 'tentative',
'claim_size': 40 }
# insert 1 claim
t1_claim_ids = self.radb.insertResourceClaims(task_id1, [t1_claim1], 'foo', 1, 1)
self.assertEqual(1, len(t1_claim_ids))
#get claim using t1_claim_ids, and check if db version is equal to original
t1_claims = self.radb.getResourceClaims(claim_ids=t1_claim_ids)
self.assertEqual(1, len(t1_claims))
for key, value in t1_claim1.items():
if key != 'status':
self.assertEqual(value, t1_claims[0][key])
#get claim again via task_id1, and check if db version is equal to original
t1_claims = self.radb.getResourceClaims(task_ids=task_id1)
self.assertEqual(1, len(t1_claims))
for key, value in t1_claim1.items():
if key != 'status':
self.assertEqual(value, t1_claims[0][key])
# try to insert a claim with the wrong (already 'claimed') status. Should rollback, and return no ids.
t1_claim2 = { 'resource_id': 117,
'starttime': task1['starttime'],
'endtime': task1['endtime'],
'status': 'claimed',
'claim_size': 10 }
t1_faulty_claim_ids = self.radb.insertResourceClaims(task_id1, [t1_claim2], 'foo', 1, 1)
self.assertEqual(1, len(self.radb.getResourceClaims(task_ids=task_id1))) #there should still be one (proper/non-faulty) claim for this task
# try to insert a claim with the wrong (already 'conflict') status. Should rollback, and return no ids.
t1_claim3 = { 'resource_id': 117,
'starttime': task1['starttime'],
'endtime': task1['endtime'],
'status': 'conflict',
'claim_size': 10 }
t1_faulty_claim_ids = self.radb.insertResourceClaims(task_id1, [t1_claim3], 'foo', 1, 1)
t1_faulty_claim_ids = self.radb.insertResourceClaims(task_id1, [t1_claim2], 'foo', 1, 1)
self.assertEqual(1, len(self.radb.getResourceClaims(task_ids=task_id1))) #there should still be one (proper/non-faulty) claim for this task
# try to update the task status to scheduled, should not succeed, since it's claims are not 'claimed' yet.
self.assertFalse(self.radb.updateTask(task_id1, task_status='scheduled'))
self.assertEqual('approved', self.radb.getTask(task_id1)['status'])
# try to update the claim status to claimed, should succeed.
self.assertTrue(self.radb.updateResourceClaims(t1_claim_ids, status='claimed'))
self.assertEqual('claimed', self.radb.getResourceClaim(t1_claim_ids[0])['status'])
# try to update the task status to scheduled again, should succeed this time.
self.assertTrue(self.radb.updateTask(task_id1, task_status='scheduled'))
self.assertEqual('scheduled', self.radb.getTask(task_id1)['status'])
self.assertEqual(0, len(self.radb.get_overlapping_claims(t1_claim_ids[0])))
self.assertEqual(0, len(self.radb.get_overlapping_tasks(t1_claim_ids[0])))
self.assertEqual(40, self.radb.get_max_resource_usage_between(117, task1['starttime'], task1['starttime'], 'claimed')['usage'])
self.assertEqual(0, self.radb.get_max_resource_usage_between(117, task1['starttime']-timedelta(hours=2), task1['starttime']-timedelta(hours=1), 'claimed')['usage'])
logger.info('------------------------------ concludes task 1 ------------------------------')
logger.info('-- now test with a 2nd task, and test resource availability, conflicts etc. --')
# another task, fully overlapping with task1
result = self.radb.insertSpecificationAndTask(1, 1, 'approved', 'observation', now, now+timedelta(hours=1), 'foo', 'CEP4')
self.assertTrue(result['inserted'])
spec_id2 = result['specification_id']
task_id2 = result['task_id']
task2 = self.radb.getTask(task_id2)
self.assertTrue(task2)
# insert a claim which won't fit, claim status after insert should be 'conflict' instead of 'tentative'
t2_claim1 = { 'resource_id': 117,
'starttime': task2['starttime'],
'endtime': task2['endtime'],
'status': 'tentative',
'claim_size': 90 }
t2_claim_ids = self.radb.insertResourceClaims(task_id2, [t2_claim1], 'foo', 1, 1)
self.assertEqual(1, len(t2_claim_ids))
# claim status after previous insert should be 'conflict' instead of 'tentative'
t2_claims = self.radb.getResourceClaims(claim_ids=t2_claim_ids)
self.assertEqual('conflict', t2_claims[0]['status'])
# and the task's status should be conflict as well
self.assertEqual('conflict', self.radb.getTask(task_id2)['status'])
self.assertEqual(set([t1_claim_ids[0]]), set(c['id'] for c in
self.radb.get_overlapping_claims(t2_claim_ids[0])))
self.assertEqual(set([task_id1]), set(t['id'] for t in
self.radb.get_overlapping_tasks(t2_claim_ids[0])))
#try to connect this claim to task1, should fail
self.assertFalse(self.radb.updateResourceClaims(t2_claim_ids, task_id=task_id1))
self.assertEqual(task_id2, t2_claims[0]['task_id'])
#try to connect this claim to other resource, should fail
self.assertFalse(self.radb.updateResourceClaims(t2_claim_ids, resource_id=118))
self.assertEqual(117, t2_claims[0]['resource_id'])
# try to update the task status to scheduled, should not succeed, since it's claims are not 'claimed' yet.
self.assertFalse(self.radb.updateTask(task_id2, task_status='scheduled'))
self.assertEqual('conflict', self.radb.getTask(task_id2)['status'])
# try to update the claim status to claimed, should not succeed, since it still won't fit
self.assertFalse(self.radb.updateResourceClaims(t2_claim_ids, status='claimed'))
self.assertEqual('conflict', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
# do conflict resolution, shift task and claims
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id2, starttime=now+timedelta(hours=2), endtime=now+timedelta(hours=3)))
# now the task and claim status should not be conflict anymore
self.assertEqual('tentative', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
self.assertEqual('approved', self.radb.getTask(task_id2)['status'])
self.assertEqual(0, len(self.radb.get_overlapping_claims(t2_claim_ids[0])))
self.assertEqual(0, len(self.radb.get_overlapping_tasks(t2_claim_ids[0])))
# try to update the claim status to claimed, should succeed now
self.assertTrue(self.radb.updateResourceClaims(t2_claim_ids, status='claimed'))
self.assertEqual('claimed', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
# and try to update the task status to scheduled, should succeed now
self.assertTrue(self.radb.updateTask(task_id2, task_status='scheduled'))
self.assertEqual('scheduled', self.radb.getTask(task_id2)['status'])
self.assertEqual(0, len(self.radb.get_overlapping_claims(t2_claim_ids[0])))
self.assertEqual(0, len(self.radb.get_overlapping_tasks(t2_claim_ids[0])))
# updating task/claim start/endtime should work, even for scheduled tasks with claimed claims
# effect might be that a scheduled tasks goes to conflict
# force conflict by moving back to original start/endtimes
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id2, starttime=task2['starttime'], endtime=task2['endtime']))
self.assertEqual('conflict', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
self.assertEqual('conflict', self.radb.getTask(task_id2)['status'])
# again do conflict resolution, shift task and claims
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id2, starttime=now+timedelta(hours=2), endtime=now+timedelta(hours=3)))
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id2, claim_status='claimed', task_status='scheduled'))
# now the task and claim status should be scheduled/claimed
self.assertEqual('scheduled', self.radb.getTask(task_id2)['status'])
self.assertEqual('claimed', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
# updating task/claim start/endtime should work, even for scheduled tasks with claimed claims
# effect might be that a scheduled tasks goes to conflict
# now, make simple endtime adjustment, task should stay scheduled
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id2, endtime=now+timedelta(hours=2.75)))
# now the task and claim status should still be scheduled/claimed
self.assertEqual('scheduled', self.radb.getTask(task_id2)['status'])
self.assertEqual('claimed', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
# now some weird corner case...
# when a task is > queued (for example, finished)
# then we don't want conflict statuses anymore if we update start/endtimes
# test here with weird starttime shift back to overlap with task1
self.assertTrue(self.radb.updateTask(task_id2, task_status='finished'))
self.assertEqual('finished', self.radb.getTask(task_id2)['status'])
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id2, starttime=task1['starttime']))
self.assertEqual('finished', self.radb.getTask(task_id2)['status'])
self.assertEqual('claimed', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
#ok, that works, now set the start/end time back to 'normal' for some later test cases
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id2, starttime=now+timedelta(hours=2), endtime=now+timedelta(hours=3)))
self.assertEqual('finished', self.radb.getTask(task_id2)['status'])
self.assertEqual('claimed', self.radb.getResourceClaim(t2_claim_ids[0])['status'])
logger.info('------------------------------ concludes task 2 ------------------------------')
logger.info('-- now test with a 3rd task, and test resource availability, conflicts etc. --')
#make sure we work with the latest info
task1 = self.radb.getTask(task_id1)
task2 = self.radb.getTask(task_id2)
# another task, partially overlapping with both task1 & task3
result = self.radb.insertSpecificationAndTask(2, 2, 'approved', 'observation',
task1['starttime'] + (task1['endtime']-task1['starttime'])/2,
task2['starttime'] + (task2['endtime']-task2['starttime'])/2,
'foo', 'CEP4')
self.assertTrue(result['inserted'])
spec_id2 = result['specification_id']
task_id3 = result['task_id']
task3 = self.radb.getTask(task_id3)
self.assertTrue(task3)
# insert a claim which won't fit, claim status after insert should be 'conflict' instead of 'tentative'
t3_claim1 = { 'resource_id': 117,
'starttime': task3['starttime'],
'endtime': task3['endtime'],
'status': 'tentative',
'claim_size': 80 }
t3_claim_ids = self.radb.insertResourceClaims(task_id3, [t3_claim1], 'foo', 1, 1)
self.assertEqual(1, len(t3_claim_ids))
# claim status after previous insert should be 'conflict' instead of 'tentative'
t3_claims = self.radb.getResourceClaims(claim_ids=t3_claim_ids)
self.assertEqual('conflict', t3_claims[0]['status'])
# and the task's status should be conflict as well
self.assertEqual('conflict', self.radb.getTask(task_id3)['status'])
self.assertEqual(set([t1_claim_ids[0], t2_claim_ids[0]]), set(c['id'] for c in
self.radb.get_overlapping_claims(t3_claim_ids[0])))
self.assertEqual(set([task_id1, task_id2]), set(t['id'] for t in
self.radb.get_overlapping_tasks(t3_claim_ids[0])))
# try to update the task status to scheduled, should not succeed, since it's claims are not 'claimed' yet.
self.assertFalse(self.radb.updateTask(task_id3, task_status='scheduled'))
self.assertEqual('conflict', self.radb.getTask(task_id3)['status'])
# try to update the claim status to claimed, should not succeed, since it still won't fit
self.assertFalse(self.radb.updateResourceClaims(t3_claim_ids, status='claimed'))
self.assertEqual('conflict', self.radb.getResourceClaim(t3_claim_ids[0])['status'])
# do conflict resolution, shift task away from task1 only (but keep overlapping with task2)
self.assertTrue(self.radb.updateTaskAndResourceClaims(task_id3, starttime=task1['endtime'] + (task2['starttime']-task1['endtime'])/2))
# now the task and claim status should still be in conflict
self.assertEqual('conflict', self.radb.getResourceClaim(t3_claim_ids[0])['status'])
self.assertEqual('conflict', self.radb.getTask(task_id3)['status'])
self.assertEqual(set([t2_claim_ids[0]]), set(c['id'] for c in
self.radb.get_overlapping_claims(t3_claim_ids[0])))
self.assertEqual(set([task_id2]), set(t['id'] for t in
self.radb.get_overlapping_tasks(t3_claim_ids[0])))
# do conflict resolution, reduce claim size (but keep overlapping with task2)
self.assertTrue(self.radb.updateResourceClaim(t3_claim_ids[0], claim_size=5))
# now the task and claim status should not be conflict anymore
self.assertEqual('tentative', self.radb.getResourceClaim(t3_claim_ids[0])['status'])
self.assertEqual('approved', self.radb.getTask(task_id3)['status'])
self.assertEqual(0, len(self.radb.get_overlapping_claims(t3_claim_ids[0])))
self.assertEqual(0, len(self.radb.get_overlapping_tasks(t3_claim_ids[0])))
# try to update the claim status to claimed, should succeed now
self.assertTrue(self.radb.updateResourceClaims(t3_claim_ids, status='claimed'))
self.assertEqual('claimed', self.radb.getResourceClaim(t3_claim_ids[0])['status'])
# and try to update the task status to scheduled, should succeed now
self.assertTrue(self.radb.updateTask(task_id3, task_status='scheduled'))
self.assertEqual('scheduled', self.radb.getTask(task_id3)['status'])
# try to trick the radb by resetting the claim_size back to 80 now that it was claimed. Should fail.
self.assertFalse(self.radb.updateResourceClaim(t3_claim_ids[0], claim_size=80))
#check if still 5, not 80
self.assertEqual(5, self.radb.getResourceClaim(t3_claim_ids[0])['claim_size'])
#and statuses should still be claimed/scheduled
self.assertEqual('claimed', self.radb.getResourceClaim(t3_claim_ids[0])['status'])
self.assertEqual('scheduled', self.radb.getTask(task_id3)['status'])
# suppose the resource_usages table is broken for some reason, fix it....
# break it first...
self._execute_query('TRUNCATE TABLE resource_allocation.resource_usage;')
#check that it's broken
self.assertNotEqual(40, self.radb.get_max_resource_usage_between(117, task1['starttime'], task1['starttime'], 'claimed')['usage'])
#fix it
self.radb.rebuild_resource_usages_table_from_claims()
#and test again that it's ok
self.assertEqual(40, self.radb.get_max_resource_usage_between(117, task1['starttime'], task1['starttime'], 'claimed')['usage'])
self.assertEqual(0, self.radb.get_max_resource_usage_between(117, task1['starttime']-timedelta(hours=2), task1['starttime']-timedelta(hours=1), 'claimed')['usage'])
def test_overlapping_claims(self):
# this is a special testcase to prove a bug found at 2017-08-16
# the bug was that a claim that should fit, does not fit according to the radb claim-methods.
# first, we'll prove that the bug exists (and that this test fails),
# and then, we'll fix the code, (so this test succeeds)
# for testing purposous let's give CEP4 storage a total size of 100
self.assertTrue(self.radb.updateResourceAvailability(117, available_capacity=100, total_capacity=100))
self.assertEqual(100, self.radb.getResources(117, include_availability=True)[0]['total_capacity'])
now = datetime.utcnow()
now -= timedelta(minutes=now.minute, seconds=now.second, microseconds=now.microsecond) # round to full hour
#insert one task, and reuse that for multiple overlapping claims
result = self.radb.insertSpecificationAndTask(0, 0, 'approved', 'observation', now, now+timedelta(hours=1), 'foo', 'CEP4')
self.assertTrue(result['inserted'])
task_id = result['task_id']
task = self.radb.getTask(task_id)
self.assertTrue(task)
self.assertEqual(task_id, task['id'])
#create two overlapping claims
claims = [ { 'resource_id': 117,
'starttime': now,
'endtime': now+timedelta(hours=0.75),
'status': 'tentative',
'claim_size': 40 },
{'resource_id': 117,
'starttime': now+timedelta(hours=0.25),
'endtime': now + timedelta(hours=1),
'status': 'tentative',
'claim_size': 40} ]
# insert the claims
claim_ids = self.radb.insertResourceClaims(task_id, claims, 'foo', 1, 1)
self.assertEqual(2, len(claim_ids))
claims_org = claims
#get claim using t1_claim_ids, and check if db version is equal to original
claims = self.radb.getResourceClaims(claim_ids=claim_ids)
self.assertEqual(2, len(claims))
for claim, claim_org in zip(claims, claims_org):
for key, value in claim_org.items():
if key != 'status':
self.assertEqual(value, claim_org[key])
# try to update the claim status to claimed, should succeed.
self.assertTrue(self.radb.updateResourceClaims(claim_ids, status='claimed'))
for claim in self.radb.getResourceClaims(claim_ids=claim_ids):
self.assertEqual('claimed', claim['status'])
# check the resource usage trend
logger.info("resource usages:\n%s", pformat(self.radb.getResourceUsages(now-timedelta(hours=1.0), now+timedelta(hours=2.0), 117)))
self.assertEqual(0, self.radb.get_max_resource_usage_between(117, now-timedelta(hours=1.0), now-timedelta(hours=0.01), 'claimed')['usage'])
self.assertEqual(40, self.radb.get_max_resource_usage_between(117, now+timedelta(hours=0.0), now+timedelta(hours=0.2), 'claimed')['usage'])
self.assertEqual(80, self.radb.get_max_resource_usage_between(117, now+timedelta(hours=0.3), now+timedelta(hours=0.6), 'claimed')['usage'])
self.assertEqual(40, self.radb.get_max_resource_usage_between(117, now+timedelta(hours=0.80), now+timedelta(hours=1.0), 'claimed')['usage'])
#check for a time range encapsulating the full task
self.assertEqual(80, self.radb.get_max_resource_usage_between(117, now+timedelta(hours=-0.1), now+timedelta(hours=1.1), 'claimed')['usage'])
#check for a time range not including the task
self.assertEqual(0, self.radb.get_max_resource_usage_between(117, now+timedelta(hours=1.1), now+timedelta(hours=2.0), 'claimed')['usage'])
self.assertEqual(0, self.radb.get_max_resource_usage_between(117, now-timedelta(hours=1.1), now-timedelta(hours=1.0), 'claimed')['usage'])
# check that there are no overlapping conflicting claims/tasks
for claim in claims:
#both claims should overlap with one (the other) claim
self.assertEqual(1, len(self.radb.get_overlapping_claims(claim['id'], 'claimed')))
self.assertEqual(1, len(self.radb.get_overlapping_tasks(claim['id'], 'claimed')))
#and there should be no overlapping claims of other status
self.assertEqual(0, len(self.radb.get_overlapping_claims(claim['id'], 'tentative')))
self.assertEqual(0, len(self.radb.get_overlapping_tasks(claim['id'], 'tentative')))
#add an extra claim, overlapping with only the last claim of size 40. So it should fit (100-40=60 and 60>30).
extra_claim = { 'resource_id': 117,
'starttime': now+timedelta(hours=0.8),
'endtime': now+timedelta(hours=0.9),
'status': 'tentative',
'claim_size': 30 }
extra_claim_ids = self.radb.insertResourceClaims(task_id, [extra_claim], 'foo', 1, 1)
self.assertEqual(1, len(extra_claim_ids))
#check the extra_claim's status, should be tentative.
for claim in self.radb.getResourceClaims(claim_ids=extra_claim_ids):
self.assertEqual('tentative', claim['status'])
# update the extra_claim status to 'claimed'. Should succeed.
self.assertTrue(self.radb.updateResourceClaims(extra_claim_ids, status='claimed'))
for claim in self.radb.getResourceClaims(claim_ids=extra_claim_ids):
self.assertEqual('claimed', claim['status'])
if __name__ == "__main__":
os.environ['TZ'] = 'UTC'
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)
unittest.main()