DB api test cases and revisions

taskflow/db/api.py

@@ -86,6 +86,9 @@ def job_get_state(context, job_id):
 def job_get_logbook(context, job_id):
     return IMPL.job_get_logbook(context, job_id)
 
+def job_create(context, job_name, job_id=None):
+	return IMPL.job_create(context, job_name, job_id)
+
 def job_destroy(context, job_id):
     return IMPL.job_destroy(context, job_id)
 

taskflow/db/sqlalchemy/api.py

@@ -21,6 +21,7 @@
 
 import logging
 
+from taskflow import states
 from taskflow.db.sqlalchemy import models
 from taskflow.db.sqlalchemy.session import get_session
 
@@ -43,7 +44,7 @@ def logbook_get(context, lb_id, session=None):
     query = model_query(context, models.LogBook, session=session).\
         filter_by(logbook_id=lb_id)
 
-    if not query:
+    if not query.first():
         raise exception.NotFound("No LogBook found with id "
                                  "%s." % (lb_id,))
 
@@ -54,7 +55,7 @@ def logbook_get_by_name(context, lb_name):
     query = model_query(context, models.LogBook).\
         filter_by(name=lb_name)
 
-    if not query:
+    if not query.all():
         raise exception.NotFound("LogBook %s not found."
                                  % (lb_name,))
 
@@ -72,7 +73,9 @@ def logbook_create(context, name, lb_id=None):
 
 def logbook_get_workflows(context, lb_id):
     """Return all workflows associated with a logbook"""
-    lb = logbook_get(context, lb_id)
+    session = get_session()
+    with session.begin():
+        lb = logbook_get(context, lb_id, session=session)
 
     return lb.workflows
 
@@ -85,14 +88,14 @@ def logbook_add_workflow(context, lb_id, wf_name):
 
         lb.workflows.append(wf)
 
-        return lb.workflows
+    return lb.workflows
 
 def logbook_destroy(context, lb_id):
     """Delete a given LogBook"""
     session = get_session()
     with session.begin():
         lb = logbook_get(context, lb_id, session=session)
-        lb.delete()
+        lb.delete(session=session)
 
 """
 JOB
@@ -100,10 +103,10 @@ JOB
 
 def job_get(context, job_id, session=None):
     """Return Job with matching job_id"""
-    query = model_query(context, models.Workflow, session=session).\
+    query = model_query(context, models.Job, session=session).\
         filter_by(job_id=job_id)
 
-    if not query:
+    if not query.first():
         raise exception.NotFound("No Job with id %s found"
                                  % (job_id,))
 
@@ -121,10 +124,10 @@ def job_add_workflow(context, job_id, wf_id):
     """Add a Workflow to given job"""
     session = get_session()
     with session.begin():
-        job = job_get(context, job_id)
-        wf = workflow_get(context, wf_id)
+        job = job_get(context, job_id, session=session)
+        wf = workflow_get(context, wf_id, session=session)
         job.workflows.append(wf)
-        return job.workflows
+    return job.workflows
 
 def job_get_owner(context, job_id):
     """Return a job's current owner"""
@@ -138,15 +141,28 @@ def job_get_state(context, job_id):
 
 def job_get_logbook(context, job_id):
     """Return the logbook associated with the given job"""
-    job = job_get(context, job_id)
+    session = get_session()
+    with session.begin():
+        job = job_get(context, job_id, session=session)
     return job.logbook
 
+def job_create(context, name, job_id=None):
+    job_ref = models.Job()
+    job_ref.name = name
+    job_ref.state = states.UNCLAIMED
+    if job_id:
+        job_ref.job_id = job_id
+        job_ref.logbook_id = job_id
+    job_ref.save()
+
+    return job_ref
+
 def job_destroy(context, job_id):
     """Delete a given Job"""
     session = get_session()
     with session.begin():
         job = job_get(context, job_id, session=session)
