Add integration tests

Change-Id: I0873da3f0ddc73722f3d8506fde463da9d39cea9
This commit is contained in:
Sergey Murashov 2014-01-23 13:55:42 +04:00
parent 4f7ba4402d
commit 6aff5394dc
6 changed files with 322 additions and 0 deletions

View File

@ -0,0 +1,100 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Mirantis, Inc.
# 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 tempest.test
import json
from tempest.common import rest_client
import os
import tempest.api.mistral.config as cfg
class MistralTest(tempest.test.BaseTestCase):
@classmethod
def setUpClass(cls):
"""
This method allows to initialize authentication before
each test case and define parameters of Mistral API Service
"""
super(MistralTest, cls).setUpClass()
if not cfg.mistral.service_available:
raise cls.skipException("Mistral tests are disabled")
user = cfg.mistral.user
password = cfg.mistral.password
tenant = cfg.mistral.tenant
auth_url = cfg.mistral.auth_url
client_args = (cls.config, user, password, auth_url, tenant)
cls.client = rest_client.RestClient(*client_args)
cls.client.service = 'identity'
cls.token = cls.client.get_auth()
cls.client.base_url = cfg.mistral.mistral_url
cls.obj = []
def tearDown(self):
super(MistralTest, self).tearDown()
for i in self.obj:
try:
self.delete_obj(i[0], i[1])
except Exception:
pass
def check_base_url(self):
resp, body = self.client.get('', self.client.headers)
return resp, json.loads(body)
def check_base_url_with_version(self):
resp, body = self.client.get('v1/', self.client.headers)
return resp, json.loads(body)
def get_list_obj(self, name):
resp, body = self.client.get('v1/%s' % name, self.client.headers)
return resp, json.loads(body)
def create_obj(self, path, name):
post_body = '{"name": "%s"}' % name
resp, body = self.client.post('v1/%s/' % path, post_body,
self.client.headers)
return resp, json.loads(body)
def delete_obj(self, path, name):
self.client.delete('v1/%s/%s' % (path, name), self.client.headers)
def update_obj(self, path, name):
post_body = '{"name": "%s"}' % (name + 'updated')
resp, body = self.client.put('v1/%s/%s' % (path, name), post_body,
self.client.headers)
return resp, json.loads(body)
def get_workbook_definition(self, name):
headers = {'X-Auth-Token': self.client.headers['X-Auth-Token']}
resp, body = self.client.get('v1/workbooks/%s/definition' % name,
headers)
return resp, body
def upload_workbook_definition(self, name):
headers = {'Content-Type': 'text/plain',
'X-Auth-Token': self.client.headers['X-Auth-Token']}
__location = os.path.realpath(os.path.join(os.getcwd(),
os.path.dirname(__file__)))
f = open(os.path.join(__location, 'demo.yaml'), 'rb').read()
resp, body = self.client.put('v1/workbooks/%s/definition' % name, f,
headers)
return resp, body

View File

@ -0,0 +1,7 @@
[mistral]
auth_url = http://127.0.0.1:5000/v2.0/
mistral_url = http://127.0.0.1:8084
user = admin
password = password
tenant = admin
service_available = True

View File

@ -0,0 +1,40 @@
import os
from oslo.config import cfg
mistral_group = cfg.OptGroup(name='mistral', title='mistral')
MistralGroup = [
cfg.StrOpt('auth_url',
default='http://127.0.0.1:5000/v2.0/',
help="keystone url"),
cfg.StrOpt('mistral_url',
default='http://127.0.0.1:8084',
help="mistral url"),
cfg.StrOpt('user',
default='admin',
help="keystone user"),
cfg.StrOpt('password',
default='password',
help="password for keystone user"),
cfg.StrOpt('tenant',
default='admin',
help='keystone tenant'),
cfg.BoolOpt('service_available',
default='True',
help='mistral available')
]
def register_config(config, config_group, config_opts):
config.register_group(config_group)
config.register_opts(config_opts, config_group)
path = os.path.join("%s/config.conf" % os.getcwd())
if os.path.exists(path):
cfg.CONF([], project='mistralintegration', default_config_files=[path])
register_config(cfg.CONF, mistral_group, MistralGroup)
mistral = cfg.CONF.mistral

