Source code for glare.tests.functional.base

# Copyright (c) 2016 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.


from oslo_serialization import jsonutils
from oslo_utils import uuidutils
import requests

from glare.tests import functional


[docs]def sort_results(lst, target='name'): return sorted(lst, key=lambda x: x[target])
[docs]class TestArtifact(functional.FunctionalTest): enabled_types = (u'sample_artifact', u'images', u'heat_templates', u'heat_environments', u'tosca_templates', u'murano_packages', u'all') users = { 'user1': { 'id': uuidutils.generate_uuid(), 'tenant_id': uuidutils.generate_uuid(), 'token': uuidutils.generate_uuid(), 'role': 'member' }, 'user2': { 'id': uuidutils.generate_uuid(), 'tenant_id': uuidutils.generate_uuid(), 'token': uuidutils.generate_uuid(), 'role': 'member' }, 'admin': { 'id': uuidutils.generate_uuid(), 'tenant_id': uuidutils.generate_uuid(), 'token': uuidutils.generate_uuid(), 'role': 'admin' }, 'anonymous': { 'id': None, 'tenant_id': None, 'token': None, 'role': None } }
[docs] def setUp(self): super(TestArtifact, self).setUp() self.set_user('user1') self.glare_server.deployment_flavor = 'noauth' self.glare_server.enabled_artifact_types = ','.join( self.enabled_types) self.glare_server.custom_artifact_types_modules = ( 'glare.tests.sample_artifact') self.start_servers(**self.__dict__.copy())
[docs] def tearDown(self): self.stop_servers() self._reset_database(self.glare_server.sql_connection) super(TestArtifact, self).tearDown()
def _url(self, path): if path.startswith('/schemas') or \ path.startswith('/quotas') or \ path.startswith('/project-quotas'): return 'http://127.0.0.1:%d%s' % (self.glare_port, path) else: return 'http://127.0.0.1:%d/artifacts%s' % (self.glare_port, path)
[docs] def set_user(self, username): if username not in self.users: raise KeyError self.current_user = username
def _headers(self, custom_headers=None): base_headers = { 'X-Identity-Status': 'Confirmed', 'X-Auth-Token': self.users[self.current_user]['token'], 'X-User-Id': self.users[self.current_user]['id'], 'X-Tenant-Id': self.users[self.current_user]['tenant_id'], 'X-Project-Id': self.users[self.current_user]['tenant_id'], 'X-Roles': self.users[self.current_user]['role'], } base_headers.update(custom_headers or {}) return base_headers
[docs] def create_artifact(self, data=None, status=201, type_name='sample_artifact'): return self.post('/' + type_name, data or {}, status=status)
def _check_artifact_method(self, method, url, data=None, status=200, headers=None): if not headers: headers = self._headers() else: headers = self._headers(headers) headers.setdefault("Content-Type", "application/json") if 'application/json' in headers['Content-Type'] and data is not None: data = jsonutils.dumps(data) response = getattr(requests, method)(self._url(url), headers=headers, data=data) self.assertEqual(status, response.status_code, response.text) if status >= 400: return response.text if ("application/json" in response.headers["content-type"] or "application/schema+json" in response.headers["content-type"]): return jsonutils.loads(response.text) return response.text
[docs] def post(self, url, data=None, status=201, headers=None): return self._check_artifact_method("post", url, data, status=status, headers=headers)
[docs] def get(self, url, status=200, headers=None): return self._check_artifact_method("get", url, status=status, headers=headers)
[docs] def delete(self, url, status=204, headers=None): return self._check_artifact_method("delete", url, status=status, headers=headers)
[docs] def patch(self, url, data, status=200, headers=None): if headers is None: headers = {} if 'Content-Type' not in headers: headers.update({'Content-Type': 'application/json-patch+json'}) return self._check_artifact_method("patch", url, data, status=status, headers=headers)
[docs] def put(self, url, data=None, status=200, headers=None): return self._check_artifact_method("put", url, data, status=status, headers=headers)
# the test cases below are written in accordance with use cases # each test tries to cover separate use case in Glare # all code inside each test tries to cover all operators and data # involved in use case execution # each tests represents part of artifact lifecycle # so we can easily define where is the failed code make_active = [{"op": "replace", "path": "/status", "value": "active"}] make_deactivated = [{"op": "replace", "path": "/status", "value": "deactivated"}] make_public = [{"op": "replace", "path": "/visibility", "value": "public"}]
[docs] def admin_action(self, artifact_id, body, status=200, type_name='sample_artifact'): cur_user = self.current_user self.set_user('admin') url = '/%s/%s' % (type_name, artifact_id) af = self.patch(url=url, data=body, status=status) self.set_user(cur_user) return af