-        job.delete()
+        job.delete(session=session)
 
 
 """
@@ -158,7 +174,7 @@ def workflow_get(context, wf_name, session=None):
     query = model_query(context, models.Workflow, session=session).\
         filter_by(name=wf_name)
 
-    if not query:
+    if not query.first():
         raise exception.NotFound("Workflow %s not found." % (wf_name,))
 
     return query.first()
@@ -180,7 +196,9 @@ def workflow_get_names(context):
 
 def workflow_get_tasks(context, wf_name):
     """Return all tasks for a given Workflow"""
-    wf = workflow_get(context, wf_name)
+    session = get_session()
+    with session.begin():
+        wf = workflow_get(context, wf_name, session=session)
 
     return wf.tasks
 
@@ -206,7 +224,7 @@ def workflow_destroy(context, wf_name):
     session = get_session()
     with session.begin():
         wf = workflow_get(context, wf_name, session=session)
-        wf.delete()
+        wf.delete(session=session)
 
 """
 TASK
@@ -214,14 +232,14 @@ TASK
 
 def task_get(context, task_id, session=None):
     """Return Task with task_id"""
-    result = model_query(context, models.Task, session=session).\
+    query = model_query(context, models.Task, session=session).\
         filter_by(task_id=task_id)
 
-    if not result:
+    if not query.first():
         raise exception.NotFound("No Task found with id "
                                  "%s." % (task_id,))
 
-    return result
+    return query.first()
 
 def task_create(context, task_name, wf_id, task_id=None):
     """Create task associated with given workflow"""
@@ -248,4 +266,4 @@ def task_destroy(context, task_id):
     session = get_session()
     with session.begin():
         task = task_get(context, task_id, session=session)
-        task.delete()
+        task.delete(session=session)

taskflow/openstack/common/exception.py

@@ -39,7 +39,11 @@ class ApiError(Error):
 
 
 class NotFound(Error):
-    pass
+    msg = " %s not found."
+
+    def __init__(self, obj):
+        msg = self.__class__.msg % obj
+        super(NotFound, self).__init__(msg)
 
 
 class UnknownScheme(Error):

taskflow/tests/unit/test_db_api.py

