Loading android/pandora/mmi2grpc/mmi2grpc/sm.py +81 −40 Original line number Diff line number Diff line Loading @@ -12,45 +12,23 @@ # See the License for the specific language governing permissions and # limitations under the License. """SMP proxy module.""" from queue import Empty, Queue from threading import Thread import sys import time from mmi2grpc._helpers import assert_description from mmi2grpc._helpers import assert_description, match_description from mmi2grpc._proxy import ProfileProxy from mmi2grpc._streaming import StreamWrapper from pandora_experimental.security_grpc import Security from pandora_experimental.host_grpc import Host # The tests needs the MMI to accept pairing confirmation request. NEEDS_PAIRING_CONFIRMATION = { "SM/CEN/EKS/BV-01-C", "SM/CEN/JW/BI-04-C", "SM/CEN/JW/BI-01-C", "SM/CEN/KDU/BV-04-C", "SM/CEN/KDU/BV-05-C", "SM/CEN/KDU/BV-06-C", "SM/CEN/KDU/BV-10-C", "SM/CEN/KDU/BV-11-C", } ACCEPTS_REMOTE_PAIRING_CONFIRMATION = { "SM/CEN/KDU/BI-01-C", "SM/CEN/KDU/BI-02-C", "SM/CEN/KDU/BI-03-C", } def debug(*args, **kwargs): print(*args, file=sys.stderr, **kwargs) def get_event(pairing_stream: StreamWrapper, addr: str): for event in pairing_stream: if event.address == addr: return event return None class SMProxy(ProfileProxy): def __init__(self, channel): Loading @@ -59,32 +37,24 @@ class SMProxy(ProfileProxy): self.host = Host(channel) self.connection = None self.pairing_stream = None self.passkey_queue = Queue() self._handle_pairing_requests() @assert_description def MMI_IUT_ENABLE_CONNECTION_SM(self, test, pts_addr: bytes, **kwargs): def MMI_IUT_ENABLE_CONNECTION_SM(self, pts_addr: bytes, **kwargs): """ Initiate an connection from the IUT to the PTS. """ self.connection = self.host.ConnectLE(address=pts_addr).connection self.pairing_stream = self.security.OnPairing() if self.connection and test in ACCEPTS_REMOTE_PAIRING_CONFIRMATION: event = get_event(pairing_stream=self.pairing_stream, addr=pts_addr) self.pairing_stream.send(event=event, confirm=True) self.pairing_stream.close() return "OK" @assert_description def MMI_ASK_IUT_PERFORM_PAIRING_PROCESS(self, test, pts_addr: bytes, **kwargs): def MMI_ASK_IUT_PERFORM_PAIRING_PROCESS(self, **kwargs): """ Please start pairing process. """ if self.connection: self.security.Pair(connection=self.connection) if test in NEEDS_PAIRING_CONFIRMATION: event = get_event(pairing_stream=self.pairing_stream, addr=pts_addr) self.pairing_stream.send(event=event, confirm=True) self.pairing_stream.close() return "OK" @assert_description Loading @@ -104,7 +74,6 @@ class SMProxy(ProfileProxy): """ Please confirm the following number matches IUT: 385874. """ return "OK" @assert_description Loading @@ -114,3 +83,75 @@ class SMProxy(ProfileProxy): """ self.host.SoftReset() return "OK" @assert_description def MMI_TESTER_ENABLE_CONNECTION_SM(self, **kwargs): """ Action: Place the IUT in connectable mode """ self.host.SetLEConnectable() return "OK" @assert_description def MMI_IUT_SMP_TIMEOUT_30_SECONDS(self, **kwargs): """ Wait for the 30 seconds. Lower tester will not send corresponding or next SMP message. """ return "OK" @assert_description def MMI_IUT_SMP_TIMEOUT_ADDITIONAL_10_SECONDS(self, **kwargs): """ Wait for an additional 10 seconds. Lower test will send corresponding or next SMP message. """ return "OK" @match_description def MMI_DISPLAY_PASSKEY_CODE(self, passkey: str, **kwargs): """ Please enter (?P<passkey>[0-9]*) in the IUT. """ self.passkey_queue.put(passkey) return "OK" @assert_description def MMI_ENTER_PASSKEY_CODE(self, **kwargs): """ Please enter 6 digit passkey code. """ return "OK" @assert_description def MMI_ENTER_WRONG_DYNAMIC_PASSKEY_CODE(self, **kwargs): """ Please enter invalid 6 digit pin code. """ return "OK" @assert_description def MMI_IUT_ABORT_PAIRING_PROCESS_DISCONNECT(self, **kwargs): """ Lower tester expects IUT aborts pairing process, and disconnect. """ return "OK" def _handle_pairing_requests(self): def task(): pairing_events = self.security.OnPairing() for event in pairing_events: if event.just_works: pairing_events.send(event=event, confirm=True) if event.passkey_entry_request: try: passkey = self.passkey_queue.get(timeout=15) pairing_events.send(event=event, passkey=int(passkey)) except Empty: debug("No passkey provided within 15 seconds") Thread(target=task).start() android/pandora/server/configs/PtsBotTest.xml +12 −0 Original line number Diff line number Diff line Loading @@ -47,5 +47,17 @@ <option name="profile" value="SM/CEN/EKS" /> <option name="profile" value="SM/CEN/JW" /> <option name="profile" value="SM/CEN/KDU" /> <option name="profile" value="SM/CEN/PROT" /> <option name="profile" value="SM/CEN/PKE" /> <option name="profile" value="SM/CEN/SCJW" /> <option name="profile" value="SM/CEN/SCPK" /> <option name="profile" value="SM/PER/PROT" /> <option name="profile" value="SM/PER/JW" /> <option name="profile" value="SM/PER/PKE" /> <option name="profile" value="SM/PER/EKS" /> <option name="profile" value="SM/PER/KDU" /> <option name="profile" value="SM/PER/SCJW" /> <option name="profile" value="SM/PER/SCPK" /> <option name="profile" value="HOGP/RH" /> </test> </configuration> android/pandora/server/configs/pts_bot_tests_config.json +37 −2 Original line number Diff line number Diff line Loading @@ -225,7 +225,34 @@ "SM/CEN/KDU/BV-06-C", "SM/CEN/KDU/BV-10-C", "SM/CEN/KDU/BI-03-C", "SM/CEN/KDU/BV-11-C" "SM/CEN/KDU/BV-11-C", "SM/CEN/PROT/BV-01-C", "SM/CEN/PKE/BV-04-C", "SM/CEN/PKE/BI-01-C", "SM/CEN/PKE/BI-02-C", "SM/CEN/SCJW/BV-04-C", "SM/CEN/SCJW/BI-01-C", "SM/CEN/SCPK/BI-01-C", "SM/PER/PROT/BV-02-C", "SM/PER/JW/BV-02-C", "SM/PER/JW/BI-03-C", "SM/PER/JW/BI-02-C", "SM/PER/PKE/BV-02-C", "SM/PER/PKE/BV-05-C", "SM/PER/PKE/BI-03-C", "SM/PER/EKS/BV-02-C", "SM/PER/EKS/BI-02-C", "SM/PER/KDU/BV-01-C", "SM/PER/KDU/BV-02-C", "SM/PER/KDU/BV-03-C", "SM/PER/KDU/BV-07-C", "SM/PER/KDU/BV-08-C", "SM/PER/KDU/BV-09-C", "SM/PER/KDU/BI-01-C", "SM/PER/SCJW/BV-03-C", "SM/PER/SCJW/BI-02-C", "SM/PER/SCPK/BV-02-C", "SM/PER/SCPK/BI-03-C" ], "skip": [ "A2DP/SRC/AS/BV-01-I", Loading Loading @@ -370,7 +397,15 @@ "L2CAP/LE/CFC/BV-07-C", "L2CAP/LE/CFC/BV-11-C", "L2CAP/LE/CFC/BV-13-C", "L2CAP/LE/CFC/BV-15-C" "L2CAP/LE/CFC/BV-15-C", "SM/CEN/PKE/BV-01-C", "SM/CEN/SCJW/BV-01-C", "SM/CEN/SCPK/BI-02-C", "SM/CEN/SCPK/BV-01-C", "SM/CEN/SCPK/BV-04-C", "SM/PER/SCJW/BV-02-C", "SM/PER/SCPK/BI-04-C", "SM/PER/SCPK/BV-03-C" ], "ics": { "TSPC_4.0HCI_1a_2": true, Loading Loading
android/pandora/mmi2grpc/mmi2grpc/sm.py +81 −40 Original line number Diff line number Diff line Loading @@ -12,45 +12,23 @@ # See the License for the specific language governing permissions and # limitations under the License. """SMP proxy module.""" from queue import Empty, Queue from threading import Thread import sys import time from mmi2grpc._helpers import assert_description from mmi2grpc._helpers import assert_description, match_description from mmi2grpc._proxy import ProfileProxy from mmi2grpc._streaming import StreamWrapper from pandora_experimental.security_grpc import Security from pandora_experimental.host_grpc import Host # The tests needs the MMI to accept pairing confirmation request. NEEDS_PAIRING_CONFIRMATION = { "SM/CEN/EKS/BV-01-C", "SM/CEN/JW/BI-04-C", "SM/CEN/JW/BI-01-C", "SM/CEN/KDU/BV-04-C", "SM/CEN/KDU/BV-05-C", "SM/CEN/KDU/BV-06-C", "SM/CEN/KDU/BV-10-C", "SM/CEN/KDU/BV-11-C", } ACCEPTS_REMOTE_PAIRING_CONFIRMATION = { "SM/CEN/KDU/BI-01-C", "SM/CEN/KDU/BI-02-C", "SM/CEN/KDU/BI-03-C", } def debug(*args, **kwargs): print(*args, file=sys.stderr, **kwargs) def get_event(pairing_stream: StreamWrapper, addr: str): for event in pairing_stream: if event.address == addr: return event return None class SMProxy(ProfileProxy): def __init__(self, channel): Loading @@ -59,32 +37,24 @@ class SMProxy(ProfileProxy): self.host = Host(channel) self.connection = None self.pairing_stream = None self.passkey_queue = Queue() self._handle_pairing_requests() @assert_description def MMI_IUT_ENABLE_CONNECTION_SM(self, test, pts_addr: bytes, **kwargs): def MMI_IUT_ENABLE_CONNECTION_SM(self, pts_addr: bytes, **kwargs): """ Initiate an connection from the IUT to the PTS. """ self.connection = self.host.ConnectLE(address=pts_addr).connection self.pairing_stream = self.security.OnPairing() if self.connection and test in ACCEPTS_REMOTE_PAIRING_CONFIRMATION: event = get_event(pairing_stream=self.pairing_stream, addr=pts_addr) self.pairing_stream.send(event=event, confirm=True) self.pairing_stream.close() return "OK" @assert_description def MMI_ASK_IUT_PERFORM_PAIRING_PROCESS(self, test, pts_addr: bytes, **kwargs): def MMI_ASK_IUT_PERFORM_PAIRING_PROCESS(self, **kwargs): """ Please start pairing process. """ if self.connection: self.security.Pair(connection=self.connection) if test in NEEDS_PAIRING_CONFIRMATION: event = get_event(pairing_stream=self.pairing_stream, addr=pts_addr) self.pairing_stream.send(event=event, confirm=True) self.pairing_stream.close() return "OK" @assert_description Loading @@ -104,7 +74,6 @@ class SMProxy(ProfileProxy): """ Please confirm the following number matches IUT: 385874. """ return "OK" @assert_description Loading @@ -114,3 +83,75 @@ class SMProxy(ProfileProxy): """ self.host.SoftReset() return "OK" @assert_description def MMI_TESTER_ENABLE_CONNECTION_SM(self, **kwargs): """ Action: Place the IUT in connectable mode """ self.host.SetLEConnectable() return "OK" @assert_description def MMI_IUT_SMP_TIMEOUT_30_SECONDS(self, **kwargs): """ Wait for the 30 seconds. Lower tester will not send corresponding or next SMP message. """ return "OK" @assert_description def MMI_IUT_SMP_TIMEOUT_ADDITIONAL_10_SECONDS(self, **kwargs): """ Wait for an additional 10 seconds. Lower test will send corresponding or next SMP message. """ return "OK" @match_description def MMI_DISPLAY_PASSKEY_CODE(self, passkey: str, **kwargs): """ Please enter (?P<passkey>[0-9]*) in the IUT. """ self.passkey_queue.put(passkey) return "OK" @assert_description def MMI_ENTER_PASSKEY_CODE(self, **kwargs): """ Please enter 6 digit passkey code. """ return "OK" @assert_description def MMI_ENTER_WRONG_DYNAMIC_PASSKEY_CODE(self, **kwargs): """ Please enter invalid 6 digit pin code. """ return "OK" @assert_description def MMI_IUT_ABORT_PAIRING_PROCESS_DISCONNECT(self, **kwargs): """ Lower tester expects IUT aborts pairing process, and disconnect. """ return "OK" def _handle_pairing_requests(self): def task(): pairing_events = self.security.OnPairing() for event in pairing_events: if event.just_works: pairing_events.send(event=event, confirm=True) if event.passkey_entry_request: try: passkey = self.passkey_queue.get(timeout=15) pairing_events.send(event=event, passkey=int(passkey)) except Empty: debug("No passkey provided within 15 seconds") Thread(target=task).start()
android/pandora/server/configs/PtsBotTest.xml +12 −0 Original line number Diff line number Diff line Loading @@ -47,5 +47,17 @@ <option name="profile" value="SM/CEN/EKS" /> <option name="profile" value="SM/CEN/JW" /> <option name="profile" value="SM/CEN/KDU" /> <option name="profile" value="SM/CEN/PROT" /> <option name="profile" value="SM/CEN/PKE" /> <option name="profile" value="SM/CEN/SCJW" /> <option name="profile" value="SM/CEN/SCPK" /> <option name="profile" value="SM/PER/PROT" /> <option name="profile" value="SM/PER/JW" /> <option name="profile" value="SM/PER/PKE" /> <option name="profile" value="SM/PER/EKS" /> <option name="profile" value="SM/PER/KDU" /> <option name="profile" value="SM/PER/SCJW" /> <option name="profile" value="SM/PER/SCPK" /> <option name="profile" value="HOGP/RH" /> </test> </configuration>
android/pandora/server/configs/pts_bot_tests_config.json +37 −2 Original line number Diff line number Diff line Loading @@ -225,7 +225,34 @@ "SM/CEN/KDU/BV-06-C", "SM/CEN/KDU/BV-10-C", "SM/CEN/KDU/BI-03-C", "SM/CEN/KDU/BV-11-C" "SM/CEN/KDU/BV-11-C", "SM/CEN/PROT/BV-01-C", "SM/CEN/PKE/BV-04-C", "SM/CEN/PKE/BI-01-C", "SM/CEN/PKE/BI-02-C", "SM/CEN/SCJW/BV-04-C", "SM/CEN/SCJW/BI-01-C", "SM/CEN/SCPK/BI-01-C", "SM/PER/PROT/BV-02-C", "SM/PER/JW/BV-02-C", "SM/PER/JW/BI-03-C", "SM/PER/JW/BI-02-C", "SM/PER/PKE/BV-02-C", "SM/PER/PKE/BV-05-C", "SM/PER/PKE/BI-03-C", "SM/PER/EKS/BV-02-C", "SM/PER/EKS/BI-02-C", "SM/PER/KDU/BV-01-C", "SM/PER/KDU/BV-02-C", "SM/PER/KDU/BV-03-C", "SM/PER/KDU/BV-07-C", "SM/PER/KDU/BV-08-C", "SM/PER/KDU/BV-09-C", "SM/PER/KDU/BI-01-C", "SM/PER/SCJW/BV-03-C", "SM/PER/SCJW/BI-02-C", "SM/PER/SCPK/BV-02-C", "SM/PER/SCPK/BI-03-C" ], "skip": [ "A2DP/SRC/AS/BV-01-I", Loading Loading @@ -370,7 +397,15 @@ "L2CAP/LE/CFC/BV-07-C", "L2CAP/LE/CFC/BV-11-C", "L2CAP/LE/CFC/BV-13-C", "L2CAP/LE/CFC/BV-15-C" "L2CAP/LE/CFC/BV-15-C", "SM/CEN/PKE/BV-01-C", "SM/CEN/SCJW/BV-01-C", "SM/CEN/SCPK/BI-02-C", "SM/CEN/SCPK/BV-01-C", "SM/CEN/SCPK/BV-04-C", "SM/PER/SCJW/BV-02-C", "SM/PER/SCPK/BI-04-C", "SM/PER/SCPK/BV-03-C" ], "ics": { "TSPC_4.0HCI_1a_2": true, Loading