Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit c8d36b85 authored by bidsharma's avatar bidsharma
Browse files

[PANDORA_TEST] Implement SMP test on pts-bot

Adds SM/PER and SM/CEN tests

Tag: #feature
Bug: 239437774
Test: atest pts-bot:SM/PER -v

Change-Id: Icfb9626d99380ab58d68de69c1514956f93471d8
parent 2157c0cb
Loading
Loading
Loading
Loading
+81 −40
Original line number Diff line number Diff line
@@ -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):
@@ -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
@@ -104,7 +74,6 @@ class SMProxy(ProfileProxy):
        """
        Please confirm the following number matches IUT: 385874.
        """

        return "OK"

    @assert_description
@@ -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()
+12 −0
Original line number Diff line number Diff line
@@ -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>
+37 −2
Original line number Diff line number Diff line
@@ -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",
@@ -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,