@@ -0,0 +1,443 @@
+""" INSERT HEADER HERE """
+
+"""Import required libraries"""
+import os
+import unittest
+
+from os import path
+from oslo.config import cfg
+from taskflow import states
+from taskflow.db import api as db_api
+from taskflow.db.sqlalchemy import models
+from taskflow.db.sqlalchemy.session import get_session
+from taskflow.openstack.common import exception
+
+db_api.configure()
+db_filepath = cfg.CONF.get('sql_connection')
+
+def setUpModule():
+	if not path.isfile('test.db'):
+		models.create_tables()
+
+def tearDownModule():
+	os.remove('test.db')
+
+"""
+JobTest
+"""
+class JobTest(unittest.TestCase):
+	wf_ids = []
+	wf_names = []
+	lb_ids = []
+	lb_names = []
+	job_ids = []
+	job_names = []
+
+	@classmethod
+	def setUpClass(cls):
+		wf_fmt = u'workflow_{}'
+		lb_tmp = db_api.logbook_create('', u'logbook_1', 1)
+		cls.lb_ids.append(1)
+		cls.lb_names.append(u'logbook_1')
+		job_tmp = db_api.job_create('', u'job_1', 1)
+		cls.job_ids.append(1)
+		cls.job_names.append(u'job_1')
+		for i in range(1, 10):
+			wf_tmp = db_api.workflow_create('', wf_fmt.format(i))
+
+			db_api.logbook_add_workflow('', 1, wf_fmt.format(i))
+			db_api.job_add_workflow('', 1, wf_fmt.format(i))
+
+			cls.wf_ids.append(i)
+			cls.wf_names.append(wf_fmt.format(i))
+
+	@classmethod
+	def tearDownClass(cls):
+		for name in cls.wf_names:
+			db_api.workflow_destroy('', name)
+		for id in cls.lb_ids:
+			db_api.logbook_destroy('', id)
+		for id in cls.job_ids:
+			db_api.job_destroy('', id)
+		cls.wf_ids = []
+		cls.wf_names = []
+		cls.lb_ids = []
+		cls.lb_names = []
+		cls.job_ids = []
+		cls.job_names = []
+
+	def test_job_get(self):
+		print '\nTesting job_get...'
+		expected = self.job_names[0]
+		actual = db_api.job_get('', self.job_ids[0]).name
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.job_get, '', 9001)
+
+	def test_job_update(self):
+		print '\nTesting job_update...'
+		db_api.job_update('', 1, dict(owner='OwnerTest', state=states.CLAIMED))
+		job = db_api.job_get('', 1)
+
+		expected = 'OwnerTest'
+		actual = job.owner
+
+		self.assertEquals(expected, actual)
+
+		expected = states.CLAIMED
+		actual = job.state
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.job_update, '', 9001, dict(owner='OwnerTest', state=states.CLAIMED))
+
+	def test_job_add_workflow(self):
+		print '\nTesting job_add_workflow...'
+		db_api.workflow_create('', u'workflow_10')
+		self.wf_ids.append(10)
+		self.wf_names.append(u'workflow_10')
+
+		expected = self.wf_ids
+		actual = []
+		temp = db_api.job_add_workflow('', 1, u'workflow_10')
+
+		for workflow in temp:
+			actual.append(workflow.id)
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.job_add_workflow, '', 9001, u'workflow_10')
+		self.assertRaises(exception.NotFound, db_api.job_add_workflow, '', 1, u'workflow_9001')
+
+	def test_job_get_owner(self):
+		print '\nTesting job_get_owner...'
+		actual = db_api.job_get_owner('', 1)
+
+		self.assertIsNone(actual)
+
+		self.assertRaises(exception.NotFound, db_api.job_get_owner, '', 9001)
+
+	def test_job_get_state(self):
+		print '\nTesting job_get_state...'
+		expected = states.UNCLAIMED
+		actual = db_api.job_get_state('', 1)
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.job_get_state, '', 9001)
+
+	def test_job_get_logbook(self):
+		print '\nTesting job_get_logbook...'
+		expected = self.lb_names[0]
+		actual = db_api.job_get_logbook('', 1).name
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.job_get_logbook, '', 9001)
+
+	def test_job_create(self):
+		print '\nTesting job_create...'
+		id = 1
+		while (self.job_ids.count(id) > 0):
+			id = id + 1
+		job_tmp = db_api.job_create('', u'job_{}'.format(id), id)
+		self.job_ids.append(id)
+		self.job_names.append(u'job_{}'.format(id))
+
+		actual = db_api.job_get('', id)
+		self.assertIsNotNone(actual)
+
+	def test_job_destroy(self):
+		print '\nTesting job_destroy...'
+		id = self.job_ids.pop()
+		db_api.job_destroy('', id)
+		self.job_names.pop()
+
+		self.assertRaises(exception.NotFound, db_api.job_get, '', id)
+
+"""
+LogBookTest
+"""
+class LogBookTest(unittest.TestCase):
+	wf_ids = []
+	wf_names = []
+	lb_ids = []
+	lb_names = []
+
+	@classmethod
+	def setUpClass(cls):
+		wf_fmt = u'workflow_{}'
+		lb_tmp = db_api.logbook_create('', u'logbook_1', 1)
+		cls.lb_ids.append(1)
+		cls.lb_names.append(u'logbook_1')
+		for i in range(1, 10):
+			wf_tmp = db_api.workflow_create('', wf_fmt.format(i))
+
+			db_api.logbook_add_workflow('', 1, wf_fmt.format(i))
+
+			cls.wf_ids.append(i)
+			cls.wf_names.append(wf_fmt.format(i))
+
+	@classmethod
+	def tearDownClass(cls):
+		for name in cls.wf_names:
+			db_api.workflow_destroy('', name)
+		for id in cls.lb_ids:
+			db_api.logbook_destroy('', id)
+		cls.wf_ids = []
+		cls.wf_names = []
+		cls.lb_ids = []
+		cls.lb_names = []
+
+	def test_logbook_get(self):
+		print '\nTesting logbook_get...'
+		expected = self.lb_names[0]
+		actual = db_api.logbook_get('', self.lb_ids[0]).name
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.logbook_get, '', 9001)
+
+	def test_logbook_get_by_name(self):
+		print '\nTesting logbook_get_by_name...'
+		expected = [self.lb_ids[0]]
+		actual = []
+		for logbook in db_api.logbook_get_by_name('', self.lb_names[0]):
+			actual.append(logbook.id)
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.logbook_get_by_name, '', u'logbook_9001')
+
+	def test_logbook_create(self):
+		print '\nTesting logbook_create...'
+		id = 1
+		while (self.lb_ids.count(id) > 0):
+			id = id + 1
+		lb_tmp = db_api.logbook_create('', u'logbook_{}'.format(id), id)
+		self.lb_ids.append(id)
+		self.lb_names.append(u'logbook_{}'.format(id))
+
+		actual = db_api.logbook_get('', id)
+
+		self.assertIsNotNone(actual)
+
+	def test_logbook_get_workflows(self):
+		print '\nTesting logbook_get_workflows...'
+		expected = self.wf_ids
+		actual = []
+		wfs = db_api.logbook_get_workflows('', self.lb_ids[0])
+
+		for workflow in wfs:
+			actual.append(workflow.id)
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.logbook_get_workflows, '', 9001)
+
+	def test_logbook_add_workflow(self):
+		print '\nTesting logbook_add_workflow...'
+		db_api.workflow_create('', u'workflow_10')
+		self.wf_ids.append(10)
+		self.wf_names.append(u'workflow_10')
+
+		expected = self.wf_ids
+		actual = []
+		temp = db_api.logbook_add_workflow('', 1, u'workflow_10')
+
+		for workflow in temp:
+			actual.append(workflow.id)
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.logbook_add_workflow, '', 9001, u'workflow_10')
+		self.assertRaises(exception.NotFound, db_api.logbook_add_workflow, '', 1, u'workflow_9001')
+
+	def test_logbook_destroy(self):
+		print '\nTesting logbook_destroy...'
+		id = self.lb_ids.pop()
+		db_api.logbook_destroy('', id)
+		self.lb_names.pop()
+
+		self.assertRaises(exception.NotFound, db_api.logbook_get, '', id)
+		
+"""
+WorkflowTest
+"""
+class WorkflowTest(unittest.TestCase):
+	tsk_ids = []
+	tsk_names = []
+	wf_ids = []
+	wf_names = []
+
+	@classmethod
+	def setUpClass(cls):
+		wf_fmt = u'workflow_{}'
+		tsk_fmt = u'task_{}'
+		for i in range(1, 10):
+			wf_tmp = db_api.workflow_create('', wf_fmt.format(i))
+			tsk_tmp = db_api.task_create('', tsk_fmt.format(i), i, i)
+
+			db_api.workflow_add_task('', wf_fmt.format(i), i)
+			
+			cls.tsk_ids.append(i)
+			cls.tsk_names.append(tsk_fmt.format(i))
+			cls.wf_ids.append(i)
+			cls.wf_names.append(wf_fmt.format(i))
+	
+	@classmethod
+	def teardownClass(cls):
+		for id in tsk_ids:
+			db_api.task_destroy('', id)
+		for name in wf_names:
+			db_api.workflow_destroy('', name)
+		cls.tsk_ids = []
+		cls.tsk_names = []
+		cls.wf_ids = []
+		cls.wf_names = []
+					
+	def test_workflow_get(self):
+		print '\nTesting workflow_get...'
+		expected =  self.wf_ids[0]
+		actual = db_api.workflow_get('', self.wf_names[0]).id
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.workflow_get, '', u'workflow_9001')
+
+	def test_workflow_get_all(self):
+		print '\nTesting workflow_get_all...'
+		expected = self.wf_ids
+		actual = []
+		temp = db_api.workflow_get_all('')
+
+		for workflow in temp:
+			actual.append(workflow.id)
+			
+		self.assertEquals(expected, actual)
+
+	def test_workflow_get_names(self):
+		print '\nTesting workflow_get_names...'
+		expected = []
+		for name in self.wf_names:
+			expected.append(name)
+		expected = tuple(expected)
+		expected = [expected]
+		actual = db_api.workflow_get_names('')
+
+		self.assertEquals(expected, actual)
+
+	def test_workflow_get_tasks(self):
+		print '\nTesting workflow_get_tasks...'
+		expected = [self.tsk_names[0], self.tsk_names[9]]
+		actual = []
+		temp = db_api.workflow_get_tasks('', u'workflow_1')
+
+		for task in temp:
+			actual.append(task.name)
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.workflow_get_tasks, '', u'workflow_9001')
+
+	def test_workflow_add_task(self):
+		print '\nTesting workflow_add_task...'
+		tsk_tmp = db_api.task_create('', u'task_10', 1, 10)
+		db_api.workflow_add_task('', u'workflow_1', 10)
+		self.tsk_ids.append(10)
+		self.tsk_names.append('task_10')
+		expected = [self.tsk_names[0], self.tsk_names[9]]
+		tsks = db_api.workflow_get_tasks('', u'workflow_1')
+		actual = [tsks[0].name, tsks[1].name]
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.workflow_add_task, '', u'workflow_9001', 10)
+		self.assertRaises(exception.NotFound, db_api.workflow_add_task, '', u'workflow_1', 9001)
+
+	def test_workflow_create(self):
+		print '\nTesting workflow_create...'
+		id = 0
+		while (self.wf_ids.count(id) > 0):
+			id = id + 1
+		wf_tmp = db_api.workflow_create('', u'workflow_{}'.format(id))
+		self.wf_ids.append(id)
+		self.wf_names.append(u'workflow_{}'.format(id))
+			
+		self.assertIsNotNone(db_api.workflow_get('', u'workflow_{}'.format(id)))
+
+	def test_workflow_destroy(self):
+		print '\nTesting workflow_destroy...'
+		name = self.wf_names.pop()
+		db_api.workflow_destroy('', name)
+		self.wf_ids.pop()
+
+		self.assertRaises(exception.NotFound, db_api.workflow_get, '', name)
+
+"""
+TaskTest
+"""
+class TaskTest(unittest.TestCase):
+	tsk_ids = []
+	tsk_names = []
+
+	@classmethod
+	def setUpClass(cls):
+		tsk_fmt = u'task_{}'
+		for i in range(1, 10):
+			tsk_tmp = db_api.task_create('', tsk_fmt.format(i), i, i)
+			cls.tsk_ids.append(i)
+			cls.tsk_names.append(tsk_fmt.format(i))
+
+	@classmethod
+	def teardownClass(cls):
+		for id in tsk_ids:
+			db_api.task_destroy('', id)
+		cls.tsk_ids = []
+		cls.tsk_names = []
+
+	def test_task_get(self):
+		print '\nTesting task_get...'
+		expected = self.tsk_names[0]
+		actual = db_api.task_get('', self.tsk_ids[0])
+
+		self.assertEquals(expected, actual.name)
+
+		self.assertRaises(exception.NotFound, db_api.task_get, '', 9001)
+
+	def test_task_create(self):
+		print '\nTesting task_create...'
+		id = 1
+		while (self.tsk_ids.count(id) > 0):
+			id = id + 1
+		tsk_tmp = db_api.task_create('', u'task_{}'.format(id), 1, id)
+		self.tsk_ids.append(id)
+		self.tsk_names.append(u'task_{}'.format(id))
+
+		self.assertIsNotNone(db_api.task_get('', id))
+
+	def test_task_update(self):
+		print '\nTesting task_update...'
+		db_api.task_update('', 1, dict(exception='ExceptionTest', stacktrace='StacktraceTest'))
+		task = db_api.task_get('', 1)
+
+		expected = 'ExceptionTest'
+		actual = task.exception
+
+		self.assertEquals(expected, actual)
+
+		expected = 'StacktraceTest'
+		actual = task.stacktrace
+
+		self.assertEquals(expected, actual)
+
+		self.assertRaises(exception.NotFound, db_api.task_update, '', 9001, dict(exception='ExceptionTest', stacktrace='StacktraceTest'))
+
+	def test_task_destroy(self):
+		print '\nTesting task_destroy...'
+		id = self.tsk_ids.pop()
+		db_api.task_destroy('', id)
+		self.tsk_names.pop()
+
+		self.assertRaises(exception.NotFound, db_api.task_get, '', id)