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 ComponentSpecificationMessage_pb2 as CompSpecMsg 25from vts.utils.python.fuzzer import FuzzerUtils 26from vts.utils.python.mirror import py2pb 27 28 29class MirrorObjectError(Exception): 30 """Raised when there is a general error in manipulating a mirror object.""" 31 pass 32 33 34class EnumMirror(object): 35 """Enum's host-side mirror instance.""" 36 37 def __init__(self, attribute): 38 logging.info(attribute) 39 for enumerator, scalar_value in zip(attribute.enum_value.enumerator, 40 attribute.enum_value.scalar_value): 41 setattr(self, enumerator, 42 getattr(scalar_value, attribute.enum_value.scalar_type)) 43 44 45class MirrorObjectForTypes(object): 46 """The class that can create host-side mirroed variable instances. 47 48 Attributes: 49 _if_spec_msg: the interface specification message of a host object to 50 mirror. 51 _parent_path: the name of a sub struct this object mirrors. 52 """ 53 54 def __init__(self, msg, parent_path=None): 55 self._if_spec_msg = msg 56 self._parent_path = parent_path 57 58 def GetSubStruct(self, sub_struct_name): 59 """Returns the Struct Specification Message. 60 61 Args: 62 sub_struct_name: string, the name of the target sub struct attribute. 63 64 Returns: 65 StructSpecificationMessage if found, None otherwise 66 """ 67 if self._if_spec_msg.interface and self._if_spec_msg.interface.sub_struct: 68 for sub_struct in self._if_spec_msg.interface.sub_struct: 69 if sub_struct.name == sub_struct_name: 70 return copy.copy(sub_struct) 71 return None 72 73 def GetCustomAggregateType(self, type_name): 74 """Returns the Argument Specification Message. 75 76 Args: 77 type_name: string, the name of the target data type. 78 79 Returns: 80 VariableSpecificationMessage if found, None otherwise 81 """ 82 try: 83 if (self._if_spec_msg.interface and 84 self._if_spec_msg.interface.attribute): 85 for attribute in self._if_spec_msg.interface.attribute: 86 if not attribute.is_const and attribute.name == type_name: 87 return copy.copy(attribute) 88 if self._if_spec_msg.attribute: 89 for attribute in self._if_spec_msg.attribute: 90 if not attribute.is_const and attribute.name == type_name: 91 return copy.copy(attribute) 92 return None 93 except AttributeError as e: 94 # TODO: check in advance whether self._if_spec_msg Interface 95 # SpecificationMessage. 96 return None 97 98 def GetAttributeSpecification(self, attribute_name): 99 """Returns the ProtoBuf specification of a requested attribute. 100 101 Args: 102 attribute_name: string, the name of a target attribute 103 (optionally excluding the namespace). 104 105 Returns: 106 VariableSpecificationMessage if found, None otherwise 107 """ 108 for attribute in self._if_spec_msg.attribute: 109 if (attribute.name == attribute_name or 110 attribute.name.endswith("::" + attribute_name)): 111 return attribute 112 return None 113 114 def Py2Pb(self, attribute_name, py_values): 115 """Returns the ProtoBuf of a give Python values. 116 117 Args: 118 attribute_name: string, the name of a target attribute. 119 py_values: Python values. 120 121 Returns: 122 Converted VariableSpecificationMessage if found, None otherwise 123 """ 124 attribute_msg = self.GetAttributeSpecification(attribute_name) 125 if attribute_msg: 126 return py2pb.Convert(attribute_msg, py_values) 127 return None 128 129 def GetConstType(self, type_name): 130 """Returns the Argument Specification Message. 131 132 Args: 133 type_name: string, the name of the target const data variable. 134 135 Returns: 136 VariableSpecificationMessage if found, None otherwise 137 """ 138 try: 139 if (self._if_spec_msg.interface and 140 self._if_spec_msg.interface.attribute): 141 for attribute in self._if_spec_msg.interface.attribute: 142 if attribute.is_const and attribute.name == type_name: 143 return copy.copy(attribute) 144 elif (attribute.type == CompSpecMsg.TYPE_ENUM and 145 attribute.name.endswith(type_name)): 146 return attribute 147 if self._if_spec_msg.attribute: 148 for attribute in self._if_spec_msg.attribute: 149 if attribute.is_const and attribute.name == type_name: 150 return copy.copy(attribute) 151 elif (attribute.type == CompSpecMsg.TYPE_ENUM and 152 attribute.name.endswith(type_name)): 153 return attribute 154 return None 155 except AttributeError as e: 156 # TODO: check in advance whether self._if_spec_msg Interface 157 # SpecificationMessage. 158 return None 159 160 # TODO: Guard against calls to this function after self.CleanUp is called. 161 def __getattr__(self, api_name, *args, **kwargs): 162 """Calls a target component's API. 163 164 Args: 165 api_name: string, the name of an API function to call. 166 *args: a list of arguments 167 **kwargs: a dict for the arg name and value pairs 168 """ 169 def MessageGenerator(*args, **kwargs): 170 """Dynamically generates a custom message instance.""" 171 arg_msg = self.GetCustomAggregateType(api_name) 172 if not arg_msg: 173 raise MirrorObjectError("arg %s unknown" % arg_msg) 174 logging.info("MessageGenerator %s %s", api_name, arg_msg) 175 logging.debug("MESSAGE %s", api_name) 176 if arg_msg.type == CompSpecMsg.TYPE_STRUCT: 177 for struct_value in arg_msg.struct_value: 178 logging.debug("for %s %s", 179 struct_value.name, struct_value.scalar_type) 180 first_vector_elem = True 181 for given_name, given_value in kwargs.iteritems(): 182 logging.debug("check %s %s", struct_value.name, given_name) 183 if given_name == struct_value.name: 184 logging.debug("match type=%s", struct_value.scalar_type) 185 if struct_value.type == CompSpecMsg.TYPE_SCALAR: 186 if struct_value.scalar_type == "uint32_t": 187 struct_value.scalar_value.uint32_t = given_value 188 elif struct_value.scalar_type == "int32_t": 189 struct_value.scalar_value.int32_t = given_value 190 else: 191 raise MirrorObjectError( 192 "support %s" % struct_value.scalar_type) 193 elif struct_value.type == CompSpecMsg.TYPE_VECTOR: 194 scalar_type = struct_value.vector_value[0].scalar_type 195 for value in given_value: 196 vector_value = struct_value.vector_value.add() 197 vector_value.type = CompSpecMsg.TYPE_SCALAR 198 vector_value.scalar_type = scalar_type 199 setattr(vector_value.scalar_value, scalar_type, value) 200 continue 201 elif arg_msg.type == CompSpecMsg.TYPE_FUNCTION_POINTER: 202 for fp_value in arg_msg.function_pointer: 203 logging.debug("for %s", fp_value.function_name) 204 for given_name, given_value in kwargs.iteritems(): 205 logging.debug("check %s %s", fp_value.function_name, given_name) 206 if given_name == fp_value.function_name: 207 fp_value.id = self.GetFunctionPointerID(given_value) 208 break 209 210 if arg_msg.type == CompSpecMsg.TYPE_STRUCT: 211 for struct_value, given_value in zip(arg_msg.struct_value, args): 212 logging.debug("arg match type=%s", struct_value.scalar_type) 213 if struct_value.type == CompSpecMsg.TYPE_SCALAR: 214 if struct_value.scalar_type == "uint32_t": 215 struct_value.scalar_value.uint32_t = given_value 216 elif struct_value.scalar_type == "int32_t": 217 struct_value.scalar_value.int32_t = given_value 218 else: 219 raise MirrorObjectError("support %s" % p_type) 220 elif arg_msg.type == CompSpecMsg.TYPE_FUNCTION_POINTER: 221 for fp_value, given_value in zip(arg_msg.function_pointer, args): 222 logging.debug("for %s", fp_value.function_name) 223 fp_value.id = self.GetFunctionPointerID(given_value) 224 logging.debug("fp %s", fp_value) 225 else: 226 logging.error("arg_msg.type %s", arg_msg.type) 227 sys.exit(-1) 228 logging.debug("generated %s", arg_msg) 229 return arg_msg 230 231 def ConstGenerator(): 232 """Dynamically generates a const variable's value.""" 233 arg_msg = self.GetConstType(api_name) 234 if not arg_msg: 235 raise MirrorObjectError("const %s unknown" % arg_msg) 236 logging.debug("check %s", api_name) 237 if arg_msg.type == CompSpecMsg.TYPE_SCALAR: 238 ret_v = getattr(arg_msg.scalar_value, arg_msg.scalar_type, None) 239 if ret_v is None: 240 raise MirrorObjectError( 241 "No value found for type %s in %s." % (p_type, value)) 242 return ret_v 243 elif arg_msg.type == CompSpecMsg.TYPE_STRING: 244 return arg_msg.string_value.message 245 elif arg_msg.type == CompSpecMsg.TYPE_ENUM: 246 return EnumMirror(arg_msg) 247 raise MirrorObjectError("const %s not found" % api_name) 248 249 struct_msg = self.GetSubStruct(api_name) 250 if struct_msg: 251 logging.debug("sub_struct %s", struct_msg) 252 if self._parent_path: 253 parent_name = "%s.%s" % (self._parent_path, api_name) 254 else: 255 parent_name = api_name 256 return MirrorObjectForTypes(struct_msg, parent_path=parent_name) 257 258 # handle attributes. 259 arg_msg = self.GetCustomAggregateType(api_name) 260 if arg_msg: 261 logging.debug("arg %s", arg_msg) 262 return MessageGenerator 263 264 arg_msg = self.GetConstType(api_name) 265 if arg_msg: 266 logging.debug("const %s *\n%s", api_name, arg_msg) 267 return ConstGenerator() 268 269 raise MirrorObjectError("unknown api name %s" % api_name) 270 271 def __str__(self): 272 """Prints all the attributes and methods.""" 273 result = "" 274 if self._if_spec_msg: 275 if self._if_spec_msg.attribute: 276 for attribute in self._if_spec_msg.attribute: 277 result += "global attribute %s\n" % attribute.name 278 if self._if_spec_msg.interface.attribute: 279 for attribute in self._if_spec_msg.interface.attribute: 280 result += "interface attribute %s\n" % attribute.name 281 return result 282 283