Move tempest tests from solum repo to solum-tempest-plugin

Partially-Implements: blueprint solum-tempest-separate-plugin
Change-Id: I8a437172646acf77a1c5007acddce8e1949b4399
This commit is contained in:
zhurong 2017-08-28 13:55:34 +08:00
parent f5a48364e3
commit ec4a098af9
28 changed files with 2543 additions and 38 deletions

View File

@ -1,19 +0,0 @@
# -*- coding: utf-8 -*-
# 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 pbr.version
__version__ = pbr.version.VersionInfo(
'solum-tempest-plugin').version_string()

View File

@ -0,0 +1,269 @@
# Copyright 2013 - Noorul Islam K M
#
# 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 copy
import json
import os
import random
import string
import time
from tempest.common import credentials_factory as common_creds
from tempest import config
from tempest.lib import auth
from tempest.lib.common import http
from tempest.lib.common import rest_client
import testtools
import yaml
from solum_tempest_plugin.common import apputils
CONF = config.CONF
assembly_sample_data = {"name": "test_assembly",
"description": "A test to create assembly",
"project_id": "project_id",
"user_id": "user_id",
"status": "status",
"application_uri": "http://localhost:5000"}
plan_sample_data = {"version": "1",
"name": "test_plan",
"description": "A test to create plan",
"project_id": "project_id",
"user_id": "user_id",
"artifacts": [{
"name": "No deus",
"artifact_type": "heroku",
"content": {
"href": "https://example.com/git/a.git",
"private": False
},
"language_pack": "auto",
}]}
solum_group = config.cfg.OptGroup(name='solum', title='Solum test options')
SolumGroup = [
config.cfg.BoolOpt('barbican_enabled',
default=False,
help="Defaults to false. Determines whether Barbican"
"is enabled."),
config.cfg.BoolOpt('camp_enabled',
default=True,
help="Defaults to true. Determines whether CAMP"
"is enabled.")
]
CONF.register_group(solum_group)
CONF.register_opts(SolumGroup, group=solum_group.name)
class SolumClient(rest_client.RestClient):
def __init__(self, auth_provider, service='application_deployment',
region='RegionOne'):
super(SolumClient, self).__init__(auth_provider, service, region)
self.endpoint_url = 'publicURL'
self.created_assemblies = []
self.created_plans = []
self.created_apps = []
self.created_lps = []
def request_without_auth(self, resource, method, headers=None, body=None):
if headers is None:
headers = {}
dscv = CONF.identity.disable_ssl_certificate_validation
http_obj = http.ClosingHttp(disable_ssl_certificate_validation=dscv)
url = '%s/%s' % (self.base_url, resource)
return http_obj.request(url, method, headers=headers, body=body)
def assembly_delete_done(self, assembly_uuid):
wait_interval = 1
growth_factor = 1.2
max_attempts = 5
for count in range(max_attempts):
try:
resp, body = self.get('v1/assemblies/%s' % assembly_uuid)
except Exception:
return True
time.sleep(wait_interval)
wait_interval *= growth_factor
return False
def create_assembly(self, plan_uuid, data=None):
assembly_data = copy.deepcopy(data or assembly_sample_data)
assembly_data['plan_uri'] = "%s/v1/plans/%s" % (self.base_url,
plan_uuid)
jsondata = json.dumps(assembly_data)
resp, body = self.post('v1/assemblies', jsondata)
assembly_resp = SolumResponse(resp=resp, body=body, body_type='json')
uuid = assembly_resp.uuid
if uuid is not None:
self.created_assemblies.append(uuid)
return assembly_resp
def create_plan(self, data=None):
plan_data = copy.deepcopy(data or plan_sample_data)
yaml_data = yaml.dump(plan_data)
resp, body = self.post('v1/plans', yaml_data,
headers={'content-type': 'application/x-yaml'})
plan_resp = SolumResponse(resp=resp, body=body, body_type='yaml')
uuid = plan_resp.uuid
if uuid is not None:
self.created_plans.append(uuid)
return plan_resp
def create_lp(self, data=None):
sample_lp = dict()
s = string.lowercase
sample_lp["name"] = "lp" + ''.join(random.sample(s, 5))
lp_url = "https://github.com/murali44/Solum-lp-Go.git"
sample_lp["source_uri"] = lp_url
jsondata = json.dumps(sample_lp)
resp, body = self.post('v1/language_packs', jsondata)
return sample_lp["name"]
def create_app(self, data=None):
app_data = copy.deepcopy(data) or apputils.get_sample_data()
json_data = json.dumps(app_data)
resp, body = self.post('v1/apps', json_data)
app_resp = SolumResponse(resp=resp, body=body, body_type='yaml')
if app_resp.id is not None:
self.created_apps.append(app_resp.id)
return app_resp
def delete_created_assemblies(self):
[self.delete_assembly(uuid) for uuid in list(self.created_assemblies)]
self.created_assemblies = []
def delete_assembly(self, uuid):
resp, body = self.delete('v1/assemblies/%s' % uuid)
if self.assembly_delete_done(uuid):
self.created_assemblies.remove(uuid)
return resp, body
def delete_created_plans(self):
self.delete_created_assemblies()
[self.delete_plan(uuid) for uuid in list(self.created_plans)]
self.created_plans = []
def delete_created_lps(self):
resp, body = self.get('v1/language_packs')
data = json.loads(body)
[self._delete_language_pack(pl['uuid']) for pl in data]
def _delete_language_pack(self, uuid):
resp, _ = self.delete('v1/language_packs/%s' % uuid)
def delete_language_pack(self, name):
resp, _ = self.delete('v1/language_packs/%s' % name)
def delete_created_apps(self):
self.delete_created_assemblies()
[self.delete_app(id) for id in list(self.created_apps)]
self.created_apps = []
def delete_app(self, id):
resp, body = self.delete(
'v1/apps/%s' % id,
headers={'content-type': 'application/json'})
if id in self.created_apps:
self.created_apps.remove(id)
return resp, body
def delete_plan(self, uuid):
resp, body = self.delete(
'v1/plans/%s' % uuid,
headers={'content-type': 'application/x-yaml'})
self.created_plans.remove(uuid)
return resp, body
class SolumResponse(object):
def __init__(self, resp, body, body_type):
self.resp = resp
self.body = body
if body_type == 'json':
self.data = json.loads(self.body)
elif body_type == 'yaml':
self.data = yaml.safe_load(self.body)
if self.data.get('uuid'):
self.uuid = self.data.get('uuid')
if self.data.get('id'):
self.id = self.data.get('id')
@property
def status(self):
return self.resp.status
@property
def yaml_data(self):
return yaml.safe_load(self.body)
@property
def json_data(self):
return json.loads(self.body)
class TestCase(testtools.TestCase):
def setUp(self):
super(TestCase, self).setUp()
credentials = common_creds.get_configured_admin_credentials(
'identity_admin')
auth_provider = get_auth_provider(credentials)
self.client = SolumClient(auth_provider)
self.builderclient = SolumClient(auth_provider, 'image_builder')
def tearDown(self):
super(TestCase, self).tearDown()
self.client.delete_created_apps()
def get_auth_provider(credentials, scope='project'):
default_params = {
'disable_ssl_certificate_validation':
CONF.identity.disable_ssl_certificate_validation,
'ca_certs': CONF.identity.ca_certificates_file,
'trace_requests': CONF.debug.trace_requests
}
if isinstance(credentials, auth.KeystoneV3Credentials):
auth_provider_class, auth_url = \
auth.KeystoneV3AuthProvider, CONF.identity.uri_v3
else:
auth_provider_class, auth_url = \
auth.KeystoneV2AuthProvider, CONF.identity.uri
_auth_provider = auth_provider_class(credentials, auth_url,
scope=scope,
**default_params)
_auth_provider.set_auth()
return _auth_provider
def is_fedora():
if os.path.exists("/etc/redhat-release"):
return True
return False
def config_set_as(config, target_value):
value = getattr(CONF.solum, config)
return value == target_value

View File

@ -0,0 +1,54 @@
# 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 json
from solum_tempest_plugin import base
class PlatformDiscoveryTestCase(base.TestCase):
def test_get_root_discovers_camp_v1_1(self):
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# get our platform_endpoints container
resp, body = (self.client.
request_without_auth('camp/platform_endpoints',
'GET'))
self.assertEqual(200, resp.status)
endpoints = json.loads(body)
self.assertEqual('platform_endpoints', endpoints['type'])
self.assertEqual('Solum_CAMP_endpoints', endpoints['name'])
pe_links = endpoints['platform_endpoint_links']
# there should be one element in the Link array
self.assertEqual(1, len(pe_links))
camp_v1_1_link = pe_links[0]
self.assertEqual('Solum_CAMP_v1_1_endpoint',
camp_v1_1_link['target_name'])
# get the URL of the platform_endpoint and strip the base URL
rel_ep_url = camp_v1_1_link['href'][len(self.client.base_url) + 1:]
# get our lone platform_endpoint resource
resp, body = (self.client.
request_without_auth(rel_ep_url,
'GET'))
self.assertEqual(200, resp.status)
endpoint = json.loads(body)
self.assertEqual('platform_endpoint', endpoint['type'])
self.assertEqual('Solum_CAMP_v1_1_endpoint', endpoint['name'])
self.assertEqual('CAMP 1.1', endpoint['specification_version'])
self.assertEqual('Solum CAMP 1.1', endpoint['implementation_version'])
self.assertEqual('KEYSTONE-2.0', endpoint['auth_scheme'])
self.assertEqual('%s/camp/v1_1/platform' % self.client.base_url,
endpoint['platform_uri'])

