functest/core/cloudify.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2018 Orange and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 """Cloudify testcase implementation."""
  11
  12 from __future__ import division
  13
  14 import logging
  15 import time
  16
  17 from cloudify_rest_client import CloudifyClient
  18
  19 from functest.core import singlevm
  20
  21
  22 class Cloudify(singlevm.SingleVm2):
  23     """Cloudify Orchestrator Case."""
  24
  25     __logger = logging.getLogger(__name__)
  26
  27     filename = ('/home/opnfv/functest/images/'
  28                 'cloudify-manager-premium-4.0.1.qcow2')
  29     flavor_ram = 4096
  30     flavor_vcpus = 2
  31     flavor_disk = 40
  32     username = 'centos'
  33     ssh_connect_loops = 12
  34     create_server_timeout = 600
  35     ports = [80, 443, 5671, 53333]
  36
  37     def __init__(self, **kwargs):
  38         """Initialize Cloudify testcase object."""
  39         if "case_name" not in kwargs:
  40             kwargs["case_name"] = "cloudify"
  41         super(Cloudify, self).__init__(**kwargs)
  42         self.cfy_client = None
  43
  44     def prepare(self):
  45         super(Cloudify, self).prepare()
  46         for port in self.ports:
  47             self.cloud.create_security_group_rule(
  48                 self.sec.id, port_range_min=port, port_range_max=port,
  49                 protocol='tcp', direction='ingress')
  50
  51     def execute(self):
  52         """
  53         Deploy Cloudify Manager.
  54         """
  55         self.cfy_client = CloudifyClient(
  56             host=self.fip.floating_ip_address,
  57             username='admin', password='admin', tenant='default_tenant',
  58             api_version='v3')
  59         self.__logger.info("Attemps running status of the Manager")
  60         for loop in range(10):
  61             try:
  62                 self.__logger.debug(
  63                     "status %s", self.cfy_client.manager.get_status())
  64                 cfy_status = self.cfy_client.manager.get_status()['status']
  65                 self.__logger.info(
  66                     "The current manager status is %s", cfy_status)
  67                 if str(cfy_status) != 'running':
  68                     raise Exception("Cloudify Manager isn't up and running")
  69                 self.cfy_client.secrets.create("foo", "bar")
  70                 self.__logger.debug(
  71                     "List secrets: %s", self.cfy_client.secrets.list())
  72                 self.cfy_client.secrets.delete("foo")
  73                 self.__logger.info("Secrets API successfully reached")
  74                 break
  75             except Exception:  # pylint: disable=broad-except
  76                 self.__logger.info(
  77                     "try %s: Cloudify Manager isn't up and running", loop + 1)
  78                 time.sleep(30)
  79         else:
  80             self.__logger.error("Cloudify Manager isn't up and running")
  81             return 1
  82         self.__logger.info("Cloudify Manager is up and running")
  83         return 0