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

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

Merge changes I1860997d,I81133db3 

* changes:
  Add default handling for behaviors
  add beginings of behavior.py, a fluent behavior composer
parents 7d7adde6 559a1bff

system/gd/cert/behavior.py

0 → 100644
+123 −0
Original line number Diff line number Diff line 
#!/usr/bin/env python3

#

#   Copyright 2020 - The Android Open Source Project

#

#   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

#

#       http://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.



from abc import ABC, abstractmethod

from mobly import signals



from cert.truth import assertThat





class IHasBehaviors(ABC):



    @abstractmethod

    def get_behaviors(self):

        pass





def anything():

    return lambda obj: True





def when(has_behaviors):

    assertThat(isinstance(has_behaviors, IHasBehaviors)).isTrue()

    return has_behaviors.get_behaviors()





def IGNORE_UNHANDLED(obj):

    pass





class SingleArgumentBehavior(object):



    def __init__(self, reply_stage_factory):

        self._reply_stage_factory = reply_stage_factory

        self._instances = []

        self.set_default_to_crash()



    def begin(self, matcher):

        return PersistenceStage(self, matcher, self._reply_stage_factory)



    def append(self, behavior_instance):

        self._instances.append(behavior_instance)



    def set_default(self, fn):

        assertThat(fn).isNotNone()

        self._default_fn = fn



    def set_default_to_crash(self):

        self._default_fn = None



    def set_default_to_ignore(self):

        self._default_fn = IGNORE_UNHANDLED



    def run(self, obj):

        for instance in self._instances:

            if instance.try_run(obj):

                return

        if self._default_fn is not None:

            self._default_fn(obj)

        else:

            raise signals.TestFailure(

                "%s: behavior for %s went unhandled" %

                (self._reply_stage_factory().__class__.__name__, obj),

                extras=None)





class PersistenceStage(object):



    def __init__(self, behavior, matcher, reply_stage_factory):

        self._behavior = behavior

        self._matcher = matcher

        self._reply_stage_factory = reply_stage_factory



    def then(self, times=1):

        reply_stage = self._reply_stage_factory()

        reply_stage.init(self._behavior, self._matcher, times)

        return reply_stage



    def always(self):

        return self.then(times=-1)





class ReplyStage(object):



    def init(self, behavior, matcher, persistence):

        self._behavior = behavior

        self._matcher = matcher

        self._persistence = persistence



    def _commit(self, fn):

        self._behavior.append(

            BehaviorInstance(self._matcher, self._persistence, fn))





class BehaviorInstance(object):



    def __init__(self, matcher, persistence, fn):

        self._matcher = matcher

        self._persistence = persistence

        self._fn = fn

        self._called_count = 0



    def try_run(self, obj):

        if not self._matcher(obj):

            return False

        if self._persistence >= 0:

            if self._called_count >= self._persistence:

                return False

        self._called_count += 1

        self._fn(obj)

        return True

system/gd/cert/cert_self_test.py

+162 −0
Original line number Diff line number Diff line
@@ -28,6 +28,11 @@ from bluetooth_packets_python3 import l2cap_packets 
from cert.event_stream import EventStream, FilteringEventStream

from cert.truth import assertThat

from cert.metadata import metadata

from cert.behavior import when

from cert.behavior import IHasBehaviors

from cert.behavior import anything

from cert.behavior import SingleArgumentBehavior

from cert.behavior import ReplyStage





class BogusProto:
@@ -88,6 +93,44 @@ class FetchEvents: 
        return None





class TestBehaviors(object):



    def __init__(self, parent):

        self.test_request_behavior = SingleArgumentBehavior(

            lambda: TestBehaviors.TestRequestReplyStage(parent))



    def test_request(self, matcher):

        return self.test_request_behavior.begin(matcher)



    class TestRequestReplyStage(ReplyStage):



        def __init__(self, parent):

            self._parent = parent



        def increment_count(self):

            self._commit(lambda obj: self._increment_count(obj))

            return self



        def _increment_count(self, obj):

            self._parent.count += 1

            self._parent.captured.append(obj)





