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

Commit 9e66ced0 authored by Zach Johnson's avatar Zach Johnson
Browse files

Remove EventStream.assert_none 

Use truth.py instead.

Bug: 157310355
Test: cert/run --host
Tag: #gd-refactor
Change-Id: I7f626670fdb31a1371db3a81a5e57c13fa4cfa46
parent 4c8c7065

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 −20
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

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: