list_parser.py

# Copyright (c) 2015 IBM Corp.
#
#    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.
#


import atexit
import logging
import os
import re
import requests
import subprocess
import tempfile


class TestListParser(object):

    """This class is for normalizing test lists to match the tests in the
    current Tempest environment.
    """

    def __init__(self, tempest_dir, insecure=False):
        """
        Initialize the TestListParser.

        :param tempest_dir: Absolute path of the Tempest directory.
        :param insecure: Whether https requests, if any, should be insecure.
        """
        self.logger = logging.getLogger(__name__)
        self.tempest_dir = tempest_dir
        self.insecure = insecure

    def _get_tempest_test_ids(self):
        """This does a 'testr list-tests' or 'stestr list' according to
        Tempest version on the Tempest directory in order to get a list
        of full test IDs for the current Tempest environment. Test ID
        mappings are then formed for these tests.
        """
        if os.path.exists(os.path.join(self.tempest_dir, '.stestr.conf')):
            init_cmd = (os.path.join(self.tempest_dir, 'tools/with_venv.sh'),
                        'stestr', 'init')
            subprocess.Popen(init_cmd, stdout=subprocess.PIPE,
                             cwd=self.tempest_dir)
            cmd = (os.path.join(self.tempest_dir, 'tools/with_venv.sh'),
                   'stestr', 'list')
        else:
            cmd = (os.path.join(self.tempest_dir, 'tools/with_venv.sh'),
                   'testr', 'list-tests')
        process = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                   cwd=self.tempest_dir)
        (stdout, stderr) = process.communicate()

        if process.returncode != 0:
            self.logger.error(stdout)
            self.logger.error(stderr)
            raise subprocess.CalledProcessError(process.returncode,
                                                ' '.join(cmd))
        try:
            return self._form_test_id_mappings(stdout.split('\n'))
        except TypeError:
            return self._form_test_id_mappings(stdout.decode().split('\n'))

    def _form_test_id_mappings(self, test_list):
        """This takes in a list of full test IDs and forms a dict containing
        base test IDs mapped to their attributes. A full test ID also contains
        test attributes such as '[gate,smoke]'
        Ex:
            'tempest.api.test1': '[gate]'
            'tempest.api.test2': ''
            'tempest.api.test3(some_scenario)': '[smoke,gate]'

        :param test_list: List of full test IDs
        """
        test_mappings = {}
        for testcase in test_list:
            if testcase.startswith("tempest"):
                # Search for any strings like '[smoke, gate]' in the test ID.
                match = re.search('(\[.*\])', testcase)

                if match:
                    testcase = re.sub('\[.*\]', '', testcase)
                    test_mappings[testcase] = match.group(1)
                else:
                    test_mappings[testcase] = ""
        return test_mappings

    def _get_base_test_ids_from_list_file(self, list_location):
        """This takes in a test list file and finds all the base test IDs
        for the tests listed.
        Ex:
            'tempest.test1[gate,id-2]' -> 'tempest.test1'
            'tempest.test2[gate,id-3](scenario)' -> 'tempest.test2(scenario)'

        :param list_location: file path or URL location of list file
        """
        try:
            response = requests.get(list_location,
                                    verify=not self.insecure)
            testcase_list = response.text.split('\n')
            test_mappings = self._form_test_id_mappings(testcase_list)
        # If the location isn't a valid URL, we assume it is a file path.
        except requests.exceptions.MissingSchema:
            try:
                with open(list_location) as data_file:
                    testcase_list = [line.rstrip('\n') for line in data_file]
                test_mappings = self._form_test_id_mappings(testcase_list)
            except Exception:
                self.logger.error("Error reading the passed in test list " +
                                  "file.")
                raise
        except Exception:
            self.logger.error("Error reading the passed in test list file.")
            raise

        return list(test_mappings.keys())

    def _get_full_test_ids(self, tempest_ids, base_ids):
        """This will remake the test ID list with the full IDs of the current
        Tempest environment. The Tempest test ID dict should have the correct
        mappings.

        :param tempest_ids: dict containing test ID mappings
        :param base_ids: list containing base test IDs
        """
        test_list = []
        for test_id in base_ids:
            try:
                attr = tempest_ids[test_id]
                # If the test has a scenario in the test ID, but also has some
                # additional attributes, the attributes need to go before the
                # scenario.
                if '(' in test_id and attr:
                    components = test_id.split('(', 1)
                    test_portion = components[0]
                    scenario = "(" + components[1]
                    test_list.append(test_portion + attr + scenario)
                else:
                    test_list.append(test_id + attr)
            except KeyError:
                self.logger.warning("Test %s not found in Tempest list." %
                                    test_id)
        self.logger.debug("Number of tests: " + str(len(test_list)))
        return test_list

    def _write_normalized_test_list(self, test_ids):
        """Create a temporary file to pass into testr containing a list of test
        IDs that should be tested.

        :param test_ids: list of full test IDs
        """
        temp = tempfile.NamedTemporaryFile(delete=False)
        for test_id in test_ids:
            temp.write(("%s\n" % test_id).encode('utf-8'))
        temp.flush()

        # Register the created file for cleanup.
        atexit.register(self._remove_test_list_file, temp.name)
        return temp.name

    def _remove_test_list_file(self, file_path):
        """Delete the given file.

        :param file_path: string containing the location of the file
        """
        if os.path.isfile(file_path):
            os.remove(file_path)

    def get_normalized_test_list(self, list_location):
        """This will take in the user's test list and will normalize it
        so that the test cases in the list map to actual full test IDS in
        the Tempest environment.

        :param list_location: file path or URL of the test list
        """
        tempest_test_ids = self._get_tempest_test_ids()
        if not tempest_test_ids:
            return None
        base_test_ids = self._get_base_test_ids_from_list_file(list_location)
        full_capability_test_ids = self._get_full_test_ids(tempest_test_ids,
                                                           base_test_ids)
        list_file = self._write_normalized_test_list(full_capability_test_ids)
        return list_file

    def create_whitelist(self, list_location):
        """This takes in a test list file, get normalized, and get whitelist
        regexes using full qualified test names (one per line).
        Ex:
            'tempest.test1[id-2,gate]' -> tempest.test1\[
            'tempest.test2[id-3,smoke](scenario)' -> tempest.test2\[
            'tempest.test3[compute,id-4]' -> tempest.test3\[

        :param list_location: file path or URL location of list file
        """
        normalized_list = open(self.get_normalized_test_list(list_location),
                               'r').read()
        # Keep the names
        tests_list = [re.sub("\[", "\[", test)
                      for test in re.findall(".*\[", normalized_list)]

        return self._write_normalized_test_list(tests_list)