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

Commit 71d71448 authored by Hui Peng's avatar Hui Peng
Browse files

Add bredr/ssp pairing tests (no_output_no_input)

Bug: 277282212
Test: atest avatar
Flag: EXEMPT, Tests
Change-Id: Icf51511c2bfee8181a1549f52e904bda0a9393ab
parent 34ca7553
Loading
Loading
Loading
Loading
+4 −0
Original line number Diff line number Diff line
@@ -14,9 +14,13 @@

from pairing.br_edr.legacy.tests import BREDRLegacyTestClass
from pairing.br_edr.misc.service_access_tests import ServiceAccessTempBondingTest

from pairing.br_edr.ssp.no_output_no_input.tests import BREDRNoOutputNoInputTestClass

from pairing.smp_test import SmpTest

_test_class_list = [
    BREDRNoOutputNoInputTestClass,
    BREDRLegacyTestClass,
    ServiceAccessTempBondingTest,
    SmpTest,
+59 −0
Original line number Diff line number Diff line
# Copyright 2024 Google LLC
#
# 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
#
#     https://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.

import asyncio
import logging

from pandora.security_pb2 import PairingEventAnswer

from pairing.br_edr.ssp.test_base import BREDRSSPPairTestBase


class BREDRNoOutputNoInputTestClass(BREDRSSPPairTestBase):

    def _setup_devices(self) -> None:

        self.ref.config.setdefault('classic_enabled', True)

        self.ref.config.setdefault('le_enabled', False)
        self.ref.config.setdefault('classic_ssp_enabled', True)
        self.ref.config.setdefault('classic_sc_enabled', False)

        self.ref.config.setdefault(
            'server',
            {
                'pairing_sc_enable': False,
                'pairing_mitm_enable': False,
                'pairing_bonding_enable': True,
                # Android IO CAP: Display_YESNO
                # Ref IO CAP:
                'io_capability': 'no_output_no_input',
            },
        )

    async def accept_pairing(self):

        # responder receives just works
        responder_ev = await anext(self.responder_pairing_event_stream)
        logging.debug(f'[{self.responder_pairing_event_stream.device.name}] responder_ev.method_variant():{responder_ev.method_variant()}')

        init_ev = await anext(self.initiator_pairing_event_stream)
        logging.debug(
            f'[{self.initiator_pairing_event_stream.device.name}] init_ev.method_variant():{init_ev.method_variant()}'
        )

        init_ev_ans = PairingEventAnswer(event=init_ev, confirm=True)
        self.initiator_pairing_event_stream.send_nowait(init_ev_ans)
        responder_ev_ans = PairingEventAnswer(event=responder_ev, confirm=True)
        self.responder_pairing_event_stream.send_nowait(responder_ev_ans)
+54 −0
Original line number Diff line number Diff line
# Copyright 2024 Google LLC
#
# 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
#
#     https://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.

import asyncio

from avatar import asynchronous

from mobly.asserts import fail

from pairing.br_edr.test_base import BREDRPairTestBase

class BREDRSSPPairTestBase(BREDRPairTestBase):

    @asynchronous
    async def test_general_pairing(self) -> None:

        self.acl_initiator = self.ref
        self.acl_responder = self.dut
        self.pairing_initiator = self.ref
        self.pairing_responder = self.dut
        self.service_initiator = self.ref
        self.service_responder = self.dut

        self.prepare_pairing()

        # first initiate an ACL connection from bumble to android
        bumble_res, android_res = await self.start_acl_connection()

        service_access_task = asyncio.create_task(
            self.start_service_access(bumble_res.connection, android_res.connection)
        )

        # at this point, android expects bumble to initiate pairing to compete the
        # service access
        pairing_task = asyncio.create_task(self.start_pairing(bumble_res.connection, android_res.connection))

        await self.accept_pairing()

        try:
            await asyncio.wait_for(service_access_task, timeout=10.0)
            await asyncio.wait_for(pairing_task, timeout=10.0)
        except:
            fail("service access and pairing should have succeeded")