aboutsummaryrefslogtreecommitdiffstats
path: root/services/api/api_virtual_function.py
diff options
context:
space:
mode:
authorLovett, Trevor <trevor.lovett@att.com>2020-03-03 14:04:39 -0600
committersteven stark <steven.stark@att.com>2020-03-03 21:49:31 +0000
commitd7d8722ce27e308defb6764d8d76f85ce7d63927 (patch)
tree701e5aee2665bbde9298ea597e8f88355ec786d5 /services/api/api_virtual_function.py
parentca6342d22a8eb94436287efb1c9276e1d307b2e9 (diff)
[VVP] Removing outdated test-engine code
Issue-ID: VVP-381 Signed-off-by: Lovett, Trevor <trevor.lovett@att.com> Change-Id: I8622794a24deb83e14334e3454194914b835d37d
Diffstat (limited to 'services/api/api_virtual_function.py')
-rw-r--r--services/api/api_virtual_function.py403
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&amp;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