View File

@ -0,0 +1,156 @@
# 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 json
from tempest.lib import exceptions as tempest_exceptions
import yaml
from solum_tempest_plugin import base
from solum_tempest_plugin.camp.v1_1 import test_plans
class TestAssembliesController(base.TestCase):
def tearDown(self):
super(TestAssembliesController, self).tearDown()
self.client.delete_created_assemblies()
self.client.delete_created_plans()
# TODO(gpilz) - this is a dup of a method in test_plans.TestPlansController
def _create_camp_plan(self, data):
yaml_data = yaml.dump(data)
resp, body = self.client.post('camp/v1_1/plans', yaml_data,
headers={'content-type':
'application/x-yaml'})
plan_resp = base.SolumResponse(resp=resp,
body=body,
body_type='json')
uuid = plan_resp.uuid
if uuid is not None:
# share the Solum client's list of created plans
self.client.created_plans.append(uuid)
return plan_resp
def test_get_solum_assembly(self):
"""Test the CAMP assemblies collection resource.
Test that an assembly resource created through the Solum API is
visible via the CAMP API.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create an assembly using Solum
p_resp = self.client.create_plan()
self.assertEqual(201, p_resp.status)
a_resp = self.client.create_assembly(plan_uuid=p_resp.uuid)
self.assertEqual(201, a_resp.status)
new_uuid = a_resp.uuid
# try to get to the newly created assembly through the CAMP assemblies
# resource. it would be more efficient to simply take the UUID of the
# newly created resource and create a CAMP API URI
# (../camp/v1_1/assemblies/<uuid>) from that, but we want to test that
# a link to the Solum-created assembly appears in in the list of links
# in the CAMP plans resource.
resp, body = self.client.get('camp/v1_1/assemblies')
self.assertEqual(200, resp.status, 'GET assemblies resource')
# pick out the assemebly link for our new assembly uuid
assemblies_dct = json.loads(body)
camp_link = None
for link in assemblies_dct['assembly_links']:
link_uuid = link['href'].split("/")[-1]
if link_uuid == new_uuid:
camp_link = link
msg = 'Unable to find link to newly created plan in CAMP plans'
self.assertIsNotNone(camp_link, msg)
url = camp_link['href'][len(self.client.base_url) + 1:]
msg = ("GET Solum assembly resource for %s" %
camp_link['target_name'])
resp, body = self.client.get(url)
self.assertEqual(200, resp.status, msg)
assembly = json.loads(body)
self.assertEqual('assembly', assembly['type'])
self.assertEqual(base.assembly_sample_data['name'], assembly['name'])
def test_create_camp_assembly(self):
"""Test creating a CAMP assembly from a local plan resource.
Creates a plan resource then uses that to create an assembly resource.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using the CAMP API
resp = self._create_camp_plan(data=test_plans.sample_data)
self.assertEqual(201, resp.status)
uri = (resp.data['uri']
[len(self.client.base_url):])
ref_obj = json.dumps({'plan_uri': uri})
resp, body = self.client.post(
'camp/v1_1/assemblies',
ref_obj,
headers={'content-type': 'application/json'})
self.assertEqual(201, resp.status)
assem_resp = base.SolumResponse(resp=resp,
body=body,
body_type='json')
uuid = assem_resp.uuid
if uuid is not None:
# share the Solum client's list of created assemblies
self.client.created_assemblies.append(uuid)
def test_delete_plan_with_assemblies(self):
"""Test deleting a plan that has assemblies associated with it.
Creates a plan, an assembly, then tries to delete the plan.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using the CAMP API
resp = self._create_camp_plan(data=test_plans.sample_data)
self.assertEqual(201, resp.status)
plan_uri = (resp.data['uri']
[len(self.client.base_url):])
ref_obj = json.dumps({'plan_uri': plan_uri})
resp, body = self.client.post(
'camp/v1_1/assemblies',
ref_obj,
headers={'content-type': 'application/json'})
self.assertEqual(201, resp.status)
assem_resp = base.SolumResponse(resp=resp,
body=body,
body_type='json')
uuid = assem_resp.uuid
if uuid is not None:
# share the Solum client's list of created assemblies
self.client.created_assemblies.append(uuid)
# try to delete the plan before deleting the assembly
# resp, body = self.client.delete(plan_uri[1:])
# self.assertEqual(409, resp.status)
self.assertRaises(tempest_exceptions.Conflict,
self.client.delete, plan_uri[1:])

View File

@ -0,0 +1,53 @@
# 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 json
from solum_tempest_plugin import base
class TestSupportedFormats(base.TestCase):
def test_formats(self):
"""Tests normative statement RE-42 from the CAMP v1.1 specification:
http://docs.oasis-open.org/camp/camp-spec/v1.1/csprd02/
camp-spec-v1.1-csprd02.pdf
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
resp, body = self.client.get('camp/v1_1/formats/')
self.assertEqual(200, resp.status, 'GET formats resource')
formats = json.loads(body)
self.assertEqual('formats', formats['type'])
self.assertEqual('Solum_CAMP_formats', formats['name'])
format_links = formats['format_links']
# there should be one element in the Link array
self.assertEqual(1, len(format_links), 'RE-42')
json_link = format_links[0]
self.assertEqual('JSON', json_link['target_name'])
# get the URL of the platform_endpoint and strip the base URL
url = json_link['href'][len(self.client.base_url) + 1:]
# get our lone platform_endpoint resource
resp, body = self.client.get(url)
self.assertEqual(200, resp.status, 'GET JSON format resource')
formatr = json.loads(body)
self.assertEqual('format', formatr['type'])
self.assertEqual('JSON', formatr['name'], 'RE-42')
self.assertEqual('application/json', formatr['mime_type'], 'RE-42')
self.assertEqual('RFC4627', formatr['version'], 'RE-42')
self.assertEqual('http://www.ietf.org/rfc/rfc4627.txt',
formatr['documentation'],
'RE-42')

View File

@ -0,0 +1,98 @@
# 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 json
from solum_tempest_plugin import base
class TestParameterDefinitions(base.TestCase):
def test_assembly_parameter_definitions(self):
"""Tests normative statement RMR-03 from the CAMP v1.1 specification:
http://docs.oasis-open.org/camp/camp-spec/v1.1/csprd02/
camp-spec-v1.1-csprd02.pdf
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
resp, body = self.client.get('camp/v1_1/assemblies/')
self.assertEqual(200, resp.status, 'GET assemblies resource')
assemblies = json.loads(body)
# get the URL of the parameter_definitions resource
url = (assemblies['parameter_definitions_uri']
[len(self.client.base_url) + 1:])
# get the parameter_definitions resource
resp, body = self.client.get(url)
self.assertEqual(200, resp.status,
'GET assembly parameter_definitions resource')
pd_resc = json.loads(body)
self.assertEqual('parameter_definitions', pd_resc['type'])
self.assertIn('parameter_definition_links', pd_resc)
pd_links = pd_resc['parameter_definition_links']
# The assembly resource must reference parameter definitions for
# the pdp_uri, plan_uri, pdp_file, and plan_file parameters. It
# can reference additional parameter definitions.
self.assertLessEqual(4,
len(pd_links),
"too few parameter definition links")
expected_pds = ['pdp_uri', 'plan_uri', 'pdp_file', 'plan_file']
for pd_link in pd_links:
expected_pds.remove(pd_link['target_name'])
self.assertEqual(0,
len(expected_pds),
('Missing parameter_definition from %s' %
pd_resc['name']))
def test_plan_parameter_definitions(self):
"""Tests normative statement RMR-06 from the CAMP v1.1 specification:
http://docs.oasis-open.org/camp/camp-spec/v1.1/csprd02/
camp-spec-v1.1-csprd02.pdf
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
resp, body = self.client.get('camp/v1_1/plans/')
self.assertEqual(200, resp.status, 'GET plans resource')
plans = json.loads(body)
# get the URL of the parameter_definitions resource
url = (plans['parameter_definitions_uri']
[len(self.client.base_url) + 1:])
# get the parameter_definitions resource
resp, body = self.client.get(url)
self.assertEqual(200, resp.status,
'GET plans parameter_definitions resource')
pd_resc = json.loads(body)
self.assertEqual('parameter_definitions', pd_resc['type'])
self.assertIn('parameter_definition_links', pd_resc)
pd_links = pd_resc['parameter_definition_links']
# The plan resource must reference parameter definitions for
# the pdp_uri, plan_uri, pdp_file, and plan_file parameters. It
# can reference additional parameter definitions.
self.assertLessEqual(4,
len(pd_links),
"too few parameter definition links")
expected_pds = ['pdp_uri', 'plan_uri', 'pdp_file', 'plan_file']
for pd_link in pd_links:
expected_pds.remove(pd_link['target_name'])
self.assertEqual(0,
len(expected_pds),
('Missing parameter_definition from %s' %
pd_resc['name']))

