functest/opnfv_tests/openstack/refstack/refstack.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 Huawei Technologies Co.,Ltd 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 """Refstack testcase implementation."""
  11
  12 import logging
  13 import os
  14 import re
  15 import subprocess
  16 import yaml
  17
  18 from functest.opnfv_tests.openstack.tempest import tempest
  19 from functest.utils import config
  20
  21
  22 class Refstack(tempest.TempestCommon):
  23     """Refstack testcase implementation class."""
  24
  25     __logger = logging.getLogger(__name__)
  26
  27     def _extract_refstack_data(self, refstack_list):
  28         yaml_data = ""
  29         with open(refstack_list, encoding='utf-8') as def_file:
  30             for line in def_file:
  31                 try:
  32                     grp = re.search(r'^([^\[]*)(\[.*\])\n*$', line)
  33                     yaml_data = f"{yaml_data}\n{grp.group(1)}: {grp.group(2)}"
  34                 except Exception:  # pylint: disable=broad-except
  35                     self.__logger.warning("Cannot parse %s", line)
  36         return yaml.full_load(yaml_data)
  37
  38     def _extract_tempest_data(self):
  39         olddir = os.getcwd()
  40         try:
  41             os.chdir(self.verifier_repo_dir)
  42             cmd = ['stestr', 'list', '^tempest.']
  43             output = subprocess.check_output(cmd)
  44         except subprocess.CalledProcessError as cpe:
  45             self.__logger.error(
  46                 "Exception when listing tempest tests: %s\n%s",
  47                 cpe.cmd, cpe.output.decode("utf-8"))
  48             raise
  49         finally:
  50             os.chdir(olddir)
  51         yaml_data2 = ""
  52         for line in output.splitlines():
  53             try:
  54                 grp = re.search(r'^([^\[]*)(\[.*\])\n*$', line.decode("utf-8"))
  55                 yaml_data2 = f"{yaml_data2}\n{grp.group(1)}: {grp.group(2)}"
  56             except Exception:  # pylint: disable=broad-except
  57                 self.__logger.warning("Cannot parse %s. skipping it", line)
  58         return yaml.full_load(yaml_data2)
  59
  60     def generate_test_list(self, **kwargs):
  61         refstack_list = os.path.join(
  62             getattr(config.CONF, 'dir_refstack_data'),
  63             f"{kwargs.get('target', 'compute')}.txt")
  64         self.backup_tempest_config(self.conf_file, '/etc')
  65         refstack_data = self._extract_refstack_data(refstack_list)
  66         tempest_data = self._extract_tempest_data()
  67         with open(self.list, 'w', encoding='utf-8') as ref_file:
  68             for key in refstack_data.keys():
  69                 try:
  70                     for data in tempest_data[key]:
  71                         if data == refstack_data[key][0]:
  72                             break
  73                     else:
  74                         self.__logger.info("%s: ids differ. skipping it", key)
  75                         continue
  76                     value = str(tempest_data[key]).replace(
  77                         "'", "").replace(", ", ",")
  78                     ref_file.write(f"{key}{value}\n")
  79                 except Exception:  # pylint: disable=broad-except
  80                     self.__logger.info("%s: not found. skipping it", key)
  81                     continue