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

Commit 9fa4d8be authored by Jizheng Chu's avatar Jizheng Chu Committed by Automerger Merge Worker
Browse files

Merge "Cert: Raise default timeout" am: a03f1c5c am: 9367d97d am: 645d41df am: a69bde8e 

Original change: https://android-review.googlesource.com/c/platform/system/bt/+/1837937

Change-Id: Ie2a29a21a7a9b8b5d8980d18c242e18a4e47422b
parents 2f8be1cb a69bde8e

system/gd/cert/event_stream.py

+2 −2
Original line number Diff line number Diff line
@@ -266,7 +266,7 @@ def NOT_FOR_YOU_assert_all_events_occur(istream, 
        asserts.assert_true(correct_order, "Events not received in correct order %s %s" % (match_fns, matched_order))





def NOT_FOR_YOU_assert_none_matching(istream, match_fn, timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):

def NOT_FOR_YOU_assert_none_matching(istream, match_fn, timeout):

    logging.debug("assert_none_matching %fs" % (timeout.total_seconds()))

    event = None

    end_time = datetime.now() + timeout
@@ -285,7 +285,7 @@ def NOT_FOR_YOU_assert_none_matching(istream, match_fn, timeout=timedelta(second 
    asserts.assert_true(event is None, msg='Expected None matching, but got {}'.format(pretty_print(event)))





def NOT_FOR_YOU_assert_none(istream, timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):

def NOT_FOR_YOU_assert_none(istream, timeout):

    logging.debug("assert_none %fs" % (timeout.total_seconds()))

    try:

        event = istream.get_event_queue().get(timeout=timeout.total_seconds())

system/gd/cert/truth.py

+3 −3
Original line number Diff line number Diff line
@@ -57,7 +57,7 @@ class ObjectSubject(object): 
            raise signals.TestFailure("Expected \"%s\" to not be None" % self._value, extras=None)





DEFAULT_TIMEOUT = timedelta(seconds=3)

DEFAULT_TIMEOUT = timedelta(seconds=10)





class EventStreamSubject(ObjectSubject):
@@ -74,7 +74,7 @@ class EventStreamSubject(ObjectSubject): 
        else:

            return MultiMatchStreamSubject(self._value, match_fns, timeout)



    def emitsNone(self, *match_fns, timeout=DEFAULT_TIMEOUT):

    def emitsNone(self, *match_fns, timeout):

        if len(match_fns) == 0:

            NOT_FOR_YOU_assert_none(self._value, timeout=timeout)

            return EventStreamContinuationSubject(self._value)
@@ -115,7 +115,7 @@ class EventStreamContinuationSubject(ObjectSubject): 
        else:

            return MultiMatchStreamSubject(self._value, match_fns, timeout)



    def thenNone(self, *match_fns, timeout=DEFAULT_TIMEOUT):

    def thenNone(self, *match_fns, timeout):

        if len(match_fns) == 0:

            NOT_FOR_YOU_assert_none(self._value, timeout=timeout)

            return EventStreamContinuationSubject(self._value)

system/gd/l2cap/classic/cert/l2cap_test_lib.py

+6 −4
Original line number Diff line number Diff line
@@ -221,7 +221,8 @@ class L2capTestBase(GeneralL2capTestBase): 


        (dut_channel, cert_channel) = self._open_unconfigured_channel_from_cert()



        assertThat(self.cert_l2cap.get_control_channel()).emitsNone(L2capMatchers.ConfigurationResponse())

        assertThat(self.cert_l2cap.get_control_channel()).emitsNone(

            L2capMatchers.ConfigurationResponse(), timeout=timedelta(seconds=3))

        # TODO: Verify that IUT sends disconnect request (not mandated)



    @metadata(pts_test_id="L2CAP/COS/CED/BV-09-C", pts_test_name="Receive Multi-Command Packet")
@@ -711,7 +712,7 @@ class L2capTestBase(GeneralL2capTestBase): 
        dut_channel.send(b'def')



        assertThat(cert_channel).emits(L2capMatchers.IFrame(tx_seq=0, payload=b'abc'))

        assertThat(cert_channel).emitsNone(L2capMatchers.IFrame(tx_seq=1, payload=b'def'))

        assertThat(cert_channel).emitsNone(L2capMatchers.IFrame(tx_seq=1, payload=b'def'), timeout=timedelta(seconds=3))



        cert_channel.send_s_frame(req_seq=1, f=Final.POLL_RESPONSE)

        assertThat(cert_channel).emits(L2capMatchers.IFrame(tx_seq=1))
@@ -1011,7 +1012,7 @@ class L2capTestBase(GeneralL2capTestBase): 
        assertThat(cert_channel).emits(L2capMatchers.SFrame(p=l2cap_packets.Poll.POLL))



        cert_channel.send_s_frame(req_seq=0, s=SupervisoryFunction.RECEIVER_NOT_READY, f=Final.POLL_RESPONSE)

        assertThat(cert_channel).emitsNone(L2capMatchers.IFrame(tx_seq=0))

        assertThat(cert_channel).emitsNone(L2capMatchers.IFrame(tx_seq=0), timeout=timedelta(seconds=3))



    @metadata(pts_test_id="L2CAP/ERM/BV-22-C", pts_test_name="Exit Local Busy Condition")

    def test_exit_local_busy_condition(self):
@@ -1232,7 +1233,8 @@ class L2capTestBase(GeneralL2capTestBase): 
        self.cert_l2cap.claim_ertm_unsupported()

        dut_channel_future = self.dut_l2cap.connect_dynamic_channel_to_cert(

            psm=0x33, mode=RetransmissionFlowControlMode.ERTM)

        assertThat(self.cert_l2cap.get_control_channel()).emitsNone(L2capMatchers.ConnectionRequest(0x33))

        assertThat(self.cert_l2cap.get_control_channel()).emitsNone(

            L2capMatchers.ConnectionRequest(0x33), timeout=timedelta(seconds=3))



    @metadata(

        pts_test_id="L2CAP/CMC/BI-01-C",