tools/emulator/vhal_emulator_test.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2017 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""
    This module tests the Vehicle HAL using adb socket.

    Protocol Buffer:
        This module relies on VehicleHalProto_pb2.py being in sync with the protobuf in the VHAL.
        If the VehicleHalProto.proto file has changed, re-generate the python version using
        a command of the form:
            protoc -I=<proto_dir> --python_out=<out_dir> <proto_dir>/VehicleHalProto.proto
        For example:
            protoDir=~/android/master/hardware/interfaces/automotive/vehicle/2.0/default/impl/vhal_v2_0/proto
            outDir=~/android/master/packages/services/Car/tools/emulator
            protoc -I=$protoDir --python_out=$outDir $protoDir/VehicleHalProto.proto
"""

from __future__ import print_function

# Suppress .pyc files
import sys
sys.dont_write_bytecode = True

import VehicleHalProto_pb2
import vhal_consts_2_0
import vhal_emulator
import logging

class VhalTest:
    # Global vars
    _badProps = [0, 0x3FFFFFFF]     # List of bad properties to try for negative tests
    _configs = 0                    # List of configs from DUT
    _log = 0                        # Logger module
    _vhal = 0                       # Handle to VHAL object that communicates over socket to DUT
    # TODO: b/38203109 - Fix OBD2 values, implement handling for complex properties
    _skipProps = [
                    vhal_consts_2_0.VEHICLEPROPERTY_OBD2_LIVE_FRAME,
                    vhal_consts_2_0.VEHICLEPROPERTY_OBD2_FREEZE_FRAME,
                    vhal_consts_2_0.VEHICLEPROPERTY_OBD2_FREEZE_FRAME_INFO,
                    vhal_consts_2_0.VEHICLEPROPERTY_OBD2_FREEZE_FRAME_CLEAR,
                    vhal_consts_2_0.VEHICLEPROPERTY_VEHICLE_MAP_SERVICE,
                    vhal_consts_2_0.VEHICLEPROPERTY_WHEEL_TICK,     # Need to support complex properties
                    0x21E00666      # FakeDataControllingProperty - an internal test property
                 ]

    def _getMidpoint(self, minVal, maxVal):
        retVal =  minVal + (maxVal - minVal)/2
        return retVal

    # Generates a test value based on the config
    def _generateTestValue(self, cfg, idx, origValue):
        valType = cfg.value_type
        if valType in self._types.TYPE_STRING:
            testValue = "test string"
        elif valType in self._types.TYPE_BYTES:
            # Generate array of integers counting from 0
            testValue = list(range(len(origValue)))
        elif valType == vhal_consts_2_0.VEHICLEPROPERTYTYPE_BOOLEAN:
            testValue = origValue ^ 1
        elif valType in self._types.TYPE_INT32:
            try:
                testValue = self._getMidpoint(cfg.area_configs[idx].min_int32_value,
                                              cfg.area_configs[idx].max_int32_value)
            except:
                # min/max values aren't set.  Set a hard-coded value
                testValue = 123
        elif valType in self._types.TYPE_INT64:
            try:
                testValue = self._getMidpoint(cfg.area_configs[idx].min_int64_value,
                                              cfg.area_configs[idx].max_int64_value)
            except:
                # min/max values aren't set.  Set a large hard-coded value
                testValue = 1 << 50
        elif valType in self._types.TYPE_FLOAT:
            try:
                testValue = self._getMidpoint(cfg.area_configs[idx].min_float_value,
                                              cfg.area_configs[idx].max_float_value)
            except:
                # min/max values aren't set.  Set a hard-coded value
                testValue = 123.456
            # Truncate float to 5 decimal places
            testValue = "%.5f" % testValue
            testValue = float(testValue)
        else:
            self._log.error("generateTestValue:  valType=0x%X is not handled", valType)
            testValue = None
        return testValue

    # Helper function to extract values array from rxMsg
    def _getValueFromMsg(self, rxMsg):
        # Check to see only one property value is returned
        if len(rxMsg.value) != 1:
            self._log.error("getValueFromMsg:  Received invalid value")
            value = None
        else:
            valType = rxMsg.value[0].value_type
            try:
                if valType in self._types.TYPE_STRING:
                    value = rxMsg.value[0].string_value
                elif valType in self._types.TYPE_BYTES:
                    value = rxMsg.value[0].bytes_value
                elif valType == vhal_consts_2_0.VEHICLEPROPERTYTYPE_BOOLEAN:
                    value = rxMsg.value[0].int32_values[0]
                elif valType in self._types.TYPE_INT32:
                    value = rxMsg.value[0].int32_values[0]
                elif valType in self._types.TYPE_INT64:
                    value = rxMsg.value[0].int64_values[0]
                elif valType in self._types.TYPE_FLOAT:
                    value = rxMsg.value[0].float_values[0]
                    # Truncate float to 5 decimal places
                    value = "%.5f" % value
                    value = float(value)
                else:
                    self._log.error("getValueFromMsg:  valType=0x%X is not handled", valType)
                    value = None
            except IndexError:
              self._log.error("getValueFromMsg:  Received malformed message: %s", str(rxMsg))
              value = None
        return value

    def _validateVmsMessage(self, rxMsg):
        return (len(rxMsg.value) == 1 and rxMsg.value[0].value_type in self._types.TYPE_MIXED and
            len(rxMsg.value[0].int32_values) > 0 and
            vhal_consts_2_0.VMSMESSAGETYPE_SUBSCRIBE <= rxMsg.value[0].int32_values[0]
                <= vhal_consts_2_0.VMSMESSAGETYPE_LAST_VMS_MESSAGE_TYPE)

    def _getVmsMessageTypeFromMsg(self, rxMsg):
        if self._validateVmsMessage(rxMsg):
            value = rxMsg.value[0].int32_values[
                vhal_consts_2_0.VMSBASEMESSAGEINTEGERVALUESINDEX_MESSAGE_TYPE]
        else:
            self._log.error("getVmsMessageTypeFromMsg:  Received invalid message")
            value = None
        return value

    # Helper function to receive a message and validate the type and status
    #   retVal = 1 if no errors
    #   retVal = 0 if errors detected
    def _rxMsgAndValidate(self, expectedType, expectedStatus):
        retVal = 1
        rxMsg = self._vhal.rxMsg()
        if rxMsg.msg_type != expectedType:
            self._log.error("rxMsg Type expected: 0x%X, received: 0x%X", expectedType, rxMsg.msg_type)
            retVal = 0
        if rxMsg.status != expectedStatus:
            self._log.error("rxMsg Status expected: 0x%X, received: 0x%X", expectedStatus, rxMsg.status)
            retVal = 0
        return rxMsg, retVal

    # Calls getConfig() on each individual property ID and verifies it matches with the config
    #   received in getConfigAll()
    def testGetConfig(self):
        self._log.info("Starting testGetConfig...")
        for cfg in self._configs:
            self._log.debug("  Getting config for propId=0x%X", cfg.prop)
            self._vhal.getConfig(cfg.prop)
            rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.GET_CONFIG_RESP,
                                                   VehicleHalProto_pb2.RESULT_OK)
            if retVal:
                if rxMsg.config[0] != cfg:
                    self._log.error("testGetConfig failed.  prop=0x%X, expected:\n%s\nreceived:\n%s",
                               cfg.prop, str(cfg), str(rxMsg.config))
        self._log.info("  Finished testGetConfig!")

    # Calls getConfig() on invalid property ID and verifies it generates an error
    def testGetBadConfig(self):
        self._log.info("Starting testGetBadConfig...")
        for prop in self._badProps:
            self._log.debug("  Testing bad propId=0x%X", prop)
            self._vhal.getConfig(prop)
            rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.GET_CONFIG_RESP,
                                                   VehicleHalProto_pb2.ERROR_INVALID_PROPERTY)
            if retVal:
                for cfg in rxMsg.config:
                    self._log.error("testGetBadConfig  prop=0x%X, expected:None, received:\n%s",
                                    cfg.prop, str(rxMsg.config))
        self._log.info("  Finished testGetBadConfig!")

    def testGetPropertyAll(self):
        self._log.info("Starting testGetPropertyAll...")
        self._vhal.getPropertyAll()
        rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.GET_PROPERTY_ALL_RESP,
                                               VehicleHalProto_pb2.RESULT_OK)
        if retVal == 0:
            self._log.error("testGetPropertyAll:  Failed to receive proper rxMsg")

        # TODO: Finish writing this test.  What should we be testing, anyway?

        self._log.info("  Finished testGetPropertyAll!")

    def testGetSet(self):
        self._log.info("Starting testGetSet()...")
        for cfg in self._configs:
            if cfg.prop in self._skipProps:
                # Skip properties that cannot be handled properly by this test.
                self._log.warning("  Skipping propId=0x%X", cfg.prop)
                continue

            areas = cfg.supported_areas
            idx = -1
            while (idx == -1) | (areas != 0):
                idx += 1
                # Get the area to test
                area = areas & (areas -1)
                area ^= areas

                # Remove the area from areas
                areas ^= area

                self._log.debug("  Testing propId=0x%X, area=0x%X", cfg.prop, area)

                # Get the current value
                self._vhal.getProperty(cfg.prop, area)
                rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.GET_PROPERTY_RESP,
                                                       VehicleHalProto_pb2.RESULT_OK)

                # Save the original value
                origValue = self._getValueFromMsg(rxMsg)
                if origValue == None:
                    self._log.error("testGetSet:  Could not get value for prop=0x%X, area=0x%X",
                                    cfg.prop, area)
                    continue

                # Generate the test value
                testValue = self._generateTestValue(cfg, idx, origValue)
                if testValue == None:
                    self._log.error("testGetSet:  Cannot generate test value for prop=0x%X, area=0x%X",
                                    cfg.prop, area)
                    continue

                # Send the new value
                self._vhal.setProperty(cfg.prop, area, testValue)
                rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.SET_PROPERTY_RESP,
                                                        VehicleHalProto_pb2.RESULT_OK)

                # Get the new value and verify it
                self._vhal.getProperty(cfg.prop, area)
                rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.GET_PROPERTY_RESP,
                                                       VehicleHalProto_pb2.RESULT_OK)
                newValue = self._getValueFromMsg(rxMsg)
                if newValue != testValue:
                    self._log.error("testGetSet: set failed for propId=0x%X, area=0x%X", cfg.prop, area)
                    print("testValue= ", testValue, "newValue= ", newValue)
                    continue

                # Reset the value to what it was before
                self._vhal.setProperty(cfg.prop, area, origValue)
                rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.SET_PROPERTY_RESP,
                                                       VehicleHalProto_pb2.RESULT_OK)
        self._log.info("  Finished testGetSet()!")

    def testGetBadProperty(self):
        self._log.info("Starting testGetBadProperty()...")
        for prop in self._badProps:
            self._log.debug("  Testing bad propId=0x%X", prop)
            self._vhal.getProperty(prop, 0)
            rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.GET_PROPERTY_RESP,
                                                   VehicleHalProto_pb2.ERROR_INVALID_PROPERTY)
            if retVal:
                for value in rxMsg.value:
                    self._log.error("testGetBadProperty  prop=0x%X, expected:None, received:\n%s",
                                    prop, str(rxMsg))
        self._log.info("  Finished testGetBadProperty()!")

    def testSetBadProperty(self):
        self._log.info("Starting testSetBadProperty()...")
        area = 1
        value = 100
        for prop in self._badProps:
            self._log.debug("  Testing bad propId=0x%X", prop)
            area = area + 1
            value = value + 1
            try:
                self._vhal.setProperty(prop, area, value)
                self._log.error("testGetBadProperty failed.  prop=0x%X, area=0x%X, value=%d",
                                prop, area, value)
            except ValueError as e:
                # Received expected error
                pass
        self._log.info("  Finished testSetBadProperty()!")

    def testGetVmsAvailability(self):
        self._log.info("Starting testVms()...")

        # Request the availability from the VmsCore.
        value = {'int32_values' : [vhal_consts_2_0.VMSMESSAGETYPE_AVAILABILITY_REQUEST] }
        self._vhal.setProperty(
            vhal_consts_2_0.VEHICLEPROPERTY_VEHICLE_MAP_SERVICE, 0, value)

        rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.SET_PROPERTY_RESP,
                                               VehicleHalProto_pb2.RESULT_OK)

        # The Vms Core should immediately respond
        rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.SET_PROPERTY_ASYNC,
                                               VehicleHalProto_pb2.RESULT_OK)

        if self._getVmsMessageTypeFromMsg(rxMsg) != vhal_consts_2_0.VMSMESSAGETYPE_AVAILABILITY_RESPONSE:
            self._log.error("testVms: VmsCore did not respond with AvailabilityResponse: %s", str(rxMsg))


        # Test that we can get the property on command
        self._vhal.getProperty(
            vhal_consts_2_0.VEHICLEPROPERTY_VEHICLE_MAP_SERVICE, 0)
        rxMsg, retVal = self._rxMsgAndValidate(VehicleHalProto_pb2.GET_PROPERTY_RESP,
                                               VehicleHalProto_pb2.RESULT_OK)

        if self._getVmsMessageTypeFromMsg(rxMsg) != vhal_consts_2_0.VMSMESSAGETYPE_AVAILABILITY_RESPONSE:
            self._log.error("testVms: VmsCore did not respond with AvailabilityResponse: %s", str(rxMsg))
        else:
            # Parse Availability Response
            layers = rxMsg.value[0].int32_values[
                vhal_consts_2_0.VMSAVAILABILITYSTATEINTEGERVALUESINDEX_NUMBER_OF_ASSOCIATED_LAYERS]
            index = vhal_consts_2_0.VMSAVAILABILITYSTATEINTEGERVALUESINDEX_LAYERS_START
            numPublishersIndex = vhal_consts_2_0.VMSMESSAGEWITHLAYERINTEGERVALUESINDEX_LAYER_VERSION
            self._log.info("testVms: %d available layers", layers)
            for layer in xrange(layers):
                self._log.info("testVms: Available layer: %s",
                               rxMsg.value[0].int32_values[index:index+numPublishersIndex])
                index += numPublishersIndex + 1 + rxMsg.value[0].int32_values[index+numPublishersIndex]

            if len(rxMsg.value[0].int32_values) != index:
              self._log.error("testVms: Malformed AvailabilityResponse: index: %d %s", index, str(rxMsg))

    def runTests(self):
        self.testGetConfig()
        self.testGetBadConfig()
        self.testGetPropertyAll()
        self.testGetSet()
        self.testGetBadProperty()
        self.testSetBadProperty()
        self.testGetVmsAvailability()
        # Add new tests here to be run


    # Valid logLevels:
    #   CRITICAL    50
    #   ERRROR      40
    #   WARNING     30
    #   INFO        20
    #   DEBUG       10
    #   NOTSET      0
    def __init__(self, types, logLevel=20):
        self._types = types
        # Configure the logger
        logging.basicConfig()
        self._log = logging.getLogger('vhal_emulator_test')
        self._log.setLevel(logLevel)
        # Start the VHAL Emulator
        self._vhal = vhal_emulator.Vhal(types)
        # Get the list of configs
        self._vhal.getConfigAll()
        self._configs = self._vhal.rxMsg().config

if __name__ == '__main__':
    v = VhalTest(vhal_consts_2_0.vhal_types_2_0)
    v.runTests()