diff options
Diffstat (limited to 'services/api/api_virtual_function.py')
| -rw-r--r-- | services/api/api_virtual_function.py | 403 |
1 files changed, 0 insertions, 403 deletions
diff --git a/services/api/api_virtual_function.py b/services/api/api_virtual_function.py deleted file mode 100644 index 193d6e2..0000000 --- a/services/api/api_virtual_function.py +++ /dev/null @@ -1,403 +0,0 @@ - -# ============LICENSE_START========================================== -# org.onap.vvp/test-engine -# =================================================================== -# Copyright © 2017 AT&T Intellectual Property. All rights reserved. -# =================================================================== -# -# Unless otherwise specified, all software contained herein is licensed -# under the Apache License, Version 2.0 (the “License”); -# you may not use this software 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. -# -# -# -# Unless otherwise specified, all documentation contained herein is licensed -# under the Creative Commons License, Attribution 4.0 Intl. (the “License”); -# you may not use this documentation except in compliance with the License. -# You may obtain a copy of the License at -# -# https://creativecommons.org/licenses/by/4.0/ -# -# Unless required by applicable law or agreed to in writing, documentation -# 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. -# -# ============LICENSE_END============================================ -# -# ECOMP is a trademark and service mark of AT&T Intellectual Property. -import datetime -import json -import time - -from django.conf import settings -import requests - -from services.api.api_user import APIUser -from services.constants import Constants, ServiceProvider -from services.database.db_general import DBGeneral -from services.helper import Helper -from services.logging_service import LoggingServiceFactory - - -logger = LoggingServiceFactory.get_logger() - - -class APIVirtualFunction: - - @staticmethod - def add_next_step(user_content, files=[]): - r1 = None - postURL = settings.ICE_EM_URL + '/v1/engmgr/engagements/' + \ - user_content['engagement_uuid'] + '/nextsteps' - logger.debug("Post add next step URL: " + postURL) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = user_content['session_token'] - data = dict() # Create JSON data for post request. - files_list = list() - if isinstance(files, list): - for file in files: - files_list.append(file) - else: - files_list.append(files) - data['files'] = files_list - data['assigneesUuids'] = [user_content['uuid']] - data['duedate'] = str(datetime.date.today()) - data['description'] = "API test - add next step." - list_data = [] - list_data.append(data) - try: - r1 = requests.post( - postURL, json=list_data, headers=headers, verify=False) - Helper.internal_assert(r1.status_code, 200) - logger.debug("Next step was added to the engagement!") - ns_uuid = r1.json() - return ns_uuid[0]['uuid'] - except BaseException: - if r1 is None: - logger.error( - "Failed to add next step to VF " + user_content['vfName']) - else: - logger.error("Failed to add next step to VF " + - user_content['vfName'] + - ", see response >>> %s %s.\nContent: %s" % - (r1.status_code, r1.reason, str( - r1.content, 'utf-8'))) - raise - - @staticmethod - def create_vf(token): - r1 = None - postUrl = settings.EM_REST_URL + "vf/" - targetVersion = DBGeneral.select_from( - "uuid", "ice_deployment_target", 1) - ecompRelease = DBGeneral.select_from("uuid", "ice_ecomp_release", 1) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = 'token ' + token - jdata = [ - { - "virtual_function": Helper.rand_string("randomString"), - "version": Helper.rand_string("randomString") + - Helper.rand_string("randomNumber"), - "target_lab_entry_date": time.strftime("%Y-%m-%d"), - "target_aic_uuid": targetVersion, - "ecomp_release": ecompRelease, - "is_service_provider_internal": False}] - try: - r1 = requests.post( - postUrl, json=jdata, headers=headers, verify=False) - Helper.internal_assert(r1.status_code, 200) - logger.debug("Virtual Function created successfully!") - content = r1.content[1:-1] - return content - except BaseException: - if r1 is None: - logger.debug("Failed to create VF >>> request failed!") - else: - logger.debug( - "Failed to create VF >>> %s %s \n %s" % - (r1.status_code, r1.reason, str( - r1.content, 'utf-8'))) - raise - - @staticmethod - def get_engagement(user_content): - r1 = None - postUrl = settings.EM_REST_URL + 'single-engagement/' + \ - str(user_content['engagement_uuid'],) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = user_content['session_token'] - try: - r1 = requests.get( - postUrl, headers=headers, verify=False) - Helper.internal_assert(r1.status_code, 200) - logger.debug("Retrieved the Engagement successfully!") - content = r1.content - return json.loads(content) - except BaseException: - if r1 is None: - logger.debug( - "Failed to Retrieve the Engagement >>> request failed!") - else: - logger.debug( - "Failed to Retrieve the Engagement >>> %s %s \n %s" % - (r1.status_code, r1.reason, str( - r1.content, 'utf-8'))) - raise - - @staticmethod - def invite_team_member(user_content): - r1 = None - postURL = settings.ICE_EM_URL + '/v1/engmgr/invite-team-members/' - logger.debug("Post invite user URL: " + postURL) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = user_content['session_token'] - data = dict() # Create JSON data for post request. - data['email'] = Helper.rand_string( - 'randomString') + "@" + ServiceProvider.email - data['eng_uuid'] = user_content['engagement_uuid'] - list_data = [] - list_data.append(data) - try: - r1 = requests.post( - postURL, json=list_data, headers=headers, verify=False) - Helper.internal_assert_boolean(r1.status_code, 200) - logger.debug("Invite sent successfully to email " + data['email']) - invite_token = DBGeneral.select_where_and( - "invitation_token", - "ice_invitation", - "email", - data['email'], - "engagement_uuid", - user_content['engagement_uuid'], - 1) - invite_url = settings.ICE_PORTAL_URL + "/#/signUp?invitation=" + \ - invite_token + "&email=" + data['email'] - logger.debug("Invitation URL is: " + invite_url) - return data['email'], invite_token, invite_url - except BaseException: - if r1 is None: - logger.error("Failed to invite team member.") - else: - logger.error( - "POST request failed to invite team member, " + - "see response >>> %s %s" % (r1.status_code, r1.reason)) - raise - - @staticmethod - def add_contact(user_content): - r1 = None - postURL = settings.ICE_EM_URL + '/v1/engmgr/add-contact/' - logger.debug("Post invite vendor contact URL: " + postURL) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = user_content['session_token'] - data = dict() # Create JSON data for post request. - data['company'] = user_content['vendor_uuid'] - data['email'] = Helper.rand_string( - 'randomString') + "@" + ServiceProvider.email - data['eng_uuid'] = user_content['engagement_uuid'] - data['full_name'] = Helper.rand_string('randomString') - data['phone_number'] = "+1201" + Helper.rand_string("randomNumber", 6) - try: - r1 = requests.post( - postURL, json=data, headers=headers, verify=False) - Helper.internal_assert_boolean(r1.status_code, 200) - logger.debug("Invite sent successfully to email " + data['email']) - invite_token = DBGeneral.select_where_and( - "invitation_token", - "ice_invitation", - "email", - data['email'], - "engagement_uuid", - user_content['engagement_uuid'], - 1) - invite_url = settings.ICE_PORTAL_URL + "/#/signUp?invitation=" +\ - invite_token + "&email=" + data['email'] +\ - "&full_name=" + data['full_name'] + \ - "&phone_number=" + data['phone_number'] + "&company=" + \ - data['company'] + "&is_contact_user=true" - logger.debug("Invitation URL is: " + invite_url) - return data['email'], invite_token, invite_url - except BaseException: - if r1 is None: - logger.error("Failed to invite vendor contact.") - else: - logger.error( - "POST request failed to invite vendor contact, " + - "see response >>> %s %s \n %s" % - (r1.status_code, r1.reason, str( - r1.content, 'utf-8'))) - raise - - @staticmethod - def edit_next_step(user_content, ns_uuid): - r1 = None - postURL = settings.ICE_EM_URL + '/v1/engmgr/nextsteps/' + ns_uuid + \ - '/engagement/' + user_content['engagement_uuid'] + '/modify/' - logger.debug("Put next step URL: " + postURL) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = user_content['session_token'] - data = dict() # Create JSON data for post request. - data['files'] = [] - data['assigneesUuids'] = [user_content['uuid']] - data['duedate'] = str(datetime.date.today()) - data['description'] = "API edit next step test " + \ - Helper.rand_string('randomString') - try: - r1 = requests.put( - postURL, json=data, headers=headers, verify=False) - Helper.internal_assert_boolean(r1.status_code, 202) - logger.debug("Next step was edited successfully!") - except BaseException: - if r1 is None: - logger.error("Failed to edit next step uuid: " + ns_uuid) - else: - logger.error( - "Failed to edit next step uuid: " + - ns_uuid + - ", see response >>> %s %s" % - (r1.status_code, - r1.reason)) - raise - - @staticmethod - def get_export_dasboard_excel(token, keywords=""): - postUrl = settings.EM_REST_URL + \ - "engagement/export/?stage=All&keyword=" + keywords - headers = {"Authorization": token} - r1 = requests.get(postUrl, headers=headers, verify=False) - Helper.internal_assert(r1.status_code, 200) - if (r1.status_code == 200): - logger.debug("APIUser activated successfully!") - return r1.content - else: - raise Exception( - "Failed to activate user >>> %s %s" % - (r1.status_code, r1.reason)) - return False - - @staticmethod - def create_engagement(wait_for_gitlab=True): - user_content = APIUser.create_new_user() - APIUser.activate_user( - user_content['uuid'], user_content['user']['activation_token']) - token = APIUser.login_user(user_content['email']) - vf_content = json.loads(APIVirtualFunction.create_vf(token)) - user_content['vfName'] = vf_content['name'] - user_content['vf_uuid'] = vf_content['uuid'] - user_content['target_aic'] = vf_content['deployment_target']['version'] - # <-- ECOMP RELEASE - user_content['ecomp_release'] = vf_content[ - 'ecomp_release']['name'] - user_content['vnf_version'] = vf_content['version'] - if(vf_content['vendor']['name'] == "AT&T"): - user_content['vendor'] = "AT&T" - else: - user_content['vendor'] = vf_content['vendor']['name'] - user_content['vendor_uuid'] = vf_content['vendor']['uuid'] - user_content['engagement_manual_id'] = vf_content[ - 'engagement']['engagement_manual_id'] - user_content['target_lab_entry_date'] = vf_content[ - 'target_lab_entry_date'] - user_content['el_email'] = vf_content[ - 'engagement']['reviewer']['email'] - user_content['el_name'] = vf_content[ - 'engagement']['reviewer']['full_name'] - user_content['pr_email'] = vf_content[ - 'engagement']['peer_reviewer']['email'] - user_content['pr_name'] = vf_content[ - 'engagement']['peer_reviewer']['full_name'] - user_content['engagement_uuid'] = vf_content['engagement']['uuid'] - user_content['session_token'] = 'token ' + token - user_content['engagement'] = vf_content['engagement'] - user_content['vfStage'] = vf_content['engagement']['engagement_stage'] - - return user_content - - @staticmethod - def set_eng_stage(user_content, requested_stage): - token = APIUser.login_user(user_content['el_email']) - r1 = None - putUrl = Constants.Default.URL.Engagement.SingleEngagement.TEXT + \ - user_content['engagement_uuid'] + "/stage/" + str(requested_stage) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = 'token ' + token - try: - r1 = requests.put( - putUrl, headers=headers, verify=False) - Helper.internal_assert(r1.status_code, 202) - logger.debug( - "Engagement stage was successfully changed to " + - str(requested_stage) + - "!") - content = r1.content[1:-1] - return content - except BaseException: - if r1 is None: - logger.debug("Failed to set eng stage >>> request failed!") - else: - logger.debug( - "Failed to set eng stage >>> %s %s \n %s" % - (r1.status_code, r1.reason, str( - r1.content, 'utf-8'))) - raise - - @staticmethod - def update_aic_version(eng_uuid, aic_version_uuid, session_token): - r1 = None - putURL = Constants.Default.URL.Engagement.EngagementOperations.TEXT + \ - eng_uuid + '/deployment-targets/' + aic_version_uuid - logger.debug("Put next step URL: " + putURL) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = session_token - try: - r1 = requests.put(putURL, headers=headers, verify=False) - Helper.internal_assert_boolean(r1.status_code, 200) - logger.debug("AIC version has changed!") - except BaseException: - if r1 is None: - msg = "Failed to edit AIC version" - else: - msg = "Failed to edit AIC version, see response >>> %s %s" % ( - r1.status_code, r1.reason) - raise msg - - @staticmethod - def update_ecomp_release(eng_uuid, ecomp_release_uuid, session_token): - r1 = None - putURL = Constants.Default.URL.Engagement.EngagementOperations.TEXT + \ - eng_uuid + '/ecomp-releases/' + ecomp_release_uuid - logger.debug("Put next step URL: " + putURL) - headers = dict() # Create header for post request. - headers['Content-type'] = 'application/json' - headers['Authorization'] = session_token - try: - r1 = requests.put(putURL, headers=headers, verify=False) - Helper.internal_assert_boolean(r1.status_code, 200) - logger.debug("AIC version has changed!") - except BaseException: - if r1 is None: - msg = "Failed to update ECOMP release" - else: - msg = "Failed to update ECOMP release," +\ - " see response >>> %s %s" % ( - r1.status_code, r1.reason) - raise msg |