ff1dc9544c
Added a SQLite backed backend API implementation that interfaces with the database using sqlalchemy. All of the unit tests associated are included as well. Change-Id: I0948fe85d381db4ef1305ed6234ca3e06f3c8760
213 lines
6.9 KiB
Python
213 lines
6.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright (C) 2013 Rackspace Hosting All Rights Reserved.
|
|
#
|
|
# 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 required libraries"""
|
|
import unittest2
|
|
|
|
from taskflow.backends import api as b_api
|
|
from taskflow import exceptions as exception
|
|
from taskflow.generics import flowdetail
|
|
from taskflow.openstack.common import uuidutils
|
|
from taskflow.patterns import graph_flow as flow
|
|
from taskflow.tests import utils
|
|
|
|
|
|
class FlowDetailTest(unittest2.TestCase):
|
|
"""This class is to test the functionality of the backend API's flowdetail
|
|
methods.
|
|
"""
|
|
wfs = []
|
|
fd_names = []
|
|
fd_ids = []
|
|
tsks = []
|
|
td_names = []
|
|
td_ids = []
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
# Create a workflow for flowdetails to use
|
|
wf_id = uuidutils.generate_uuid()
|
|
wf_name = 'wf-%s' % (wf_id)
|
|
|
|
wf = flow.Flow(wf_name, None, wf_id)
|
|
cls.wfs.append(wf)
|
|
|
|
# Create a task for taskdetails to use
|
|
task_id = uuidutils.generate_uuid()
|
|
task_name = 'task-%s' % (task_id)
|
|
|
|
tsk = utils.DummyTask(task_name, task_id)
|
|
cls.tsks.append(tsk)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
# Clear out the lists of workflows and tasks
|
|
utils.drain(cls.wfs)
|
|
utils.drain(cls.tsks)
|
|
|
|
def setUp(self):
|
|
# Create a flowdetail and record its uuid and name
|
|
fd_id = uuidutils.generate_uuid()
|
|
fd_name = 'fd-%s' % (fd_id)
|
|
|
|
b_api.flowdetail_create(fd_name, self.wfs[0], fd_id)
|
|
self.fd_names.append(fd_name)
|
|
self.fd_ids.append(fd_id)
|
|
|
|
# Create a taskdetail and record its uuid and name
|
|
td_id = uuidutils.generate_uuid()
|
|
td_name = 'td-%s' % (td_id)
|
|
|
|
b_api.taskdetail_create(td_name, self.tsks[0], td_id)
|
|
self.td_names.append(td_name)
|
|
self.td_ids.append(td_id)
|
|
|
|
def tearDown(self):
|
|
# Destroy all taskdetails and flowdetails in the backend
|
|
for id in self.td_ids:
|
|
b_api.taskdetail_destroy(id)
|
|
for id in self.fd_ids:
|
|
b_api.flowdetail_destroy(id)
|
|
|
|
# Drain the lists of taskdetail and flowdetail uuids and names
|
|
utils.drain(self.fd_names)
|
|
utils.drain(self.fd_ids)
|
|
utils.drain(self.td_names)
|
|
utils.drain(self.td_ids)
|
|
|
|
def test_flowdetail_create(self):
|
|
# Create a flowdetail and record its uuid and name
|
|
fd_id = uuidutils.generate_uuid()
|
|
fd_name = 'fd-%s' % (fd_id)
|
|
|
|
b_api.flowdetail_create(fd_name, self.wfs[0], fd_id)
|
|
self.fd_names.append(fd_name)
|
|
self.fd_ids.append(fd_id)
|
|
|
|
# Check to see that the created flowdetail is there
|
|
actual = b_api.flowdetail_get(fd_id)
|
|
|
|
self.assertIsNotNone(actual)
|
|
|
|
def test_flowdetail_destroy(self):
|
|
# Destroy the last added flowdetail
|
|
id = self.fd_ids.pop()
|
|
b_api.flowdetail_destroy(id)
|
|
self.fd_names.pop()
|
|
|
|
# Check to make sure the removed flowdetail is no longer there
|
|
self.assertRaises(exception.NotFound, b_api.flowdetail_get,
|
|
id)
|
|
|
|
def test_flowdetail_save(self):
|
|
# Create a generic flowdetail to save
|
|
fd_id = uuidutils.generate_uuid()
|
|
fd_name = 'fd-%s' % (fd_id)
|
|
wf = self.wfs[0]
|
|
fd = flowdetail.FlowDetail(fd_name, wf, fd_id)
|
|
|
|
# Save the generic flowdetail to the backend and record its uuid/name
|
|
b_api.flowdetail_save(fd)
|
|
self.fd_names.append(fd_name)
|
|
self.fd_ids.append(fd_id)
|
|
|
|
# Check that the saved flowdetail is in the backend
|
|
actual = b_api.flowdetail_get(fd_id)
|
|
|
|
self.assertIsNotNone(actual)
|
|
# Check that the saved flowdetail has no taskdetails
|
|
self.assertEquals(len(actual), 0)
|
|
|
|
# Add a generic taskdetail to the flowdetail
|
|
td = b_api.taskdetail_get(self.td_ids[0])
|
|
fd.add_task_detail(td)
|
|
|
|
# Save the updated flowdetail
|
|
b_api.flowdetail_save(fd)
|
|
|
|
# Check that the saved flowdetail is still there
|
|
actual = b_api.flowdetail_get(fd_id)
|
|
|
|
self.assertIsNotNone(actual)
|
|
# Check that the addition of a taskdetail was recorded
|
|
self.assertEquals(len(actual), 1)
|
|
|
|
def test_flowdetail_delete(self):
|
|
# Get the flowdetail to delete
|
|
id = self.fd_ids.pop()
|
|
fd = b_api.flowdetail_get(id)
|
|
# Delete the flowdetail
|
|
b_api.flowdetail_delete(fd)
|
|
self.fd_names.pop()
|
|
|
|
# Make sure it is not there anymore
|
|
self.assertRaises(exception.NotFound, b_api.flowdetail_get,
|
|
id)
|
|
|
|
def test_flowdetail_get(self):
|
|
# Get the first flowdetail
|
|
actual = b_api.flowdetail_get(self.fd_ids[0])
|
|
|
|
# Check that it is a flowdetail
|
|
self.assertIsInstance(actual, flowdetail.FlowDetail)
|
|
# Check that its name matches what is expected
|
|
self.assertEquals(actual.name, self.fd_names[0])
|
|
|
|
def test_flowdetail_add_task_detail(self):
|
|
# Get the first flowdetail
|
|
actual = b_api.flowdetail_get(self.fd_ids[0])
|
|
|
|
# Make sure it has no taskdetails
|
|
self.assertEquals(len(actual), 0)
|
|
|
|
# Add a taskdetail to the flowdetail
|
|
b_api.flowdetail_add_task_detail(self.fd_ids[0], self.td_ids[0])
|
|
|
|
# Get the flowdetail again
|
|
actual = b_api.flowdetail_get(self.fd_ids[0])
|
|
|
|
# Check that the flowdetail has one taskdetail
|
|
self.assertEquals(len(actual), 1)
|
|
|
|
def test_flowdetail_remove_taskdetail(self):
|
|
# Add a taskdetail to the first flowdetail
|
|
b_api.flowdetail_add_task_detail(self.fd_ids[0], self.td_ids[0])
|
|
|
|
# Get the first flowdetail
|
|
actual = b_api.flowdetail_get(self.fd_ids[0])
|
|
|
|
# Check that the first flowdetail has exactly one taskdetail
|
|
self.assertEquals(len(actual), 1)
|
|
|
|
# Remove the taskdetail from the first flowdetail
|
|
b_api.flowdetail_remove_taskdetail(self.fd_ids[0], self.td_ids[0])
|
|
|
|
# Get the first flowdetail
|
|
actual = b_api.flowdetail_get(self.fd_ids[0])
|
|
|
|
# Check that the first flowdetail no longer has any taskdetails
|
|
self.assertEquals(len(actual), 0)
|
|
|
|
def test_flowdetail_get_ids_names(self):
|
|
# Get a list of all uuids and names for flowdetails
|
|
actual = b_api.flowdetail_get_ids_names()
|
|
|
|
# Match it to our in-memory records
|
|
self.assertEquals(actual.values(), self.fd_names)
|
|
self.assertEquals(actual.keys(), self.fd_ids)
|