Cert: Handle None filter function in FilteringStream (d9fa6cfc) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/cert/cert_self_test.py

+9 −1

Original line number	Diff line number	Diff line
		@@ -20,7 +20,7 @@ import time
		from mobly import asserts
		from datetime import datetime, timedelta
		from acts.base_test import BaseTestClass
		from cert.event_stream import EventStream
		from cert.event_stream import EventStream, FilteringEventStream
		from cert.truth import assertThat

		# Test packet nesting
		@@ -386,3 +386,11 @@ class CertSelfTest(BaseTestClass):
		logging.debug(e)
		return True # Failed as expected
		return False

		def test_filtering_event_stream_none_filter_function(self):
		with EventStream(FetchEvents(events=[1, 2, 3],
		delay_ms=50)) as event_stream:
		filtered_event_stream = FilteringEventStream(event_stream, None)
		assertThat(filtered_event_stream)\
		.emits(lambda data: data.value_ == 1)\
		.then(lambda data: data.value_ == 3)

+3 −2

Original line number	Diff line number	Diff line
		@@ -39,7 +39,7 @@ class IEventStream(ABC):
		class FilteringEventStream(IEventStream):

		def __init__(self, stream, filter_fn):
		self.filter_fn = filter_fn
		self.filter_fn = filter_fn if filter_fn else lambda x: x
		self.event_queue = SimpleQueue()
		self.stream = stream

		@@ -264,11 +264,12 @@ def NOT_FOR_YOU_assert_event_occurs(
		try:
		current_event = istream.get_event_queue().get(
		timeout=remaining.total_seconds())
		logging.debug("current_event: %s", current_event)
		if match_fn(current_event):
		event_list.append(current_event)
		except Empty:
		continue
		logging.debug("Done waiting for event")
		logging.debug("Done waiting for event, received %d", len(event_list))
		asserts.assert_true(
		len(event_list) >= at_least_times,
		msg=("Expected at least %d events, but got %d" % (at_least_times,