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

Commit 6573026e authored by Zach Johnson's avatar Zach Johnson Committed by Gerrit Code Review
Browse files

Merge changes I287d8d04,I7f626670 

* changes:
  Remove EventStream.assert_all_events_occur
  Remove EventStream.assert_none
parents 2f4605a9 2d39db4d

system/gd/cert/cert_self_test.py

+0 −34
Original line number Diff line number Diff line
@@ -140,40 +140,6 @@ class CertSelfTest(BaseTestClass): 
    def teardown_test(self):

        return True



    def test_assert_none_passes(self):

        with EventStream(FetchEvents(events=[], delay_ms=50)) as event_stream:

            event_stream.assert_none(timeout=timedelta(milliseconds=10))



    def test_assert_none_passes_after_one_second(self):

        with EventStream(FetchEvents([1], delay_ms=1500)) as event_stream:

            event_stream.assert_none(timeout=timedelta(seconds=1.0))



    def test_assert_none_fails(self):

        try:

            with EventStream(FetchEvents(events=[17], delay_ms=50)) as event_stream:

                event_stream.assert_none(timeout=timedelta(seconds=1))

        except Exception as e:

            logging.debug(e)

            return True  # Failed as expected

        return False



    def test_assert_none_matching_passes(self):

        with EventStream(FetchEvents(events=[1, 2, 3], delay_ms=50)) as event_stream:

            event_stream.assert_none_matching(lambda data: data.value_ == 4, timeout=timedelta(seconds=0.15))



    def test_assert_none_matching_passes_after_1_second(self):

        with EventStream(FetchEvents(events=[1, 2, 3, 4], delay_ms=400)) as event_stream:

            event_stream.assert_none_matching(lambda data: data.value_ == 4, timeout=timedelta(seconds=1))



    def test_assert_none_matching_fails(self):

        try:

            with EventStream(FetchEvents(events=[1, 2, 3], delay_ms=50)) as event_stream:

                event_stream.assert_none_matching(lambda data: data.value_ == 2, timeout=timedelta(seconds=1))

        except Exception as e:

            logging.debug(e)

            return True  # Failed as expected

        return False



    def test_assert_occurs_at_least_passes(self):

        with EventStream(FetchEvents(events=[1, 2, 3, 1, 2, 3], delay_ms=40)) as event_stream:

            event_stream.assert_event_occurs(

system/gd/cert/event_stream.py

+0 −23
Original line number Diff line number Diff line
@@ -160,26 +160,6 @@ class EventStream(IEventStream, Closable): 
            else:

                raise exp



    def assert_none(self, timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):

        """

        Assert no event happens within timeout period



        :param timeout: a timedelta object

        :return:

        """

        NOT_FOR_YOU_assert_none(self, timeout)



    def assert_none_matching(self, match_fn, timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):

        """

        Assert no events where match_fn(event) is True happen within timeout

        period



        :param match_fn: return True/False on match_fn(event)

        :param timeout: a timedelta object

        :return:

        """

        NOT_FOR_YOU_assert_none_matching(self, match_fn, timeout)



    def assert_event_occurs(self, match_fn, at_least_times=1, timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):

        """

        Assert at least |at_least_times| instances of events happen where
@@ -221,9 +201,6 @@ class EventStream(IEventStream, Closable): 
            len(event_list) <= at_most_times,

            msg=("Expected at most %d events, but got %d" % (at_most_times, len(event_list))))



    def assert_all_events_occur(self, match_fns, order_matters, timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):

        NOT_FOR_YOU_assert_all_events_occur(self, match_fns, order_matters, timeout)





def static_remaining_time_delta(end_time):

    remaining = end_time - datetime.now()

system/gd/hal/cert/simple_hal_test.py

+1 −1
Original line number Diff line number Diff line
@@ -65,7 +65,7 @@ class SimpleHalTest(GdBaseTestClass): 


    def test_none_event(self):

        with EventStream(self.dut.hal.FetchHciEvent(empty_pb2.Empty())) as hci_event_stream:

            hci_event_stream.assert_none(timeout=timedelta(seconds=1))

            assertThat(hci_event_stream).emitsNone(timeout=timedelta(seconds=1))



    def test_fetch_hci_event(self):

        with EventStream(self.dut.hal.FetchHciEvent(empty_pb2.Empty())) as hci_event_stream: