--- /dev/null
+# Copyright 2019 AT&T Intellectual Property. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ONAPLibrary.BaseAAIKeywords import BaseAAIKeywords
+from ONAPLibrary.robotlibcore import HybridCore
+
+
+class AAI(HybridCore):
+ """The main interface for interacting with A&AI. It handles low level stuff like managing the http request
+ library and A&AI required fields """
+
+ ROBOT_LIBRARY_SCOPE = "GLOBAL"
+
+ def __init__(self):
+ self.keyword_implementors = [
+ BaseAAIKeywords()
+ ]
+ HybridCore.__init__(self, self.keyword_implementors)
--- /dev/null
+# Copyright 2019 AT&T Intellectual Property. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from robot.api import logger
+from robot.api.deco import keyword
+from robot.libraries.BuiltIn import BuiltIn
+import time
+
+from ONAPLibrary.RequestsHelper import RequestsHelper
+
+
+class BaseAAIKeywords(object):
+ """The main interface for interacting with AAI. It handles low level stuff like managing the http request library
+ and required fields. """
+
+ def __init__(self):
+ super(BaseAAIKeywords, self).__init__()
+ self.reqs = RequestsHelper()
+ self.builtin = BuiltIn()
+
+ @keyword
+ def run_get_request(self, endpoint, data_path, accept="application/json", auth=None):
+ """Runs an AAI get request"""
+ resp = self.reqs.get_request("aai", endpoint, data_path, sdc_user=None, accept=accept, auth=auth)
+ self.builtin.should_be_equal_as_strings(resp.status_code, "200")
+ return resp
+
+ @keyword
+ def run_post_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an AAI post request"""
+ return self.reqs.post_request("aai", endpoint, data_path, data, sdc_user=None, files=None,
+ accept=accept, auth=auth)
+
+ @keyword
+ def run_put_request(self, endpoint, data_path, data, accept="application/json", auth=None):
+ """Runs an AAI post request"""
+ return self.reqs.put_request("aai", endpoint, data_path, data, sdc_user=None, accept=accept, auth=auth)
+
+ @keyword
+ def run_delete_request(self, endpoint, data_path, resource_version, accept="application/json", auth=None):
+ """Runs an AAI delete request"""
+ return self.reqs.delete_request("aai", endpoint, data_path + '?resource-version=' + resource_version, data=None,
+ sdc_user=None, accept=accept, auth=auth)
+
+ @keyword
+ def wait_for_node_to_exist(self, endpoint, search_node_type, key, uuid, auth=None):
+ logger.info('Waiting for AAI traversal to complete...')
+ for i in range(30):
+ time.sleep(1)
+ result = self.find_node(endpoint, search_node_type, key, uuid, auth=auth)
+ if result:
+ return result
+
+ error_message = "AAI traversal didn't finish in 30 seconds. Something is wrong. Type {0}, UUID {1}".format(
+ search_node_type, uuid)
+ logger.error(error_message)
+ self.builtin.fail(error_message)
+
+ @keyword
+ def find_node(self, endpoint, search_node_type, key, node_uuid, auth=None):
+ data_path = '/aai/v11/search/nodes-query?search-node-type={0}&filter={1}:EQUALS:{2}'.format(
+ search_node_type, key, node_uuid)
+ resp = self.reqs.get_request("aai", endpoint, data_path, accept="application/json", auth=auth)
+ response = resp.json()
+ return 'result-data' in response
import time
from vcpeutils.csar_parser import *
-import requests
from robot.api import logger
from datetime import datetime
-import urllib3
import sys
from ONAPLibrary.PreloadSDNCKeywords import PreloadSDNCKeywords
from ONAPLibrary.RequestSOKeywords import RequestSOKeywords
+from ONAPLibrary.BaseAAIKeywords import BaseAAIKeywords
+from ONAPLibrary.UUIDKeywords import UUIDKeywords
class SoUtils:
self.global_subscriber_id = 'Demonstration'
self.vgw_VfModuleModelInvariantUuid = '26d6a718-17b2-4ba8-8691-c44343b2ecd2'
self.so = RequestSOKeywords()
+ self.aai = BaseAAIKeywords()
+ self.uuid = UUIDKeywords()
@staticmethod
def add_req_info(req_details, instance_name, product_family_id=None):
auth=self.so_userpass)
return done
- def create_entire_service(self, csar_file, vnf_template_file, preload_dict, name_suffix, region_name, tenant_id,
- ssh_key):
+ def create_entire_service(self, csar_file, vnf_template_file, preload_dict, region_name, tenant_id, ssh_key):
"""
:param csar_file:
:param vnf_template_file:
:param preload_dict:
- :param name_suffix:
:param region_name:
:param tenant_id
:param ssh_key
:return: service instance UUID
"""
+
+ name_suffix = str(self.uuid.generate_timestamp())
+
logger.info('\n----------------------------------------------------------------------------------')
logger.info('Start to create entire service defined in csar: {0}'.format(csar_file))
parser = CsarParser()
return None
# wait for AAI to complete traversal
- self.wait_for_aai('service', svc_instance_id)
+ endpoint = "https://{0}:{1}".format(self.aai_host, self.aai_query_port)
+ self.aai.wait_for_node_to_exist(endpoint, 'service-instance', 'service-instance-id', svc_instance_id)
# create networks
for model in parser.net_models:
if not vnf_instance_id:
logger.error('No VNF instance ID returned!')
sys.exit()
- self.wait_for_aai('vnf', vnf_instance_id)
+ endpoint = "https://{0}:{1}".format(self.aai_host, self.aai_query_port)
+ self.aai.wait_for_node_to_exist(endpoint, 'generic-vnf', 'vnf-id', vnf_instance_id)
# SDNC Preload
preloader = PreloadSDNCKeywords()
return svc_instance_id
- def wait_for_aai(self, node_type, uuid):
- logger.info('Waiting for AAI traversal to complete...')
- for i in range(30):
- time.sleep(1)
- if self.is_node_in_aai(node_type, uuid):
- return
-
- logger.error("AAI traversal didn't finish in 30 seconds. Something is wrong. Type {0}, UUID {1}".format(
- node_type, uuid))
- sys.exit()
-
- def is_node_in_aai(self, node_type, node_uuid):
- key = None
- search_node_type = None
- if node_type == 'service':
- search_node_type = 'service-instance'
- key = 'service-instance-id'
- elif node_type == 'vnf':
- search_node_type = 'generic-vnf'
- key = 'vnf-id'
- else:
- logging.error('Invalid node_type: ' + node_type)
- sys.exit()
-
- url = 'https://{0}:{1}/aai/v11/search/nodes-query?search-node-type={2}&filter={3}:EQUALS:{4}'.format(
- self.aai_host, self.aai_query_port, search_node_type, key, node_uuid)
-
- headers = {'Content-Type': 'application/json', 'Accept': 'application/json', 'X-FromAppID': 'vCPE-Robot', 'X-TransactionId': 'get_aai_subscr'}
- urllib3.disable_warnings()
- r = requests.get(url, headers=headers, auth=self.aai_userpass, verify=False)
- response = r.json()
- logger.debug('aai query: ' + url)
- logger.debug('aai response:\n' + json.dumps(response, indent=4, sort_keys=True))
- return 'result-data' in response
-
@staticmethod
def network_name_to_subnet_name(network_name):
"""
Code Review / testsuite / python-testing-utils.git / commitdiff