1#
2# Copyright (C) 2016 The Android Open Source Project
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8#      http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16
17import os
18import logging
19import itertools
20
21from vts.runners.host import const
22from vts.testcases.kernel.ltp import ltp_configs
23from vts.testcases.kernel.ltp import ltp_enums
24from vts.testcases.kernel.ltp import test_case
25from vts.testcases.kernel.ltp.configs import stable_tests
26from vts.testcases.kernel.ltp.configs import disabled_tests
27from vts.utils.python.common import filter_utils
28
29
30class TestCasesParser(object):
31    """Load a ltp vts testcase definition file and parse it into a generator.
32
33    Attributes:
34        _data_path: string, the vts data path on host side
35        _filter_func: function, a filter method that will emit exception if a test is filtered
36        _ltp_tests_filter: list of string, filter for tests that are stable and disabled
37    """
38
39    def __init__(self, data_path, filter_func):
40        self._data_path = data_path
41        self._filter_func = filter_func
42        self._ltp_tests_filter = filter_utils.Filter(
43            [ i[0] for i in stable_tests.STABLE_TESTS ],
44            disabled_tests.DISABLED_TESTS,
45            enable_regex=True)
46        self._ltp_tests_filter.ExpandBitness()
47
48    def ValidateDefinition(self, line):
49        """Validate a tab delimited test case definition.
50
51        Will check whether the given line of definition has three parts
52        separated by tabs.
53        It will also trim leading and ending white spaces for each part
54        in returned tuple (if valid).
55
56        Returns:
57            A tuple in format (test suite, test name, test command) if
58            definition is valid. None otherwise.
59        """
60        items = [
61            item.strip()
62            for item in line.split(ltp_enums.Delimiters.TESTCASE_DEFINITION)
63        ]
64        if not len(items) == 3 or not items:
65            return None
66        else:
67            return items
68
69    def Load(self,
70             ltp_dir,
71             n_bit,
72             test_filter,
73             run_staging=False,
74             is_low_mem=False):
75        """Read the definition file and yields a TestCase generator.
76
77        Args:
78            ltp_dir: string, directory that contains ltp binaries and scripts
79            n_bit: int, bitness
80            test_filter: Filter object, test name filter from base_test
81            run_staging: bool, whether to use staging configuration
82            is_low_mem: bool, whether to use low memory device configuration
83        """
84        scenario_groups = (ltp_configs.TEST_SUITES_LOW_MEM
85                           if is_low_mem else ltp_configs.TEST_SUITES)
86        logging.info('LTP scenario groups: %s', scenario_groups)
87
88        run_scritp = self.GenerateLtpRunScript(scenario_groups)
89
90        for line in run_scritp:
91            items = self.ValidateDefinition(line)
92            if not items:
93                continue
94
95            testsuite, testname, command = items
96            if is_low_mem and testsuite.endswith(
97                    ltp_configs.LOW_MEMORY_SCENARIO_GROUP_SUFFIX):
98                testsuite = testsuite[:-len(
99                    ltp_configs.LOW_MEMORY_SCENARIO_GROUP_SUFFIX)]
100
101            # Tests failed to build will have prefix "DISABLED_"
102            if testname.startswith("DISABLED_"):
103                logging.info("[Parser] Skipping test case {}-{}. Reason: "
104                             "not built".format(testsuite, testname))
105                continue
106
107            # Some test cases contain semicolons in their commands,
108            # and we replace them with &&
109            command = command.replace(';', '&&')
110
111            # Some test cases have hardcoded "/tmp" in the command
112            # we replace that with ltp_configs.TMPDIR
113            command = command.replace('/tmp', ltp_configs.TMPDIR)
114
115            testcase = test_case.TestCase(
116                testsuite=testsuite, testname=testname, command=command)
117
118            test_display_name = "{}_{}bit".format(str(testcase), n_bit)
119
120            # Check runner's base_test filtering method
121            try:
122                self._filter_func(test_display_name)
123            except:
124                logging.info("[Parser] Skipping test case %s. Reason: "
125                             "filtered" % testcase.fullname)
126                testcase.is_filtered = True
127                testcase.note = "filtered"
128
129            # For skipping tests that are not designed or ready for Android,
130            # check for bit specific test in disabled list as well as non-bit specific
131            if ((self._ltp_tests_filter.IsInExcludeFilter(str(testcase)) or
132                 self._ltp_tests_filter.IsInExcludeFilter(test_display_name)) and
133                    not test_filter.IsInIncludeFilter(test_display_name)):
134                logging.info("[Parser] Skipping test case %s. Reason: "
135                             "disabled" % testcase.fullname)
136                continue
137
138            # For separating staging tests from stable tests
139            if not self._ltp_tests_filter.IsInIncludeFilter(test_display_name):
140                if not run_staging and not test_filter.IsInIncludeFilter(
141                        test_display_name):
142                    # Skip staging tests in stable run
143                    continue
144                else:
145                    testcase.is_staging = True
146                    testcase.note = "staging"
147            else:
148                if run_staging:
149                    # Skip stable tests in staging run
150                    continue
151
152            if not testcase.is_staging:
153                for x in stable_tests.STABLE_TESTS:
154                    if x[0] == test_display_name and x[1]:
155                        testcase.is_mandatory = True
156                        break
157
158            logging.info("[Parser] Adding test case %s." % testcase.fullname)
159            yield testcase
160
161    def ReadCommentedTxt(self, filepath):
162        '''Read a lines of a file that are not commented by #.
163
164        Args:
165            filepath: string, path of file to read
166
167        Returns:
168            A set of string representing non-commented lines in given file
169        '''
170        if not filepath:
171            logging.error('Invalid file path')
172            return None
173
174        with open(filepath, 'r') as f:
175            lines_gen = (line.strip() for line in f)
176            return set(
177                line for line in lines_gen
178                if line and not line.startswith('#'))
179
180    def GenerateLtpTestCases(self, testsuite, disabled_tests_list):
181        '''Generate test cases for each ltp test suite.
182
183        Args:
184            testsuite: string, test suite name
185
186        Returns:
187            A list of string
188        '''
189        testsuite_script = os.path.join(self._data_path,
190                                        ltp_configs.LTP_RUNTEST_DIR, testsuite)
191
192        result = []
193        for line in open(testsuite_script, 'r'):
194            line = line.strip()
195            if not line or line.startswith('#'):
196                continue
197
198            testname = line.split()[0]
199            testname_prefix = ('DISABLED_'
200                               if testname in disabled_tests_list else '')
201            testname_modified = testname_prefix + testname
202
203            result.append("\t".join(
204                [testsuite, testname_modified, line[len(testname):].strip()]))
205        return result
206
207    def GenerateLtpRunScript(self, scenario_groups):
208        '''Given a scenario group generate test case script.
209
210        Args:
211            scenario_groups: list of string, name of test scenario groups to use
212
213        Returns:
214            A list of string
215        '''
216        disabled_tests_path = os.path.join(
217            self._data_path, ltp_configs.LTP_DISABLED_BUILD_TESTS_CONFIG_PATH)
218        disabled_tests_list = self.ReadCommentedTxt(disabled_tests_path)
219
220        result = []
221        for testsuite in scenario_groups:
222            result.extend(
223                self.GenerateLtpTestCases(testsuite, disabled_tests_list))
224
225        #TODO(yuexima): remove duplicate
226
227        return result
228