openstack/mistral
Code Issues Proposed changes

Add functional tests for workflow query

Browse Source 
Change-Id: I5ffff57d0d4c4fdbd8b62b4ee652074060a137bb
Partially-Implements: blueprint mistral-query-enhancement
This commit is contained in:
LingxianKong 2015-07-19 15:54:38 +08:00 committed by Lingxian Kong
parent 8137664379
commit 6d707f2306
1 changed files with 79 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
mistral/tests/functional/api/v2/test_mistral_basic_v2.py
View File

@ -19,6 +19,7 @@ from tempest_lib import decorators
from tempest_lib import exceptions
from mistral.tests.functional import base
from mistral import utils
class WorkbookTestsV2(base.TestCase):
@ -145,6 +146,84 @@ class WorkflowTestsV2(base.TestCase):
self.assertIn('std.create_instance', names)
self.assertNotIn('next', body)
@test.attr(type='smoke')
def test_get_list_workflows_with_pagination(self):
resp, body = self.client.get_list_obj(
'workflows?limit=1&sort_keys=name&sort_dirs=desc'
)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body['workflows']))
self.assertIn('next', body)
name_1 = body['workflows'][0].get('name')
next = body.get('next')
param_dict = utils.get_dict_from_string(
next.split('?')[1],
delimiter='&'
)
expected_sub_dict = {
'limit': 1,
'sort_keys': 'name',
'sort_dirs': 'desc'
}
self.assertDictContainsSubset(expected_sub_dict, param_dict)
# Query again using 'next' hint
url_param = next.split('/')[-1]
resp, body = self.client.get_list_obj(url_param)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body['workflows']))
name_2 = body['workflows'][0].get('name')
self.assertGreater(name_1, name_2)
@test.attr(type='negative')
def test_get_list_workflows_nonexist_sort_dirs(self):
context = self.assertRaises(
exceptions.BadRequest,
self.client.get_list_obj,
'workflows?limit=1&sort_keys=id&sort_dirs=nonexist'
)
self.assertIn(
'Unknown sort direction',
context.resp_body.get('faultstring')
)
@test.attr(type='negative')
def test_get_list_workflows_invalid_limit(self):
context = self.assertRaises(
exceptions.BadRequest,
self.client.get_list_obj,
'workflows?limit=-1&sort_keys=id&sort_dirs=asc'
)
self.assertIn(
'Limit must be positive',
context.resp_body.get('faultstring')
)
@test.attr(type='negative')
def test_get_list_workflows_duplicate_sort_keys(self):
context = self.assertRaises(
exceptions.BadRequest,
self.client.get_list_obj,
'workflows?limit=1&sort_keys=id,id&sort_dirs=asc,asc'
)
self.assertIn(
'Length of sort_keys must be equal or greater than sort_dirs',
context.resp_body.get('faultstring')
)
@test.attr(type='sanity')
def test_create_and_delete_workflow(self):
resp, body = self.client.create_workflow('wf_v2.yaml')