Merge "Add event stream to cert/truth, and start using in AclManagerTest" am: 7aad4e4bd7 (9b1598d9) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/cert/cert_self_test.py

+37 −0

Original line number	Original line	Diff line number	Diff line
	@@ -284,3 +284,40 @@ class CertSelfTest(BaseTestClass):
	except Exception as e:		except Exception as e:
	return True		return True
	return False		return False

			def test_assertThat_eventStream_emits_passes(self):
			with EventCallbackStream(FetchEvents(events=[1, 2, 3],
			delay_ms=50)) as event_stream:
			event_asserts = EventAsserts(event_stream)
			assertThat(event_asserts).emits(lambda data: data.value_ == 1)

			def test_assertThat_eventStream_emits_then_passes(self):
			with EventCallbackStream(FetchEvents(events=[1, 2, 3],
			delay_ms=50)) as event_stream:
			event_asserts = EventAsserts(event_stream)
			assertThat(event_asserts).emits(lambda data: data.value_ == 1).then(
			lambda data: data.value_ == 3)

			def test_assertThat_eventStream_emits_fails(self):
			try:
			with EventCallbackStream(
			FetchEvents(events=[1, 2, 3], delay_ms=50)) as event_stream:
			event_asserts = EventAsserts(event_stream)
			assertThat(event_asserts).emits(lambda data: data.value_ == 4)
			except Exception as e:
			logging.debug(e)
			return True # Failed as expected
			return False

			def test_assertThat_eventStream_emits_then_fails(self):
			try:
			with EventCallbackStream(
			FetchEvents(events=[1, 2, 3], delay_ms=50)) as event_stream:
			event_asserts = EventAsserts(event_stream)
			assertThat(event_asserts).emits(
			lambda data: data.value_ == 1).emits(
			lambda data: data.value_ == 4)
			except Exception as e:
			logging.debug(e)
			return True # Failed as expected
			return False

system/gd/cert/truth.py

+23 −0

Original line number	Original line	Diff line number	Diff line
	@@ -20,6 +20,7 @@ from mobly.asserts import assert_true
	from mobly.asserts import assert_false		from mobly.asserts import assert_false

	from mobly import signals		from mobly import signals
			from cert.event_asserts import EventAsserts

	import sys, traceback		import sys, traceback

	@@ -53,6 +54,26 @@ class ObjectSubject(object):
	"Expected \"%s\" to not be None" % self._value, extras=None)		"Expected \"%s\" to not be None" % self._value, extras=None)


			class EventStreamSubject(ObjectSubject):

			def __init__(self, value):
			super().__init__(value)

			def emits(self, match_fn):
			self._value.assert_event_occurs(match_fn)
			return EventStreamContinuationSubject(self._value)


			class EventStreamContinuationSubject(ObjectSubject):

			def __init__(self, value):
			super().__init__(value)

			def then(self, match_fn):
			self._value.assert_event_occurs(match_fn)
			return EventStreamContinuationSubject(self._value)


	class BooleanSubject(ObjectSubject):		class BooleanSubject(ObjectSubject):

	def __init__(self, value):		def __init__(self, value):
	@@ -68,5 +89,7 @@ class BooleanSubject(ObjectSubject):
	def assertThat(subject):		def assertThat(subject):
	if type(subject) is bool:		if type(subject) is bool:
	return BooleanSubject(subject)		return BooleanSubject(subject)
			elif isinstance(subject, EventAsserts):
			return EventStreamSubject(subject)
	else:		else:
	return ObjectSubject(subject)		return ObjectSubject(subject)

system/gd/hci/cert/acl_manager_test.py

+15 −17

