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

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

Merge changes I5578a52e,I36446c92 

* changes:
  Simplify packet parsing/filtering for L2capTest
  Replace direct usages of acl manager stream with acl link instead
parents 3cc6444f e3ad81b0

system/gd/cert/py_acl_manager.py

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

        self.device = device

        self.handle = handle

        self.handle = handle

        # todo enable filtering after sorting out handles

        # 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:

        if remote_addr:

            remote_addr_bytes = bytes(

            remote_addr_bytes = bytes(

system/gd/cert/truth.py

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

#   limitations under the License.





import time

import time

from datetime import timedelta





from mobly.asserts import assert_true

from mobly.asserts import assert_true

from mobly.asserts import assert_false

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

                "Expected \"%s\" to not be None" % self._value, extras=None)









DEFAULT_TIMEOUT = timedelta(seconds=3)





class EventStreamSubject(ObjectSubject):

class EventStreamSubject(ObjectSubject):





    def __init__(self, value):

    def __init__(self, value):

        super().__init__(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:

        if len(match_fns) == 0:

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

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

        elif len(match_fns) == 1:

        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)

            return EventStreamContinuationSubject(self._value)

        else:

        else:

            return MultiMatchStreamSubject(self._value, match_fns)

            return MultiMatchStreamSubject(self._value, match_fns, timeout)









class MultiMatchStreamSubject(object):

class MultiMatchStreamSubject(object):





    def __init__(self, stream, match_fns):

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

        self._stream = stream

        self._stream = stream

        self._match_fns = match_fns

        self._match_fns = match_fns

        self._timeout = timeout





    def inAnyOrder(self):

    def inAnyOrder(self):

        NOT_FOR_YOU_assert_all_events_occur(

        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)

        return EventStreamContinuationSubject(self._stream)





    def inOrder(self):

    def inOrder(self):

        NOT_FOR_YOU_assert_all_events_occur(

        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)

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

    def __init__(self, value):

        super().__init__(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:

        if len(match_fns) == 0:

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

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

        elif len(match_fns) == 1:

        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)

            return EventStreamContinuationSubject(self._value)

        else:

        else:

            return MultiMatchStreamSubject(self._value, match_fns)

            return MultiMatchStreamSubject(self._value, match_fns, timeout)









class BooleanSubject(ObjectSubject):

class BooleanSubject(ObjectSubject):

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

+167 −269

File changed.

Preview size limit exceeded, changes collapsed.