class ObjectWithBehaviors(IHasBehaviors):



    def __init__(self):

        self.behaviors = TestBehaviors(self)

        self.count = 0

        self.captured = []

        self.unhandled_count = 0



    def get_behaviors(self):

        return self.behaviors



    def increment_unhandled(self):

        self.unhandled_count += 1





class CertSelfTest(BaseTestClass):



    def setup_test(self):
@@ -484,3 +527,122 @@ class CertSelfTest(BaseTestClass): 
            asserts.assert_equal(e.extras["pts_test_name"], "Hello world")

        else:

            asserts.fail("Must throw an exception using @metadata decorator")



    def test_fluent_behavior_simple(self):

        thing = ObjectWithBehaviors()



        when(thing).test_request(anything()).then().increment_count()



        thing.behaviors.test_request_behavior.run("A")



        assertThat(thing.count).isEqualTo(1)

        assertThat(thing.captured).isEqualTo(["A"])



    def test_fluent_behavior__then_single__captures_one(self):

        thing = ObjectWithBehaviors()



        thing.behaviors.test_request_behavior.set_default_to_ignore()



        when(thing).test_request(anything()).then().increment_count()



        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("A")



        assertThat(thing.count).isEqualTo(1)

        assertThat(thing.captured).isEqualTo(["A"])



    def test_fluent_behavior__then_times__captures_all(self):

        thing = ObjectWithBehaviors()



        when(thing).test_request(anything()).then(times=3).increment_count()



        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("B")

        thing.behaviors.test_request_behavior.run("C")



        assertThat(thing.count).isEqualTo(3)

        assertThat(thing.captured).isEqualTo(["A", "B", "C"])



    def test_fluent_behavior__always__captures_all(self):

        thing = ObjectWithBehaviors()



        when(thing).test_request(anything()).always().increment_count()



        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("B")

        thing.behaviors.test_request_behavior.run("C")



        assertThat(thing.count).isEqualTo(3)

        assertThat(thing.captured).isEqualTo(["A", "B", "C"])



    def test_fluent_behavior__matcher__captures_relevant(self):

        thing = ObjectWithBehaviors()

        thing.behaviors.test_request_behavior.set_default_to_ignore()



        when(thing).test_request(

            lambda obj: obj == "B").always().increment_count()



        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("B")

        thing.behaviors.test_request_behavior.run("C")



        assertThat(thing.count).isEqualTo(1)

        assertThat(thing.captured).isEqualTo(["B"])



    def test_fluent_behavior__then_repeated__captures_relevant(self):

        thing = ObjectWithBehaviors()

        thing.behaviors.test_request_behavior.set_default_to_ignore()



        when(thing).test_request(

            anything()).then().increment_count().increment_count()



        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("B")

        thing.behaviors.test_request_behavior.run("A")



        assertThat(thing.count).isEqualTo(2)

        assertThat(thing.captured).isEqualTo(["A", "B"])



    def test_fluent_behavior__fallback__captures_relevant(self):

        thing = ObjectWithBehaviors()

        thing.behaviors.test_request_behavior.set_default_to_ignore()



        when(thing).test_request(lambda obj: obj == "B").then(

            times=1).increment_count()

        when(thing).test_request(

            lambda obj: obj == "C").always().increment_count()



        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("B")

        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("C")

        thing.behaviors.test_request_behavior.run("B")

        thing.behaviors.test_request_behavior.run("C")



        assertThat(thing.count).isEqualTo(3)

        assertThat(thing.captured).isEqualTo(["B", "C", "C"])



    def test_fluent_behavior__default_unhandled_crash(self):

        thing = ObjectWithBehaviors()



        when(thing).test_request(anything()).then().increment_count()



        thing.behaviors.test_request_behavior.run("A")

        try:

            thing.behaviors.test_request_behavior.run("A")

        except Exception as e:

            logging.debug(e)

            return True  # Failed as expected

        return False



    def test_fluent_behavior__set_default_works(self):

        thing = ObjectWithBehaviors()

        thing.behaviors.test_request_behavior.set_default(

            lambda obj: thing.increment_unhandled())



        when(thing).test_request(anything()).then().increment_count()



        thing.behaviors.test_request_behavior.run("A")

        thing.behaviors.test_request_behavior.run("A")

        assertThat(thing.unhandled_count).isEqualTo(1)