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 copy
18import logging
19import itertools
20import operator
21
22from vts.runners.host import const
23from vts.utils.python.common import cmd_utils
24from vts.utils.python.os import path_utils
25
26from vts.testcases.kernel.ltp.shell_environment import shell_environment
27from vts.testcases.kernel.ltp import ltp_enums
28from vts.testcases.kernel.ltp import ltp_configs
29from vts.testcases.kernel.ltp import requirements
30
31
32class EnvironmentRequirementChecker(object):
33    """LTP testcase environment checker.
34
35    This class contains a dictionary for some known environment
36    requirements for a set of test cases and several environment
37    check functions to be mapped with. All check functions' results
38    are cached in a dictionary for multiple use.
39
40    Attributes:
41        _REQUIREMENT_DEFINITIONS: dictionary {string, obj}, a map between
42            requirement name and the actual definition class object
43        _result_cache: dictionary {requirement_check_method_name:
44            (bool, string)}, a map between check method name and cached result
45            tuples (boolean, note)
46        _executable_available: dict {string, bool}, a map between executable
47                                path and its existance on target
48        _shell_env: ShellEnvironment object, which checks and sets
49            shell environments given a shell mirror
50        shell: shell mirror object, can be used to execute shell
51            commands on target side through runner
52        ltp_bin_host_path: string, host path of ltp binary
53    """
54
55    def __init__(self, shell):
56        self.shell = shell
57        self._result_cache = {}
58        self._executable_available = {}
59        self._shell_env = shell_environment.ShellEnvironment(self.shell)
60        self._REQUIREMENT_DEFINITIONS = requirements.GetRequrementDefinitions()
61
62    @property
63    def shell(self):
64        """Get the runner's shell mirror object to execute commands"""
65        return self._shell
66
67    @shell.setter
68    def shell(self, shell):
69        """Set the runner's shell mirror object to execute commands"""
70        self._shell = shell
71
72    def Cleanup(self):
73        """Run all cleanup jobs at the end of tests"""
74        self._shell_env.Cleanup()
75
76    def GetRequirements(self, test_case):
77        """Get a list of requirements for a fiven test case
78
79        Args:
80            test_case: TestCase object, the test case to query
81        """
82        result = copy.copy(ltp_configs.REQUIREMENT_FOR_ALL)
83
84        result.extend(
85            rule
86            for rule, tests in ltp_configs.REQUIREMENTS_TO_TESTCASE.iteritems()
87            if test_case.fullname in tests)
88
89        result.extend(
90            rule
91            for rule, tests in ltp_configs.REQUIREMENT_TO_TESTSUITE.iteritems()
92            if test_case.testsuite in tests)
93
94        return list(set(result))
95
96    def Check(self, test_case):
97        """Check whether a given test case's requirement has been satisfied.
98        Skip the test if not.
99
100        If check failed, this method returns False and the reason is set
101        to test_case.note.
102
103        Args:
104            test_case: TestCase object, a given test case to check
105
106        Returns:
107            True if check pass; False otherwise
108        """
109        if (test_case.requirement_state ==
110                ltp_enums.RequirementState.UNSATISFIED or
111                not self.TestBinaryExists(test_case)):
112            return False
113
114        for requirement in self.GetRequirements(test_case):
115            if requirement not in self._result_cache:
116                definitions = self._REQUIREMENT_DEFINITIONS[requirement]
117                self._result_cache[
118                    requirement] = self._shell_env.ExecuteDefinitions(
119                        definitions)
120
121            result, note = self._result_cache[requirement]
122            logging.info("Result for %s's requirement %s is %s", test_case,
123                         requirement, result)
124            if result is False:
125                test_case.requirement_state = ltp_enums.RequirementState.UNSATISFIED
126                test_case.note = note
127                return False
128
129        test_case.requirement_state = ltp_enums.RequirementState.SATISFIED
130        return True
131
132    def CheckAllTestCaseExecutables(self, test_cases):
133        """Run a batch job to check executable exists and set permissions.
134
135        The result will be stored in self._executable_available for use in
136        TestBinaryExists method.
137
138        Args:
139            test_case: list of TestCase objects.
140        """
141        executables_generators = (
142            test_case.GetRequiredExecutablePaths(self.ltp_bin_host_path)
143            for test_case in test_cases)
144        executables = list(
145            set(itertools.chain.from_iterable(executables_generators)))
146
147        # Set all executables executable permission using chmod.
148        logging.info("Setting permissions on device")
149        permission_command = "chmod 775 %s" % path_utils.JoinTargetPath(
150            ltp_configs.LTPBINPATH, '*')
151        permission_result = self.shell.Execute(permission_command)
152        if permission_result[const.EXIT_CODE][0]:
153            logging.error("Permission command '%s' failed.",
154                          permission_command)
155
156        # Check existence of all executables used in test definition.
157        # Some executables needed by test cases but not listed in test
158        # definition will not be checked here
159        executable_exists_commands = [
160            "ls %s" % executable for executable in executables
161            if executable not in ltp_configs.INTERNAL_BINS
162        ]
163        logging.info("Checking binary existence on host: %s",
164                     executable_exists_commands)
165
166        cmd_results = cmd_utils.ExecuteShellCommand(executable_exists_commands)
167        executable_exists_results = map(operator.not_,
168                                        cmd_results[cmd_utils.EXIT_CODE])
169        logging.info("Finished checking binary existence on host: %s",
170                     cmd_results)
171
172        self._executable_available = dict(
173            zip(executables, executable_exists_results))
174
175        # Check whether all the internal binaries in path needed exist
176        bin_path_exist_commands = ["which %s" % bin
177                                   for bin in ltp_configs.INTERNAL_BINS]
178        bin_path_results = map(
179            operator.not_,
180            self.shell.Execute(bin_path_exist_commands)[const.EXIT_CODE])
181
182        bin_path_results = map(
183            operator.not_,
184            self.shell.Execute(bin_path_exist_commands)[const.EXIT_CODE])
185
186        self._executable_available.update(
187            dict(zip(ltp_configs.INTERNAL_BINS, bin_path_results)))
188
189    def TestBinaryExists(self, test_case):
190        """Check whether the given test case's binary exists.
191
192        Args:
193            test_case: TestCase, the object representing the test case
194
195        Return:
196            True if exists, False otherwise
197        """
198        if test_case.requirement_state == ltp_enums.RequirementState.UNSATISFIED:
199            logging.warn("[Checker] Attempting to run test case that has "
200                         "already been checked and requirement not satisfied."
201                         "%s" % test_case)
202            return False
203
204        executables = test_case.GetRequiredExecutablePaths(
205            self.ltp_bin_host_path)
206        results = [self._executable_available[executable]
207                   for executable in executables]
208
209        if not all(results):
210            test_case.requirement_state = ltp_enums.RequirementState.UNSATISFIED
211            test_case.note = "Some executables not exist: {}".format(
212                zip(executables, results))
213            logging.error("[Checker] Binary existance check failed for {}. "
214                          "Reason: {}".format(test_case, test_case.note))
215            return False
216        else:
217            return True
218