View File

@ -0,0 +1,57 @@
Services:
MyRest:
type: REST_API
parameters:
baseUrl: http://localhost:8988
actions:
task1:
parameters:
url: /tasks/task1
method: GET
task-parameters:
task2:
parameters:
url: /tasks/task2
method: GET
task-parameters:
task3:
parameters:
url: /tasks/task3
method: GET
task-parameters:
task4:
parameters:
url: /tasks/task4
method: GET
task-parameters:
Workflow:
tasks:
task1:
action: MyRest:task1
parameters:
task2:
dependsOn: [task1]
action: MyRest:task2
parameters:
task3:
dependsOn: [task1]
action: MyRest:task3
parameters:
task4:
dependsOn: [task2, task3]
action: MyRest:task4
parameters:
events:
task4:
type: periodic
tasks: task4
parameters:
cron-pattern: "*/1 * * * *"

View File

@ -0,0 +1,118 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Mirantis, Inc.
#
# 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 testtools
from tempest.api.mistral import base
from tempest.test import attr
from tempest import exceptions
class SanityTests(base.MistralTest):
@attr(type='smoke')
def test_check_base_url(self):
resp, _ = self.check_base_url()
self.assertEqual(resp['status'], '200')
@attr(type='smoke')
def test_get_list_obj(self):
resp, _ = self.check_base_url_with_version()
self.assertEqual(resp['status'], '200')
@attr(type='smoke')
def test_get_list_workbooks(self):
resp, body = self.get_list_obj('workbooks')
self.assertEqual(resp['status'], '200')
self.assertEqual(body['workbooks'], [])
@attr(type='smoke')
def test_get_workbook(self):
self.create_obj('workbooks', 'test')
self.obj.append(['workbooks', 'test'])
resp, body = self.get_list_obj('workbooks/test')
self.assertEqual(resp['status'], '200')
self.assertEqual(body['name'], 'test')
@attr(type='smoke')
def test_get_executions(self):
self.create_obj('workbooks', 'test')
self.obj.append(['workbooks', 'test'])
resp, body = self.get_list_obj('workbooks/test/executions')
self.assertEqual(resp['status'], '200')
self.assertEqual(body['executions'], [])
@attr(type='smoke')
def test_create_and_delete_workbook(self):
resp, body = self.create_obj('workbooks', 'test')
self.obj.append(['workbooks', 'test'])
self.assertEqual(resp['status'], '201')
self.assertEqual(body['name'], 'test')
resp, body = self.get_list_obj('workbooks')
self.assertEqual(resp['status'], '200')
self.assertEqual(body['workbooks'][0]['name'], 'test')
self.delete_obj('workbooks', 'test')
_, body = self.get_list_obj('workbooks')
self.assertEqual(body['workbooks'], [])
@attr(type='smoke')
def test_update_workbook(self):
self.create_obj('workbooks', 'test')
self.obj.append(['workbooks', 'test'])
resp, body = self.update_obj('workbooks', 'test')
self.assertEqual(resp['status'], '200')
self.assertEqual(body['name'], 'testupdated')
self.obj.append(['workbooks', 'testupdated'])
@attr(type='smoke')
def test_get_workbook_definition(self):
self.create_obj('workbooks', 'test')
self.obj.append(['workbooks', 'test'])
resp, body = self.get_workbook_definition('test')
self.assertEqual(resp['status'], '200')
self.assertIsNotNone(body)
@testtools.skip('It is not implemented')
@attr(type='smoke')
def test_upload_workbook_definition(self):
self.create_obj('workbooks', 'test1')
self.obj.append(['workbooks', 'test1'])
resp, body = self.upload_workbook_definition('test1')
self.assertEqual(resp['status'], '200')
@attr(type='negative')
def test_double_create_obj(self):
self.create_obj('workbooks', 'test')
self.obj.append(['workbooks', 'test'])
self.assertRaises(exceptions.BadRequest, self.create_obj, 'workbooks',
'test')
self.delete_obj('workbooks', 'test')
_, body = self.get_list_obj('workbooks')
self.assertEqual(body['workbooks'], [])