diff options
Diffstat (limited to 'cvp/opnfv_testapi/resources')
19 files changed, 0 insertions, 2655 deletions
diff --git a/cvp/opnfv_testapi/resources/__init__.py b/cvp/opnfv_testapi/resources/__init__.py deleted file mode 100644 index 05c0c939..00000000 --- a/cvp/opnfv_testapi/resources/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## diff --git a/cvp/opnfv_testapi/resources/application_handlers.py b/cvp/opnfv_testapi/resources/application_handlers.py deleted file mode 100644 index 258c1aa2..00000000 --- a/cvp/opnfv_testapi/resources/application_handlers.py +++ /dev/null @@ -1,233 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -import logging -import json - -from tornado import web -from tornado import gen -from bson import objectid - -from opnfv_testapi.common.config import CONF -from opnfv_testapi.common import utils -from opnfv_testapi.resources import handlers -from opnfv_testapi.resources import application_models -from opnfv_testapi.tornado_swagger import swagger -from opnfv_testapi.ui.auth import constants as auth_const - - -class GenericApplicationHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericApplicationHandler, self).__init__(application, - request, - **kwargs) - self.table = "applications" - self.table_cls = application_models.Application - - -class ApplicationsLogoHandler(GenericApplicationHandler): - @web.asynchronous - @gen.coroutine - def post(self): - role = self.get_secure_cookie(auth_const.ROLE) - if role.find('administrator') == -1: - msg = 'Only administrator is allowed to upload logos' - self.finish_request({'code': '-1', 'msg': msg}) - return - - fileinfo = self.request.files['file'][0] - fname = fileinfo['filename'] - location = '3rd_party/static/testapi-ui/assets/img/' - fh = open(location + fname, 'w') - fh.write(fileinfo['body']) - msg = 'Successfully uploaded logo: ' + fname - resp = {'code': '1', 'msg': msg} - self.finish_request(resp) - - -class ApplicationsGetLogoHandler(GenericApplicationHandler): - def get(self, filename): - location = '3rd_party/static/testapi-ui/assets/img/' + filename - self.set_header('Content-Type', 'application/force-download') - self.set_header('Content-Disposition', - 'attachment; filename=%s' % filename) - try: - with open(location, "rb") as f: - try: - while True: - _buffer = f.read(4096) - if _buffer: - self.write(_buffer) - else: - f.close() - self.finish() - return - except Exception: - raise web.HTTPError(404) - except Exception: - raise web.HTTPError(500) - - -class ApplicationsCLHandler(GenericApplicationHandler): - @swagger.operation(nickname="queryApplications") - @web.asynchronous - @gen.coroutine - def get(self): - """ - @description: Retrieve result(s) for a application project - on a specific pod. - @notes: Retrieve application. - Available filters for this request are : - - id : Application id - - period : x last days, incompatible with from/to - - from : starting time in 2016-01-01 or 2016-01-01 00:01:23 - - to : ending time in 2016-01-01 or 2016-01-01 00:01:23 - - signed : get logined user result - - @return 200: all application results consist with query, - empty list if no result is found - @rtype: L{Applications} - """ - def descend_limit(): - descend = self.get_query_argument('descend', 'true') - return -1 if descend.lower() == 'true' else 1 - - def last_limit(): - return self.get_int('last', self.get_query_argument('last', 0)) - - def page_limit(): - return self.get_int('page', self.get_query_argument('page', 0)) - - limitations = { - 'sort': {'_id': descend_limit()}, - 'last': last_limit(), - 'page': page_limit(), - 'per_page': CONF.api_results_per_page - } - - query = yield self.set_query() - yield self._list(query=query, **limitations) - logging.debug('list end') - - @swagger.operation(nickname="createApplication") - @web.asynchronous - def post(self): - """ - @description: create a application - @param body: application to be created - @type body: L{ApplicationCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: application is created. - @raise 404: pod/project/applicationcase not exist - @raise 400: body/pod_name/project_name/case_name not provided - """ - openid = self.get_secure_cookie(auth_const.OPENID) - if openid: - self.json_args['owner'] = openid - - self._post() - - @gen.coroutine - def _post(self): - miss_fields = [] - carriers = [] - - role = self.get_secure_cookie(auth_const.ROLE) - if role.find('administrator') == -1: - self.finish_request({'code': '403', 'msg': 'Only administrator \ - is allowed to submit application.'}) - return - - query = {"openid": self.json_args['user_id']} - table = "users" - ret, msg = yield self._check_if_exists(table=table, query=query) - logging.debug('ret:%s', ret) - if not ret: - self.finish_request({'code': '403', 'msg': msg}) - return - self._create(miss_fields=miss_fields, carriers=carriers) - - self._send_email() - - def _send_email(self): - - data = self.table_cls.from_dict(self.json_args) - subject = "[OPNFV CVP]New OPNFV CVP Application Submission" - content = """Hi CVP Reviewer, - -This is a new application: - - Organization Name: {}, - Organization Website: {}, - Product Name: {}, - Product Specifications: {}, - Product Documentation: {}, - Product Categories: {}, - Primary Name: {}, - Primary Email: {}, - Primary Address: {}, - Primary Phone: {}, - User ID Type: {}, - User ID: {} - -Best Regards, -CVP Team - """.format(data.organization_name, - data.organization_web, - data.product_name, - data.product_spec, - data.product_documentation, - data.product_categories, - data.prim_name, - data.prim_email, - data.prim_address, - data.prim_phone, - data.id_type, - data.user_id) - - utils.send_email(subject, content) - - -class ApplicationsGURHandler(GenericApplicationHandler): - @swagger.operation(nickname="deleteAppById") - def delete(self, id): - query = {'_id': objectid.ObjectId(id)} - self._delete(query=query) - - @swagger.operation(nickname="updateApplicationById") - def put(self, application_id): - """ - @description: update a single application by id - @param body: fields to be updated - @type body: L{ApplicationUpdateRequest} - @in body: body - @rtype: L{Application} - @return 200: update success - @raise 404: Application not exist - @raise 403: nothing to update - """ - data = json.loads(self.request.body) - item = data.get('item') - value = data.get(item) - logging.debug('%s:%s', item, value) - try: - self.update(application_id, item, value) - except Exception as e: - logging.error('except:%s', e) - return - - @web.asynchronous - @gen.coroutine - def update(self, application_id, item, value): - self.json_args = {} - self.json_args[item] = value - query = {'_id': application_id, 'owner': - self.get_secure_cookie(auth_const.OPENID)} - db_keys = ['_id', 'owner'] - self._update(query=query, db_keys=db_keys) diff --git a/cvp/opnfv_testapi/resources/application_models.py b/cvp/opnfv_testapi/resources/application_models.py deleted file mode 100644 index e2bb652d..00000000 --- a/cvp/opnfv_testapi/resources/application_models.py +++ /dev/null @@ -1,39 +0,0 @@ -############################################################################## -# Copyright (c) 2015 -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - -from datetime import datetime - - -@swagger.model() -class Application(models.ModelBase): - """ - @property trust_indicator: used for long duration test case - @ptype trust_indicator: L{TI} - """ - def __init__(self, _id=None, owner=None, status="created", - creation_date=[], trust_indicator=None): - self._id = _id - self.owner = owner - self.creation_date = datetime.now() - self.status = status - - -@swagger.model() -class Applications(models.ModelBase): - """ - @property applications: - @ptype tests: C{list} of L{Application} - """ - def __init__(self): - self.applications = list() - - @staticmethod - def attr_parser(): - return {'applications': Application} diff --git a/cvp/opnfv_testapi/resources/handlers.py b/cvp/opnfv_testapi/resources/handlers.py deleted file mode 100644 index 9b156e17..00000000 --- a/cvp/opnfv_testapi/resources/handlers.py +++ /dev/null @@ -1,331 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -# feng.xiaowei@zte.com.cn refactor db.pod to db.pods 5-19-2016 -# feng.xiaowei@zte.com.cn refactor test_project to project 5-19-2016 -# feng.xiaowei@zte.com.cn refactor response body 5-19-2016 -# feng.xiaowei@zte.com.cn refactor pod/project response info 5-19-2016 -# feng.xiaowei@zte.com.cn refactor testcase related handler 5-20-2016 -# feng.xiaowei@zte.com.cn refactor result related handler 5-23-2016 -# feng.xiaowei@zte.com.cn refactor dashboard related handler 5-24-2016 -# feng.xiaowei@zte.com.cn add methods to GenericApiHandler 5-26-2016 -# feng.xiaowei@zte.com.cn remove PodHandler 5-26-2016 -# feng.xiaowei@zte.com.cn remove ProjectHandler 5-26-2016 -# feng.xiaowei@zte.com.cn remove TestcaseHandler 5-27-2016 -# feng.xiaowei@zte.com.cn remove ResultHandler 5-29-2016 -# feng.xiaowei@zte.com.cn remove DashboardHandler 5-30-2016 -############################################################################## - -import json -from datetime import datetime -from datetime import timedelta - -import logging -from tornado import gen -from tornado import web - -from opnfv_testapi.common import check -from opnfv_testapi.common import message -from opnfv_testapi.common import raises -from opnfv_testapi.db import api as dbapi -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger -from opnfv_testapi.ui.auth import constants as auth_const - -DEFAULT_REPRESENTATION = "application/json" - - -class GenericApiHandler(web.RequestHandler): - def __init__(self, application, request, **kwargs): - super(GenericApiHandler, self).__init__(application, request, **kwargs) - self.json_args = None - self.table = None - self.table_cls = None - self.db_projects = 'projects' - self.db_pods = 'pods' - self.db_testcases = 'testcases' - self.db_results = 'results' - self.db_scenarios = 'scenarios' - self.auth = self.settings["auth"] - - def get_int(self, key, value): - try: - value = int(value) - except: - raises.BadRequest(message.must_int(key)) - return value - - @gen.coroutine - def set_query(self): - query = dict() - date_range = dict() - for k in self.request.query_arguments.keys(): - v = self.get_query_argument(k) - if k == 'period': - v = self.get_int(k, v) - if v > 0: - period = datetime.now() - timedelta(days=v) - obj = {"$gte": str(period)} - query['start_date'] = obj - elif k == 'from': - date_range.update({'$gte': str(v)}) - elif k == 'to': - date_range.update({'$lt': str(v)}) - elif k == 'signed': - openid = self.get_secure_cookie(auth_const.OPENID) - user = yield dbapi.db_find_one("users", {'openid': openid}) - role = self.get_secure_cookie(auth_const.ROLE) - logging.info('role:%s', role) - if role: - query['$or'] = [ - { - "shared": { - "$elemMatch": {"$eq": openid} - } - }, - {"owner": openid}, - { - "shared": { - "$elemMatch": {"$eq": user.get("email")} - } - } - ] - - if role.find("reviewer") != -1: - query['$or'].append({"status": {"$ne": "private"}}) - elif k not in ['last', 'page', 'descend', 'per_page']: - query[k] = v - if date_range: - query['start_date'] = date_range - - # if $lt is not provided, - # empty/None/null/'' start_date will also be returned - if 'start_date' in query and '$lt' not in query['start_date']: - query['start_date'].update({'$lt': str(datetime.now())}) - - logging.debug("query:%s", query) - raise gen.Return((query)) - - def prepare(self): - if self.request.method != "GET" and self.request.method != "DELETE": - if self.request.headers.get("Content-Type") is not None: - if self.request.headers["Content-Type"].startswith( - DEFAULT_REPRESENTATION): - try: - self.json_args = json.loads(self.request.body) - except (ValueError, KeyError, TypeError) as error: - raises.BadRequest(message.bad_format(str(error))) - - def finish_request(self, json_object=None): - if json_object: - self.write(json.dumps(json_object)) - self.set_header("Content-Type", DEFAULT_REPRESENTATION) - self.finish() - - def _create_response(self, resource): - href = self.request.full_url() + '/' + str(resource) - return models.CreateResponse(href=href).format() - - def format_data(self, data): - cls_data = self.table_cls.from_dict(data) - return cls_data.format_http() - - @gen.coroutine - @check.no_body - @check.miss_fields - @check.carriers_exist - @check.new_not_exists - def _inner_create(self, **kwargs): - data = self.table_cls.from_dict(self.json_args) - for k, v in kwargs.iteritems(): - if k != 'query': - data.__setattr__(k, v) - - if self.table != 'results': - data.creation_date = datetime.now() - _id = yield dbapi.db_save(self.table, data.format()) - logging.warning("_id:%s", _id) - raise gen.Return(_id) - - def _create_only(self, **kwargs): - resource = self._inner_create(**kwargs) - logging.warning("resource:%s", resource) - - @check.authenticate - @check.no_body - @check.miss_fields - @check.carriers_exist - @check.new_not_exists - def _create(self, **kwargs): - # resource = self._inner_create(**kwargs) - data = self.table_cls.from_dict(self.json_args) - for k, v in kwargs.iteritems(): - if k != 'query': - data.__setattr__(k, v) - - if self.table != 'results': - data.creation_date = datetime.now() - _id = yield dbapi.db_save(self.table, data.format()) - if 'name' in self.json_args: - resource = data.name - else: - resource = _id - - self.finish_request(self._create_response(resource)) - - @gen.coroutine - def _check_if_exists(self, *args, **kwargs): - query = kwargs['query'] - table = kwargs['table'] - if query and table: - data = yield dbapi.db_find_one(table, query) - if data: - raise gen.Return((True, 'Data alreay exists. %s' % (query))) - raise gen.Return((False, 'Data does not exist. %s' % (query))) - - # @web.asynchronous - @gen.coroutine - def _list(self, query=None, res_op=None, *args, **kwargs): - logging.debug("_list query:%s", query) - sort = kwargs.get('sort') - page = kwargs.get('page', 0) - last = kwargs.get('last', 0) - per_page = kwargs.get('per_page', 0) - if query is None: - query = {} - - total_pages = 0 - if page > 0: - cursor = dbapi.db_list(self.table, query) - records_count = yield cursor.count() - total_pages = self._calc_total_pages(records_count, - last, - page, - per_page) - pipelines = self._set_pipelines(query, sort, last, page, per_page) - cursor = dbapi.db_aggregate(self.table, pipelines) - data = list() - while (yield cursor.fetch_next): - data.append(self.format_data(cursor.next_object())) - if res_op is None: - res = {self.table: data} - else: - res = res_op(data, *args) - if page > 0: - res.update({ - 'pagination': { - 'current_page': kwargs.get('page'), - 'total_pages': total_pages - } - }) - self.finish_request(res) - logging.debug('_list end') - - @staticmethod - def _calc_total_pages(records_count, last, page, per_page): - logging.debug("totalItems:%d per_page:%d", records_count, per_page) - records_nr = records_count - if (records_count > last) and (last > 0): - records_nr = last - - total_pages, remainder = divmod(records_nr, per_page) - if remainder > 0: - total_pages += 1 - if page > 1 and page > total_pages: - raises.BadRequest( - 'Request page > total_pages [{}]'.format(total_pages)) - return total_pages - - @staticmethod - def _set_pipelines(query, sort, last, page, per_page): - pipelines = list() - if query: - pipelines.append({'$match': query}) - if sort: - pipelines.append({'$sort': sort}) - - if page > 0: - pipelines.append({'$skip': (page - 1) * per_page}) - pipelines.append({'$limit': per_page}) - elif last > 0: - pipelines.append({'$limit': last}) - - return pipelines - - @web.asynchronous - @gen.coroutine - @check.not_exist - def _get_one(self, data, query=None): - self.finish_request(self.format_data(data)) - - @check.authenticate - @check.not_exist - def _delete(self, data, query=None): - yield dbapi.db_delete(self.table, query) - self.finish_request() - - @check.authenticate - @check.no_body - @check.not_exist - @check.updated_one_not_exist - def _update(self, data, query=None, **kwargs): - logging.debug("_update") - data = self.table_cls.from_dict(data) - update_req = self._update_requests(data) - yield dbapi.db_update(self.table, query, update_req) - update_req['_id'] = str(data._id) - self.finish_request(update_req) - - def _update_requests(self, data): - request = dict() - for k, v in self.json_args.iteritems(): - request = self._update_request(request, k, v, - data.__getattribute__(k)) - if not request: - raises.Forbidden(message.no_update()) - - edit_request = data.format() - edit_request.update(request) - return edit_request - - @staticmethod - def _update_request(edit_request, key, new_value, old_value): - """ - This function serves to prepare the elements in the update request. - We try to avoid replace the exact values in the db - edit_request should be a dict in which we add an entry (key) after - comparing values - """ - if not (new_value is None): - if new_value != old_value: - edit_request[key] = new_value - - return edit_request - - def _update_query(self, keys, data): - query = dict() - equal = True - for key in keys: - new = self.json_args.get(key) - old = data.get(key) - if new is None: - new = old - elif new != old: - equal = False - query[key] = new - return query if not equal else dict() - - -class VersionHandler(GenericApiHandler): - @swagger.operation(nickname='listAllVersions') - def get(self): - """ - @description: list all supported versions - @rtype: L{Versions} - """ - versions = [{'version': 'api.cvp.0.7.0', 'description': 'basics'}] - self.finish_request({'versions': versions}) diff --git a/cvp/opnfv_testapi/resources/models.py b/cvp/opnfv_testapi/resources/models.py deleted file mode 100644 index e8fc532b..00000000 --- a/cvp/opnfv_testapi/resources/models.py +++ /dev/null @@ -1,115 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -# feng.xiaowei@zte.com.cn mv Pod to pod_models.py 5-18-2016 -# feng.xiaowei@zte.com.cn add MetaCreateResponse/MetaGetResponse 5-18-2016 -# feng.xiaowei@zte.com.cn mv TestProject to project_models.py 5-19-2016 -# feng.xiaowei@zte.com.cn delete meta class 5-19-2016 -# feng.xiaowei@zte.com.cn add CreateResponse 5-19-2016 -# feng.xiaowei@zte.com.cn mv TestCase to testcase_models.py 5-20-2016 -# feng.xiaowei@zte.com.cn mv TestResut to result_models.py 5-23-2016 -# feng.xiaowei@zte.com.cn add ModelBase 12-20-2016 -############################################################################## -import ast -import copy - -from opnfv_testapi.tornado_swagger import swagger - - -class ModelBase(object): - - def format(self): - return self._format(['_id']) - - def format_http(self): - return self._format([]) - - @classmethod - def from_dict(cls, a_dict): - if a_dict is None: - return None - - attr_parser = cls.attr_parser() - t = cls() - for k, v in a_dict.iteritems(): - value = v - if isinstance(v, dict) and k in attr_parser: - value = attr_parser[k].from_dict(v) - elif isinstance(v, list) and k in attr_parser: - value = [] - for item in v: - value.append(attr_parser[k].from_dict(item)) - - t.__setattr__(k, value) - - return t - - @staticmethod - def attr_parser(): - return {} - - def _format(self, excludes): - new_obj = copy.deepcopy(self) - dicts = new_obj.__dict__ - for k in dicts.keys(): - if k in excludes: - del dicts[k] - elif dicts[k]: - dicts[k] = self._obj_format(dicts[k]) - return dicts - - def _obj_format(self, obj): - if self._has_format(obj): - obj = obj.format() - elif isinstance(obj, unicode): - try: - obj = self._obj_format(ast.literal_eval(obj)) - except: - try: - obj = str(obj) - except: - obj = obj - elif isinstance(obj, list): - hs = list() - for h in obj: - hs.append(self._obj_format(h)) - obj = hs - elif not isinstance(obj, (str, int, float, dict)): - obj = str(obj) - return obj - - @staticmethod - def _has_format(obj): - return not isinstance(obj, (str, unicode)) and hasattr(obj, 'format') - - -@swagger.model() -class CreateResponse(ModelBase): - def __init__(self, href=''): - self.href = href - - -@swagger.model() -class Versions(ModelBase): - """ - @property versions: - @ptype versions: C{list} of L{Version} - """ - - def __init__(self): - self.versions = list() - - @staticmethod - def attr_parser(): - return {'versions': Version} - - -@swagger.model() -class Version(ModelBase): - def __init__(self, version=None, description=None): - self.version = version - self.description = description diff --git a/cvp/opnfv_testapi/resources/pod_handlers.py b/cvp/opnfv_testapi/resources/pod_handlers.py deleted file mode 100644 index 50298875..00000000 --- a/cvp/opnfv_testapi/resources/pod_handlers.py +++ /dev/null @@ -1,78 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -import handlers -from opnfv_testapi.resources import pod_models -from opnfv_testapi.tornado_swagger import swagger - - -class GenericPodHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericPodHandler, self).__init__(application, request, **kwargs) - self.table = 'pods' - self.table_cls = pod_models.Pod - - -class PodCLHandler(GenericPodHandler): - @swagger.operation(nickname='listAllPods') - def get(self): - """ - @description: list all pods - @return 200: list all pods, empty list is no pod exist - @rtype: L{Pods} - """ - self._list() - - @swagger.operation(nickname='createPod') - def post(self): - """ - @description: create a pod - @param body: pod to be created - @type body: L{PodCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: pod is created. - @raise 403: pod already exists - @raise 400: body or name not provided - """ - def query(): - return {'name': self.json_args.get('name')} - miss_fields = ['name'] - self._create(miss_fields=miss_fields, query=query) - - -class PodGURHandler(GenericPodHandler): - @swagger.operation(nickname='getPodByName') - def get(self, pod_name): - """ - @description: get a single pod by pod_name - @rtype: L{Pod} - @return 200: pod exist - @raise 404: pod not exist - """ - self._get_one(query={'name': pod_name}) - - def delete(self, pod_name): - """ Remove a POD - - # check for an existing pod to be deleted - mongo_dict = yield self.db.pods.find_one( - {'name': pod_name}) - pod = TestProject.pod(mongo_dict) - if pod is None: - raise HTTPError(HTTP_NOT_FOUND, - "{} could not be found as a pod to be deleted" - .format(pod_name)) - - # just delete it, or maybe save it elsewhere in a future - res = yield self.db.projects.remove( - {'name': pod_name}) - - self.finish_request(answer) - """ - pass diff --git a/cvp/opnfv_testapi/resources/pod_models.py b/cvp/opnfv_testapi/resources/pod_models.py deleted file mode 100644 index 2c3ea978..00000000 --- a/cvp/opnfv_testapi/resources/pod_models.py +++ /dev/null @@ -1,52 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - - -# name: name of the POD e.g. zte-1 -# mode: metal or virtual -# details: any detail -# role: ci-pod or community-pod or single-node - - -@swagger.model() -class PodCreateRequest(models.ModelBase): - def __init__(self, name, mode='', details='', role=""): - self.name = name - self.mode = mode - self.details = details - self.role = role - - -@swagger.model() -class Pod(models.ModelBase): - def __init__(self, - name='', mode='', details='', - role="", _id='', create_date=''): - self.name = name - self.mode = mode - self.details = details - self.role = role - self._id = _id - self.creation_date = create_date - - -@swagger.model() -class Pods(models.ModelBase): - """ - @property pods: - @ptype pods: C{list} of L{Pod} - """ - def __init__(self): - self.pods = list() - - @staticmethod - def attr_parser(): - return {'pods': Pod} diff --git a/cvp/opnfv_testapi/resources/project_handlers.py b/cvp/opnfv_testapi/resources/project_handlers.py deleted file mode 100644 index be295070..00000000 --- a/cvp/opnfv_testapi/resources/project_handlers.py +++ /dev/null @@ -1,86 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## - -from opnfv_testapi.resources import handlers -from opnfv_testapi.resources import project_models -from opnfv_testapi.tornado_swagger import swagger - - -class GenericProjectHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericProjectHandler, self).__init__(application, - request, - **kwargs) - self.table = 'projects' - self.table_cls = project_models.Project - - -class ProjectCLHandler(GenericProjectHandler): - @swagger.operation(nickname="listAllProjects") - def get(self): - """ - @description: list all projects - @return 200: return all projects, empty list is no project exist - @rtype: L{Projects} - """ - self._list() - - @swagger.operation(nickname="createProject") - def post(self): - """ - @description: create a project - @param body: project to be created - @type body: L{ProjectCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: project is created. - @raise 403: project already exists - @raise 400: body or name not provided - """ - def query(): - return {'name': self.json_args.get('name')} - miss_fields = ['name'] - self._create(miss_fields=miss_fields, query=query) - - -class ProjectGURHandler(GenericProjectHandler): - @swagger.operation(nickname='getProjectByName') - def get(self, project_name): - """ - @description: get a single project by project_name - @rtype: L{Project} - @return 200: project exist - @raise 404: project not exist - """ - self._get_one(query={'name': project_name}) - - @swagger.operation(nickname="updateProjectByName") - def put(self, project_name): - """ - @description: update a single project by project_name - @param body: project to be updated - @type body: L{ProjectUpdateRequest} - @in body: body - @rtype: L{Project} - @return 200: update success - @raise 404: project not exist - @raise 403: new project name already exist or nothing to update - """ - query = {'name': project_name} - db_keys = ['name'] - self._update(query=query, db_keys=db_keys) - - @swagger.operation(nickname='deleteProjectByName') - def delete(self, project_name): - """ - @description: delete a project by project_name - @return 200: delete success - @raise 404: project not exist - """ - self._delete(query={'name': project_name}) diff --git a/cvp/opnfv_testapi/resources/project_models.py b/cvp/opnfv_testapi/resources/project_models.py deleted file mode 100644 index 3243882b..00000000 --- a/cvp/opnfv_testapi/resources/project_models.py +++ /dev/null @@ -1,48 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - - -@swagger.model() -class ProjectCreateRequest(models.ModelBase): - def __init__(self, name, description=''): - self.name = name - self.description = description - - -@swagger.model() -class ProjectUpdateRequest(models.ModelBase): - def __init__(self, name='', description=''): - self.name = name - self.description = description - - -@swagger.model() -class Project(models.ModelBase): - def __init__(self, - name=None, _id=None, description=None, create_date=None): - self._id = _id - self.name = name - self.description = description - self.creation_date = create_date - - -@swagger.model() -class Projects(models.ModelBase): - """ - @property projects: - @ptype projects: C{list} of L{Project} - """ - def __init__(self): - self.projects = list() - - @staticmethod - def attr_parser(): - return {'projects': Project} diff --git a/cvp/opnfv_testapi/resources/result_handlers.py b/cvp/opnfv_testapi/resources/result_handlers.py deleted file mode 100644 index 2e65ba44..00000000 --- a/cvp/opnfv_testapi/resources/result_handlers.py +++ /dev/null @@ -1,308 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -import logging -from datetime import datetime -from datetime import timedelta -import json -import tarfile -import io - -from tornado import gen -from tornado import web -from bson import objectid - -from opnfv_testapi.common.config import CONF -from opnfv_testapi.common import message -from opnfv_testapi.common import raises -from opnfv_testapi.resources import handlers -from opnfv_testapi.resources import result_models -from opnfv_testapi.tornado_swagger import swagger -from opnfv_testapi.ui.auth import constants as auth_const - - -class GenericResultHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericResultHandler, self).__init__(application, - request, - **kwargs) - self.table = self.db_results - self.table_cls = result_models.TestResult - - def get_int(self, key, value): - try: - value = int(value) - except: - raises.BadRequest(message.must_int(key)) - return value - - def set_query(self): - query = dict() - date_range = dict() - - query['public'] = {'$not': {'$eq': 'false'}} - for k in self.request.query_arguments.keys(): - v = self.get_query_argument(k) - if k == 'project' or k == 'pod' or k == 'case': - query[k + '_name'] = v - elif k == 'period': - v = self.get_int(k, v) - if v > 0: - period = datetime.now() - timedelta(days=v) - obj = {"$gte": str(period)} - query['start_date'] = obj - elif k == 'trust_indicator': - query[k + '.current'] = float(v) - elif k == 'from': - date_range.update({'$gte': str(v)}) - elif k == 'to': - date_range.update({'$lt': str(v)}) - elif k == 'signed': - openid = self.get_secure_cookie(auth_const.OPENID) - role = self.get_secure_cookie(auth_const.ROLE) - logging.info('role:%s', role) - if role: - del query['public'] - query['user'] = openid - if role.find("reviewer") != -1: - del query['user'] - query['review'] = 'true' - elif k not in ['last', 'page', 'descend']: - query[k] = v - if date_range: - query['start_date'] = date_range - - # if $lt is not provided, - # empty/None/null/'' start_date will also be returned - if 'start_date' in query and '$lt' not in query['start_date']: - query['start_date'].update({'$lt': str(datetime.now())}) - - return query - - -class ResultsCLHandler(GenericResultHandler): - @swagger.operation(nickname="queryTestResults") - def get(self): - """ - @description: Retrieve result(s) for a test project - on a specific pod. - @notes: Retrieve result(s) for a test project on a specific pod. - Available filters for this request are : - - project : project name - - case : case name - - pod : pod name - - version : platform version (Arno-R1, ...) - - installer : fuel/apex/compass/joid/daisy - - build_tag : Jenkins build tag name - - period : x last days, incompatible with from/to - - from : starting time in 2016-01-01 or 2016-01-01 00:01:23 - - to : ending time in 2016-01-01 or 2016-01-01 00:01:23 - - scenario : the test scenario (previously version) - - criteria : the global criteria status passed or failed - - trust_indicator : evaluate the stability of the test case - to avoid running systematically long and stable test case - - signed : get logined user result - - GET /results/project=functest&case=vPing&version=Arno-R1 \ - &pod=pod_name&period=15&signed - @return 200: all test results consist with query, - empty list if no result is found - @rtype: L{TestResults} - @param pod: pod name - @type pod: L{string} - @in pod: query - @required pod: False - @param project: project name - @type project: L{string} - @in project: query - @required project: False - @param case: case name - @type case: L{string} - @in case: query - @required case: False - @param version: i.e. Colorado - @type version: L{string} - @in version: query - @required version: False - @param installer: fuel/apex/joid/compass - @type installer: L{string} - @in installer: query - @required installer: False - @param build_tag: i.e. v3.0 - @type build_tag: L{string} - @in build_tag: query - @required build_tag: False - @param scenario: i.e. odl - @type scenario: L{string} - @in scenario: query - @required scenario: False - @param criteria: i.e. passed - @type criteria: L{string} - @in criteria: query - @required criteria: False - @param period: last days - @type period: L{string} - @in period: query - @required period: False - @param from: i.e. 2016-01-01 or 2016-01-01 00:01:23 - @type from: L{string} - @in from: query - @required from: False - @param to: i.e. 2016-01-01 or 2016-01-01 00:01:23 - @type to: L{string} - @in to: query - @required to: False - @param last: last records stored until now - @type last: L{string} - @in last: query - @required last: False - @param page: which page to list - @type page: L{int} - @in page: query - @required page: False - @param trust_indicator: must be float - @type trust_indicator: L{float} - @in trust_indicator: query - @required trust_indicator: False - @param signed: user results or all results - @type signed: L{string} - @in signed: query - @required signed: False - @param descend: true, newest2oldest; false, oldest2newest - @type descend: L{string} - @in descend: query - @required descend: False - """ - def descend_limit(): - descend = self.get_query_argument('descend', 'true') - return -1 if descend.lower() == 'true' else 1 - - def last_limit(): - return self.get_int('last', self.get_query_argument('last', 0)) - - def page_limit(): - return self.get_int('page', self.get_query_argument('page', 0)) - - limitations = { - 'sort': {'_id': descend_limit()}, - 'last': last_limit(), - 'page': page_limit(), - 'per_page': CONF.api_results_per_page - } - - self._list(query=self.set_query(), **limitations) - - @swagger.operation(nickname="createTestResult") - def post(self): - """ - @description: create a test result - @param body: result to be created - @type body: L{ResultCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: result is created. - @raise 404: pod/project/testcase not exist - @raise 400: body/pod_name/project_name/case_name not provided - """ - self._post() - - def _post(self): - def pod_query(): - return {'name': self.json_args.get('pod_name')} - - def project_query(): - return {'name': self.json_args.get('project_name')} - - def testcase_query(): - return {'project_name': self.json_args.get('project_name'), - 'name': self.json_args.get('case_name')} - - miss_fields = ['pod_name', 'project_name', 'case_name'] - carriers = [('pods', pod_query), - ('projects', project_query), - ('testcases', testcase_query)] - - self._create(miss_fields=miss_fields, carriers=carriers) - - -class ResultsUploadHandler(ResultsCLHandler): - @swagger.operation(nickname="uploadTestResult") - @web.asynchronous - @gen.coroutine - def post(self): - """ - @description: upload and create a test result - @param body: result to be created - @type body: L{ResultCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: result is created. - @raise 404: pod/project/testcase not exist - @raise 400: body/pod_name/project_name/case_name not provided - """ - fileinfo = self.request.files['file'][0] - tar_in = tarfile.open(fileobj=io.BytesIO(fileinfo['body']), - mode="r:gz") - try: - results = tar_in.extractfile('results/results.json').read() - except KeyError: - msg = 'Uploaded results must contain at least one passing test.' - self.finish_request({'code': 403, 'msg': msg}) - return - results = results.split('\n') - result_ids = [] - for result in results: - if result == '': - continue - self.json_args = json.loads(result).copy() - build_tag = self.json_args['build_tag'] - _id = yield self._inner_create() - result_ids.append(str(_id)) - test_id = build_tag[13:49] - log_path = '/home/testapi/logs/%s' % (test_id) - tar_in.extractall(log_path) - log_filename = "/home/testapi/logs/log_%s.tar.gz" % (test_id) - with open(log_filename, "wb") as tar_out: - tar_out.write(fileinfo['body']) - resp = {'id': test_id, 'results': result_ids} - self.finish_request(resp) - - -class ResultsGURHandler(GenericResultHandler): - @swagger.operation(nickname='DeleteTestResultById') - def delete(self, result_id): - query = {'_id': objectid.ObjectId(result_id)} - self._delete(query=query) - - @swagger.operation(nickname='getTestResultById') - def get(self, result_id): - """ - @description: get a single result by result_id - @rtype: L{TestResult} - @return 200: test result exist - @raise 404: test result not exist - """ - query = dict() - query["_id"] = objectid.ObjectId(result_id) - self._get_one(query=query) - - @swagger.operation(nickname="updateTestResultById") - def put(self, result_id): - """ - @description: update a single result by _id - @param body: fields to be updated - @type body: L{ResultUpdateRequest} - @in body: body - @rtype: L{Result} - @return 200: update success - @raise 404: result not exist - @raise 403: nothing to update - """ - query = {'_id': objectid.ObjectId(result_id)} - db_keys = [] - self._update(query=query, db_keys=db_keys) diff --git a/cvp/opnfv_testapi/resources/result_models.py b/cvp/opnfv_testapi/resources/result_models.py deleted file mode 100644 index 698f4981..00000000 --- a/cvp/opnfv_testapi/resources/result_models.py +++ /dev/null @@ -1,133 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - - -@swagger.model() -class TIHistory(models.ModelBase): - """ - @ptype step: L{float} - """ - def __init__(self, date=None, step=0): - self.date = date - self.step = step - - -@swagger.model() -class TI(models.ModelBase): - """ - @property histories: trust_indicator update histories - @ptype histories: C{list} of L{TIHistory} - @ptype current: L{float} - """ - def __init__(self, current=0): - self.current = current - self.histories = list() - - @staticmethod - def attr_parser(): - return {'histories': TIHistory} - - -@swagger.model() -class ResultCreateRequest(models.ModelBase): - """ - @property trust_indicator: - @ptype trust_indicator: L{TI} - """ - def __init__(self, - pod_name=None, - project_name=None, - case_name=None, - installer=None, - version=None, - start_date=None, - stop_date=None, - details=None, - build_tag=None, - scenario=None, - criteria=None, - user=None, - public="true", - review="false", - trust_indicator=None): - self.pod_name = pod_name - self.project_name = project_name - self.case_name = case_name - self.installer = installer - self.version = version - self.start_date = start_date - self.stop_date = stop_date - self.details = details - self.build_tag = build_tag - self.scenario = scenario - self.criteria = criteria - self.user = user - self.public = public - self.review = review - self.trust_indicator = trust_indicator if trust_indicator else TI(0) - - -@swagger.model() -class ResultUpdateRequest(models.ModelBase): - """ - @property trust_indicator: - @ptype trust_indicator: L{TI} - """ - def __init__(self, trust_indicator=None): - self.trust_indicator = trust_indicator - - -@swagger.model() -class TestResult(models.ModelBase): - """ - @property trust_indicator: used for long duration test case - @ptype trust_indicator: L{TI} - """ - def __init__(self, _id=None, case_name=None, project_name=None, - pod_name=None, installer=None, version=None, - start_date=None, stop_date=None, details=None, - build_tag=None, scenario=None, criteria=None, - user=None, public="true", review="false", - trust_indicator=None): - self._id = _id - self.case_name = case_name - self.project_name = project_name - self.pod_name = pod_name - self.installer = installer - self.version = version - self.start_date = start_date - self.stop_date = stop_date - self.details = details - self.build_tag = build_tag - self.scenario = scenario - self.criteria = criteria - self.user = user - self.public = public - self.review = review - self.trust_indicator = trust_indicator - - @staticmethod - def attr_parser(): - return {'trust_indicator': TI} - - -@swagger.model() -class TestResults(models.ModelBase): - """ - @property results: - @ptype results: C{list} of L{TestResult} - """ - def __init__(self): - self.results = list() - - @staticmethod - def attr_parser(): - return {'results': TestResult} diff --git a/cvp/opnfv_testapi/resources/scenario_handlers.py b/cvp/opnfv_testapi/resources/scenario_handlers.py deleted file mode 100644 index 5d420a56..00000000 --- a/cvp/opnfv_testapi/resources/scenario_handlers.py +++ /dev/null @@ -1,282 +0,0 @@ -import functools - -from opnfv_testapi.common import message -from opnfv_testapi.common import raises -from opnfv_testapi.resources import handlers -import opnfv_testapi.resources.scenario_models as models -from opnfv_testapi.tornado_swagger import swagger - - -class GenericScenarioHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericScenarioHandler, self).__init__(application, - request, - **kwargs) - self.table = self.db_scenarios - self.table_cls = models.Scenario - - -class ScenariosCLHandler(GenericScenarioHandler): - @swagger.operation(nickname="queryScenarios") - def get(self): - """ - @description: Retrieve scenario(s). - @notes: Retrieve scenario(s) - Available filters for this request are : - - name : scenario name - - GET /scenarios?name=scenario_1 - @param name: scenario name - @type name: L{string} - @in name: query - @required name: False - @param installer: installer type - @type installer: L{string} - @in installer: query - @required installer: False - @param version: version - @type version: L{string} - @in version: query - @required version: False - @param project: project name - @type project: L{string} - @in project: query - @required project: False - @return 200: all scenarios satisfy queries, - empty list if no scenario is found - @rtype: L{Scenarios} - """ - - def _set_query(): - query = dict() - elem_query = dict() - for k in self.request.query_arguments.keys(): - v = self.get_query_argument(k) - if k == 'installer': - elem_query["installer"] = v - elif k == 'version': - elem_query["versions.version"] = v - elif k == 'project': - elem_query["versions.projects.project"] = v - else: - query[k] = v - if elem_query: - query['installers'] = {'$elemMatch': elem_query} - return query - - self._list(query=_set_query()) - - @swagger.operation(nickname="createScenario") - def post(self): - """ - @description: create a new scenario by name - @param body: scenario to be created - @type body: L{ScenarioCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: scenario is created. - @raise 403: scenario already exists - @raise 400: body or name not provided - """ - def query(): - return {'name': self.json_args.get('name')} - miss_fields = ['name'] - self._create(miss_fields=miss_fields, query=query) - - -class ScenarioGURHandler(GenericScenarioHandler): - @swagger.operation(nickname='getScenarioByName') - def get(self, name): - """ - @description: get a single scenario by name - @rtype: L{Scenario} - @return 200: scenario exist - @raise 404: scenario not exist - """ - self._get_one(query={'name': name}) - pass - - @swagger.operation(nickname="updateScenarioByName") - def put(self, name): - """ - @description: update a single scenario by name - @param body: fields to be updated - @type body: L{ScenarioUpdateRequest} - @in body: body - @rtype: L{Scenario} - @return 200: update success - @raise 404: scenario not exist - @raise 403: nothing to update - """ - query = {'name': name} - db_keys = ['name'] - self._update(query=query, db_keys=db_keys) - - @swagger.operation(nickname="deleteScenarioByName") - def delete(self, name): - """ - @description: delete a scenario by name - @return 200: delete success - @raise 404: scenario not exist: - """ - - self._delete(query={'name': name}) - - def _update_query(self, keys, data): - query = dict() - if self._is_rename(): - new = self._term.get('name') - if data.get('name') != new: - query['name'] = new - - return query - - def _update_requests(self, data): - updates = { - ('name', 'update'): self._update_requests_rename, - ('installer', 'add'): self._update_requests_add_installer, - ('installer', 'delete'): self._update_requests_delete_installer, - ('version', 'add'): self._update_requests_add_version, - ('version', 'delete'): self._update_requests_delete_version, - ('owner', 'update'): self._update_requests_change_owner, - ('project', 'add'): self._update_requests_add_project, - ('project', 'delete'): self._update_requests_delete_project, - ('customs', 'add'): self._update_requests_add_customs, - ('customs', 'delete'): self._update_requests_delete_customs, - ('score', 'add'): self._update_requests_add_score, - ('trust_indicator', 'add'): self._update_requests_add_ti, - } - - updates[(self._field, self._op)](data) - - return data.format() - - def _iter_installers(xstep): - @functools.wraps(xstep) - def magic(self, data): - [xstep(self, installer) - for installer in self._filter_installers(data.installers)] - return magic - - def _iter_versions(xstep): - @functools.wraps(xstep) - def magic(self, installer): - [xstep(self, version) - for version in (self._filter_versions(installer.versions))] - return magic - - def _iter_projects(xstep): - @functools.wraps(xstep) - def magic(self, version): - [xstep(self, project) - for project in (self._filter_projects(version.projects))] - return magic - - def _update_requests_rename(self, data): - data.name = self._term.get('name') - if not data.name: - raises.BadRequest(message.missing('name')) - - def _update_requests_add_installer(self, data): - data.installers.append(models.ScenarioInstaller.from_dict(self._term)) - - def _update_requests_delete_installer(self, data): - data.installers = self._remove_installers(data.installers) - - @_iter_installers - def _update_requests_add_version(self, installer): - installer.versions.append(models.ScenarioVersion.from_dict(self._term)) - - @_iter_installers - def _update_requests_delete_version(self, installer): - installer.versions = self._remove_versions(installer.versions) - - @_iter_installers - @_iter_versions - def _update_requests_change_owner(self, version): - version.owner = self._term.get('owner') - - @_iter_installers - @_iter_versions - def _update_requests_add_project(self, version): - version.projects.append(models.ScenarioProject.from_dict(self._term)) - - @_iter_installers - @_iter_versions - def _update_requests_delete_project(self, version): - version.projects = self._remove_projects(version.projects) - - @_iter_installers - @_iter_versions - @_iter_projects - def _update_requests_add_customs(self, project): - project.customs = list(set(project.customs + self._term)) - - @_iter_installers - @_iter_versions - @_iter_projects - def _update_requests_delete_customs(self, project): - project.customs = filter( - lambda f: f not in self._term, - project.customs) - - @_iter_installers - @_iter_versions - @_iter_projects - def _update_requests_add_score(self, project): - project.scores.append( - models.ScenarioScore.from_dict(self._term)) - - @_iter_installers - @_iter_versions - @_iter_projects - def _update_requests_add_ti(self, project): - project.trust_indicators.append( - models.ScenarioTI.from_dict(self._term)) - - def _is_rename(self): - return self._field == 'name' and self._op == 'update' - - def _remove_installers(self, installers): - return self._remove('installer', installers) - - def _filter_installers(self, installers): - return self._filter('installer', installers) - - def _remove_versions(self, versions): - return self._remove('version', versions) - - def _filter_versions(self, versions): - return self._filter('version', versions) - - def _remove_projects(self, projects): - return self._remove('project', projects) - - def _filter_projects(self, projects): - return self._filter('project', projects) - - def _remove(self, field, fields): - return filter( - lambda f: getattr(f, field) != self._locate.get(field), - fields) - - def _filter(self, field, fields): - return filter( - lambda f: getattr(f, field) == self._locate.get(field), - fields) - - @property - def _field(self): - return self.json_args.get('field') - - @property - def _op(self): - return self.json_args.get('op') - - @property - def _locate(self): - return self.json_args.get('locate') - - @property - def _term(self): - return self.json_args.get('term') diff --git a/cvp/opnfv_testapi/resources/scenario_models.py b/cvp/opnfv_testapi/resources/scenario_models.py deleted file mode 100644 index 467cff24..00000000 --- a/cvp/opnfv_testapi/resources/scenario_models.py +++ /dev/null @@ -1,204 +0,0 @@ -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - - -def list_default(value): - return value if value else list() - - -def dict_default(value): - return value if value else dict() - - -@swagger.model() -class ScenarioTI(models.ModelBase): - def __init__(self, date=None, status='silver'): - self.date = date - self.status = status - - -@swagger.model() -class ScenarioScore(models.ModelBase): - def __init__(self, date=None, score='0'): - self.date = date - self.score = score - - -@swagger.model() -class ScenarioProject(models.ModelBase): - """ - @property customs: - @ptype customs: C{list} of L{string} - @property scores: - @ptype scores: C{list} of L{ScenarioScore} - @property trust_indicators: - @ptype trust_indicators: C{list} of L{ScenarioTI} - """ - def __init__(self, - project='', - customs=None, - scores=None, - trust_indicators=None): - self.project = project - self.customs = list_default(customs) - self.scores = list_default(scores) - self.trust_indicators = list_default(trust_indicators) - - @staticmethod - def attr_parser(): - return {'scores': ScenarioScore, - 'trust_indicators': ScenarioTI} - - def __eq__(self, other): - return [self.project == other.project and - self._customs_eq(other) and - self._scores_eq(other) and - self._ti_eq(other)] - - def __ne__(self, other): - return not self.__eq__(other) - - def _customs_eq(self, other): - return set(self.customs) == set(other.customs) - - def _scores_eq(self, other): - return set(self.scores) == set(other.scores) - - def _ti_eq(self, other): - return set(self.trust_indicators) == set(other.trust_indicators) - - -@swagger.model() -class ScenarioVersion(models.ModelBase): - """ - @property projects: - @ptype projects: C{list} of L{ScenarioProject} - """ - def __init__(self, version=None, projects=None): - self.version = version - self.projects = list_default(projects) - - @staticmethod - def attr_parser(): - return {'projects': ScenarioProject} - - def __eq__(self, other): - return [self.version == other.version and self._projects_eq(other)] - - def __ne__(self, other): - return not self.__eq__(other) - - def _projects_eq(self, other): - for s_project in self.projects: - for o_project in other.projects: - if s_project.project == o_project.project: - if s_project != o_project: - return False - - return True - - -@swagger.model() -class ScenarioInstaller(models.ModelBase): - """ - @property versions: - @ptype versions: C{list} of L{ScenarioVersion} - """ - def __init__(self, installer=None, versions=None): - self.installer = installer - self.versions = list_default(versions) - - @staticmethod - def attr_parser(): - return {'versions': ScenarioVersion} - - def __eq__(self, other): - return [self.installer == other.installer and self._versions_eq(other)] - - def __ne__(self, other): - return not self.__eq__(other) - - def _versions_eq(self, other): - for s_version in self.versions: - for o_version in other.versions: - if s_version.version == o_version.version: - if s_version != o_version: - return False - - return True - - -@swagger.model() -class ScenarioCreateRequest(models.ModelBase): - """ - @property installers: - @ptype installers: C{list} of L{ScenarioInstaller} - """ - def __init__(self, name='', installers=None): - self.name = name - self.installers = list_default(installers) - - @staticmethod - def attr_parser(): - return {'installers': ScenarioInstaller} - - -@swagger.model() -class ScenarioUpdateRequest(models.ModelBase): - """ - @property field: update field - @property op: add/delete/update - @property locate: information used to locate the field - @property term: new value - """ - def __init__(self, field=None, op=None, locate=None, term=None): - self.field = field - self.op = op - self.locate = dict_default(locate) - self.term = dict_default(term) - - -@swagger.model() -class Scenario(models.ModelBase): - """ - @property installers: - @ptype installers: C{list} of L{ScenarioInstaller} - """ - def __init__(self, name='', create_date='', _id='', installers=None): - self.name = name - self._id = _id - self.creation_date = create_date - self.installers = list_default(installers) - - @staticmethod - def attr_parser(): - return {'installers': ScenarioInstaller} - - def __ne__(self, other): - return not self.__eq__(other) - - def __eq__(self, other): - return [self.name == other.name and self._installers_eq(other)] - - def _installers_eq(self, other): - for s_install in self.installers: - for o_install in other.installers: - if s_install.installer == o_install.installer: - if s_install != o_install: - return False - - return True - - -@swagger.model() -class Scenarios(models.ModelBase): - """ - @property scenarios: - @ptype scenarios: C{list} of L{Scenario} - """ - def __init__(self): - self.scenarios = list() - - @staticmethod - def attr_parser(): - return {'scenarios': Scenario} diff --git a/cvp/opnfv_testapi/resources/sut_handlers.py b/cvp/opnfv_testapi/resources/sut_handlers.py deleted file mode 100644 index 16c50b83..00000000 --- a/cvp/opnfv_testapi/resources/sut_handlers.py +++ /dev/null @@ -1,112 +0,0 @@ -############################################################################## -# Copyright (c) 2017 -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -import logging -import json -import os - -from opnfv_testapi.resources import handlers -from opnfv_testapi.resources import sut_models -from opnfv_testapi.tornado_swagger import swagger - -LOG = logging.getLogger(__name__) -LOG.setLevel(logging.DEBUG) - -RESULT_PATH = '/home/testapi/logs/{}/results' - - -class GenericSutHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericSutHandler, self).__init__(application, - request, - **kwargs) - self.table = "suts" - self.table_cls = sut_models.Sut - - -class HardwareHandler(GenericSutHandler): - @swagger.operation(nickname="getHardwareById") - def get(self, id): - endpoint_info = self._read_endpoint_info(id) - LOG.debug('Endpoint info: %s', endpoint_info) - - all_info = self._read_sut_info(id) - LOG.debug('All SUT info: %s', all_info) - - hardware_info = {k: self._get_single_host_info(v) - for k, v in all_info.items()} - LOG.debug('SUT info: %s', hardware_info) - - data = { - 'endpoint_info': endpoint_info, - 'hardware_info': hardware_info - } - - self.write(data) - - def _read_endpoint_info(self, id): - path = os.path.join(RESULT_PATH.format(id), 'endpoint_info.json') - try: - with open(path) as f: - endpoint_info = json.load(f) - except Exception: - endpoint_info = [] - - return endpoint_info - - def _read_sut_info(self, id): - path = os.path.join(RESULT_PATH.format(id), 'all_hosts_info.json') - try: - with open(path) as f: - all_info = json.load(f) - except Exception: - all_info = {} - return all_info - - def _get_single_host_info(self, single_info): - info = [] - facts = single_info.get('ansible_facts', {}) - - info.append(['hostname', facts.get('ansible_hostname')]) - - info.append(['product_name', facts.get('ansible_product_name')]) - info.append(['product_version', facts.get('ansible_product_version')]) - - processors = facts.get('ansible_processor', []) - try: - processor_type = '{} {}'.format(processors[0], processors[1]) - except IndexError: - LOG.exception('No Processor in SUT data') - processor_type = None - info.append(['processor_type', processor_type]) - info.append(['architecture', facts.get('ansible_architecture')]) - info.append(['processor_cores', facts.get('ansible_processor_cores')]) - info.append(['processor_vcpus', facts.get('ansible_processor_vcpus')]) - - memory = facts.get('ansible_memtotal_mb') - memory = round(memory * 1.0 / 1024, 2) if memory else None - info.append(['memory', '{} GB'.format(memory)]) - - devices = facts.get('ansible_devices', {}) - info.extend([self._get_device_info(k, v) for k, v in devices.items()]) - - lsb_description = facts.get('ansible_lsb', {}).get('description') - info.append(['OS', lsb_description]) - - interfaces = facts.get('ansible_interfaces') - info.append(['interfaces', interfaces]) - info.extend([self._get_interface_info(facts, i) for i in interfaces]) - info = [i for i in info if i] - - return info - - def _get_interface_info(self, facts, name): - mac = facts.get('ansible_{}'.format(name), {}).get('macaddress') - return [name, mac] if mac else [] - - def _get_device_info(self, name, info): - return ['disk_{}'.format(name), info.get('size')] diff --git a/cvp/opnfv_testapi/resources/sut_models.py b/cvp/opnfv_testapi/resources/sut_models.py deleted file mode 100644 index b4a869b6..00000000 --- a/cvp/opnfv_testapi/resources/sut_models.py +++ /dev/null @@ -1,31 +0,0 @@ -############################################################################## -# Copyright (c) 2017 -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - - -@swagger.model() -class Sut(models.ModelBase): - """ - """ - def __init__(self): - pass - - -@swagger.model() -class Suts(models.ModelBase): - """ - @property suts: - @ptype tests: C{list} of L{Sut} - """ - def __init__(self): - self.suts = list() - - @staticmethod - def attr_parser(): - return {'suts': Sut} diff --git a/cvp/opnfv_testapi/resources/test_handlers.py b/cvp/opnfv_testapi/resources/test_handlers.py deleted file mode 100644 index 82cf9ae6..00000000 --- a/cvp/opnfv_testapi/resources/test_handlers.py +++ /dev/null @@ -1,307 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -import logging -import os -import json - -from tornado import web -from tornado import gen -from bson import objectid - -from opnfv_testapi.common.config import CONF -from opnfv_testapi.common import message -from opnfv_testapi.common import raises -from opnfv_testapi.resources import handlers -from opnfv_testapi.resources import test_models -from opnfv_testapi.tornado_swagger import swagger -from opnfv_testapi.ui.auth import constants as auth_const -from opnfv_testapi.db import api as dbapi - -DOVETAIL_LOG_PATH = '/home/testapi/logs/{}/results/dovetail.log' - - -class GenericTestHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericTestHandler, self).__init__(application, - request, - **kwargs) - self.table = "tests" - self.table_cls = test_models.Test - - -class TestsCLHandler(GenericTestHandler): - @swagger.operation(nickname="queryTests") - @web.asynchronous - @gen.coroutine - def get(self): - """ - @description: Retrieve result(s) for a test project - on a specific pod. - @notes: Retrieve result(s) for a test project on a specific pod. - Available filters for this request are : - - id : Test id - - period : x last days, incompatible with from/to - - from : starting time in 2016-01-01 or 2016-01-01 00:01:23 - - to : ending time in 2016-01-01 or 2016-01-01 00:01:23 - - signed : get logined user result - - GET /results/project=functest&case=vPing&version=Arno-R1 \ - &pod=pod_name&period=15&signed - @return 200: all test results consist with query, - empty list if no result is found - @rtype: L{Tests} - """ - def descend_limit(): - descend = self.get_query_argument('descend', 'true') - return -1 if descend.lower() == 'true' else 1 - - def last_limit(): - return self.get_int('last', self.get_query_argument('last', 0)) - - def page_limit(): - return self.get_int('page', self.get_query_argument('page', 0)) - - limitations = { - 'sort': {'_id': descend_limit()}, - 'last': last_limit(), - 'page': page_limit(), - 'per_page': CONF.api_results_per_page - } - - query = yield self.set_query() - yield self._list(query=query, **limitations) - logging.debug('list end') - - @swagger.operation(nickname="createTest") - @web.asynchronous - def post(self): - """ - @description: create a test - @param body: test to be created - @type body: L{TestCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: test is created. - @raise 404: pod/project/testcase not exist - @raise 400: body/pod_name/project_name/case_name not provided - """ - openid = self.get_secure_cookie(auth_const.OPENID) - if openid: - self.json_args['owner'] = openid - - self._post() - - @gen.coroutine - def _post(self): - miss_fields = [] - carriers = [] - query = {'owner': self.json_args['owner'], 'id': self.json_args['id']} - ret, msg = yield self._check_if_exists(table="tests", query=query) - if ret: - self.finish_request({'code': '403', 'msg': msg}) - return - - self._create(miss_fields=miss_fields, carriers=carriers) - - -class TestsGURHandler(GenericTestHandler): - - @swagger.operation(nickname="getTestById") - @web.asynchronous - @gen.coroutine - def get(self, test_id): - query = dict() - query["_id"] = objectid.ObjectId(test_id) - - data = yield dbapi.db_find_one(self.table, query) - if not data: - raises.NotFound(message.not_found(self.table, query)) - - validation = yield self._check_api_response_validation(data['id']) - - data.update({'validation': validation}) - - self.finish_request(self.format_data(data)) - - @gen.coroutine - def _check_api_response_validation(self, test_id): - log_path = DOVETAIL_LOG_PATH.format(test_id) - if not os.path.exists(log_path): - raises.Forbidden('dovetail.log not found, please check') - - with open(log_path) as f: - log_content = f.read() - - warning_keyword = 'Strict API response validation DISABLED' - if warning_keyword in log_content: - raise gen.Return('API response validation disabled') - else: - raise gen.Return('API response validation enabled') - - @swagger.operation(nickname="deleteTestById") - def delete(self, test_id): - query = {'_id': objectid.ObjectId(test_id)} - self._delete(query=query) - - @swagger.operation(nickname="updateTestById") - @web.asynchronous - def put(self, _id): - """ - @description: update a single test by id - @param body: fields to be updated - @type body: L{TestUpdateRequest} - @in body: body - @rtype: L{Test} - @return 200: update success - @raise 404: Test not exist - @raise 403: nothing to update - """ - logging.debug('put') - data = json.loads(self.request.body) - item = data.get('item') - value = data.get(item) - logging.debug('%s:%s', item, value) - try: - self.update(_id, item, value) - except Exception as e: - logging.error('except:%s', e) - return - - @gen.coroutine - def _convert_to_id(self, email): - query = {"email": email} - table = "users" - if query and table: - data = yield dbapi.db_find_one(table, query) - if data: - raise gen.Return((True, 'Data alreay exists. %s' % (query), - data.get("openid"))) - raise gen.Return((False, 'Data does not exist. %s' % (query), None)) - - @gen.coroutine - def update(self, _id, item, value): - logging.debug("update") - if item == "shared": - new_list = [] - for user in value: - ret, msg, user_id = yield self._convert_to_id(user) - if ret: - user = user_id - new_list.append(user) - query = {"$or": [{"openid": user}, {"email": user}]} - table = "users" - ret, msg = yield self._check_if_exists(table=table, - query=query) - logging.debug('ret:%s', ret) - if not ret: - self.finish_request({'code': '403', 'msg': msg}) - return - - if len(new_list) != len(set(new_list)): - msg = "Already shared with this user" - self.finish_request({'code': '403', 'msg': msg}) - return - - logging.debug("before _update") - self.json_args = {} - self.json_args[item] = value - ret, msg = yield self.check_auth(item, value) - if not ret: - self.finish_request({'code': '404', 'msg': msg}) - return - - query = {'_id': objectid.ObjectId(_id)} - db_keys = ['_id', ] - - test = yield dbapi.db_find_one("tests", query) - if not test: - msg = 'Record does not exist' - self.finish_request({'code': 404, 'msg': msg}) - return - - curr_user = self.get_secure_cookie(auth_const.OPENID) - if item in {"shared", "label", "sut_label"}: - query['owner'] = curr_user - db_keys.append('owner') - - if item == 'sut_label': - if test['status'] != 'private' and not value: - msg = 'SUT version cannot be changed to None after submitting.' - self.finish_request({'code': 403, 'msg': msg}) - return - - if item == "status": - if value in {'approved', 'not approved'}: - if test['status'] == 'private': - msg = 'Not allowed to approve/not approve' - self.finish_request({'code': 403, 'msg': msg}) - return - - user = yield dbapi.db_find_one("users", {'openid': curr_user}) - if 'administrator' not in user['role']: - msg = 'No permission to operate' - self.finish_request({'code': 403, 'msg': msg}) - return - elif value == 'review': - if test['status'] != 'private': - msg = 'Not allowed to submit to review' - self.finish_request({'code': 403, 'msg': msg}) - return - - if not test['sut_label']: - msg = 'Please fill out SUT version before submission' - self.finish_request({'code': 403, 'msg': msg}) - return - - query['owner'] = curr_user - db_keys.append('owner') - - test_query = { - 'id': test['id'], - '$or': [ - {'status': 'review'}, - {'status': 'approved'}, - {'status': 'not approved'} - ] - } - record = yield dbapi.db_find_one("tests", test_query) - if record: - msg = ('{} has already submitted one record with the same ' - 'Test ID: {}'.format(record['owner'], test['id'])) - self.finish_request({'code': 403, 'msg': msg}) - return - else: - query['owner'] = curr_user - db_keys.append('owner') - - logging.debug("before _update 2") - self._update(query=query, db_keys=db_keys) - - @gen.coroutine - def check_auth(self, item, value): - logging.debug('check_auth') - user = self.get_secure_cookie(auth_const.OPENID) - query = {} - if item == "status": - if value == "private" or value == "review": - logging.debug('check review') - query['user_id'] = user - data = yield dbapi.db_find_one('applications', query) - if not data: - logging.debug('not found') - raise gen.Return((False, message.no_auth())) - if value == "approve" or value == "not approved": - logging.debug('check approve') - query['role'] = {"$regex": ".*reviewer.*"} - query['openid'] = user - data = yield dbapi.db_find_one('users', query) - if not data: - logging.debug('not found') - raise gen.Return((False, message.no_auth())) - raise gen.Return((True, {})) diff --git a/cvp/opnfv_testapi/resources/test_models.py b/cvp/opnfv_testapi/resources/test_models.py deleted file mode 100644 index 3829cd6d..00000000 --- a/cvp/opnfv_testapi/resources/test_models.py +++ /dev/null @@ -1,90 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - -from datetime import datetime - - -@swagger.model() -class TestCreateRequest(models.ModelBase): - """ - @property trust_indicator: - @ptype trust_indicator: L{TI} - """ - def __init__(self, - _id=None, - owner=None, - results=[], - public="false", - review="false", - status="private", - shared=[]): - self._id = _id - self.owner = owner - self.results = results.copy() - self.public = public - self.review = review - self.upload_date = datetime.now() - self.status = status - self.shared = shared - - -class ResultUpdateRequest(models.ModelBase): - """ - @property trust_indicator: - @ptype trust_indicator: L{TI} - """ - def __init__(self, trust_indicator=None): - self.trust_indicator = trust_indicator - - -@swagger.model() -class Test(models.ModelBase): - """ - @property trust_indicator: used for long duration test case - @ptype trust_indicator: L{TI} - """ - def __init__(self, - _id=None, - owner=None, - results=[], - public="false", - review="false", - status="private", - shared=[], - filename="", - label="", - sut_label="", - trust_indicator=None): - self._id = _id - self.owner = owner - self.results = results - self.public = public - self.review = review - self.upload_date = datetime.now() - self.status = status - self.shared = shared - self.filename = filename - self.label = label - self.sut_label = sut_label - - -@swagger.model() -class Tests(models.ModelBase): - """ - @property tests: - @ptype tests: C{list} of L{Test} - """ - def __init__(self): - self.tests = list() - - @staticmethod - def attr_parser(): - return {'tests': Test} diff --git a/cvp/opnfv_testapi/resources/testcase_handlers.py b/cvp/opnfv_testapi/resources/testcase_handlers.py deleted file mode 100644 index 9399326f..00000000 --- a/cvp/opnfv_testapi/resources/testcase_handlers.py +++ /dev/null @@ -1,103 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## - -from opnfv_testapi.resources import handlers -from opnfv_testapi.resources import testcase_models -from opnfv_testapi.tornado_swagger import swagger - - -class GenericTestcaseHandler(handlers.GenericApiHandler): - def __init__(self, application, request, **kwargs): - super(GenericTestcaseHandler, self).__init__(application, - request, - **kwargs) - self.table = self.db_testcases - self.table_cls = testcase_models.Testcase - - -class TestcaseCLHandler(GenericTestcaseHandler): - @swagger.operation(nickname="listAllTestCases") - def get(self, project_name): - """ - @description: list all testcases of a project by project_name - @return 200: return all testcases of this project, - empty list is no testcase exist in this project - @rtype: L{TestCases} - """ - self._list(query={'project_name': project_name}) - - @swagger.operation(nickname="createTestCase") - def post(self, project_name): - """ - @description: create a testcase of a project by project_name - @param body: testcase to be created - @type body: L{TestcaseCreateRequest} - @in body: body - @rtype: L{CreateResponse} - @return 200: testcase is created in this project. - @raise 403: project not exist - or testcase already exists in this project - @raise 400: body or name not provided - """ - def project_query(): - return {'name': project_name} - - def testcase_query(): - return {'project_name': project_name, - 'name': self.json_args.get('name')} - miss_fields = ['name'] - carriers = [(self.db_projects, project_query)] - self._create(miss_fields=miss_fields, - carriers=carriers, - query=testcase_query, - project_name=project_name) - - -class TestcaseGURHandler(GenericTestcaseHandler): - @swagger.operation(nickname='getTestCaseByName') - def get(self, project_name, case_name): - """ - @description: get a single testcase - by case_name and project_name - @rtype: L{Testcase} - @return 200: testcase exist - @raise 404: testcase not exist - """ - query = dict() - query['project_name'] = project_name - query["name"] = case_name - self._get_one(query=query) - - @swagger.operation(nickname="updateTestCaseByName") - def put(self, project_name, case_name): - """ - @description: update a single testcase - by project_name and case_name - @param body: testcase to be updated - @type body: L{TestcaseUpdateRequest} - @in body: body - @rtype: L{Project} - @return 200: update success - @raise 404: testcase or project not exist - @raise 403: new testcase name already exist in project - or nothing to update - """ - query = {'project_name': project_name, 'name': case_name} - db_keys = ['name', 'project_name'] - self._update(query=query, db_keys=db_keys) - - @swagger.operation(nickname='deleteTestCaseByName') - def delete(self, project_name, case_name): - """ - @description: delete a testcase by project_name and case_name - @return 200: delete success - @raise 404: testcase not exist - """ - query = {'project_name': project_name, 'name': case_name} - self._delete(query=query) diff --git a/cvp/opnfv_testapi/resources/testcase_models.py b/cvp/opnfv_testapi/resources/testcase_models.py deleted file mode 100644 index 2379dfc4..00000000 --- a/cvp/opnfv_testapi/resources/testcase_models.py +++ /dev/null @@ -1,95 +0,0 @@ -############################################################################## -# Copyright (c) 2015 Orange -# guyrodrigue.koffi@orange.com / koffirodrigue@gmail.com -# All rights reserved. This program and the accompanying materials -# are made available under the terms of the Apache License, Version 2.0 -# which accompanies this distribution, and is available at -# http://www.apache.org/licenses/LICENSE-2.0 -############################################################################## -from opnfv_testapi.resources import models -from opnfv_testapi.tornado_swagger import swagger - - -@swagger.model() -class TestcaseCreateRequest(models.ModelBase): - def __init__(self, name, url=None, description=None, - catalog_description=None, tier=None, ci_loop=None, - criteria=None, blocking=None, dependencies=None, run=None, - domains=None, tags=None, version=None): - self.name = name - self.url = url - self.description = description - self.catalog_description = catalog_description - self.tier = tier - self.ci_loop = ci_loop - self.criteria = criteria - self.blocking = blocking - self.dependencies = dependencies - self.run = run - self.domains = domains - self.tags = tags - self.version = version - self.trust = "Silver" - - -@swagger.model() -class TestcaseUpdateRequest(models.ModelBase): - def __init__(self, name=None, description=None, project_name=None, - catalog_description=None, tier=None, ci_loop=None, - criteria=None, blocking=None, dependencies=None, run=None, - domains=None, tags=None, version=None, trust=None): - self.name = name - self.description = description - self.catalog_description = catalog_description - self.project_name = project_name - self.tier = tier - self.ci_loop = ci_loop - self.criteria = criteria - self.blocking = blocking - self.dependencies = dependencies - self.run = run - self.domains = domains - self.tags = tags - self.version = version - self.trust = trust - - -@swagger.model() -class Testcase(models.ModelBase): - def __init__(self, _id=None, name=None, project_name=None, - description=None, url=None, creation_date=None, - catalog_description=None, tier=None, ci_loop=None, - criteria=None, blocking=None, dependencies=None, run=None, - domains=None, tags=None, version=None, - trust=None): - self._id = None - self.name = None - self.project_name = None - self.description = None - self.catalog_description = None - self.url = None - self.creation_date = None - self.tier = None - self.ci_loop = None - self.criteria = None - self.blocking = None - self.dependencies = None - self.run = None - self.domains = None - self.tags = None - self.version = None - self.trust = None - - -@swagger.model() -class Testcases(models.ModelBase): - """ - @property testcases: - @ptype testcases: C{list} of L{Testcase} - """ - def __init__(self): - self.testcases = list() - - @staticmethod - def attr_parser(): - return {'testcases': Testcase} |