View File

@ -0,0 +1,423 @@
# 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 copy
import json
from tempest.lib import exceptions as tempest_exceptions
import yaml
from solum_tempest_plugin import base
from solum_tempest_plugin.v1 import test_plan as solum_tests
sample_data = {"camp_version": "CAMP 1.1",
"name": "camp_test_plan",
"description": "A test to create CAMP plan",
"artifacts": [{
"name": "train spotter service",
"artifact_type": "org.python:python",
"language_pack": "python1",
"content": {"href": "https://sporgil.com/git/spotter.git"},
"requirements": [{
"requirement_type": "org.python:runtime",
"fulfillment": {
"name": "python runtime",
"description": "python 2.7.x runtime",
"characteristics": [{
"characteristic_type": "org.python:version",
"version": "[2.7, 3,0)"
}]
}
}]
}]}
class TestPlansController(base.TestCase):
def setUp(self):
super(TestPlansController, self).setUp()
def tearDown(self):
super(TestPlansController, self).tearDown()
self.client.delete_created_plans()
def _assert_output_expected(self, body_data, data):
self.assertEqual(body_data['name'], data['name'])
self.assertEqual(body_data['description'], data['description'])
if body_data['artifacts']:
self.assertEqual(body_data['artifacts'][0]['content']['href'],
data['artifacts'][0]['content']['href'])
self.assertIsNotNone(body_data['uuid'])
def _create_camp_plan(self, data):
yaml_data = yaml.dump(data)
resp, body = self.client.post('camp/v1_1/plans', yaml_data,
headers={'content-type':
'application/x-yaml'})
plan_resp = base.SolumResponse(resp=resp,
body=body,
body_type='json')
uuid = plan_resp.uuid
if uuid is not None:
# share the Solum client's list of created plans
self.client.created_plans.append(uuid)
return plan_resp
def test_get_solum_plan(self):
"""Test the visibility of Solum-created plans
Test that an plan resource created through the Solum API is
visible via the CAMP API.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using Solum
p_resp = self.client.create_plan()
self.assertEqual(201, p_resp.status)
new_plan = p_resp.yaml_data
new_uuid = new_plan['uuid']
# try to get to the newly plan created through the CAMP plans
# resource. it would be more efficient to simply take the UUID of the
# newly created resource and create a CAMP API URI
# (../camp/v1_1/plans/<uuid>) from that, but we want to test that a
# link to the Solum-created plan appears in in the list of links in
# the CAMP plans resource.
resp, body = self.client.get('camp/v1_1/plans')
self.assertEqual(200, resp.status, 'GET plans resource')
# pick out the plan link for our new plan uuid
plans_dct = json.loads(body)
camp_link = None
for link in plans_dct['plan_links']:
link_uuid = link['href'].split("/")[-1]
if link_uuid == new_uuid:
camp_link = link
msg = 'Unable to find link to newly created plan in CAMP plans'
self.assertIsNotNone(camp_link, msg)
url = camp_link['href'][len(self.client.base_url) + 1:]
msg = ("GET Solum plan resource for %s" %
camp_link['target_name'])
resp, body = self.client.get(url)
self.assertEqual(200, resp.status, msg)
# CAMP plans are rendered in JSON
plan = json.loads(body)
self.assertEqual(base.plan_sample_data['name'], plan['name'])
self.assertEqual(base.plan_sample_data['description'],
plan['description'])
def test_create_camp_plan(self):
"""Test the visibility of CAMP-created plans
Test that an plan resource created through the CAMP API is
visible through the Solum API.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using the CAMP API
resp = self._create_camp_plan(data=sample_data)
self.assertEqual(201, resp.status)
self._assert_output_expected(resp.data, sample_data)
uuid = resp.data['uuid']
# get the plan using the Solum API
resp, body = self.client.get(
'v1/plans/%s' % uuid,
headers={'content-type': 'application/x-yaml'})
self.assertEqual(200, resp.status)
yaml_data = yaml.safe_load(body)
self._assert_output_expected(yaml_data, sample_data)
def test_create_camp_plan_with_private_github_repo(self):
"""Test CAMP support for private git repos
Test that CAMP support the Solum private github case.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# copy the Solum test data and add a camp_version
camp_data = copy.copy(solum_tests.sample_data_private)
camp_data['camp_version'] = 'CAMP 1.1'
resp = self._create_camp_plan(data=camp_data)
self.assertEqual(201, resp.status)
self._assert_output_expected(resp.data, camp_data)
def test_get_no_plan(self):
"""Try to GET a CAMP plan that doesn't exist
Test the CAMP API's ability to respond with an HTTP 404 when doing a
GET on a non-existent plan.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
self.assertRaises(tempest_exceptions.NotFound,
self.client.get, 'camp/v1_1/plans/no_plan')
def test_create_bad_content_type(self):
"""Try to create a CAMP plan with a bogus Content-Type
Test that an attempt to create a plan with an incorrect Content-Type
header results in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
yaml_data = yaml.dump(sample_data)
self.assertRaises(tempest_exceptions.InvalidContentType,
self.client.post, 'camp/v1_1/plans', yaml_data,
headers={'content-type': 'image/jpeg'})
def test_create_no_camp_version(self):
"""Try to create a CAMP plan from input lacking 'camp_version'
Test that an attempt to create a plan with no 'camp_version' results
in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
no_version_data = copy.copy(sample_data)
del no_version_data['camp_version']
no_version_str = yaml.dump(no_version_data)
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'camp/v1_1/plans',
no_version_str,
headers={'content-type': 'application/x-yaml'})
def test_create_bad_camp_version(self):
"""Try to create a CAMP plan from input with incorrect 'camp_version'
Test that an attempt to create a plan with an incorrect 'camp_version'
results in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
bad_version_data = copy.copy(sample_data)
bad_version_data['camp_version'] = 'CAMP 8.1'
bad_version_str = yaml.dump(bad_version_data)
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'camp/v1_1/plans',
bad_version_str,
headers={'content-type': 'application/x-yaml'})
def test_create_empty_yaml(self):
"""Try to create a CAMP plan from an empty YAML document
Test that an attempt to create a plan using an empty yaml document
results in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'camp/v1_1/plans', '{}',
headers={'content-type': 'application/x-yaml'})
def test_create_invalid_yaml(self):
"""Try to create a CAMP plan from invalid YAML
Test that an attempt to create a plan using an invalid document
results in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'camp/v1_1/plans', 'invalid type',
headers={'content-type': 'application/x-yaml'})
def test_create_invalid_syntax(self):
"""Try to create a CAMP plan from garbled YAML
Test that an attempt to create a plan using yaml with an invalid syntax
results in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'camp/v1_1/plans',
"}invalid: y'm'l3!",
headers={'content-type': 'application/x-yaml'})
def test_delete_solum_plan_from_camp(self):
"""Test the ability to DELETE a Solum-created plan
Test that an plan resource created through the Solum API can
be deleted through the CAMP API.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using Solum
create_resp = self.client.create_plan()
self.assertEqual(201, create_resp.status)
uuid = create_resp.uuid
# delete the plan using CAMP
resp, body = self.client.delete('camp/v1_1/plans/%s' % uuid)
self.assertEqual(204, resp.status)
# remove the plan from the list of plans so we don't try to remove it
# twice
self.client.created_plans.remove(uuid)
def test_delete_camp_plan_from_solum(self):
"""Test the ability of the Solum API to delete a CAMP-created plan
Test that an plan resource created through the CAMP API can
be deleted through the Solum API.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using the CAMP API
resp = self._create_camp_plan(data=sample_data)
self.assertEqual(201, resp.status)
self._assert_output_expected(resp.data, sample_data)
uuid = resp.data['uuid']
# delete the plan using the Solum API
resp, body = self.client.delete('v1/plans/%s' % uuid)
self.assertEqual(202, resp.status)
# remove the plan from the list of plans so we don't try to remove it
# twice
self.client.created_plans.remove(uuid)
def test_delete_no_plan(self):
"""Try to DELTE a plan that doesn't exist
Test the ability of CAMP to respond with an HTTP 404 when the client
tries to DELETE a plan that doesn' exist
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
self.assertRaises(tempest_exceptions.NotFound,
self.client.delete, 'camp/v1_1/plans/no_plan')
def test_patch_plan(self):
"""PATCH a CAMP plan.
Test the ability to modify a CAMP plan using the HTTP PATCH
method with a JSON Patch request.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using the CAMP API
resp = self._create_camp_plan(data=sample_data)
self.assertEqual(201, resp.status)
uri = (resp.data['uri']
[len(self.client.base_url) + 1:])
patch_data = [
{"op": "add", "path": "/tags", "value": ["foo", "baz"]}
]
patch_json = json.dumps(patch_data)
resp, body = self.client.patch(
uri, patch_json,
headers={'content-type': 'application/json-patch+json'})
self.assertEqual(200, resp.status)
json_data = json.loads(body)
self.assertIn('tags', json_data)
tags = json_data['tags']
self.assertEqual(2, len(tags))
self.assertEqual('foo', tags[0])
self.assertEqual('baz', tags[1])
def test_patch_no_plan(self):
"""Try to PATCH a non-existent CAMP plan
Test that an attempt to PATCH a plan that doesn't exist results in
an HTTP 404 "Not Found" error
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# use an invalid JSON Patch document to test correct error precedence
patch_data = [
{"op": "add", "value": ["foo", "baz"]}
]
patch_json = json.dumps(patch_data)
# use a bad Content-Type to further test correct error precedence
self.assertRaises(tempest_exceptions.NotFound,
self.client.patch, 'camp/v1_1/plans/no_plan',
patch_json,
headers={'content-type':
'application/x-not-a-type'})
def test_patch_bad_content_type(self):
"""PATCH a CAMP plan using an incorrect content-type
Test that an attempt to patch a plan with an incorrect Content-Type
results in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using the CAMP API
resp = self._create_camp_plan(data=sample_data)
self.assertEqual(201, resp.status)
uri = (resp.data['uri']
[len(self.client.base_url) + 1:])
patch_data = [
{"op": "add", "path": "/tags", "value": ["foo", "baz"]}
]
patch_json = json.dumps(patch_data)
self.assertRaises(tempest_exceptions.InvalidContentType,
self.client.patch, uri, patch_json,
headers={'content-type': 'application/x-not-a-type'})
def test_patch_bad_json_patch(self):
"""PATCH a CAMP plan using invalid JSON Patch document
Test that an attempt to patch a plan with a mal-formed JSON Patch
request (missing 'path') results in the proper error.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# create a plan using the CAMP API
resp = self._create_camp_plan(data=sample_data)
self.assertEqual(201, resp.status)
uri = (resp.data['uri']
[len(self.client.base_url) + 1:])
patch_data = [
{"op": "add", "value": ["foo", "baz"]}
]
patch_json = json.dumps(patch_data)
self.assertRaises(tempest_exceptions.BadRequest,
self.client.patch, uri, patch_json,
headers={'content-type':
'application/json-patch+json'})

View File

@ -0,0 +1,73 @@
# 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 json
from solum_tempest_plugin import base
class TestPlatformAndContainers(base.TestCase):
def _test_get_resource(self, url, rtype, name):
resp, body = self.client.get(url)
self.assertEqual(200, resp.status, 'GET %s resource' % rtype)
resource = json.loads(body)
self.assertEqual(rtype, resource['type'])
self.assertEqual(name, resource['name'])
def test_get_platform_and_containers(self):
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
# get and test our platform resource
resp, body = self.client.get('camp/v1_1/platform/')
self.assertEqual(200, resp.status, 'GET platform resource')
platform = json.loads(body)
self.assertEqual('platform', platform['type'])
self.assertEqual('Solum_CAMP_v1_1_platform', platform['name'])
self.assertEqual('CAMP 1.1', platform['specification_version'])
self.assertEqual('Solum CAMP 1.1', platform['implementation_version'])
# get and test the supported formats resource
url = (platform['supported_formats_uri']
[len(self.client.base_url) + 1:])
self._test_get_resource(url, 'formats', 'Solum_CAMP_formats')
# get and test the extensions resource
url = (platform['extensions_uri']
[len(self.client.base_url) + 1:])
self._test_get_resource(url, 'extensions', 'Solum_CAMP_extensions')
# get and test the type_definitions resource
url = (platform['type_definitions_uri']
[len(self.client.base_url) + 1:])
self._test_get_resource(url,
'type_definitions',
'Solum_CAMP_type_definitions')
# get and test the platform_endpoints resource
url = (platform['platform_endpoints_uri']
[len(self.client.base_url) + 1:])
self._test_get_resource(url,
'platform_endpoints',
'Solum_CAMP_endpoints')
# get and test the assemblies collection resource
url = platform['assemblies_uri'][len(self.client.base_url) + 1:]
self._test_get_resource(url, 'assemblies', 'Solum_CAMP_assemblies')
# get and test the services collection resource
url = platform['services_uri'][len(self.client.base_url) + 1:]
self._test_get_resource(url, 'services', 'Solum_CAMP_services')
# get and test the plans collection resource
url = platform['plans_uri'][len(self.client.base_url) + 1:]
self._test_get_resource(url, 'plans', 'Solum_CAMP_plans')

View File

@ -0,0 +1,58 @@
# 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 json
from solum_tempest_plugin import base
class TestTypeDefinitions(base.TestCase):
def _test_get_resource(self, abs_url, msg, rtype, name):
url = abs_url[len(self.client.base_url) + 1:]
resp, body = self.client.get(url)
self.assertEqual(200, resp.status, msg)
resource = json.loads(body)
self.assertEqual(rtype, resource['type'])
self.assertEqual(name, resource['name'])
return body
def test_type_definitions(self):
"""Test the CAMP type_definition metadata.
Crawls tree rooted in type_definitions and verifies that all the
resources exist and that all the links to the attribute_definition
resources are valid and the attribute_definitions resources exist.
"""
if base.config_set_as('camp_enabled', False):
self.skipTest('CAMP not enabled.')
resp, body = self.client.get('camp/v1_1/type_definitions')
self.assertEqual(200, resp.status, 'GET type_definitions resource')
defs_dct = json.loads(body)
for link_dct in defs_dct['type_definition_links']:
msg = ("GET type_definition resource for %s" %
link_dct['target_name'])
body = self._test_get_resource(link_dct['href'],
msg,
'type_definition',
link_dct['target_name'])
def_dct = json.loads(body)
for adl_dct in def_dct['attribute_definition_links']:
msg = ("GET attribute_definition resource for %s" %
link_dct['target_name'])
self._test_get_resource(adl_dct['href'],
msg,
'attribute_definition',
adl_dct['target_name'])

View File

View File

@ -0,0 +1,38 @@
# 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 random
import string
def get_sample_data(languagepack=''):
data = dict()
s = string.lowercase
data["name"] = "test_app" + ''.join(random.sample(s, 5))
data["description"] = "descp"
data["languagepack"] = languagepack
data["trigger_actions"] = ["test", "build", "deploy"]
data["ports"] = [80]
source = {}
source['repository'] = "https://github.com/a/b.git"
source['revision'] = "master"
data["source"] = source
workflow = {}
workflow["test_cmd"] = "./unit_tests.sh"
workflow["run_cmd"] = "python app.py"
data["workflow_config"] = workflow
data["repo_token"] = 'abc'
return data

View File

@ -0,0 +1,22 @@
# Copyright 2015 - Rackspace Hosting
#
# 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.
from solum_tempest_plugin import base
class ReleaseVersionTests(base.TestCase):
def test_release_reported(self):
resp, body = self.client.get('/')
release_version = resp.get('x-solum-release')
self.assertTrue(release_version is not None)

View File

@ -0,0 +1,88 @@
#
# Copyright 2013 - Red Hat, 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 json
from solum_tempest_plugin import base
class VersionDiscoveryTestCase(base.TestCase):
def test_get_root_discovers_v1(self):
resp, body = self.client.get('/')
body = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body))
v1 = body[0]
self.assertEqual('v1.0', v1['id'])
self.assertEqual('CURRENT', v1['status'])
self.assertEqual('v1', v1['link']['target_name'])
self.assertEqual('%s/v1' % self.client.base_url, v1['link']['href'])
def test_delete_root_discovers_v1(self):
resp, body = self.client.delete('/')
body = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body))
v1 = body[0]
self.assertEqual('v1.0', v1['id'])
self.assertEqual('CURRENT', v1['status'])
self.assertEqual('v1', v1['link']['target_name'])
self.assertEqual('%s/v1' % self.client.base_url, v1['link']['href'])
def test_post_root_discovers_v1(self):
resp, body = self.client.post('/', '{}')
body = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body))
v1 = body[0]
self.assertEqual('v1.0', v1['id'])
self.assertEqual('CURRENT', v1['status'])
self.assertEqual('v1', v1['link']['target_name'])
self.assertEqual('%s/v1' % self.client.base_url, v1['link']['href'])
def test_put_root_discovers_v1(self):
resp, body = self.client.put('/', '{}')
body = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body))
v1 = body[0]
self.assertEqual('v1.0', v1['id'])
self.assertEqual('CURRENT', v1['status'])
self.assertEqual('v1', v1['link']['target_name'])
self.assertEqual('%s/v1' % self.client.base_url, v1['link']['href'])
def test_post_no_body_root_discovers_v1(self):
self.skipTest("POST without body will hang request: #1367470")
resp, body = self.client.post('/', None)
body = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body))
v1 = body[0]
self.assertEqual('v1.0', v1['id'])
self.assertEqual('CURRENT', v1['status'])
self.assertEqual('v1', v1['link']['target_name'])
self.assertEqual('%s/v1' % self.client.base_url, v1['link']['href'])
def test_put_no_body_root_discovers_v1(self):
self.skipTest("PUT without body will hang request: #1367470")
resp, body = self.client.put('/', None)
body = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual(1, len(body))
v1 = body[0]
self.assertEqual('v1.0', v1['id'])
self.assertEqual('CURRENT', v1['status'])
self.assertEqual('v1', v1['link']['target_name'])
self.assertEqual('%s/v1' % self.client.base_url, v1['link']['href'])

View File

View File

@ -0,0 +1,59 @@
# 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 json
import time
import requests
from solum_tempest_plugin import base
from solum_tempest_plugin.common import apputils
class TestTriggerController(base.TestCase):
def test_trigger_post(self):
lp_name = self.client.create_lp()
data = apputils.get_sample_data(languagepack=lp_name)
resp = self.client.create_app(data=data)
bdy = json.loads(resp.body)
trigger_uri = bdy['trigger_uri']
# Using requests instead of self.client to test unauthenticated request
status_url = 'https://api.github.com/repos/u/r/statuses/{sha}'
body_dict = {'sender': {'url': 'https://api.github.com'},
'pull_request': {'head': {'sha': 'asdf'}},
'repository': {'statuses_url': status_url}}
body = json.dumps(body_dict)
resp = requests.post(trigger_uri, data=body)
self.assertEqual(202, resp.status_code)
self.client.delete_created_apps()
# since app delete is an async operation, wait few seconds for app
# delete and then delete language pack (otherwise language pack
# cannot be deleted)
time.sleep(2)
self.client.delete_created_lps()
def test_trigger_post_with_empty_body(self):
lp_name = self.client.create_lp()
data = apputils.get_sample_data(languagepack=lp_name)
resp = self.client.create_app(data=data)
bdy = json.loads(resp.body)
trigger_uri = bdy['trigger_uri']
# Using requests instead of self.client to test unauthenticated request
resp = requests.post(trigger_uri)
self.assertEqual(400, resp.status_code)
self.client.delete_created_apps()
# since app delete is an async operation, wait few seconds for app
# delete and then delete language pack (otherwise language pack
# cannot be deleted)
time.sleep(2)
self.client.delete_created_lps()

View File

@ -0,0 +1,154 @@
#
# Copyright 2015 - Rackspace US, 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 json
import time
from tempest.lib import exceptions as tempest_exceptions
import yaml
from solum_tempest_plugin import base
from solum_tempest_plugin.common import apputils
class TestAppController(base.TestCase):
def _assert_app_data(self, actual, expected):
self.assertEqual(expected["name"], actual["name"])
self.assertEqual(expected["description"], actual["description"])
self.assertEqual(expected["languagepack"], actual["languagepack"])
self.assertEqual(expected["trigger_actions"],
actual["trigger_actions"])
self.assertEqual(expected["ports"], actual["ports"])
self.assertEqual(expected["source"]["repository"],
actual["source"]["repository"])
self.assertEqual(expected["source"]["revision"],
actual["source"]["revision"])
self.assertEqual(expected["workflow_config"]["test_cmd"],
actual["workflow_config"]["test_cmd"])
self.assertEqual(expected["workflow_config"]["run_cmd"],
actual["workflow_config"]["run_cmd"])
def setUp(self):
super(TestAppController, self).setUp()
def tearDown(self):
super(TestAppController, self).tearDown()
def test_app_create(self):
lp_name = self.client.create_lp()
data = apputils.get_sample_data(languagepack=lp_name)
resp = self.client.create_app(data=data)
self.assertEqual(201, resp.status)
self.client.delete_app(resp.id)
time.sleep(2)
self.client.delete_language_pack(lp_name)
def test_app_create_bad_port_data(self):
try:
bad_data = apputils.get_sample_data()
bad_data["ports"][0] = -1
self.client.create_plan(data=bad_data)
except tempest_exceptions.BadRequest:
self.assertTrue(True)
def test_app_create_empty_body(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/apps', '{}',
headers={'content-type': 'application/json'})
def test_app_patch(self):
lp_name = self.client.create_lp()
data = apputils.get_sample_data(languagepack=lp_name)
create_resp = self.client.create_app(data=data)
self.assertEqual(201, create_resp.status)
json_update = {
'name': 'newfakeappname',
'workflow_config': {
'run_cmd': 'newruncmd',
},
'source': {
'repository': 'newrepo',
},
}
uri = 'v1/apps/%s' % create_resp.id
resp, body = self.client.patch(
uri, json.dumps(json_update),
headers={'content-type': 'application/json'})
self.assertEqual(200, resp.status)
app_body = json.loads(body)
self.assertEqual('newfakeappname', app_body["name"])
self.assertEqual("newruncmd", app_body["workflow_config"]["run_cmd"])
self.assertEqual("newrepo", app_body["source"]["repository"])
self.client.delete_app(create_resp.id)
time.sleep(2)
self.client.delete_language_pack(lp_name)
def test_app_get(self):
lp_name = self.client.create_lp()
data = apputils.get_sample_data(languagepack=lp_name)
create_resp = self.client.create_app(data=data)
self.assertEqual(201, create_resp.status)
id = create_resp.id
resp, body = self.client.get(
'v1/apps/%s' % id,
headers={'content-type': 'application/json'})
self.assertEqual(200, resp.status)
yaml_data = yaml.safe_load(body)
self._assert_app_data(yaml_data, data)
self.client.delete_app(create_resp.id)
time.sleep(2)
self.client.delete_language_pack(lp_name)
def test_apps_get_all(self):
lp_name = self.client.create_lp()
data = apputils.get_sample_data(languagepack=lp_name)
create_resp = self.client.create_app(data)
self.assertEqual(201, create_resp.status)
resp, body = self.client.get(
'v1/apps', headers={'content-type': 'application/json'})
resp_data = yaml.safe_load(body)
self.assertEqual(200, resp.status)
id = create_resp.id
filtered = [app for app in resp_data if app['id'] == id]
self.assertEqual(filtered[0]['id'], id)
self.client.delete_app(id)
time.sleep(2)
self.client.delete_language_pack(lp_name)
def test_app_delete(self):
lp_name = self.client.create_lp()
data = apputils.get_sample_data(languagepack=lp_name)
create_resp = self.client.create_app(data)
self.assertEqual(201, create_resp.status)
id = create_resp.id
resp, body = self.client.delete_app(id)
self.assertEqual(202, resp.status)
self.assertEqual('', body)
time.sleep(2)
self.client.delete_language_pack(lp_name)
def test_app_delete_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.delete, 'v1/apps/not_found')

View File

@ -0,0 +1,212 @@
#
# Copyright 2013 - Noorul Islam K M
#
# 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 json
from tempest.lib import exceptions as tempest_exceptions
from solum_tempest_plugin import base
sample_data = {"name": "test_assembly",
"description": "A test to create assembly",
"project_id": "project_id",
"user_id": "user_id",
"status": "QUEUED",
"application_uri": "http://localhost:5000"}
plan_sample_data = {"version": "1",
"name": "test_plan",
"description": "A test to create plan",
"project_id": "project_id",
"user_id": "user_id"}
class TestAssemblyController(base.TestCase):
def setUp(self):
super(TestAssemblyController, self).setUp()
def tearDown(self):
super(TestAssemblyController, self).tearDown()
self.client.delete_created_assemblies()
self.client.delete_created_plans()
def _assert_output_expected(self, body_data, data):
self.assertEqual(body_data['description'], data['description'])
self.assertEqual(body_data['plan_uri'], data['plan_uri'])
self.assertEqual(body_data['name'], data['name'])
self.assertIsNotNone(body_data['uuid'])
self.assertEqual(body_data['status'], data['status'])
self.assertEqual(body_data['application_uri'], data['application_uri'])
def test_assemblies_get_all(self):
# Create assemblies to find
p_resp_1 = self.client.create_plan()
self.assertEqual(201, p_resp_1.status)
a_resp_1 = self.client.create_assembly(data=sample_data,
plan_uuid=p_resp_1.uuid)
self.assertEqual(201, a_resp_1.status)
p_resp_2 = self.client.create_plan()
self.assertEqual(201, p_resp_2.status)
a_resp_2 = self.client.create_assembly(data=sample_data,
plan_uuid=p_resp_2.uuid)
self.assertEqual(201, a_resp_2.status)
# Get list of all assemblies
resp, body = self.client.get('v1/assemblies')
self.assertEqual(200, resp.status)
# Search for uuids of created assemblies
assembly_list = json.loads(body)
found_uuid_1 = False
found_uuid_2 = False
for assembly in assembly_list:
uuid = json.dumps(assembly['uuid'])
if a_resp_1.uuid in uuid:
found_uuid_1 = True
elif a_resp_2.uuid in uuid:
found_uuid_2 = True
self.assertTrue(found_uuid_1,
'Cannot find assembly [%s] in list of all assemblies.'
% a_resp_1.uuid)
self.assertTrue(found_uuid_2,
'Cannot find assembly [%s] in list of all assemblies.'
% a_resp_2.uuid)
def test_assemblies_create(self):
plan_resp = self.client.create_plan()
self.assertEqual(201, plan_resp.status)
assembly_resp = self.client.create_assembly(
plan_uuid=plan_resp.uuid,
data=sample_data)
self.assertEqual(201, assembly_resp.status)
sample_data['plan_uri'] = "%s/v1/plans/%s" % (self.client.base_url,
plan_resp.uuid)
self._assert_output_expected(assembly_resp.data, sample_data)
def test_assemblies_create_none(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/assemblies', "{}")
def test_assemblies_get(self):
plan_resp = self.client.create_plan(data=plan_sample_data)
self.assertEqual(201, plan_resp.status)
plan_uuid = plan_resp.uuid
assembly_resp = self.client.create_assembly(
plan_uuid=plan_uuid,
data=sample_data)
self.assertEqual(201, assembly_resp.status)
uuid = assembly_resp.uuid
sample_data['plan_uri'] = "%s/v1/plans/%s" % (self.client.base_url,
plan_uuid)
resp, body = self.client.get('v1/assemblies/%s' % uuid)
self.assertEqual(200, resp.status)
json_data = json.loads(body)
self._assert_output_expected(json_data, sample_data)
# Now check that HTTPS is respected. No new assemblies are created.
for k in ['plan_uri', 'trigger_uri']:
if k in sample_data:
sample_data[k] = sample_data[k].replace('http:', 'https:', 1)
use_https = {'X-Forwarded-Proto': 'https'}
resp, body = self.client.get('v1/assemblies/%s' % uuid,
headers=use_https)
self.assertEqual(200, resp.status)
json_data = json.loads(body)
self._assert_output_expected(json_data, sample_data)
def test_assemblies_get_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.get, 'v1/assemblies/not_found')
def test_assemblies_put(self):
plan_resp = self.client.create_plan()
self.assertEqual(201, plan_resp.status)
plan_uuid = plan_resp.uuid
assembly_resp = self.client.create_assembly(
plan_uuid=plan_uuid,
data=sample_data)
self.assertEqual(201, assembly_resp.status)
uuid = assembly_resp.uuid
uri = "%s/v1/plans/%s" % (self.client.base_url, plan_uuid)
updated_data = {"name": "test_assembly_updated",
"description": "A test to create assembly updated",
"plan_uri": uri,
"user_id": "user_id updated",
"status": "new_status",
"application_uri": "new_uri"}
updated_json = json.dumps(updated_data)
resp, body = self.client.put('v1/assemblies/%s' % uuid, updated_json)
self.assertEqual(200, resp.status)
json_data = json.loads(body)
self._assert_output_expected(json_data, updated_data)
def test_assemblies_put_not_found(self):
updated_data = {"name": "test_assembly_updated",
"description": "A test to create assembly updated",
"plan_uri": 'fake_uri',
"project_id": "project_id updated",
"user_id": "user_id updated",
"status": "new_status",
"application_uri": "new_uri"}
updated_json = json.dumps(updated_data)
self.assertRaises(tempest_exceptions.NotFound,
self.client.put, 'v1/assemblies/not_found',
updated_json)
def test_assemblies_put_none(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.put, 'v1/assemblies/any', "{}")
def test_assemblies_put_cannot_update(self):
plan_resp = self.client.create_plan()
self.assertEqual(201, plan_resp.status)
plan_uuid = plan_resp.uuid
assembly_resp = self.client.create_assembly(
plan_uuid=plan_uuid,
data=sample_data)
self.assertEqual(201, assembly_resp.status)
uuid = assembly_resp.uuid
immutables = [
('id', 'new_assembly_id'),
('uuid', 'new_assembly_uuid'),
('project_id', 'new_project_id'),
]
for key_value in immutables:
updated_data = dict([key_value])
updated_json = json.dumps(updated_data)
self.assertRaises(tempest_exceptions.BadRequest,
self.client.put,
'v1/assemblies/%s' % uuid,
updated_json)
def test_assemblies_delete(self):
plan_resp = self.client.create_plan()
self.assertEqual(201, plan_resp.status)
assembly_resp = self.client.create_assembly(
plan_uuid=plan_resp.uuid,
data=sample_data)
self.assertEqual(201, assembly_resp.status)
uuid = assembly_resp.uuid
resp, body = self.client.delete_assembly(uuid)
self.assertEqual(204, resp.status)
self.assertEqual('', body)
def test_assemblies_delete_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.delete, 'v1/assemblies/not_found')

View File

@ -0,0 +1,153 @@
#
# Copyright 2013 - Noorul Islam K M
#
# 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 json
from tempest.lib import exceptions as tempest_exceptions
from solum_tempest_plugin import base
sample_data = {'name': 'test_component',
'description': 'desc'}
assembly_sample_data = {'name': 'test_assembly',
'description': 'desc assembly'}
plan_sample_data = {'version': '1',
'name': 'test_plan',
'description': 'A test to create plan'}
class TestComponentController(base.TestCase):
def setUp(self):
super(TestComponentController, self).setUp()
def tearDown(self):
super(TestComponentController, self).tearDown()
self.client.delete_created_assemblies()
self.client.delete_created_plans()
def _assert_output_expected(self, body_data, data):
self.assertEqual(body_data['name'], data['name'])
self.assertEqual(body_data['assembly_uuid'], data['assembly_uuid'])
self.assertIsNotNone(body_data['uuid'])
self.assertIsNotNone(body_data['project_id'])
self.assertIsNotNone(body_data['user_id'])
def _delete_component(self, uuid):
resp, body = self.client.delete('v1/components/%s' % uuid)
self.assertEqual(204, resp.status)
def _create_component(self):
plan_resp = self.client.create_plan()
self.assertEqual(201, plan_resp.status,)
assembly_resp = self.client.create_assembly(plan_uuid=plan_resp.uuid)
self.assertEqual(201, assembly_resp.status)
plan_uuid = plan_resp.uuid
assembly_uuid = assembly_resp.uuid
sample_data['assembly_uuid'] = assembly_uuid
data = json.dumps(sample_data)
resp, body = self.client.post('v1/components', data)
self.assertEqual(201, resp.status)
out_data = json.loads(body)
uuid = out_data['uuid']
self.assertIsNotNone(uuid)
return uuid, assembly_uuid, plan_uuid
def test_components_get_all(self):
uuid, assembly_uuid, plan_uuid = self._create_component()
resp, body = self.client.get('v1/components')
data = json.loads(body)
self.assertEqual(200, resp.status)
filtered = [com for com in data if com['uuid'] == uuid]
self.assertEqual(1, len(filtered))
self.assertEqual(uuid, filtered[0]['uuid'])
self._delete_component(uuid)
def test_components_create(self):
plan_resp = self.client.create_plan()
self.assertEqual(201, plan_resp.status)
assembly_resp = self.client.create_assembly(plan_uuid=plan_resp.uuid)
self.assertEqual(201, assembly_resp.status)
plan_uuid = plan_resp.uuid
assembly_uuid = assembly_resp.uuid
sample_data['assembly_uuid'] = assembly_uuid
sample_data['plan_uri'] = "%s/v1/plans/%s" % (self.client.base_url,
plan_uuid)
sample_json = json.dumps(sample_data)
resp, body = self.client.post('v1/components', sample_json)
self.assertEqual(201, resp.status)
json_data = json.loads(body)
self._assert_output_expected(json_data, sample_data)
self._delete_component(json_data['uuid'])
def test_components_create_none(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/components', "{}")
def test_components_get(self):
uuid, assembly_uuid, plan_uuid = self._create_component()
sample_data['plan_uri'] = "%s/v1/plans/%s" % (self.client.base_url,
plan_uuid)
resp, body = self.client.get('v1/components/%s' % uuid)
self.assertEqual(200, resp.status)
json_data = json.loads(body)
self._assert_output_expected(json_data, sample_data)
self._delete_component(uuid)
def test_components_get_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.get, 'v1/components/not_found')
def test_components_put(self):
uuid, assembly_uuid, plan_uuid = self._create_component()
updated_data = {'name': 'test_service_updated',
'description': 'desc updated',
'plan_uri': "%s/v1/plans/%s" % (self.client.base_url,
plan_uuid),
'assembly_uuid': assembly_uuid}
updated_json = json.dumps(updated_data)
resp, body = self.client.put('v1/components/%s' % uuid, updated_json)
self.assertEqual(200, resp.status)
json_data = json.loads(body)
self._assert_output_expected(json_data, updated_data)
self._delete_component(uuid)
def test_components_put_not_found(self):
updated_data = {'name': 'test_service_updated',
'description': 'desc updated',
'plan_uri': "%s/v1/plans/%s" % (self.client.base_url,
'not_found'),
'assembly_uuid': 'not_found'}
updated_json = json.dumps(updated_data)
self.assertRaises(tempest_exceptions.NotFound,
self.client.put, 'v1/components/not_found',
updated_json)
def test_components_put_none(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.put, 'v1/components/any', "{}")
def test_components_delete(self):
uuid, assembly_uuid, plan_uuid = self._create_component()
resp, body = self.client.delete('v1/components/%s' % uuid)
self.assertEqual(204, resp.status)
self.assertEqual('', body)
def test_components_delete_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.delete, 'v1/components/not_found')

View File

@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
#
# Copyright 2013 - Noorul Islam K M
#
# 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
@ -12,17 +13,15 @@
# License for the specific language governing permissions and limitations
# under the License.
"""
test_solum_tempest_plugin
----------------------------------
import json
Tests for `solum_tempest_plugin` module.
"""
from solum_tempest_plugin.tests import base
from solum_tempest_plugin import base
class TestSolum_tempest_plugin(base.TestCase):
class TestExtensionController(base.TestCase):
def test_something(self):
pass
def test_extensions_get_all(self):
resp, body = self.client.get('v1/extensions')
data = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual([], data)

View File

@ -0,0 +1,143 @@
# Copyright 2014 - Rackspace
#
# 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 json
import random
import string
import time
from tempest.lib import exceptions as tempest_exceptions
from solum_tempest_plugin import base
from solum_tempest_plugin.common import apputils
sample_plan = {"version": "1",
"name": "test_plan",
"artifacts": [{
"name": "No deus",
"language_pack": "language_pack_name"
}]}
class TestLanguagePackController(base.TestCase):
def _get_sample_languagepack(self):
sample_lp = dict()
s = string.lowercase
sample_lp["name"] = "lp" + ''.join(random.sample(s, 5))
lp_url = "https://github.com/murali44/Solum-lp-Go.git"
sample_lp["source_uri"] = lp_url
return sample_lp
def setUp(self):
super(TestLanguagePackController, self).setUp()
def _delete_all(self):
resp, body = self.client.get('v1/language_packs')
data = json.loads(body)
self.assertEqual(200, resp.status)
[self._delete_language_pack(pl['uuid']) for pl in data]
def _delete_language_pack(self, uuid):
resp, _ = self.client.delete('v1/language_packs/%s' % uuid)
self.assertEqual(204, resp.status)
def _create_language_pack(self):
sample_lp = self._get_sample_languagepack()
jsondata = json.dumps(sample_lp)
resp, body = self.client.post('v1/language_packs', jsondata)
self.assertEqual(201, resp.status)
out_data = json.loads(body)
uuid = out_data['uuid']
self.assertIsNotNone(uuid)
return uuid, sample_lp
def test_language_packs_get_all(self):
uuid, sample_lp = self._create_language_pack()
resp, body = self.client.get('v1/language_packs')
data = json.loads(body)
self.assertEqual(200, resp.status)
filtered = [pl for pl in data if pl['uuid'] == uuid]
self.assertEqual(uuid, filtered[0]['uuid'])
self._delete_language_pack(uuid)
def test_language_packs_create(self):
sample_lp = self._get_sample_languagepack()
sample_json = json.dumps(sample_lp)
resp, body = self.client.post('v1/language_packs', sample_json)
self.assertEqual(201, resp.status)
json_data = json.loads(body)
self.assertEqual("QUEUED", json_data["status"])
self.assertEqual(sample_lp['name'], json_data["name"])
self._delete_language_pack(json_data["uuid"])
def test_language_packs_create_none(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/language_packs', "{}")
def test_language_packs_get(self):
uuid, sample_lp = self._create_language_pack()
resp, body = self.client.get('v1/language_packs/%s' % uuid)
self.assertEqual(200, resp.status)
json_data = json.loads(body)
self.assertEqual(sample_lp['source_uri'], json_data['source_uri'])
self.assertEqual(sample_lp['name'], json_data['name'])
self._delete_language_pack(uuid)
def test_language_packs_get_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.get, 'v1/language_packs/not_found')
def test_language_packs_delete(self):
uuid, sample_lp = self._create_language_pack()
resp, body = self.client.delete('v1/language_packs/%s' % uuid)
self.assertEqual(204, resp.status)
self.assertEqual('', body)
def test_language_packs_delete_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.delete, 'v1/language_packs/not_found')
def test_language_packs_delete_used_by_plan(self):
uuid, sample_lp = self._create_language_pack()
artifacts = sample_plan['artifacts']
artifacts[0]['language_pack'] = sample_lp['name']
sample_plan['artifacts'] = artifacts
resp = self.client.create_plan(data=sample_plan)
self.assertRaises(tempest_exceptions.Conflict,
self.client.delete, 'v1/language_packs/%s' % uuid)
self.client.delete_plan(resp.uuid)
# Sleep for a few seconds to make sure plans are deleted.
time.sleep(5)
self._delete_language_pack(uuid)
def test_language_packs_delete_used_by_app(self):
uuid, sample_lp = self._create_language_pack()
sample_app = apputils.get_sample_data()
sample_app["languagepack"] = sample_lp["name"]
resp = self.client.create_app(data=sample_app)
self.assertRaises(tempest_exceptions.Conflict,
self.client.delete, 'v1/language_packs/%s' % uuid)
bdy = json.loads(resp.body)
self.client.delete_app(bdy["id"])
# Sleep for a few seconds to make sure plans are deleted.
time.sleep(5)
self._delete_language_pack(uuid)

View File

@ -0,0 +1,27 @@
#
# Copyright 2013 - Noorul Islam K M
#
# 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 json
from solum_tempest_plugin import base
class TestOperationController(base.TestCase):
def test_operations_get_all(self):
resp, body = self.client.get('v1/operations')
data = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual([], data)

View File

@ -0,0 +1,232 @@
#
# Copyright 2013 - Rackspace US, 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.
from tempest.lib import exceptions as tempest_exceptions
import yaml
from solum_tempest_plugin import base
sample_data = {"version": "1",
"name": "test_plan",
"description": "A test to create plan",
"artifacts": [{
"name": "No deus",
"artifact_type": "heroku",
"content": {
"href": "https://example.com/git/a.git",
"private": False,
},
"language_pack": "auto",
"ports": 123
}]}
bad_data = {"version": "1",
"name": "test_plan",
"description": "A test to create plan",
"artifacts": [{
"name": "No deus",
"artifact_type": "heroku",
"content": {
"href": "https://example.com/git/a.git",
"private": False,
},
"language_pack": "auto",
"ports": -1
}]}
sample_data_private = {"version": "1",
"name": "test_plan",
"description": "A test to create plan",
"artifacts": [{
"name": "No deus",
"artifact_type": "heroku",
"content": {
"href": "https://example.com/git/a.git",
"private": True,
},
"language_pack": "auto",
}]}
class TestPlanController(base.TestCase):
def setUp(self):
super(TestPlanController, self).setUp()
def tearDown(self):
super(TestPlanController, self).tearDown()
self.client.delete_created_plans()
def _delete_all(self):
resp, body = self.client.get(
'v1/plans', headers={'accept-type': 'application/x-yaml'})
data = yaml.safe_load(body)
self.assertEqual(200, resp.status)
[self._delete_plan(pl['uuid']) for pl in data]
def _assert_output_expected(self, body_data, data):
self.assertEqual(body_data['description'], data['description'])
self.assertEqual(body_data['name'], data['name'])
if body_data['artifacts']:
self.assertEqual(body_data['artifacts'][0]['content']['href'],
data['artifacts'][0]['content']['href'])
self.assertIsNotNone(body_data['uuid'])
def test_plans_get_all(self):
create_resp = self.client.create_plan()
self.assertEqual(201, create_resp.status)
resp, body = self.client.get(
'v1/plans', headers={'content-type': 'application/x-yaml'})
data = yaml.safe_load(body)
self.assertEqual(200, resp.status)
uuid = create_resp.uuid
filtered = [pl for pl in data if pl['uuid'] == uuid]
self.assertEqual(uuid, filtered[0]['uuid'])
def test_plans_create(self):
resp = self.client.create_plan(data=sample_data)
self.assertEqual(201, resp.status)
self._assert_output_expected(resp.data, sample_data)
def test_plans_create_bad_port_data(self):
try:
self.client.create_plan(data=bad_data)
except tempest_exceptions.BadRequest:
self.assertTrue(True)
def test_plans_create_with_private_github_repo(self):
resp = self.client.create_plan(data=sample_data_private)
self.assertEqual(201, resp.status)
self._assert_output_expected(resp.data, sample_data)
def test_plans_create_empty_yaml(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/plans', '{}',
headers={'content-type': 'application/x-yaml'})
def test_plans_create_invalid_yaml_type(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/plans', 'invalid type',
headers={'content-type': 'application/x-yaml'})
def test_plans_create_invalid_yaml_syntax(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/plans', "}invalid: y'm'l3!",
headers={'content-type': 'application/x-yaml'})
def test_plans_get(self):
create_resp = self.client.create_plan(data=sample_data)
self.assertEqual(201, create_resp.status)
uuid = create_resp.uuid
resp, body = self.client.get(
'v1/plans/%s' % uuid,
headers={'content-type': 'application/x-yaml'})
self.assertEqual(200, resp.status, )
yaml_data = yaml.safe_load(body)
self._assert_output_expected(yaml_data, sample_data)
def test_plans_get_with_private_github_repo(self):
create_resp = self.client.create_plan(data=sample_data_private)
self.assertEqual(201, create_resp.status)
uuid = create_resp.uuid
resp, body = self.client.get(
'v1/plans/%s' % uuid,
headers={'content-type': 'application/x-yaml'})
self.assertEqual(200, resp.status)
yaml_data = yaml.safe_load(body)
public_key = yaml_data['artifacts'][0]['content']['public_key']
self.assertIsNotNone(public_key)
self._assert_output_expected(yaml_data, sample_data)
def test_plans_get_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.get, 'v1/plans/not_found',
headers={'content-type': 'application/x-yaml'})
def test_plans_put(self):
create_resp = self.client.create_plan()
self.assertEqual(201, create_resp.status)
uuid = create_resp.uuid
updated_data = {"version": "1",
"name": "test_plan_updated",
"description": "A test to create plan updated",
"type": "plan",
"artifacts": []}
updated_yaml = yaml.dump(updated_data)
resp, body = self.client.put(
'v1/plans/%s' % uuid, updated_yaml,
headers={'content-type': 'application/x-yaml'})
self.assertEqual(200, resp.status)
yaml_data = yaml.safe_load(body)
self._assert_output_expected(yaml_data, updated_data)
def test_plans_put_not_found(self):
updated_data = {"name": "test_plan updated",
"description": "A test to create plan updated",
"type": "plan",
"artifacts": []}
updated_yaml = yaml.dump(updated_data)
self.assertRaises(tempest_exceptions.NotFound,
self.client.put, 'v1/plans/not_found', updated_yaml,
headers={'content-type': 'application/x-yaml'})
def test_plans_put_empty_yaml(self):
create_resp = self.client.create_plan()
self.assertEqual(201, create_resp.status)
# get the URI of the newly created plan
uri = (create_resp.data['uri']
[len(self.client.base_url) + 1:])
self.assertRaises(tempest_exceptions.BadRequest,
self.client.put, uri, '{}',
headers={'content-type': 'application/x-yaml'})
def test_plans_put_invalid_yaml_type(self):
create_resp = self.client.create_plan()
self.assertEqual(201, create_resp.status)
# get the URI of the newly created plan
uri = (create_resp.data['uri']
[len(self.client.base_url) + 1:])
self.assertRaises(tempest_exceptions.BadRequest,
self.client.put, uri, 'invalid type',
headers={'content-type': 'application/x-yaml'})
def test_plans_put_invalid_yaml_syntax(self):
create_resp = self.client.create_plan()
self.assertEqual(201, create_resp.status)
# get the URI of the newly created plan
uri = (create_resp.data['uri']
[len(self.client.base_url) + 1:])
self.assertRaises(tempest_exceptions.BadRequest,
self.client.put, uri, "}invalid: y'm'l3!",
headers={'content-type': 'application/x-yaml'})
def test_plans_delete(self):
create_resp = self.client.create_plan()
self.assertEqual(201, create_resp.status)
uuid = create_resp.uuid
resp, body = self.client.delete_plan(uuid)
self.assertEqual(202, resp.status)
self.assertEqual('', body)
def test_plans_delete_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.delete, 'v1/plans/not_found')

View File

@ -0,0 +1,77 @@
#
# Copyright 2013 - Noorul Islam K M
#
# 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 json
from solum import version
from solum_tempest_plugin import base
class TestRootController(base.TestCase):
def test_index(self):
resp, body = self.client.request_without_auth('', 'GET')
self.assertEqual(200, resp.status)
data = json.loads(body)
self.assertEqual(data[0]['id'], 'v1.0')
self.assertEqual(data[0]['status'], 'CURRENT')
self.assertEqual(data[0]['link'],
{'href': '%s/v1' % self.client.base_url,
'target_name': 'v1'})
def test_platform(self):
resp, body = self.client.request_without_auth('v1', 'GET')
self.assertEqual(200, resp.status)
data = json.loads(body)
self.assertEqual(data['uri'], '%s/v1' % self.client.base_url)
self.assertEqual(data['type'], 'platform')
self.assertEqual(data['name'], 'solum')
self.assertEqual(data['description'], 'solum native implementation')
self.assertEqual(data['implementation_version'],
version.version_string())
self.assertEqual(data['plans_uri'],
'%s/v1/plans' % self.client.base_url)
self.assertEqual(data['assemblies_uri'],
'%s/v1/assemblies' % self.client.base_url)
self.assertEqual(data['services_uri'],
'%s/v1/services' % self.client.base_url)
self.assertEqual(data['components_uri'],
'%s/v1/components' % self.client.base_url)
self.assertEqual(data['extensions_uri'],
'%s/v1/extensions' % self.client.base_url)
self.assertEqual(data['operations_uri'],
'%s/v1/operations' % self.client.base_url)
self.assertEqual(data['sensors_uri'],
'%s/v1/sensors' % self.client.base_url)
self.assertEqual(data['language_packs_uri'],
'%s/v1/language_packs' % self.client.base_url)
self.assertEqual(data['pipelines_uri'],
'%s/v1/pipelines' % self.client.base_url)
self.assertEqual(data['triggers_uri'],
'%s/v1/triggers' % self.client.base_url)
self.assertEqual(data['infrastructure_uri'],
'%s/v1/infrastructure' % self.client.base_url)
def test_request_without_auth(self):
resp, body = self.client.request_without_auth('v1', 'GET')
self.assertEqual(200, resp.status)
resp, body = self.client.get('v1')
self.assertEqual(200, resp.status)
resp, body = self.client.request_without_auth(
'v1/plans', 'GET', headers={'content-type': 'application/x-yaml'})
self.assertEqual(401, resp.status)
resp, body = self.client.get(
'v1/plans', headers={'content-type': 'application/x-yaml'})
self.assertEqual(200, resp.status)

View File

@ -1,7 +1,5 @@
# -*- coding: utf-8 -*-
# Copyright 2010-2011 OpenStack Foundation
# Copyright (c) 2013 Hewlett-Packard Development Company, L.P.
#
# Copyright 2013 - Noorul Islam K M
#
# 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
@ -15,9 +13,15 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslotest import base
import json
from solum_tempest_plugin import base
class TestCase(base.BaseTestCase):
class TestSensorController(base.TestCase):
"""Test case base class for all unit tests."""
def test_sensors_get_all(self):
resp, body = self.client.get('v1/sensors')
data = json.loads(body)
self.assertEqual(200, resp.status)
self.assertEqual([], data)

View File

@ -0,0 +1,132 @@
#
# Copyright 2013 - Noorul Islam K M
#
# 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 json
from tempest.lib import exceptions as tempest_exceptions
from solum_tempest_plugin import base
sample_data = {"name": "test_service",
"description": "A test to create service",
"project_id": "project_id",
"user_id": "user_id",
"service_type": "mysql",
"read_only": True,
"type": "service"}
class TestServiceController(base.TestCase):
def setUp(self):
super(TestServiceController, self).setUp()
self.addCleanup(self._delete_all)
def _delete_all(self):
resp, body = self.client.get('v1/services')
data = json.loads(body)
self.assertEqual(resp.status, 200)
[self._delete_service(ser['uuid']) for ser in data]
def _assert_output_expected(self, body_data, data):
self.assertEqual(body_data['description'], data['description'])
self.assertEqual(body_data['name'], data['name'])
self.assertEqual(body_data['service_type'], data['service_type'])
self.assertEqual(body_data['read_only'], data['read_only'])
self.assertEqual(body_data['type'], 'service')
self.assertIsNotNone(body_data['uuid'])
def _delete_service(self, uuid):
resp, _ = self.client.delete('v1/services/%s' % uuid)
self.assertEqual(resp.status, 204)
def _create_service(self):
jsondata = json.dumps(sample_data)
resp, body = self.client.post('v1/services', jsondata)
self.assertEqual(resp.status, 201)
out_data = json.loads(body)
uuid = out_data['uuid']
self.assertIsNotNone(uuid)
return uuid
def test_services_get_all(self):
uuid = self._create_service()
resp, body = self.client.get('v1/services')
data = json.loads(body)
self.assertEqual(resp.status, 200)
filtered = [ser for ser in data if ser['uuid'] == uuid]
self.assertEqual(filtered[0]['uuid'], uuid)
def test_services_create(self):
sample_json = json.dumps(sample_data)
resp, body = self.client.post('v1/services', sample_json)
self.assertEqual(resp.status, 201)
json_data = json.loads(body)
self._assert_output_expected(json_data, sample_data)
self._delete_service(json_data['uuid'])
def test_services_create_none(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.post, 'v1/services', "{}")
def test_services_get(self):
uuid = self._create_service()
resp, body = self.client.get('v1/services/%s' % uuid)
self.assertEqual(resp.status, 200)
json_data = json.loads(body)
self._assert_output_expected(json_data, sample_data)
self._delete_service(uuid)
def test_services_get_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.get, 'v1/services/not_found')
def test_services_put(self):
uuid = self._create_service()
updated_data = {"name": "test_service_updated",
"description": "A test to create service updated",
"user_id": "user_id updated",
"service_type": "mysql updated",
"read_only": False}
updated_json = json.dumps(updated_data)
resp, body = self.client.put('v1/services/%s' % uuid, updated_json)
self.assertEqual(resp.status, 200)
json_data = json.loads(body)
self._assert_output_expected(json_data, updated_data)
self._delete_service(uuid)
def test_services_put_not_found(self):
updated_data = {"name": "test_service_updated",
"description": "A test to create service updated",
"user_id": "user_id updated",
"service_type": "mysql updated",
"read_only": False}
updated_json = json.dumps(updated_data)
self.assertRaises(tempest_exceptions.NotFound,
self.client.put, 'v1/services/not_found',
updated_json)
def test_services_put_none(self):
self.assertRaises(tempest_exceptions.BadRequest,
self.client.put, 'v1/services/any', "{}")
def test_services_delete(self):
uuid = self._create_service()
resp, body = self.client.delete('v1/services/%s' % uuid)
self.assertEqual(resp.status, 204)
self.assertEqual(body, '')
def test_services_delete_not_found(self):
self.assertRaises(tempest_exceptions.NotFound,
self.client.delete, 'v1/services/not_found')