Testcases for Edge cloud Environment check

This testcases only specific to OpenStack.

JIRA: AUTO-31

Change-Id: I283ffcfef97320765e7a5d76898aecd1077bf12e
Signed-off-by: Mohankumar Navaneethan <mnavaneethan@mvista.com>

diff --git a/lib/auto/testcase/EdgeCloud/AutoOSPlatCheck.py b/lib/auto/testcase/EdgeCloud/AutoOSPlatCheck.py
new file mode 100644 (file)
index 0000000..5a19109
--- /dev/null
+++ b/lib/auto/testcase/EdgeCloud/AutoOSPlatCheck.py
@@ -0,0 +1,164 @@
+# !/usr/bin/python
+#
+# Copyright (c) 2018 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 os
+import re
+
+import logging
+import json
+import requests
+
+logger = logging.getLogger(__name__)
+
+class OS_env_check:
+   """Edge Clould Basic Env Function definition"""
+
+
+   def __init__(self):
+       """Variable Intitialization"""
+       self.osver = "v2.0"
+       self.imagever = "v2"
+       self.keystone_ver = 'v3'
+       self.tacker_ver = 'v1.0'
+
+   def ping_os_endpoints(self):
+       "Simple ping check to OpenStack endpoint"
+
+       os_auth_url = os.environ.get('OS_AUTH_URL', None)
+       password = os.environ.get('OS_PASSWORD', None)
+       if os_auth_url is None:
+           logger.error("Source the OpenStack credentials first")
+           exit(0)
+       try:
+           if os_auth_url:
+               endpoint_ip = re.search(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', os_auth_url).group()
+               response = os.system("ping -c 1 " + endpoint_ip + ">/dev/null")
+               if response == 0:
+                  return 0
+               else:
+                    logger.error("Cannot talk to the OpenStack endpoint %s\n" % endpoint_ip)
+                    exit(0)
+       except Exception:
+          logger.exception('Errors when verifying connectivity to %s', endpoint_ip)
+       return False
+
+   def fetch_token(self):
+       "Fetch OS_AUTH_TOKEN from OpenStack Service"
+
+       #(e.g.)url = 'http://10.164.16.100:5000/identity/v3/auth/tokens'
+       url = 'http://'+self.endpoint_ip+':5000/'+self.keystone_ver+'/auth/tokens'
+       data = '{"auth":{"identity":{"methods":["password"],"password":{"user":' \
+              '{"domain":{"name":"default"},"name":"admin",' \
+              '"password":"admin"}}},"scope":{"project":' \
+              '{"domain":{"name":"default"},"name":"admin"}}}}'
+       headers = {"Accept": "application/json"}
+       try:
+           response = requests.post(url, headers=headers,  data=data)
+           header_data = (response.headers)
+           token = header_data['X-Subject-Token']
+           response_body = response.content
+       except Exception:
+           logger.error(" Failure: Not able to send API request for creating token")
+       if (response.status_code == 201):
+           response_body = response.content.decode('utf-8')
+           res = json.loads(response_body)
+           admin_user_id= res['token']['user']['id']
+           return response.status_code,token
+
+   def check_os_running_services(self):
+       "Get active/running OpenStack Services"
+
+       url = 'http://' + self.endpoint_ip + ':5000/' + self.keystone_ver + '/auth/tokens'
+       data = '{"auth": {"identity": {"methods": ["password"],"password": \
+               {"user": {"domain": {"name": "default"},"name": "admin",\
+               "password": "admin"}}},\
+               "scope": {"project": {"domain": {"name": "default"},"name": "admin"}}}}'
+       headers = {"Accept": "application/json"}
+       response = requests.post(url, headers=headers, data=data)
+       service = []
+       url_ep = []
+       if (response.status_code == 201):
+           response_body = response.content.decode('utf-8')
+           res = json.loads(response_body)
+           catalogs = res['token']['catalog']
+           for x in catalogs:
+               services = x['name']
+               if x['name'] is not None:
+                   service.append(x['name'])
+               endpoints = x['endpoints']
+               for y in endpoints:
+                   url = y['url']
+                   if y['url'] not in url_ep:
+                      url_ep.append(url)
+       return response.status_code,service,url_ep
+
+   def check_nova_service(self, endpoints, token):
+       """ checks that a simple nova operation works """
+
+       try:
+           nova_url = endpoints.get('nova')
+           url = nova_url+ '/servers/detail'
+           headers = {"Content-Type": "application/json", "X-Auth-Token": token}
+           response = requests.get(url, headers=headers)
+           if (response.status_code == 200):
+              logger.info("Nova service is Active")
+       except Exception as error:
+           logger.error("Nova service is FAILED")
+           raise error
+       return response.status_code
+
+   def check_neutron_service (self, endpoints, token):
+       """ checks that a simple neutron operation works """
+
+       try:
+           neutron_url = endpoints.get('neutron')
+           url = neutron_url +self.osver+'/networks'
+           headers = {"Content-Type": "application/json", "X-Auth-Token": token}
+           response = requests.get(url, headers=headers)
+           if (response.status_code == 200):
+              logger.info("Neutron service is Active")
+       except Exception as error:
+           logger.error("Neutron service is FAILED")
+           raise error
+       return response.status_code
+
+   def check_glance_service(self, endpoints, token):
+       """ checks that a simple glance operation works """
+
+       try:
+           glance_url = endpoints.get('glance')
+           url = glance_url + '/' + self.imagever + '/images'
+           headers = {"Content-Type": "application/json", "X-Auth-Token": token}
+           response = requests.get(url, headers=headers)
+           if (response.status_code == 200):
+              logger.info("Glance:Image service is Active")
+       except Exception as error:
+           logger.error("Glance:Image service is FAILED")
+           raise error
+       return response.status_code
+
+   def check_tacker_service(self, endpoints, token):
+       """ checks that a simple tacker operation works """
+
+       try:
+           if 'tacker' in endpoints.keys():
+               logger.info("Tacker VNF Manager service is running")
+           else:
+               logger.error("No Tacker VNF Manager service is running")
+               return (0)
+           tacker_url = endpoints.get('tacker')
+           url = tacker_url + '/' + self.tacker_ver + '/vnf.json'
+           headers = {"Content-Type": "application/json", "X-Auth-Token": token}
+           response = requests.get(url, headers=headers)
+           if (response.status_code == 200):
+              logger.info("Tacker:VNF Manager has active VNFs")
+       except Exception as error:
+           logger.error("Tacker:No Active VNFs")
+           raise error
+       return response.status_code

diff --git a/lib/auto/testcase/EdgeCloud/AutoOSPlatTest.py b/lib/auto/testcase/EdgeCloud/AutoOSPlatTest.py
new file mode 100644 (file)
index 0000000..ef99ce5
--- /dev/null
+++ b/lib/auto/testcase/EdgeCloud/AutoOSPlatTest.py
@@ -0,0 +1,80 @@
+"""Script to Test the AUTO Edge Cloud OpenStack Services."""
+# !/usr/bin/python
+#
+# Copyright (c) 2018 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
+#fetch_token
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# ###########################################################################
+#                          OPNFV AUTO Edge Cloud Script
+# **** Scripted by Mohankumar Navaneethan  - mnavaneethan@mvista.com ******
+# ###########################################################################
+
+# Testcase 1 : Ping OpenStack Endpoints
+# Testcase 2 : Creation of Auth-Token
+# TestCase 3 : Check OpenStack Active Services
+# TestCase 4 : Check OpenStack Nova Service
+# TestCase 5 : Check OpenStack Neutron Service
+# TestCase 6 : Check OpenStack Glance Service
+# TestCase 7 : Check OpenStack Tacker Service.
+# ###########################################################################
+#
+import logging
+from AutoOSPlatCheck import OS_env_check
+
+
+class Env_check:
+    """Script to Test AUTO Edge Cloud OpenStack Services."""
+    logger = logging.getLogger(__name__)
+    Env_obj = OS_env_check()
+    print("################################################################")
+    print("                    OPNFV AUTO Script             ")
+    print("################################################################")
+    logger.info("Prerequisites OpenStack configuration for AUTO")
+    #########################################################################
+    logger.info("\t1. Ping OpenStack Endpoints")
+    if (Env_obj.ping_endpoints == 0):
+        logger.info("\t\tPing to OpenStack Endpoint is successfull")
+    else:
+        logger.error("\t\tPing to OpenStack Endpoint is NOT successfull")
+
+    logger.info("\t2. Creation of Auth-Token")
+    response_code , token = Env_obj.fetch_token()
+    if (response_code == 201):
+        logger.info("\t\tCreation of Token is successfull")
+    else:
+        logger.error("\t\t  :  Creation of Token is NOT successfull")
+    logger.info("\t3. Check OpenStack Active Services")
+    status, services, endpoint = Env_obj.check_os_running_services()
+    endpoints = dict(zip(services, endpoint))
+    if (status == 201):
+        logger.info("\t\tCheck OpenStack Active Services is successfull")
+    else:
+        logger.error("\t\tCheck OpenStack Active Services is NOT successfull")
+
+    logger.info("\t4. Check OpenStack Nova Service")
+    if (Env_obj.check_nova_service(endpoints, token) == 200):
+        logger.info("\t\tNova service is responsive")
+    else:
+        logger.error("\t\tNova service is NOT responsive")
+
+    logger.info("\t5. Check OpenStack Neutron Service")
+    if (Env_obj.check_neutron_service(endpoints, token) == 200):
+        logger.info("\t\tNeutron service is responsive")
+    else:
+        logger.error("\t\tNeutron service is NOT responsive")
+
+    logger.info("\t6. Check OpenStack Glance Service")
+    if (Env_obj.check_glance_service(endpoints, token) == 200):
+        logger.info("\t\tGlance service is responsive")
+    else:
+        logger.error("\t\tGlance service is NOT responsive")
+
+    logger.info("\t7. Check OpenStack Tacker Service")
+    if (Env_obj.check_glance_service(endpoints, token) == 200):
+        logger.info("\t\tTacker VNF Manager service is responsive")
+    else:
+        logger.error("\t\tTacker VNF Manager is NOT responsive")

diff --git a/lib/auto/testcase/EdgeCloud/__init__.py b/lib/auto/testcase/EdgeCloud/__init__.py
new file mode 100644 (file)
index 0000000..e69de29