# Copyright (c) 2014 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 json import os import requests import uuid from tempest import clients from tempest.common import rest_client from tempest import config from tempest import exceptions import tempest.test CONF = config.CONF class MuranoClient(rest_client.RestClient): def __init__(self, auth_provider): super(MuranoClient, self).__init__(auth_provider) self.service = 'application_catalog' self.endpoint_url = 'publicURL' def get_environments_list(self): resp, body = self.get('v1/environments') return resp, json.loads(body) def create_environment(self, name): post_body = '{"name": "%s"}' % name resp, body = self.post('v1/environments', post_body) return resp, json.loads(body) def delete_environment(self, environment_id): return self.delete('v1/environments/{0}'.format(environment_id)) def update_environment(self, environment_id): post_body = '{"name": "%s"}' % ("changed-environment-name") resp, body = self.put('v1/environments/{0}'.format(environment_id), post_body) return resp, json.loads(body) def get_environment(self, environment_id): resp, body = self.get('v1/environments/{0}'.format(environment_id)) return resp, json.loads(body) def create_session(self, environment_id): post_body = None resp, body = self.post( 'v1/environments/{0}/configure'.format(environment_id), post_body ) return resp, json.loads(body) def delete_session(self, environment_id, session_id): return self.delete( 'v1/environments/{0}/sessions/{1}'.format(environment_id, session_id)) def get_session(self, environment_id, session_id): resp, body = self.get( 'v1/environments/{0}/sessions/{1}'.format(environment_id, session_id)) return resp, json.loads(body) def create_service(self, environment_id, session_id, post_body): post_body = json.dumps(post_body) headers = self.get_headers() headers.update( {'X-Configuration-Session': session_id} ) resp, body = self.post( 'v1/environments/{0}/services'.format(environment_id), post_body, headers ) return resp, json.loads(body) def delete_service(self, environment_id, session_id, service_id): headers = self.get_headers() headers.update( {'X-Configuration-Session': session_id} ) return self.delete( 'v1/environments/{0}/services/{1}'.format(environment_id, service_id), headers ) def get_services_list(self, environment_id, session_id): headers = self.get_headers() headers.update( {'X-Configuration-Session': session_id} ) resp, body = self.get( 'v1/environments/{0}/services'.format(environment_id), headers ) return resp, json.loads(body) def get_service(self, environment_id, session_id, service_id): headers = self.get_headers() headers.update( {'X-Configuration-Session': session_id} ) resp, body = self.get( 'v1/environments/{0}/services/{1}'.format(environment_id, service_id), headers ) return resp, json.loads(body) def get_list_packages(self): resp, body = self.get('v1/catalog/packages') return resp, json.loads(body) def get_package(self, id): resp, body = self.get('v1/catalog/packages/{0}'.format(id)) return resp, json.loads(body) def upload_package(self, package_name, body): __location__ = os.path.realpath(os.path.join( os.getcwd(), os.path.dirname(__file__))) headers = {'X-Auth-Token': self.auth_provider.get_token()} files = {'%s' % package_name: open( os.path.join(__location__, 'v1/DummyTestApp.zip'), 'rb')} post_body = {'JsonString': json.dumps(body)} request_url = '{endpoint}{url}'.format(endpoint=self.base_url, url='/v1/catalog/packages') resp = requests.post(request_url, files=files, data=post_body, headers=headers) return resp def update_package(self, id, post_body): headers = { 'X-Auth-Token': self.auth_provider.get_token(), 'content-type': 'application/murano-packages-json-patch' } resp, body = self.patch('v1/catalog/packages/{0}'.format(id), json.dumps(post_body), headers=headers) return resp, json.loads(body) def delete_package(self, id): return self.delete('v1/catalog/packages/{0}'.format(id)) def download_package(self, id): return self.get('v1/catalog/packages/{0}/download'.format(id)) def get_ui_definition(self, id): return self.get('v1/catalog/packages/{0}/ui'.format(id)) def get_logo(self, id): return self.get('v1/catalog/packages/{0}/logo'.format(id)) def list_categories(self): resp, body = self.get('v1/catalog/packages/categories') return resp, json.loads(body) class TestCase(tempest.test.BaseTestCase): @classmethod def setUpClass(cls): super(TestCase, cls).setUpClass() # If no credentials are provided, the Manager will use those # in CONF.identity and generate an auth_provider from them mgr = clients.Manager() cls.client = MuranoClient(mgr.auth_provider) def setUp(self): super(TestCase, self).setUp() self.environments = [] def tearDown(self): super(TestCase, self).tearDown() for environment in self.environments: try: self.client.delete_environment(environment['id']) except exceptions.NotFound: pass def create_environment(self, name): environment = self.client.create_environment(name)[1] self.environments.append(environment) return environment def create_demo_service(self, environment_id, session_id): post_body = { "?": { "id": uuid.uuid4().hex, "type": "io.murano.tests.demoService" }, "availabilityZone": "nova", "name": "demo", "unitNamingPattern": "host", "osImage": { "type": "cirros.demo", "name": "demo", "title": "Demo" }, "units": [{}], "flavor": "m1.small", "configuration": "standalone" } return self.client.create_service(environment_id, session_id, post_body) class NegativeTestCase(TestCase): @classmethod def setUpClass(cls): super(NegativeTestCase, cls).setUpClass() # If no credentials are provided, the Manager will use those # in CONF.identity and generate an auth_provider from them mgr = clients.Manager() cls.alt_client = MuranoClient(mgr.auth_provider)