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

Commit 4ba2386b authored by Automerger Merge Worker's avatar Automerger Merge Worker
Browse files

Merge changes I5578a52e,I36446c92 am: b95450dc 

Change-Id: I433a494bf6d17ee2f158b07192091020aafaa942
parents 14050d9d b95450dc

system/gd/cert/py_acl_manager.py

+2 −1
Original line number Diff line number Diff line
@@ -35,7 +35,8 @@ class PyAclManagerAclConnection(IEventStream, Closable): 
        self.device = device

        self.handle = handle

        # todo enable filtering after sorting out handles

        self.our_acl_stream = FilteringEventStream(acl_stream, None)

        #self.our_acl_stream = FilteringEventStream(acl_stream, None)

        self.our_acl_stream = acl_stream



        if remote_addr:

            remote_addr_bytes = bytes(

system/gd/cert/truth.py

+28 −9
Original line number Diff line number Diff line
@@ -15,6 +15,7 @@ 
#   limitations under the License.



import time

from datetime import timedelta



from mobly.asserts import assert_true

from mobly.asserts import assert_false
@@ -56,35 +57,49 @@ class ObjectSubject(object): 
                "Expected \"%s\" to not be None" % self._value, extras=None)





DEFAULT_TIMEOUT = timedelta(seconds=3)





class EventStreamSubject(ObjectSubject):



    def __init__(self, value):

        super().__init__(value)



    def emits(self, *match_fns):

    def emits(self, *match_fns, at_least_times=1, timeout=DEFAULT_TIMEOUT):

        if len(match_fns) == 0:

            raise signals.TestFailure("Must specify a match function")

        elif len(match_fns) == 1:

            NOT_FOR_YOU_assert_event_occurs(self._value, match_fns[0])

            NOT_FOR_YOU_assert_event_occurs(

                self._value,

                match_fns[0],

                at_least_times=at_least_times,

                timeout=timeout)

            return EventStreamContinuationSubject(self._value)

        else:

            return MultiMatchStreamSubject(self._value, match_fns)

            return MultiMatchStreamSubject(self._value, match_fns, timeout)





class MultiMatchStreamSubject(object):



    def __init__(self, stream, match_fns):

    def __init__(self, stream, match_fns, timeout):

        self._stream = stream

        self._match_fns = match_fns

        self._timeout = timeout



    def inAnyOrder(self):

        NOT_FOR_YOU_assert_all_events_occur(

            self._stream, self._match_fns, order_matters=False)

            self._stream,

            self._match_fns,

            order_matters=False,

            timeout=self._timeout)

        return EventStreamContinuationSubject(self._stream)



    def inOrder(self):

        NOT_FOR_YOU_assert_all_events_occur(

            self._stream, self._match_fns, order_matters=True)

            self._stream,

            self._match_fns,

            order_matters=True,

            timeout=self._timeout)

        return EventStreamContinuationSubject(self._stream)
@@ -93,14 +108,18 @@ class EventStreamContinuationSubject(ObjectSubject): 
    def __init__(self, value):

        super().__init__(value)



    def then(self, *match_fns):

    def then(self, *match_fns, at_least_times=1, timeout=DEFAULT_TIMEOUT):

        if len(match_fns) == 0:

            raise signals.TestFailure("Must specify a match function")

        elif len(match_fns) == 1:

            NOT_FOR_YOU_assert_event_occurs(self._value, match_fns[0])

            NOT_FOR_YOU_assert_event_occurs(

                self._value,

                match_fns[0],

                at_least_times=at_least_times,

                timeout=timeout)

            return EventStreamContinuationSubject(self._value)

        else:

            return MultiMatchStreamSubject(self._value, match_fns)

            return MultiMatchStreamSubject(self._value, match_fns, timeout)





class BooleanSubject(ObjectSubject):

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

+167 −269

File changed.

Preview size limit exceeded, changes collapsed.