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 random
20import sys
21
22from google.protobuf import text_format
23
24from vts.proto import AndroidSystemControlMessage_pb2 as ASysCtrlMsg
25from vts.proto import ComponentSpecificationMessage_pb2 as CompSpecMsg
26from vts.utils.python.fuzzer import FuzzerUtils
27from vts.utils.python.mirror import native_entity_mirror
28from vts.utils.python.mirror import py2pb
29
30_DEFAULT_TARGET_BASE_PATHS = ["/system/lib64/hw"]
31_DEFAULT_HWBINDER_SERVICE = "default"
32
33INTERFACE = "interface"
34API = "api"
35
36
37class MirrorObjectError(Exception):
38    """Raised when there is a general error in manipulating a mirror object."""
39    pass
40
41
42class HalMirror(native_entity_mirror.NativeEntityMirror):
43    """The class that acts as the mirror to an Android device's HAL layer.
44
45    This class exists on the host and can be used to communicate to a
46    particular HIDL HAL on the target side.
47
48    Attributes:
49        _callback_server: the instance of a callback server.
50    """
51
52    def __init__(self,
53                 client,
54                 callback_server,
55                 hal_driver_id=None,
56                 if_spec_message=None,
57                 caller_uid=None):
58        super(HalMirror, self).__init__(client, hal_driver_id, if_spec_message,
59                                        caller_uid)
60        self._callback_server = callback_server
61
62    def InitHalDriver(self, target_type, target_version_major,
63                      target_version_minor, target_package,
64                      target_component_name, hw_binder_service_name,
65                      handler_name, bits, is_test_hal):
66        """Initiates the driver for a HIDL HAL on the target device and loads
67        the interface specification message.
68
69        Args:
70            target_type: string, the target type name (e.g., light, camera).
71            target_version_major:
72              int, the target component major version (e.g., 1.0 -> 1).
73            target_version_minor:
74              int, the target component minor version (e.g., 1.0 -> 0).
75            target_package: . separated string (e.g., a.b.c) to denote the
76                            package name of target component.
77            target_component_name: string, the target componet name (e.g., INfc).
78            hw_binder_service_name: string, name of the HAL service instance
79                                    (e.g. default)
80            handler_name: string, the name of the handler. target_type is used
81                          by default.
82            bits: integer, processor architecture indicator: 32 or 64.
83            is_test_hal: bool, whether the HAL service is a test HAL
84                         (e.g. msgq).
85
86        Raises:
87            errors.ComponentLoadingError is raised when error occurs trying to
88            create a MirrorObject.
89        """
90        driver_id = self.LaunchMirrorDriver(
91            ASysCtrlMsg.VTS_DRIVER_TYPE_HAL_HIDL,
92            "hal_hidl",
93            target_type,
94            target_version_major,
95            target_version_minor,
96            target_package=target_package,
97            target_component_name=target_component_name,
98            handler_name=handler_name,
99            hw_binder_service_name=hw_binder_service_name,
100            bits=bits,
101            is_test_hal=is_test_hal)
102        self._driver_id = driver_id
103
104        #TODO: ListApis assumes only one HAL is loaded at a time, need to
105        #      figure out a way to get the api_spec when we want to test
106        #      multiple HALs together.
107        found_api_spec = self._client.ListApis()
108        if not found_api_spec:
109            raise errors.ComponentLoadingError(
110                "No API found for %s" % target_type)
111        if_spec_msg = CompSpecMsg.ComponentSpecificationMessage()
112        text_format.Merge(found_api_spec, if_spec_msg)
113
114        self._if_spec_msg = if_spec_msg
115
116    def GetCallbackFunctionID(self, function_pointer):
117        """Gets registsred callback function id for the given function_pointer.
118
119        Args:
120            function_pointer: the callback function pointer.
121
122        Returns:
123            Id for the call back function registered with callback server.
124        """
125        if self._callback_server:
126            id = self._callback_server.GetCallbackId(function_pointer)
127            if id is None:
128                id = self._callback_server.RegisterCallback(function_pointer)
129            return str(id)
130        else:
131            raise MirrorObjectError("callback server is not started.")
132
133    def GetHidlCallbackInterface(self, interface_name, **kwargs):
134        """Gets the ProtoBuf message for a callback interface based on args.
135
136        Args:
137            interface_name: string, the callback interface name.
138            **kwargs: a dict for the arg name and value pairs
139
140        Returns:
141            VariableSpecificationMessage that contains the callback interface
142            description.
143        """
144        var_msg = CompSpecMsg.VariableSpecificationMessage()
145        var_msg.name = interface_name
146        var_msg.type = CompSpecMsg.TYPE_FUNCTION_POINTER
147        var_msg.is_callback = True
148
149        msg = self._if_spec_msg
150        specification = self._client.ReadSpecification(
151            interface_name, msg.component_class, msg.component_type,
152            msg.component_type_version_major, msg.component_type_version_minor,
153            msg.package)
154        logging.debug("specification: %s", specification)
155        interface = getattr(specification, INTERFACE, None)
156        apis = getattr(interface, API, [])
157        for api in apis:
158            function_pointer = None
159            if api.name in kwargs:
160                function_pointer = kwargs[api.name]
161            else:
162
163                def dummy(*args):
164                    """Dummy implementation for any callback function."""
165                    logging.debug(
166                        "Entering dummy implementation"
167                        " for callback function: %s", api.name)
168                    for arg_index in range(len(args)):
169                        logging.debug("arg%s: %s", arg_index, args[arg_index])
170
171                function_pointer = dummy
172            func_pt_msg = var_msg.function_pointer.add()
173            func_pt_msg.function_name = api.name
174            func_pt_msg.id = self.GetCallbackFunctionID(function_pointer)
175
176        return var_msg
177
178    def GetHidlTypeInterface(self, interface_name):
179        """Gets a HalMirror for HIDL HAL types specification.
180
181        Args:
182            interface_name: string, the name of a target interface to read.
183        """
184        return self.GetHalMirrorForInterface(interface_name)
185
186    def GetHalMirrorForInterface(self, interface_name, driver_id=None):
187        """Gets a HalMirror for a HIDL HAL interface.
188
189        Args:
190            interface_name: string, the name of a target interface to read.
191            driver_id: int, driver is of the corresponding HIDL HAL interface.
192
193        Returns:
194            a host-side mirror of a HIDL HAL interface.
195        """
196        if not self._if_spec_msg:
197            raise MirrorObjectError("specification is not loaded")
198        msg = self._if_spec_msg
199        found_api_spec = self._client.ReadSpecification(
200            interface_name,
201            msg.component_class,
202            msg.component_type,
203            msg.component_type_version_major,
204            msg.component_type_version_minor,
205            msg.package,
206            recursive=True)
207
208        logging.debug("found_api_spec %s", found_api_spec)
209        if not driver_id:
210            driver_id = self._driver_id
211        # Instantiate a MirrorObject and return it.
212        return HalMirror(self._client, self._callback_server, driver_id,
213                         found_api_spec)
214