diff options
Diffstat (limited to 'services/api/api_user.py')
| -rw-r--r-- | services/api/api_user.py | 308 |
1 files changed, 308 insertions, 0 deletions
diff --git a/services/api/api_user.py b/services/api/api_user.py new file mode 100644 index 0000000..3e38fd2 --- /dev/null +++ b/services/api/api_user.py @@ -0,0 +1,308 @@ + +# ============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 re + +from django.conf import settings +import requests + +from services.api.api_bridge import APIBridge +from services.api.api_gitlab import APIGitLab +from services.constants import Constants, ServiceProvider +from services.helper import Helper +from services.logging_service import LoggingServiceFactory + + +logger = LoggingServiceFactory.get_logger() + +class APIUser: + + @staticmethod + # Update account API - only adds new SSH key! + def update_account(user_content): + r1 = None + token = APIUser.login_user(user_content['email']) + user_content['session_token'] = 'token ' + token + sshKey = Helper.generate_sshpub_key() + putURL = settings.ICE_EM_URL + '/v1/engmgr/users/account' + logger.debug("Put user URL: " + putURL) + headers = dict() # Create header for put request. + headers['Content-type'] = 'application/json' + headers['Authorization'] = user_content['session_token'] + # headers['Authorization'] = user_content['activation_token'] + put_data = dict() # Create JSON data for put request. + user_content['vendor'] = user_content['company']['name'] + if user_content['vendor'] == "AT&T": + put_data['company'] = "AT&T" + else: + put_data['company'] = user_content['vendor'] + put_data['email'] = user_content['email'] + put_data['full_name'] = user_content['full_name'] + put_data['password'] = "" + put_data['phone_number'] = "+1201" + \ + Helper.rand_string("randomNumber", 6) + put_data['ssh_key'] = sshKey + try: + r1 = requests.put( + putURL, json=put_data, headers=headers, verify=False) + Helper.internal_assert(r1.status_code, 200) + logger.debug( + "SSH Key was added successfully to user " + user_content['full_name']) + if not APIBridge.is_gitlab_ready(user_content): + raise + return sshKey + except: + if r1 is None: + logger.error("Failed to add public SSH key to user.") + else: + logger.error("PUT request failed to add SSH key to user, see response >>> %s %s \n %s" % ( + r1.status_code, r1.reason, str(r1.content, 'utf-8'))) + raise + + @staticmethod + # Update account API - only adds new SSH key! + def update_account_injec_script(user_content): + r1 = None + putURL = settings.ICE_EM_URL + '/v1/engmgr/users/account' + logger.debug("Put user URL: " + putURL) + headers = dict() # Create header for put request. + headers['Content-type'] = 'application/json' + headers['Authorization'] = user_content['session_token'] + put_data = dict() # Create JSON data for put request. + if user_content['vendor'] == "AT&T": + put_data['company'] = "AT&T" + else: + put_data['company'] = user_content['vendor'] + put_data['email'] = user_content['email'] + script = "<script>;</script>" + put_data['full_name'] = script + put_data['password'] = "" + put_data['phone_number'] = "+1201" + \ + Helper.rand_string("randomNumber", 6) + try: + r1 = requests.put( + putURL, json=put_data, headers=headers, verify=False) + Helper.internal_assert(r1.status_code, 200) + msg = "Testing for Cross site scripting successfully : " + \ + user_content['full_name'] + \ + "stattus Code = " + str(r1.status_code) + logger.debug(msg) + if not APIBridge.is_gitlab_ready(user_content): + raise + return True + except: + if r1 is None: + logger.error("Failed to add public SSH key to user.") + else: + logger.error("PUT request failed to add SSH key to user, see response >>> %s %s \n %s" % ( + r1.status_code, r1.reason, str(r1.content, 'utf-8'))) + raise + + @staticmethod + def create_new_user(company=False, activate=False): + signupUrl = settings.EM_REST_URL + "signup/" + signupParams = APIUser.create_signup_param(company) + r1 = requests.post(signupUrl, json=signupParams, verify=False) + if (r1.status_code == 201 or r1.status_code == 200): + logger.debug("Moving to next test case. status=" + + str(r1.status_code)) + pass # Need to break here. + try: + user_data = r1.json() + except Exception as e: + logger.error("=========== json error ========") + logger.error("r1.content = " + str(r1.content) + + ", status=" + str(r1.status_code)) + logger.error("=========== json error end ========") + raise e + if activate: + APIUser.activate_user( + user_data['uuid'], user_data["user"]["activation_token"]) + return user_data + + @staticmethod + def login_user(email): + postUrl = settings.EM_REST_URL + "login" + user_data = dict() # Create JSON data for post request. + user_data['email'] = email + user_data['password'] = "iceusers" + try: + headers = {'Content-type': 'application/json'} + r = requests.post( + postUrl, json=user_data, headers=headers, verify=False) + logger.debug(str(r.status_code) + " " + r.reason) + decoded_response = r.json() + return decoded_response['token'] + except: + logger.debug("Failed to login.") + raise + + @staticmethod + def set_ssh(user_content, sshKey=None): # Set SSH key to user. + r1 = None + if sshKey is None: + logger.debug("About to generate an ssh key for the user: %s" % + user_content['email']) + sshKey = Helper.generate_sshpub_key() + postURL = settings.ICE_EM_URL + '/v1/engmgr/users/ssh' + logger.debug("Post user URL: " + postURL) + headers = dict() # Create header for post request. + headers['Content-type'] = 'application/json' + headers['Authorization'] = user_content['session_token'] + post_data = dict() # Create JSON data for post request. + post_data['ssh_key'] = sshKey + try: + r1 = requests.post( + postURL, json=post_data, headers=headers, verify=False) + Helper.internal_assert(r1.status_code, 200) + logger.debug( + "SSH Key was added successfully") + if not APIBridge.is_gitlab_ready(user_content): + raise + return sshKey + except: + if r1 is None: + logger.error("Failed to add public SSH key.") + else: + logger.error( + "POST request failed to add SSH key to user, see response >>> %s %s" % (r1.status_code, r1.reason)) + raise + + @staticmethod + def create_signup_param(company=False): + try: # Click on element in UI, by xPath locator. + if not company: + company = ServiceProvider.MainServiceProvider + email_domain = ServiceProvider.email + else: + email_domain = company.lower() + ".com" # + data = { + "company": company, + "full_name": Helper.rand_string("randomString"), + "email": Helper.rand_string("randomString") + "@" + email_domain, + "phone_number": Constants.Default.Phone.TEXT, + "password": Constants.Default.Password.TEXT, + "regular_email_updates": "True" + } + + return data + # If failed - count the failure and add the error to list of errors. + except Exception as e: + errorMsg = "Could not create Sign Up parametrs" + raise Exception(errorMsg, e) + + @staticmethod + def activate_user(userUuid, activationToken): + postUrl = settings.ICE_EM_URL + "/v1/engmgr/users/activate/" + \ + userUuid + "/" + activationToken + logger.debug(postUrl) + r1 = requests.get(postUrl, verify=False) + if (r1.status_code == 200): + logger.debug("APIUser activated successfully!") + return True + else: + raise Exception( + "Failed to activate user >>> %s %s" % (r1.status_code, r1.reason)) + + @staticmethod + def signup_invited_user(company, invited_email, invite_token, invite_url, user_content, is_contact_user="false", activate=False, wait_for_gitlab=True): + r1 = None + postURL = settings.ICE_EM_URL + '/v1/engmgr/signup' + logger.debug("Post signup URL: " + postURL) + if is_contact_user == "true": + fullName = re.sub("http.*full_name=", "", invite_url) + fullName = re.sub("&.*", "", fullName) + logger.debug( + "Invited contact full name is (according to url): " + fullName) + else: + fullName = Helper.rand_string('randomString') + + post_data = dict() # Create JSON data for post request. + post_data['company'] = company + post_data['email'] = invited_email + post_data['full_name'] = fullName + post_data['invitation'] = invite_token + post_data['is_contact_user'] = is_contact_user + post_data['password'] = "iceusers" + post_data['phone_number'] = "+1201" + \ + Helper.rand_string("randomNumber", 6) + post_data['regular_email_updates'] = "False" + post_data['terms'] = "True" + try: + requests.get(invite_url, verify=False) + r1 = requests.post( + postURL, json=post_data, verify=False) + Helper.internal_assert(r1.status_code, 200) + logger.debug("Invited user signed-up successfully!") + + user_data = r1.json() + if activate: + APIUser.activate_user( + user_data['uuid'], user_data["user"]["activation_token"]) + + if wait_for_gitlab: + if not APIBridge.is_gitlab_ready(user_content): + raise + return post_data + except: + if r1 is None: + logger.error("Failed to sign up the invited team member.") + else: + logger.error("POST request failed to sign up the invited team member, see response >>> %s %s \n %s" % ( + r1.status_code, r1.reason, str(r1.content, 'utf-8'))) + raise + + @staticmethod + def create_new_user_content(): + user_content = APIBridge.create_engagement() + APIGitLab.git_clone_push(user_content) + APIBridge.frontend_login( + user_content['email'], Constants.Default.Password.TEXT) + vfName = user_content['vfName'] + uuid = user_content['uuid'] + inviteEmail = Helper.rand_invite_email() + newObj = [vfName, uuid, inviteEmail] + return newObj, user_content + + @staticmethod + def create_new_user_content_login_with_api(): + user_content = APIBridge.create_engagement() + APIGitLab.git_clone_push(user_content) + token = "token " + APIBridge.login_user(user_content['el_email']) + user_content['session_token'] = token + return user_content |