Loading android/pandora/mmi2grpc/mmi2grpc/gap.py +47 −48 Original line number Diff line number Diff line Loading @@ -6,9 +6,9 @@ from time import sleep from pandora_experimental.gatt_grpc import GATT from pandora_experimental.gatt_pb2 import GattServiceParams, GattCharacteristicParams from pandora.host_grpc import Host from pandora.host_pb2 import ConnectabilityMode, DataTypes, DiscoverabilityMode, OwnAddressType from pandora.security_grpc import Security, SecurityStorage, PairingEventAnswer from pandora.security_pb2 import LESecurityLevel from pandora.host_pb2 import PUBLIC, RANDOM, DISCOVERABLE_GENERAL, NOT_DISCOVERABLE, DISCOVERABLE_LIMITED, NOT_CONNECTABLE, DataTypes from pandora.security_grpc import Security, SecurityStorage from pandora.security_pb2 import LE_LEVEL3, PairingEventAnswer class GAPProxy(ProfileProxy): Loading Loading @@ -81,7 +81,7 @@ class GAPProxy(ProfileProxy): self.advertise = self.host.Advertise( connectable=True, own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, ) self.pairing_events = self.security.OnPairing() Loading Loading @@ -134,9 +134,9 @@ class GAPProxy(ProfileProxy): if self.counter == 0: self.counter += 1 self.security_storage.DeleteBond(public=pts_addr) self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" if test == "GAP/SEC/AUT/BV-21-C" and self.connection is not None: Loading @@ -158,9 +158,9 @@ class GAPProxy(ProfileProxy): scans.cancel() break self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=address).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=address).connection if test in {"GAP/BOND/BON/BV-04-C"}: self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" Loading Loading @@ -206,7 +206,7 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, data=DataTypes(complete_service_class_uuids128=["955798ce-3022-455c-b759-ee8edcd73d1a"],)) return "OK" Loading @@ -216,7 +216,7 @@ class GAPProxy(ProfileProxy): Please prepare IUT to send an advertising report with Local Name. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC, self.advertise = self.host.Advertise(own_address_type=PUBLIC, data=DataTypes( include_complete_local_name=True, include_shortened_local_name=True, Loading @@ -232,7 +232,7 @@ class GAPProxy(ProfileProxy): self.advertise = self.host.Advertise( connectable=True, own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, ) self.pairing_events = self.security.OnPairing() Loading @@ -246,7 +246,7 @@ class GAPProxy(ProfileProxy): Specific Data. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC, self.advertise = self.host.Advertise(own_address_type=PUBLIC, data=DataTypes(manufacturer_specific_data=b"d0n't b3 3v1l!",)) return "OK" Loading @@ -257,7 +257,7 @@ class GAPProxy(ProfileProxy): Please prepare IUT to send an advertising report with TX Power Level. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC, self.advertise = self.host.Advertise(own_address_type=PUBLIC, data=DataTypes(include_tx_power_level=True,)) return "OK" Loading @@ -269,7 +269,7 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, connectable=True, ) Loading @@ -284,7 +284,7 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, connectable=True, ) Loading Loading @@ -358,8 +358,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading @@ -376,8 +376,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading @@ -393,8 +393,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.NOT_DISCOVERABLE), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=NOT_DISCOVERABLE), own_address_type=PUBLIC, connectable=True, ) Loading @@ -407,8 +407,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading Loading @@ -458,7 +458,7 @@ class GAPProxy(ProfileProxy): for response in self.scan_responses: assert response.HasField("public") if (response.public == pts_addr and response.data.le_discoverability_mode == DiscoverabilityMode.DISCOVERABLE_LIMITED): response.data.le_discoverability_mode == DISCOVERABLE_LIMITED): self.scan_responses.cancel() return "OK" Loading @@ -477,7 +477,7 @@ class GAPProxy(ProfileProxy): for response in self.scan_responses: assert response.HasField("public") if (response.public == pts_addr and response.data.le_discoverability_mode == DiscoverabilityMode.DISCOVERABLE_GENERAL): response.data.le_discoverability_mode == DISCOVERABLE_GENERAL): self.scan_responses.cancel() discovered = True return Loading @@ -504,7 +504,7 @@ class GAPProxy(ProfileProxy): for response in self.scan_responses: assert response.HasField("public") if (response.public == pts_addr and response.data.le_discoverability_mode == DiscoverabilityMode.DISCOVERABLE_LIMITED): response.data.le_discoverability_mode == DISCOVERABLE_LIMITED): self.inquiry_responses.cancel() discovered = True return Loading @@ -525,7 +525,7 @@ class GAPProxy(ProfileProxy): send an advertising report. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC,) self.advertise = self.host.Advertise(own_address_type=PUBLIC,) return "OK" Loading Loading @@ -566,7 +566,7 @@ class GAPProxy(ProfileProxy): if test not in {"GAP/SEC/AUT/BV-21-C"}: self.security_storage.DeleteBond(public=pts_addr) self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" Loading Loading @@ -627,11 +627,11 @@ class GAPProxy(ProfileProxy): Please make IUT general discoverable. Press OK to continue. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.DISCOVERABLE_GENERAL) self.host.SetDiscoverabilityMode(mode=DISCOVERABLE_GENERAL) self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading @@ -658,9 +658,8 @@ class GAPProxy(ProfileProxy): Please make IUT not connectable. Press OK to continue. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetDiscoverabilityMode(mode=NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading @@ -670,8 +669,8 @@ class GAPProxy(ProfileProxy): Please make IUT not discoverable. Press OK to continue. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetDiscoverabilityMode(mode=NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading Loading @@ -704,7 +703,7 @@ class GAPProxy(ProfileProxy): """ # No idea how we can bond in non-bondable mode, but this passes the tests... self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" Loading Loading @@ -741,7 +740,7 @@ class GAPProxy(ProfileProxy): Please enter Non-Connectable mode. """ self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading @@ -751,8 +750,8 @@ class GAPProxy(ProfileProxy): Please enter General Discoverable and Non-Connectable mode. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.DISCOVERABLE_GENERAL) self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetDiscoverabilityMode(mode=DISCOVERABLE_GENERAL) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading @@ -765,8 +764,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=False, ) Loading @@ -780,8 +779,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.NOT_DISCOVERABLE), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=NOT_DISCOVERABLE), own_address_type=PUBLIC, connectable=True, ) Loading @@ -795,8 +794,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading Loading @@ -879,7 +878,7 @@ class GAPProxy(ProfileProxy): """ Please send an ATT connect request to establish an L2CAP channel. """ self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=pts_addr).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection return "OK" Loading Loading @@ -929,7 +928,7 @@ class GAPProxy(ProfileProxy): if test != "GAP/SEC/SEM/BV-08-C": # we already started in the Connect MMI self.pairing_events = self.security.OnPairing() self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) def after_that(): self.host.WaitConnection() # this really waits for bonding Loading android/pandora/mmi2grpc/mmi2grpc/gatt.py +49 −34 Original line number Diff line number Diff line Loading @@ -21,8 +21,23 @@ from mmi2grpc._proxy import ProfileProxy from pandora_experimental.gatt_grpc import GATT from pandora.host_grpc import Host from pandora.host_pb2 import ConnectabilityMode, OwnAddressType from pandora_experimental.gatt_pb2 import AttStatusCode, AttProperties, AttPermissions from pandora.host_pb2 import PUBLIC, RANDOM from pandora_experimental.gatt_pb2 import ( INVALID_HANDLE, READ_NOT_PERMITTED, UNKNOWN_ERROR, INSUFFICIENT_AUTHENTICATION, ATTRIBUTE_NOT_FOUND, APPLICATION_ERROR, INVALID_ATTRIBUTE_LENGTH, WRITE_NOT_PERMITTED, PERMISSION_NONE, PROPERTY_READ, PROPERTY_WRITE, PERMISSION_READ, PERMISSION_WRITE, PERMISSION_READ_ENCRYPTED, ) from pandora_experimental.gatt_pb2 import GattServiceParams from pandora_experimental.gatt_pb2 import GattCharacteristicParams from pandora_experimental.gatt_pb2 import ReadCharacteristicResponse Loading Loading @@ -71,7 +86,7 @@ class GATTProxy(ProfileProxy): PTS. """ self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=pts_addr).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection if test in NEEDS_CACHE_CLEARED: self.gatt.ClearCache(connection=self.connection) return "OK" Loading Loading @@ -474,10 +489,10 @@ class GATTProxy(ProfileProxy): """ if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.INVALID_HANDLE assert self.read_response.status == INVALID_HANDLE elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None assert AttStatusCode.INVALID_HANDLE in\ assert INVALID_HANDLE in\ list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) return "Yes" Loading @@ -496,14 +511,14 @@ class GATTProxy(ProfileProxy): # Android read error doesn't return an error code so we have to also # compare to the generic error code here. if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.READ_NOT_PERMITTED or\ self.read_response.status == AttStatusCode.UNKNOWN_ERROR assert self.read_response.status == READ_NOT_PERMITTED or\ self.read_response.status == UNKNOWN_ERROR elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None status_list = list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) assert AttStatusCode.READ_NOT_PERMITTED in status_list or\ AttStatusCode.UNKNOWN_ERROR in status_list assert READ_NOT_PERMITTED in status_list or\ UNKNOWN_ERROR in status_list return "Yes" @assert_description Loading @@ -518,10 +533,10 @@ class GATTProxy(ProfileProxy): """ if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.INSUFFICIENT_AUTHENTICATION assert self.read_response.status == INSUFFICIENT_AUTHENTICATION elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None assert AttStatusCode.INSUFFICIENT_AUTHENTICATION in\ assert INSUFFICIENT_AUTHENTICATION in\ list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) return "Yes" Loading Loading @@ -557,14 +572,14 @@ class GATTProxy(ProfileProxy): # Android read error doesn't return an error code so we have to also # compare to the generic error code here. if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.ATTRIBUTE_NOT_FOUND or\ self.read_response.status == AttStatusCode.UNKNOWN_ERROR assert self.read_response.status == ATTRIBUTE_NOT_FOUND or\ self.read_response.status == UNKNOWN_ERROR elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None status_list = list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) assert AttStatusCode.ATTRIBUTE_NOT_FOUND in status_list or\ AttStatusCode.UNKNOWN_ERROR in status_list assert ATTRIBUTE_NOT_FOUND in status_list or\ UNKNOWN_ERROR in status_list return "Yes" def MMI_IUT_SEND_READ_GREATER_OFFSET(self, description: str, **kwargs): Loading Loading @@ -609,10 +624,10 @@ class GATTProxy(ProfileProxy): """ if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.APPLICATION_ERROR assert self.read_response.status == APPLICATION_ERROR elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None assert AttStatusCode.APPLICATION_ERROR in\ assert APPLICATION_ERROR in\ list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) return "Yes" Loading Loading @@ -766,7 +781,7 @@ class GATTProxy(ProfileProxy): """ assert self.write_response is not None assert self.write_response.status == AttStatusCode.INVALID_HANDLE assert self.write_response.status == INVALID_HANDLE return "Yes" @assert_description Loading @@ -781,7 +796,7 @@ class GATTProxy(ProfileProxy): """ assert self.write_response is not None assert self.write_response.status == AttStatusCode.WRITE_NOT_PERMITTED assert self.write_response.status == WRITE_NOT_PERMITTED return "Yes" def MMI_IUT_SEND_PREPARE_WRITE(self, description: str, **kwargs): Loading Loading @@ -870,7 +885,7 @@ class GATTProxy(ProfileProxy): assert self.write_response is not None # See MMI_IUT_SEND_PREPARE_WRITE_GREATER_OFFSET if self.written_over_length == True: assert self.write_response.status == AttStatusCode.INVALID_ATTRIBUTE_LENGTH assert self.write_response.status == INVALID_ATTRIBUTE_LENGTH return "OK" def MMI_IUT_SEND_WRITE_REQUEST_GREATER(self, description: str, **kwargs): Loading Loading @@ -902,7 +917,7 @@ class GATTProxy(ProfileProxy): """ assert self.write_response is not None assert self.write_response.status == AttStatusCode.INVALID_ATTRIBUTE_LENGTH assert self.write_response.status == INVALID_ATTRIBUTE_LENGTH return "OK" def MMI_IUT_SEND_PREPARE_WRITE_REQUEST_GREATER(self, description: str, **kwargs): Loading Loading @@ -949,7 +964,7 @@ class GATTProxy(ProfileProxy): """ self.advertise = self.host.Advertise( connectable=True, own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, ) self.gatt.RegisterService( service=GattServiceParams( Loading @@ -957,13 +972,13 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=BASE_READ_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_READ, properties=PROPERTY_READ, permissions=PERMISSION_READ, ), GattCharacteristicParams( uuid=BASE_WRITE_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_WRITE, permissions=AttPermissions.PERMISSION_WRITE, properties=PROPERTY_WRITE, permissions=PERMISSION_WRITE, ), ], )) Loading Loading @@ -1042,8 +1057,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_NONE, properties=PROPERTY_READ, permissions=PERMISSION_NONE, ), ], )) Loading Loading @@ -1086,8 +1101,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_READ_ENCRYPTED, properties=PROPERTY_READ, permissions=PERMISSION_READ_ENCRYPTED, ), ], )) Loading Loading @@ -1119,8 +1134,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_NONE, properties=PROPERTY_READ, permissions=PERMISSION_NONE, ), ], )) Loading Loading @@ -1152,8 +1167,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_WRITE, permissions=AttPermissions.PERMISSION_NONE, properties=PROPERTY_WRITE, permissions=PERMISSION_NONE, ), ], )) Loading android/pandora/mmi2grpc/mmi2grpc/hfp.py +9 −8 Original line number Diff line number Diff line Loading @@ -18,9 +18,10 @@ from mmi2grpc._proxy import ProfileProxy from pandora_experimental.hfp_grpc import HFP from pandora.host_grpc import Host from pandora.host_pb2 import ConnectabilityMode, DiscoverabilityMode from pandora.security_grpc import Security, SecurityStorage, PairingEventAnswer from pandora_experimental.hfp_pb2 import AudioPath from pandora.host_pb2 import DISCOVERABLE_GENERAL, CONNECTABLE from pandora.security_grpc import Security, SecurityStorage from pandora.security_pb2 import PairingEventAnswer from pandora_experimental.hfp_pb2 import AUDIO_PATH_HANDSFREE, AUDIO_PATH_SPEAKERS import sys import threading Loading Loading @@ -85,7 +86,7 @@ class HFPProxy(ProfileProxy): time.sleep(2) if test == "HFP/AG/SLC/BV-02-C": self.host.SetConnectabilityMode(mode=ConnectabilityMode.CONNECTABLE) self.host.SetConnectabilityMode(mode=CONNECTABLE) self.connection = self.host.Connect(address=pts_addr).connection else: if not self.connection: Loading Loading @@ -130,7 +131,7 @@ class HFPProxy(ProfileProxy): Make the Implementation Under Test (IUT) connectable, then click Ok. """ self.host.SetConnectabilityMode(mode=ConnectabilityMode.CONNECTABLE) self.host.SetConnectabilityMode(mode=CONNECTABLE) return "OK" Loading Loading @@ -236,7 +237,7 @@ class HFPProxy(ProfileProxy): if "HFP/HF" in test: self.hfp.DisconnectToAudioAsHandsfree(connection=self.connection) else: self.hfp.SetAudioPath(audio_path=AudioPath.AUDIO_PATH_SPEAKERS) self.hfp.SetAudioPath(audio_path=AUDIO_PATH_SPEAKERS) threading.Thread(target=disable_audio).start() Loading @@ -262,7 +263,7 @@ class HFPProxy(ProfileProxy): if "HFP/HF" in test: self.hfp.ConnectToAudioAsHandsfree(connection=self.connection) else: self.hfp.SetAudioPath(audio_path=AudioPath.AUDIO_PATH_HANDSFREE) self.hfp.SetAudioPath(audio_path=AUDIO_PATH_HANDSFREE) threading.Thread(target=enable_audio).start() Loading Loading @@ -772,7 +773,7 @@ class HFPProxy(ProfileProxy): click Ok. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.DISCOVERABLE_GENERAL) self.host.SetDiscoverabilityMode(mode=DISCOVERABLE_GENERAL) return "OK" Loading android/pandora/mmi2grpc/mmi2grpc/hogp.py +5 −5 Original line number Diff line number Diff line Loading @@ -7,9 +7,9 @@ from mmi2grpc._helpers import assert_description, match_description from mmi2grpc._proxy import ProfileProxy from pandora.host_grpc import Host from pandora.host_pb2 import OwnAddressType from pandora.security_grpc import Security, PairingEventAnswer from pandora.security_pb2 import LESecurityLevel from pandora.host_pb2 import RANDOM from pandora.security_grpc import Security from pandora.security_pb2 import LE_LEVEL3, PairingEventAnswer from pandora_experimental.gatt_grpc import GATT BASE_UUID = uuid.UUID("00000000-0000-1000-8000-00805F9B34FB") Loading Loading @@ -40,10 +40,10 @@ class HOGPProxy(ProfileProxy): to the PTS. """ self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=pts_addr).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection self.pairing_stream = self.security.OnPairing() def secure(): self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) threading.Thread(target=secure).start() return "OK" Loading android/pandora/mmi2grpc/mmi2grpc/l2cap.py +5 −4 File changed.Preview size limit exceeded, changes collapsed. Show changes Loading
android/pandora/mmi2grpc/mmi2grpc/gap.py +47 −48 Original line number Diff line number Diff line Loading @@ -6,9 +6,9 @@ from time import sleep from pandora_experimental.gatt_grpc import GATT from pandora_experimental.gatt_pb2 import GattServiceParams, GattCharacteristicParams from pandora.host_grpc import Host from pandora.host_pb2 import ConnectabilityMode, DataTypes, DiscoverabilityMode, OwnAddressType from pandora.security_grpc import Security, SecurityStorage, PairingEventAnswer from pandora.security_pb2 import LESecurityLevel from pandora.host_pb2 import PUBLIC, RANDOM, DISCOVERABLE_GENERAL, NOT_DISCOVERABLE, DISCOVERABLE_LIMITED, NOT_CONNECTABLE, DataTypes from pandora.security_grpc import Security, SecurityStorage from pandora.security_pb2 import LE_LEVEL3, PairingEventAnswer class GAPProxy(ProfileProxy): Loading Loading @@ -81,7 +81,7 @@ class GAPProxy(ProfileProxy): self.advertise = self.host.Advertise( connectable=True, own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, ) self.pairing_events = self.security.OnPairing() Loading Loading @@ -134,9 +134,9 @@ class GAPProxy(ProfileProxy): if self.counter == 0: self.counter += 1 self.security_storage.DeleteBond(public=pts_addr) self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" if test == "GAP/SEC/AUT/BV-21-C" and self.connection is not None: Loading @@ -158,9 +158,9 @@ class GAPProxy(ProfileProxy): scans.cancel() break self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=address).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=address).connection if test in {"GAP/BOND/BON/BV-04-C"}: self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" Loading Loading @@ -206,7 +206,7 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, data=DataTypes(complete_service_class_uuids128=["955798ce-3022-455c-b759-ee8edcd73d1a"],)) return "OK" Loading @@ -216,7 +216,7 @@ class GAPProxy(ProfileProxy): Please prepare IUT to send an advertising report with Local Name. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC, self.advertise = self.host.Advertise(own_address_type=PUBLIC, data=DataTypes( include_complete_local_name=True, include_shortened_local_name=True, Loading @@ -232,7 +232,7 @@ class GAPProxy(ProfileProxy): self.advertise = self.host.Advertise( connectable=True, own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, ) self.pairing_events = self.security.OnPairing() Loading @@ -246,7 +246,7 @@ class GAPProxy(ProfileProxy): Specific Data. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC, self.advertise = self.host.Advertise(own_address_type=PUBLIC, data=DataTypes(manufacturer_specific_data=b"d0n't b3 3v1l!",)) return "OK" Loading @@ -257,7 +257,7 @@ class GAPProxy(ProfileProxy): Please prepare IUT to send an advertising report with TX Power Level. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC, self.advertise = self.host.Advertise(own_address_type=PUBLIC, data=DataTypes(include_tx_power_level=True,)) return "OK" Loading @@ -269,7 +269,7 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, connectable=True, ) Loading @@ -284,7 +284,7 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, connectable=True, ) Loading Loading @@ -358,8 +358,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading @@ -376,8 +376,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading @@ -393,8 +393,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.NOT_DISCOVERABLE), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=NOT_DISCOVERABLE), own_address_type=PUBLIC, connectable=True, ) Loading @@ -407,8 +407,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading Loading @@ -458,7 +458,7 @@ class GAPProxy(ProfileProxy): for response in self.scan_responses: assert response.HasField("public") if (response.public == pts_addr and response.data.le_discoverability_mode == DiscoverabilityMode.DISCOVERABLE_LIMITED): response.data.le_discoverability_mode == DISCOVERABLE_LIMITED): self.scan_responses.cancel() return "OK" Loading @@ -477,7 +477,7 @@ class GAPProxy(ProfileProxy): for response in self.scan_responses: assert response.HasField("public") if (response.public == pts_addr and response.data.le_discoverability_mode == DiscoverabilityMode.DISCOVERABLE_GENERAL): response.data.le_discoverability_mode == DISCOVERABLE_GENERAL): self.scan_responses.cancel() discovered = True return Loading @@ -504,7 +504,7 @@ class GAPProxy(ProfileProxy): for response in self.scan_responses: assert response.HasField("public") if (response.public == pts_addr and response.data.le_discoverability_mode == DiscoverabilityMode.DISCOVERABLE_LIMITED): response.data.le_discoverability_mode == DISCOVERABLE_LIMITED): self.inquiry_responses.cancel() discovered = True return Loading @@ -525,7 +525,7 @@ class GAPProxy(ProfileProxy): send an advertising report. """ self.advertise = self.host.Advertise(own_address_type=OwnAddressType.PUBLIC,) self.advertise = self.host.Advertise(own_address_type=PUBLIC,) return "OK" Loading Loading @@ -566,7 +566,7 @@ class GAPProxy(ProfileProxy): if test not in {"GAP/SEC/AUT/BV-21-C"}: self.security_storage.DeleteBond(public=pts_addr) self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" Loading Loading @@ -627,11 +627,11 @@ class GAPProxy(ProfileProxy): Please make IUT general discoverable. Press OK to continue. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.DISCOVERABLE_GENERAL) self.host.SetDiscoverabilityMode(mode=DISCOVERABLE_GENERAL) self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading @@ -658,9 +658,8 @@ class GAPProxy(ProfileProxy): Please make IUT not connectable. Press OK to continue. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetDiscoverabilityMode(mode=NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading @@ -670,8 +669,8 @@ class GAPProxy(ProfileProxy): Please make IUT not discoverable. Press OK to continue. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetDiscoverabilityMode(mode=NOT_DISCOVERABLE) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading Loading @@ -704,7 +703,7 @@ class GAPProxy(ProfileProxy): """ # No idea how we can bond in non-bondable mode, but this passes the tests... self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) return "OK" Loading Loading @@ -741,7 +740,7 @@ class GAPProxy(ProfileProxy): Please enter Non-Connectable mode. """ self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading @@ -751,8 +750,8 @@ class GAPProxy(ProfileProxy): Please enter General Discoverable and Non-Connectable mode. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.DISCOVERABLE_GENERAL) self.host.SetConnectabilityMode(mode=ConnectabilityMode.NOT_CONNECTABLE) self.host.SetDiscoverabilityMode(mode=DISCOVERABLE_GENERAL) self.host.SetConnectabilityMode(mode=NOT_CONNECTABLE) return "OK" Loading @@ -765,8 +764,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=False, ) Loading @@ -780,8 +779,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.NOT_DISCOVERABLE), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=NOT_DISCOVERABLE), own_address_type=PUBLIC, connectable=True, ) Loading @@ -795,8 +794,8 @@ class GAPProxy(ProfileProxy): """ self.advertise = self.host.Advertise( data=DataTypes(le_discoverability_mode=DiscoverabilityMode.DISCOVERABLE_GENERAL), own_address_type=OwnAddressType.PUBLIC, data=DataTypes(le_discoverability_mode=DISCOVERABLE_GENERAL), own_address_type=PUBLIC, connectable=True, ) Loading Loading @@ -879,7 +878,7 @@ class GAPProxy(ProfileProxy): """ Please send an ATT connect request to establish an L2CAP channel. """ self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=pts_addr).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection return "OK" Loading Loading @@ -929,7 +928,7 @@ class GAPProxy(ProfileProxy): if test != "GAP/SEC/SEM/BV-08-C": # we already started in the Connect MMI self.pairing_events = self.security.OnPairing() self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) def after_that(): self.host.WaitConnection() # this really waits for bonding Loading
android/pandora/mmi2grpc/mmi2grpc/gatt.py +49 −34 Original line number Diff line number Diff line Loading @@ -21,8 +21,23 @@ from mmi2grpc._proxy import ProfileProxy from pandora_experimental.gatt_grpc import GATT from pandora.host_grpc import Host from pandora.host_pb2 import ConnectabilityMode, OwnAddressType from pandora_experimental.gatt_pb2 import AttStatusCode, AttProperties, AttPermissions from pandora.host_pb2 import PUBLIC, RANDOM from pandora_experimental.gatt_pb2 import ( INVALID_HANDLE, READ_NOT_PERMITTED, UNKNOWN_ERROR, INSUFFICIENT_AUTHENTICATION, ATTRIBUTE_NOT_FOUND, APPLICATION_ERROR, INVALID_ATTRIBUTE_LENGTH, WRITE_NOT_PERMITTED, PERMISSION_NONE, PROPERTY_READ, PROPERTY_WRITE, PERMISSION_READ, PERMISSION_WRITE, PERMISSION_READ_ENCRYPTED, ) from pandora_experimental.gatt_pb2 import GattServiceParams from pandora_experimental.gatt_pb2 import GattCharacteristicParams from pandora_experimental.gatt_pb2 import ReadCharacteristicResponse Loading Loading @@ -71,7 +86,7 @@ class GATTProxy(ProfileProxy): PTS. """ self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=pts_addr).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection if test in NEEDS_CACHE_CLEARED: self.gatt.ClearCache(connection=self.connection) return "OK" Loading Loading @@ -474,10 +489,10 @@ class GATTProxy(ProfileProxy): """ if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.INVALID_HANDLE assert self.read_response.status == INVALID_HANDLE elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None assert AttStatusCode.INVALID_HANDLE in\ assert INVALID_HANDLE in\ list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) return "Yes" Loading @@ -496,14 +511,14 @@ class GATTProxy(ProfileProxy): # Android read error doesn't return an error code so we have to also # compare to the generic error code here. if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.READ_NOT_PERMITTED or\ self.read_response.status == AttStatusCode.UNKNOWN_ERROR assert self.read_response.status == READ_NOT_PERMITTED or\ self.read_response.status == UNKNOWN_ERROR elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None status_list = list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) assert AttStatusCode.READ_NOT_PERMITTED in status_list or\ AttStatusCode.UNKNOWN_ERROR in status_list assert READ_NOT_PERMITTED in status_list or\ UNKNOWN_ERROR in status_list return "Yes" @assert_description Loading @@ -518,10 +533,10 @@ class GATTProxy(ProfileProxy): """ if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.INSUFFICIENT_AUTHENTICATION assert self.read_response.status == INSUFFICIENT_AUTHENTICATION elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None assert AttStatusCode.INSUFFICIENT_AUTHENTICATION in\ assert INSUFFICIENT_AUTHENTICATION in\ list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) return "Yes" Loading Loading @@ -557,14 +572,14 @@ class GATTProxy(ProfileProxy): # Android read error doesn't return an error code so we have to also # compare to the generic error code here. if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.ATTRIBUTE_NOT_FOUND or\ self.read_response.status == AttStatusCode.UNKNOWN_ERROR assert self.read_response.status == ATTRIBUTE_NOT_FOUND or\ self.read_response.status == UNKNOWN_ERROR elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None status_list = list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) assert AttStatusCode.ATTRIBUTE_NOT_FOUND in status_list or\ AttStatusCode.UNKNOWN_ERROR in status_list assert ATTRIBUTE_NOT_FOUND in status_list or\ UNKNOWN_ERROR in status_list return "Yes" def MMI_IUT_SEND_READ_GREATER_OFFSET(self, description: str, **kwargs): Loading Loading @@ -609,10 +624,10 @@ class GATTProxy(ProfileProxy): """ if type(self.read_response) is ReadCharacteristicResponse: assert self.read_response.status == AttStatusCode.APPLICATION_ERROR assert self.read_response.status == APPLICATION_ERROR elif type(self.read_response) is ReadCharacteristicsFromUuidResponse: assert self.read_response.characteristics_read is not None assert AttStatusCode.APPLICATION_ERROR in\ assert APPLICATION_ERROR in\ list(map(lambda characteristic_read: characteristic_read.status,\ self.read_response.characteristics_read)) return "Yes" Loading Loading @@ -766,7 +781,7 @@ class GATTProxy(ProfileProxy): """ assert self.write_response is not None assert self.write_response.status == AttStatusCode.INVALID_HANDLE assert self.write_response.status == INVALID_HANDLE return "Yes" @assert_description Loading @@ -781,7 +796,7 @@ class GATTProxy(ProfileProxy): """ assert self.write_response is not None assert self.write_response.status == AttStatusCode.WRITE_NOT_PERMITTED assert self.write_response.status == WRITE_NOT_PERMITTED return "Yes" def MMI_IUT_SEND_PREPARE_WRITE(self, description: str, **kwargs): Loading Loading @@ -870,7 +885,7 @@ class GATTProxy(ProfileProxy): assert self.write_response is not None # See MMI_IUT_SEND_PREPARE_WRITE_GREATER_OFFSET if self.written_over_length == True: assert self.write_response.status == AttStatusCode.INVALID_ATTRIBUTE_LENGTH assert self.write_response.status == INVALID_ATTRIBUTE_LENGTH return "OK" def MMI_IUT_SEND_WRITE_REQUEST_GREATER(self, description: str, **kwargs): Loading Loading @@ -902,7 +917,7 @@ class GATTProxy(ProfileProxy): """ assert self.write_response is not None assert self.write_response.status == AttStatusCode.INVALID_ATTRIBUTE_LENGTH assert self.write_response.status == INVALID_ATTRIBUTE_LENGTH return "OK" def MMI_IUT_SEND_PREPARE_WRITE_REQUEST_GREATER(self, description: str, **kwargs): Loading Loading @@ -949,7 +964,7 @@ class GATTProxy(ProfileProxy): """ self.advertise = self.host.Advertise( connectable=True, own_address_type=OwnAddressType.PUBLIC, own_address_type=PUBLIC, ) self.gatt.RegisterService( service=GattServiceParams( Loading @@ -957,13 +972,13 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=BASE_READ_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_READ, properties=PROPERTY_READ, permissions=PERMISSION_READ, ), GattCharacteristicParams( uuid=BASE_WRITE_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_WRITE, permissions=AttPermissions.PERMISSION_WRITE, properties=PROPERTY_WRITE, permissions=PERMISSION_WRITE, ), ], )) Loading Loading @@ -1042,8 +1057,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_NONE, properties=PROPERTY_READ, permissions=PERMISSION_NONE, ), ], )) Loading Loading @@ -1086,8 +1101,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_READ_ENCRYPTED, properties=PROPERTY_READ, permissions=PERMISSION_READ_ENCRYPTED, ), ], )) Loading Loading @@ -1119,8 +1134,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_READ, permissions=AttPermissions.PERMISSION_NONE, properties=PROPERTY_READ, permissions=PERMISSION_NONE, ), ], )) Loading Loading @@ -1152,8 +1167,8 @@ class GATTProxy(ProfileProxy): characteristics=[ GattCharacteristicParams( uuid=CUSTOM_CHARACTERISTIC_UUID, properties=AttProperties.PROPERTY_WRITE, permissions=AttPermissions.PERMISSION_NONE, properties=PROPERTY_WRITE, permissions=PERMISSION_NONE, ), ], )) Loading
android/pandora/mmi2grpc/mmi2grpc/hfp.py +9 −8 Original line number Diff line number Diff line Loading @@ -18,9 +18,10 @@ from mmi2grpc._proxy import ProfileProxy from pandora_experimental.hfp_grpc import HFP from pandora.host_grpc import Host from pandora.host_pb2 import ConnectabilityMode, DiscoverabilityMode from pandora.security_grpc import Security, SecurityStorage, PairingEventAnswer from pandora_experimental.hfp_pb2 import AudioPath from pandora.host_pb2 import DISCOVERABLE_GENERAL, CONNECTABLE from pandora.security_grpc import Security, SecurityStorage from pandora.security_pb2 import PairingEventAnswer from pandora_experimental.hfp_pb2 import AUDIO_PATH_HANDSFREE, AUDIO_PATH_SPEAKERS import sys import threading Loading Loading @@ -85,7 +86,7 @@ class HFPProxy(ProfileProxy): time.sleep(2) if test == "HFP/AG/SLC/BV-02-C": self.host.SetConnectabilityMode(mode=ConnectabilityMode.CONNECTABLE) self.host.SetConnectabilityMode(mode=CONNECTABLE) self.connection = self.host.Connect(address=pts_addr).connection else: if not self.connection: Loading Loading @@ -130,7 +131,7 @@ class HFPProxy(ProfileProxy): Make the Implementation Under Test (IUT) connectable, then click Ok. """ self.host.SetConnectabilityMode(mode=ConnectabilityMode.CONNECTABLE) self.host.SetConnectabilityMode(mode=CONNECTABLE) return "OK" Loading Loading @@ -236,7 +237,7 @@ class HFPProxy(ProfileProxy): if "HFP/HF" in test: self.hfp.DisconnectToAudioAsHandsfree(connection=self.connection) else: self.hfp.SetAudioPath(audio_path=AudioPath.AUDIO_PATH_SPEAKERS) self.hfp.SetAudioPath(audio_path=AUDIO_PATH_SPEAKERS) threading.Thread(target=disable_audio).start() Loading @@ -262,7 +263,7 @@ class HFPProxy(ProfileProxy): if "HFP/HF" in test: self.hfp.ConnectToAudioAsHandsfree(connection=self.connection) else: self.hfp.SetAudioPath(audio_path=AudioPath.AUDIO_PATH_HANDSFREE) self.hfp.SetAudioPath(audio_path=AUDIO_PATH_HANDSFREE) threading.Thread(target=enable_audio).start() Loading Loading @@ -772,7 +773,7 @@ class HFPProxy(ProfileProxy): click Ok. """ self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.DISCOVERABLE_GENERAL) self.host.SetDiscoverabilityMode(mode=DISCOVERABLE_GENERAL) return "OK" Loading
android/pandora/mmi2grpc/mmi2grpc/hogp.py +5 −5 Original line number Diff line number Diff line Loading @@ -7,9 +7,9 @@ from mmi2grpc._helpers import assert_description, match_description from mmi2grpc._proxy import ProfileProxy from pandora.host_grpc import Host from pandora.host_pb2 import OwnAddressType from pandora.security_grpc import Security, PairingEventAnswer from pandora.security_pb2 import LESecurityLevel from pandora.host_pb2 import RANDOM from pandora.security_grpc import Security from pandora.security_pb2 import LE_LEVEL3, PairingEventAnswer from pandora_experimental.gatt_grpc import GATT BASE_UUID = uuid.UUID("00000000-0000-1000-8000-00805F9B34FB") Loading Loading @@ -40,10 +40,10 @@ class HOGPProxy(ProfileProxy): to the PTS. """ self.connection = self.host.ConnectLE(own_address_type=OwnAddressType.RANDOM, public=pts_addr).connection self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection self.pairing_stream = self.security.OnPairing() def secure(): self.security.Secure(connection=self.connection, le=LESecurityLevel.LE_LEVEL3) self.security.Secure(connection=self.connection, le=LE_LEVEL3) threading.Thread(target=secure).start() return "OK" Loading
android/pandora/mmi2grpc/mmi2grpc/l2cap.py +5 −4 File changed.Preview size limit exceeded, changes collapsed. Show changes