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

Commit 44e07188 authored by Treehugger Robot's avatar Treehugger Robot Committed by Gerrit Code Review
Browse files

Merge changes I3f250bf8,Ib845ba56 

* changes:
  [Pandora] Classic HID host-side tests
  [Pandora] "Safely" reconnect phone
parents 7dc6c3d2 67dfaab6

android/pandora/mmi2grpc/mmi2grpc/__init__.py

+23 −7
Original line number Diff line number Diff line
@@ -26,10 +26,12 @@ from mmi2grpc.a2dp import A2DPProxy 
from mmi2grpc.avrcp import AVRCPProxy

from mmi2grpc.gatt import GATTProxy

from mmi2grpc.hfp import HFPProxy

from mmi2grpc.hid import HIDProxy

from mmi2grpc.hogp import HOGPProxy

from mmi2grpc.sdp import SDPProxy

from mmi2grpc.sm import SMProxy

from mmi2grpc._helpers import format_proxy

from mmi2grpc._rootcanal import RootCanal



from pandora_experimental.host_grpc import Host
@@ -55,31 +57,40 @@ class IUT: 
        """

        self.port = port

        self.test = test

        self.rootcanal = None



        # Profile proxies.

        self._a2dp = None

        self._avrcp = None

        self._gatt = None

        self._hfp = None

        self._hid = None

        self._hogp = None

        self._sdp = None

        self._sm = None

        self._hogp = None



    def __enter__(self):

        """Resets the IUT when starting a PTS test."""

        self.rootcanal = RootCanal()

        self.rootcanal.reconnect_phone()



        # Note: we don't keep a single gRPC channel instance in the IUT class

        # because reset is allowed to close the gRPC server.

        with grpc.insecure_channel(f'localhost:{self.port}') as channel:

            self._retry(Host(channel).HardReset)(wait_for_ready=True)



    def __exit__(self, exc_type, exc_value, exc_traceback):

        self.rootcanal.close()

        self.rootcanal = None



        self._a2dp = None

        self._avrcp = None

        self._gatt = None

        self._hfp = None

        self._hid = None

        self._hogp = None

        self._sdp = None

        self._sm = None

        self._hogp = None



    def _retry(self, func):
@@ -154,6 +165,16 @@ class IUT: 
            if not self._hfp:

                self._hfp = HFPProxy(grpc.insecure_channel(f'localhost:{self.port}'))

            return self._hfp.interact(test, interaction, description, pts_address)

        # Handles HID MMIs.

        if profile in ('HID'):

            if not self._hid:

                self._hid = HIDProxy(grpc.insecure_channel(f'localhost:{self.port}'), self.rootcanal)

            return self._hid.interact(test, interaction, description, pts_address)

        # Handles HOGP MMIs.

        if profile in ('HOGP'):

            if not self._hogp:

                self._hogp = HOGPProxy(grpc.insecure_channel(f'localhost:{self.port}'))

            return self._hogp.interact(test, interaction, description, pts_address)

        # Handles SDP MMIs.

        if profile in ('SDP'):

            if not self._sdp:
@@ -164,11 +185,6 @@ class IUT: 
            if not self._sm:

                self._sm = SMProxy(grpc.insecure_channel(f'localhost:{self.port}'))

            return self._sm.interact(test, interaction, description, pts_address)

        # Handles HOGP MMIs.

        if profile in ('HOGP'):

            if not self._hogp:

                self._hogp = HOGPProxy(grpc.insecure_channel(f'localhost:{self.port}'))

            return self._hogp.interact(test, interaction, description, pts_address)



        # Handles unsupported profiles.

        code = format_proxy(profile, interaction, description)

android/pandora/mmi2grpc/mmi2grpc/_rootcanal.py

+22 −15
Original line number Diff line number Diff line
@@ -84,6 +84,9 @@ class RootCanal: 
        # discard initialization messages

        self.channel.receive_response()



    def close(self):

        self.channel.close()



    @staticmethod

    def _parse_device_list(raw):

        # time for some cursed parsing!
@@ -112,8 +115,10 @@ class RootCanal: 
        return transport, idxs



    def reconnect_phone(self):

        devices = self.channel.send_command("list", [])

        devices = self._parse_device_list(devices)

        raw_devices = None

        try:

            raw_devices = self.channel.send_command("list", [])

            devices = self._parse_device_list(raw_devices)



            for dev_i, name in enumerate(devices["Devices"]):

                # the default transports are always 0 and 1
@@ -127,6 +132,8 @@ class RootCanal: 
                for phy in target_phys:

                    if dev_i not in self._parse_phy(devices["Phys"][phy])[1]:

                        self.channel.send_command("add_device_to_phy", [dev_i, phy])

        except Exception as e:

            print(raw_devices, e)



    def disconnect_phy(self):

        # first, list all devices

android/pandora/mmi2grpc/mmi2grpc/hid.py

0 → 100644
+185 −0
Original line number Diff line number Diff line 
from threading import Thread

from time import sleep

from mmi2grpc._helpers import assert_description, match_description

from mmi2grpc._proxy import ProfileProxy



from pandora_experimental.hid_grpc import HID

from pandora_experimental.host_grpc import Host

from pandora_experimental.hid_pb2 import HID_REPORT_TYPE_OUTPUT

from mmi2grpc._rootcanal import RootCanal





class HIDProxy(ProfileProxy):



    def __init__(self, channel, rootcanal):

        super().__init__()

        self.hid = HID(channel)

        self.host = Host(channel)

        self.rootcanal = rootcanal

        self.connection = None



    @assert_description

    def TSC_MMI_iut_enable_connection(self, pts_addr: bytes, **kwargs):

        """

        Click Ok, then using the Implementation Under Test (IUT) connect to the

        PTS.

        """



        self.rootcanal.reconnect_phy_if_needed()

        self.connection = self.host.Connect(address=pts_addr).connection



        return "OK"



    @assert_description

    def TSC_MMI_iut_release_connection(self, pts_addr: bytes, **kwargs):

        """

        Click Ok, then release the HID connection from the Implementation Under

        Test (IUT) by closing the Interrupt Channel followed by the Control

        Channel.



        Description:  This can be done using the anticipated L2CAP

        Disconnection Requests.  If the host is unable to perform the connection

        request, the IUT may break the ACL or Baseband Link by going out of

        range.

        """



        self.host.Disconnect(connection=self.connection)



        return "OK"



    @assert_description

    def TSC_MMI_iut_disable_connection(self, pts_addr: bytes, **kwargs):

        """

        Disable the connection using the Implementation UnderTest (IUT).



        Note:

        The IUT may either disconnect the Interupt Control Channels or send a

        host initiated virtual cable unplug and wait for the PTS to disconnect

        the channels.

        """



        self.host.Disconnect(connection=self.connection)

        self.connection = None



        return "OK"



    @assert_description

    def TSC_HID_MMI_iut_accept_connection_ready_confirm(self, **kwargs):

        """

        Please prepare the IUT to accept connection from PTS and then click OK.

        """



        self.rootcanal.reconnect_phy_if_needed()



        return "OK"



    @assert_description

    def TSC_MMI_iut_connectable_enter_pw_dev(self, **kwargs):

        """

        Make the Implementation Under Test (IUT) connectable, then click Ok.

        """



        self.rootcanal.reconnect_phy_if_needed()



        return "OK"



    @assert_description

    def TSC_HID_MMI_iut_accept_control_channel(self, pts_addr: bytes, **kwargs):

        """

        Accept the control channel connection from the Implementation Under Test

        (IUT).

        """



        return "OK"



    @assert_description

    def TSC_MMI_tester_release_connection(self, **kwargs):

        """

        Place the Implementation Under Test (IUT) in a state which will allow

        the PTS to perform an HID connection release, then click Ok.



        Note:  The

        PTS will send an L2CAP disconnect request for the Interrupt channel,

        then the control channel.

        """



        return "OK"



    @assert_description

    def TSC_MMI_host_iut_prepare_to_receive_pointing_data(self, **kwargs):

        """

        Place the Implementation Under Test (IUT) in a state to receive and

        verify HID pointing data, then click Ok.

        """



        return "OK"



    @assert_description

    def TSC_MMI_host_iut_verify_pointing_data(self, **kwargs):

        """

        Verify that the pointer on the Implementation Under Test (IUT) moved to

        the left (X< 0), then click Ok.

        """



        # TODO: implement!



        return "OK"



    @assert_description

    def TSC_MMI_host_send_output_report(self, pts_addr: bytes, **kwargs):

        """

        Send an output report from the HOST.

        """



        self.hid.SendHostReport(

            address=pts_addr,

            report_type=HID_REPORT_TYPE_OUTPUT,

            report="8",  # keyboard enable num-lock

        )



        return "OK"



    @match_description

    def TSC_MMI_verify_output_report(self, **kwargs):

        """

        Verify that the output report is correct.  nnOutput Report =0(?:0|1)'

        """



        # TODO: check the report matches the num-lock setting



        return "OK"



    @assert_description

    def TSC_MMI_rf_shield_iut_or_tester(self, pts_addr: bytes, **kwargs):

        """

        Click Ok, then perform one of the following actions:



        1. Move the PTS

        and Implementation Under Test (IUT) out of range of each other.

        2. Place

        either the PTS or IUT in an RF sheild box.

        """



        def disconnect():

            sleep(2)

            self.rootcanal.disconnect_phy()



        Thread(target=disconnect).start()



        return "OK"



    @assert_description

    def TSC_MMI_iut_auto_connection(self, pts_addr: bytes, **kwargs):

        """

        Click OK, then initiate a HID connection automatically from the IUT to

        the PTS

        """



        def connect():

            sleep(1)

            self.rootcanal.reconnect_phy_if_needed()

            self.connection = self.host.Connect(address=pts_addr).connection



        Thread(target=connect).start()



        return "OK"

android/pandora/server/Android.bp

+3 −0
Original line number Diff line number Diff line
@@ -134,6 +134,7 @@ genrule { 
        "proto/pandora_experimental/avrcp.proto",

        "proto/pandora_experimental/gatt.proto",

        "proto/pandora_experimental/hfp.proto",

        "proto/pandora_experimental/hid.proto",

        "proto/pandora_experimental/host.proto",

        "proto/pandora_experimental/security.proto",

    ],
@@ -146,6 +147,8 @@ genrule { 
        "pandora_experimental/gatt_pb2.py",

        "pandora_experimental/hfp_grpc.py",

        "pandora_experimental/hfp_pb2.py",

        "pandora_experimental/hid_grpc.py",

        "pandora_experimental/hid_pb2.py",

        "pandora_experimental/host_grpc.py",

        "pandora_experimental/host_pb2.py",

        "pandora_experimental/security_grpc.py",

android/pandora/server/configs/PtsBotTest.xml

+2 −1
Original line number Diff line number Diff line
@@ -32,10 +32,11 @@ 
        <option name="profile" value="HFP/AG/HFI" />

        <option name="profile" value="HFP/AG/SLC" />

        <option name="profile" value="HFP/AG/TCA" />

        <option name="profile" value="HID/HOS" />

        <option name="profile" value="HOGP/RH" />

        <option name="profile" value="SDP/SR" />

        <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="HOGP/RH" />

    </test>

</configuration>