1#
2# Copyright (C) 2017 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
19
20from vts.runners.host import const
21from vts.runners.host import keys
22from vts.runners.host import test_runner
23from vts.testcases.template.gtest_binary_test import gtest_binary_test
24from vts.testcases.template.gtest_binary_test import gtest_test_case
25from vts.utils.python.cpu import cpu_frequency_scaling
26from vts.utils.python.hal import hal_service_name_utils
27from vts.utils.python.precondition import precondition_utils
28
29# The pattern indicating a full hal test name including the service name info.
30# e.g. TM.TC(default)_32bit
31_HAL_TEST_NAME_PATTERN = ".*\(.*\).*"
32
33class HidlHalGTest(gtest_binary_test.GtestBinaryTest):
34    """Base class to run a VTS target-side HIDL HAL test.
35
36    Attributes:
37        DEVICE_TEST_DIR: string, temp location for storing binary
38        TAG_PATH_SEPARATOR: string, separator used to separate tag and path
39        tags: all the tags that appeared in binary list
40        testcases: list of GtestTestCase objects, list of test cases to run
41        _cpu_freq: CpuFrequencyScalingController instance of a target device.
42        _dut: AndroidDevice, the device under test as config
43        _initial_test_case_cnt: Number of initial test cases.
44        _target_hals: List of String, the targeting hal service of the test.
45                      e.g (["android.hardware.foo@1.0::IFoo"])
46    """
47
48    def setUpClass(self):
49        """Checks precondition."""
50        self._initial_test_case_cnt = 0
51        super(HidlHalGTest, self).setUpClass()
52        if not hasattr(self, "_target_hals"):
53            self._target_hals = []
54
55        opt_params = [keys.ConfigKeys.IKEY_SKIP_IF_THERMAL_THROTTLING,
56                      keys.ConfigKeys.IKEY_DISABLE_CPU_FREQUENCY_SCALING]
57        self.getUserParams(opt_param_names=opt_params)
58
59        self._skip_if_thermal_throttling = self.getUserParam(
60            keys.ConfigKeys.IKEY_SKIP_IF_THERMAL_THROTTLING,
61            default_value=False)
62        self._disable_cpu_frequency_scaling = self.getUserParam(
63            keys.ConfigKeys.IKEY_DISABLE_CPU_FREQUENCY_SCALING,
64            default_value=True)
65
66        if not self.isSkipAllTests():
67            self._cpu_freq = cpu_frequency_scaling.CpuFrequencyScalingController(
68                self._dut)
69            if self._disable_cpu_frequency_scaling:
70                logging.debug("Disabling CPU frequency scaling")
71                self._cpu_freq.DisableCpuScaling()
72        else:
73            self._cpu_freq = None
74
75        if not self.isSkipAllTests():
76            ret = precondition_utils.CanRunHidlHalTest(
77                self, self._dut, self.shell, self.run_as_compliance_test)
78            if not ret:
79                self.skipAllTests("HIDL HAL precondition check failed.")
80
81        # Extend timeout if there are multiple service instance combinations.
82        if (not self.isSkipAllTests() and self._initial_test_case_cnt and
83                len(self.testcases) > self._initial_test_case_cnt):
84            self.resetTimeout(self.timeout * len(self.testcases) /
85                              float(self._initial_test_case_cnt))
86
87        if self.sancov.enabled and self._target_hals:
88            self.sancov.InitializeDeviceCoverage(self._dut,
89                                                 self._target_hals)
90        if self.coverage.enabled and self._target_hals:
91            self.coverage.SetHalNames(self._target_hals)
92            self.coverage.SetCoverageReportFilePrefix(self.test_module_name + self.abi_bitness)
93            self.coverage.InitializeDeviceCoverage(self._dut)
94
95    def CreateTestCases(self):
96        """Create testcases and conditionally enable passthrough mode.
97
98        Create testcases as defined in HidlHalGtest. If the passthrough option
99        is provided in the configuration or if coverage is enabled, enable
100        passthrough mode on the test environment.
101        """
102        super(HidlHalGTest, self).CreateTestCases()
103
104        passthrough_opt = self.getUserParam(
105            keys.ConfigKeys.IKEY_PASSTHROUGH_MODE, default_value=False)
106
107        # Enable coverage if specified in the configuration.
108        if passthrough_opt:
109            self._EnablePassthroughMode()
110
111    # @Override
112    def CreateTestCase(self, path, tag=''):
113        """Create a list of GtestTestCase objects from a binary path.
114
115        Support testing against different service names by first executing a
116        dummpy test case which lists all the registered hal services. Then
117        query the service name(s) for each registered service with lshal.
118        For each service name, create a new test case each with the service
119        name as an additional argument.
120
121        Args:
122            path: string, absolute path of a gtest binary on device
123            tag: string, a tag that will be appended to the end of test name
124
125        Returns:
126            A list of GtestTestCase objects.
127        """
128        initial_test_cases = super(HidlHalGTest, self).CreateTestCase(path,
129                                                                      tag)
130        self._initial_test_case_cnt += len(initial_test_cases)
131        if not initial_test_cases:
132            return initial_test_cases
133        # first, run one test with --list_registered_services.
134        list_service_test_case = copy.copy(initial_test_cases[0])
135        list_service_test_case.args = " --list_registered_services"
136        results = self.shell.Execute(
137            list_service_test_case.GetRunCommand(raw_command=True))
138        if (results[const.EXIT_CODE][0]):
139            logging.error("Failed to list test cases from binary %s",
140                          list_service_test_case.path)
141        # parse the results to get the registered service list.
142        registered_services = []
143        comb_mode = hal_service_name_utils.CombMode.FULL_PERMUTATION
144        # TODO: consider to use a standard data format (e.g. json) instead of
145        # parsing the print output.
146        for line in results[const.STDOUT][0].split("\n"):
147            line = str(line)
148            if line.startswith("hal_service: "):
149                service = line[len("hal_service: "):]
150                registered_services.append(service)
151            if line.startswith("service_comb_mode: "):
152                comb_mode = int(line[len("service_comb_mode: "):])
153
154        # If no service registered, return the initial test cases directly.
155        if not registered_services:
156            logging.error("No hal service registered.")
157            return initial_test_cases
158
159        self._target_hals = copy.copy(registered_services)
160
161        # find the correponding service name(s) for each registered service and
162        # store the mapping in dict service_instances.
163        service_instances = {}
164        for service in registered_services:
165            testable, service_names = hal_service_name_utils.GetHalServiceName(
166                self.shell, service, self.abi_bitness,
167                self.run_as_compliance_test)
168            if not testable:
169                self.skipAllTests("Hal: %s is not testable, "
170                                  "skip all tests." % service)
171                return initial_test_cases
172            if not service_names:
173                self.skipAllTests("No service name found for: %s, skip all tests." % service)
174                # If any of the test services are not available, return the
175                # initial test cases directly.
176                return initial_test_cases
177            else:
178                service_instances[service] = service_names
179        logging.debug("registered service instances: %s", service_instances)
180        logging.debug("service comb mode: %d", comb_mode)
181
182        # If request NO_COMBINATION mode, return the initial test cases directly.
183        if comb_mode == hal_service_name_utils.CombMode.NO_COMBINATION:
184            return initial_test_cases
185
186        # get all the combination of service instances.
187        service_instance_combinations = hal_service_name_utils.GetServiceInstancesCombinations(
188            registered_services, service_instances, comb_mode);
189
190        new_test_cases = []
191        appendix_list = []
192        for instance_combination in service_instance_combinations:
193            for test_case in initial_test_cases:
194                new_test_case = copy.copy(test_case)
195                service_name_list = []
196                for instance in instance_combination:
197                    new_test_case.args += " --hal_service_instance=" + instance
198                    service_name_list.append(instance[instance.find('/')+1:])
199                name_appendix = "({0})".format(",".join(service_name_list))
200                new_test_case.name_appendix = name_appendix
201                new_test_cases.append(new_test_case)
202            appendix_list.append(name_appendix)
203        self.test_filter.ExpandAppendix(appendix_list, _HAL_TEST_NAME_PATTERN)
204        return new_test_cases
205
206    def _EnablePassthroughMode(self):
207        """Enable passthrough mode by setting getStub to true.
208
209        This funciton should be called after super class' setupClass method.
210        If called before setupClass, user_params will be changed in order to
211        trigger setupClass method to invoke this method again.
212        """
213        if self.testcases:
214            for test_case in self.testcases:
215                envp = ' %s=true' % const.VTS_HAL_HIDL_GET_STUB
216                test_case.envp += envp
217        else:
218            logging.warn('No test cases are defined yet. Maybe setupClass '
219                         'has not been called. Changing user_params to '
220                         'enable passthrough mode option.')
221            self.user_params[keys.ConfigKeys.IKEY_PASSTHROUGH_MODE] = True
222
223    def setUp(self):
224        """Skips the test case if thermal throttling lasts for 30 seconds."""
225        super(HidlHalGTest, self).setUp()
226
227        if (self._skip_if_thermal_throttling and
228                getattr(self, "_cpu_freq", None)):
229            self._cpu_freq.SkipIfThermalThrottling(retry_delay_secs=30)
230
231    def tearDown(self):
232        """Skips the test case if there is thermal throttling."""
233        if (self._skip_if_thermal_throttling and
234                getattr(self, "_cpu_freq", None)):
235            self._cpu_freq.SkipIfThermalThrottling()
236
237        super(HidlHalGTest, self).tearDown()
238
239    def tearDownClass(self):
240        """Turns off CPU frequency scaling."""
241        if (not self.isSkipAllTests() and getattr(self, "_cpu_freq", None)
242            and self._disable_cpu_frequency_scaling):
243            logging.debug("Enabling CPU frequency scaling")
244            self._cpu_freq.EnableCpuScaling()
245
246        if self.sancov.enabled and self._target_hals:
247            self.sancov.FlushDeviceCoverage(self._dut, self._target_hals)
248            self.sancov.ProcessDeviceCoverage(self._dut,
249                                              self._target_hals)
250            self.sancov.Upload()
251
252        super(HidlHalGTest, self).tearDownClass()
253
254
255if __name__ == "__main__":
256    test_runner.main()
257