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