Original line number	Original line	Diff line number	Diff line
	@@ -21,6 +21,7 @@ import logging
	from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass		from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass
	from cert.event_callback_stream import EventCallbackStream		from cert.event_callback_stream import EventCallbackStream
	from cert.event_asserts import EventAsserts		from cert.event_asserts import EventAsserts
			from cert.truth import assertThat
	from google.protobuf import empty_pb2 as empty_proto		from google.protobuf import empty_pb2 as empty_proto
	from hci.facade import acl_manager_facade_pb2 as acl_manager_facade		from hci.facade import acl_manager_facade_pb2 as acl_manager_facade
	from neighbor.facade import facade_pb2 as neighbor_facade		from neighbor.facade import facade_pb2 as neighbor_facade
	@@ -90,8 +91,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):

	self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True)		self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True)

	cert_hci_event_asserts.assert_event_occurs(		assertThat(cert_hci_event_asserts).emits(get_address_from_complete)
	get_address_from_complete)

	with EventCallbackStream(		with EventCallbackStream(
	self.dut.hci_acl_manager.CreateConnection(		self.dut.hci_acl_manager.CreateConnection(
	@@ -115,7 +115,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
	return False		return False

	# Cert Accepts		# Cert Accepts
	cert_hci_event_asserts.assert_event_occurs(get_connect_request)		assertThat(cert_hci_event_asserts).emits(get_connect_request)
	self.enqueue_hci_command(		self.enqueue_hci_command(
	hci_packets.AcceptConnectionRequestBuilder(		hci_packets.AcceptConnectionRequestBuilder(
	connection_request.GetBdAddr(),		connection_request.GetBdAddr(),
	@@ -137,7 +137,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
	return True		return True
	return False		return False

	cert_hci_event_asserts.assert_event_occurs(get_handle)		assertThat(cert_hci_event_asserts).emits(get_handle)
	cert_handle = handle		cert_handle = handle

	self.enqueue_acl_data(		self.enqueue_acl_data(
	@@ -159,9 +159,9 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
	b'\x29\x00\x07\x00This is just SomeMoreAclData from the DUT'		b'\x29\x00\x07\x00This is just SomeMoreAclData from the DUT'
	)))		)))

	cert_acl_data_asserts.assert_event_occurs(		assertThat(cert_acl_data_asserts).emits(
	lambda packet: b'SomeMoreAclData' in packet.data)		lambda packet: b'SomeMoreAclData' in packet.data)
	acl_data_asserts.assert_event_occurs(		assertThat(acl_data_asserts).emits(
	lambda packet: b'SomeAclData' in packet.payload)		lambda packet: b'SomeAclData' in packet.payload)

	def test_cert_connects(self):		def test_cert_connects(self):
	@@ -224,7 +224,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):

	conn_handle = 0xfff		conn_handle = 0xfff

	cert_hci_event_asserts.assert_event_occurs(get_handle)		assertThat(cert_hci_event_asserts).emits(get_handle)
	cert_handle = conn_handle		cert_handle = conn_handle

	self.enqueue_acl_data(		self.enqueue_acl_data(
	@@ -234,9 +234,9 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
	bytes(		bytes(
	b'\x26\x00\x07\x00This is just SomeAclData from the Cert'))		b'\x26\x00\x07\x00This is just SomeAclData from the Cert'))

	cert_acl_data_asserts.assert_event_occurs(		assertThat(cert_acl_data_asserts).emits(
	lambda packet: b'SomeMoreAclData' in packet.data)		lambda packet: b'SomeMoreAclData' in packet.data)
	acl_data_asserts.assert_event_occurs(		assertThat(acl_data_asserts).emits(
	lambda packet: b'SomeAclData' in packet.payload)		lambda packet: b'SomeAclData' in packet.payload)

	def test_recombination_l2cap_packet(self):		def test_recombination_l2cap_packet(self):
	@@ -273,8 +273,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):

	self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True)		self.enqueue_hci_command(hci_packets.ReadBdAddrBuilder(), True)

	cert_hci_event_asserts.assert_event_occurs(		assertThat(cert_hci_event_asserts).emits(get_address_from_complete)
	get_address_from_complete)

	with EventCallbackStream(		with EventCallbackStream(
	self.dut.hci_acl_manager.CreateConnection(		self.dut.hci_acl_manager.CreateConnection(
	@@ -298,7 +297,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
	return False		return False

	# Cert Accepts		# Cert Accepts
	cert_hci_event_asserts.assert_event_occurs(get_connect_request)		assertThat(cert_hci_event_asserts).emits(get_connect_request)
	self.enqueue_hci_command(		self.enqueue_hci_command(
	hci_packets.AcceptConnectionRequestBuilder(		hci_packets.AcceptConnectionRequestBuilder(
	connection_request.GetBdAddr(),		connection_request.GetBdAddr(),
	@@ -320,7 +319,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
	return True		return True
	return False		return False

	cert_hci_event_asserts.assert_event_occurs(get_handle)		assertThat(cert_hci_event_asserts).emits(get_handle)
	cert_handle = handle		cert_handle = handle

	self.enqueue_acl_data(		self.enqueue_acl_data(
	@@ -341,7 +340,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
	# DUT gets a connection complete event and sends and receives		# DUT gets a connection complete event and sends and receives
	connection_event_asserts.assert_event_occurs(get_handle)		connection_event_asserts.assert_event_occurs(get_handle)

	acl_data_asserts.assert_event_occurs(		assertThat(acl_data_asserts).emits(
	lambda packet: b'Hello!' in packet.payload)		lambda packet: b'Hello!' in packet.payload).then(
	acl_data_asserts.assert_event_occurs(
	lambda packet: b'Hello' * 200 in packet.payload)		lambda packet: b'Hello' * 200 in packet.payload)