Unbind GD Cert test classes from GdBaseTestClass (630c24bd) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/cert/gd_base_test.py

+1 −1

Original line number	Original line	Diff line number	Diff line
	@@ -64,6 +64,7 @@ class GdBaseTestClass(BaseTestClass):
	self.rootcanal_running = self.info['rootcanal_running']		self.rootcanal_running = self.info['rootcanal_running']
	self.rootcanal_logpath = self.info['rootcanal_logpath']		self.rootcanal_logpath = self.info['rootcanal_logpath']
	self.rootcanal_process = self.info['rootcanal_process']		self.rootcanal_process = self.info['rootcanal_process']
			self.rootcanal_logger = self.info['rootcanal_logger']

	if 'rootcanal' in self.controller_configs:		if 'rootcanal' in self.controller_configs:
	asserts.assert_true(self.info['rootcanal_exist'],		asserts.assert_true(self.info['rootcanal_exist'],
	@@ -77,7 +78,6 @@ class GdBaseTestClass(BaseTestClass):
	msg="Cannot start root-canal at " + str(self.info['rootcanal']))		msg="Cannot start root-canal at " + str(self.info['rootcanal']))
	asserts.assert_true(self.info['is_subprocess_alive'], msg="root-canal stopped immediately after running")		asserts.assert_true(self.info['is_subprocess_alive'], msg="root-canal stopped immediately after running")

	self.rootcanal_logger = self.info['rootcanal_logger']
	self.controller_configs = self.info['controller_configs']		self.controller_configs = self.info['controller_configs']

	# Parse and construct GD device objects		# Parse and construct GD device objects

system/gd/cert/gd_base_test_lib.py

+3 −0

Original line number	Original line	Diff line number	Diff line
	@@ -42,6 +42,9 @@ def setup_class_core(dut_module, cert_module, verbose_mode, log_path_base, contr

	# Start root-canal if needed		# Start root-canal if needed
	info['rootcanal_running'] = False		info['rootcanal_running'] = False
			info['rootcanal_logpath'] = None
			info['rootcanal_process'] = None
			info['rootcanal_logger'] = None
	if 'rootcanal' in info['controller_configs']:		if 'rootcanal' in info['controller_configs']:
	info['rootcanal_running'] = True		info['rootcanal_running'] = True
	# Get root canal binary		# Get root canal binary

system/gd/hal/cert/simple_hal_test.py

+7 −114

Original line number	Original line	Diff line number	Diff line
	@@ -14,128 +14,21 @@
	# See the License for the specific language governing permissions and		# See the License for the specific language governing permissions and
	# limitations under the License.		# limitations under the License.

	from datetime import timedelta

	from cert.gd_base_test import GdBaseTestClass		from cert.gd_base_test import GdBaseTestClass
	from cert.event_stream import EventStream		from hal.cert.simple_hal_test_lib import SimpleHalTestBase
	from cert.truth import assertThat
	from cert.py_hal import PyHal
	from cert.matchers import HciMatchers
	from cert.captures import HciCaptures
	from google.protobuf import empty_pb2
	from facade import rootservice_pb2 as facade_rootservice_pb2
	from hal import hal_facade_pb2 as hal_facade_pb2
	from bluetooth_packets_python3 import hci_packets
	import bluetooth_packets_python3 as bt_packets
	from bluetooth_packets_python3.hci_packets import AclBuilder
	from bluetooth_packets_python3 import RawBuilder

	_GRPC_TIMEOUT = 10		_GRPC_TIMEOUT = 10


	class SimpleHalTest(GdBaseTestClass):		class SimpleHalTest(GdBaseTestClass, SimpleHalTestBase):

	def setup_class(self):		def setup_class(self):
	super().setup_class(dut_module='HAL', cert_module='HAL')		GdBaseTestClass.setup_class(self, dut_module='HAL', cert_module='HAL')

	def setup_test(self):		def setup_test(self):
	super().setup_test()		GdBaseTestClass.setup_test(self)
			SimpleHalTestBase.setup_test(self, self.dut, self.cert)
	self.dut_hal = PyHal(self.dut)
	self.cert_hal = PyHal(self.cert)

	self.dut_hal.reset()
	self.cert_hal.reset()

	def teardown_test(self):		def teardown_test(self):
	self.dut_hal.close()		SimpleHalTestBase.teardown_test(self)
	self.cert_hal.close()		GdBaseTestClass.teardown_test(self)
	super().teardown_test()

	def test_stream_events(self):
	self.dut_hal.send_hci_command(
	hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))

	assertThat(self.dut_hal.get_hci_event_stream()).emits(
	HciMatchers.Exactly(hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)))

	def test_loopback_hci_command(self):
	self.dut_hal.send_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))

	command = hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,
	'0C:05:04:03:02:01')
	self.dut_hal.send_hci_command(command)

	assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LoopbackOf(command))

	def test_inquiry_from_dut(self):
	self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))

	lap = hci_packets.Lap()
	lap.lap = 0x33
	self.dut_hal.send_hci_command(hci_packets.InquiryBuilder(lap, 0x30, 0xff))

	assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'\x02\x0f' in packet.payload
	# Expecting an HCI Event (code 0x02, length 0x0f)
	)

	def test_le_ad_scan_cert_advertises(self):
	self.dut_hal.set_random_le_address('0D:05:04:03:02:01')

	self.dut_hal.set_scan_parameters()
	self.dut_hal.start_scanning()

	advertisement = self.cert_hal.create_advertisement(
	0,
	'0C:05:04:03:02:01',
	min_interval=512,
	max_interval=768,
	peer_address='A6:A5:A4:A3:A2:A1',
	tx_power=0x7f,
	sid=1)
	advertisement.set_data(b'Im_A_Cert')
	advertisement.start()

	assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'Im_A_Cert' in packet.payload)

	advertisement.stop()

	self.dut_hal.stop_scanning()

	def test_le_connection_dut_advertises(self):
	self.cert_hal.set_random_le_address('0C:05:04:03:02:01')
	self.cert_hal.initiate_le_connection('0D:05:04:03:02:01')

	# DUT Advertises
	advertisement = self.dut_hal.create_advertisement(0, '0D:05:04:03:02:01')
	advertisement.set_data(b'Im_The_DUT')
	advertisement.set_scan_response(b'Im_The_D')
	advertisement.start()

	cert_acl = self.cert_hal.complete_le_connection()
	dut_acl = self.dut_hal.complete_le_connection()

	dut_acl.send_first(b'Just SomeAclData')
	cert_acl.send_first(b'Just SomeMoreAclData')

	assertThat(self.cert_hal.get_acl_stream()).emits(lambda packet: b'SomeAclData' in packet.payload)
	assertThat(self.dut_hal.get_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.payload)

	def test_le_connect_list_connection_cert_advertises(self):
	self.dut_hal.set_random_le_address('0D:05:04:03:02:01')
	self.dut_hal.add_to_connect_list('0C:05:04:03:02:01')
	self.dut_hal.initiate_le_connection_by_connect_list('BA:D5:A4:A3:A2:A1')

	advertisement = self.cert_hal.create_advertisement(
	1,
	'0C:05:04:03:02:01',
	min_interval=512,
	max_interval=768,
	peer_address='A6:A5:A4:A3:A2:A1',
	tx_power=0x7F,
	sid=0)
	advertisement.set_data(b'Im_A_Cert')
	advertisement.start()

	assertThat(self.cert_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())
	assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())

system/gd/hal/cert/simple_hal_test_lib.py

0 → 100644

+134 −0

Original line number	Original line	Diff line number	Diff line
			#!/usr/bin/env python3
			#
			# Copyright 2019 - 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 datetime import timedelta

			from cert.event_stream import EventStream
			from cert.truth import assertThat
			from cert.py_hal import PyHal
			from cert.matchers import HciMatchers
			from cert.captures import HciCaptures
			from google.protobuf import empty_pb2
			from facade import rootservice_pb2 as facade_rootservice_pb2
			from hal import hal_facade_pb2 as hal_facade_pb2
			from bluetooth_packets_python3 import hci_packets
			import bluetooth_packets_python3 as bt_packets
			from bluetooth_packets_python3.hci_packets import AclBuilder
			from bluetooth_packets_python3 import RawBuilder

			_GRPC_TIMEOUT = 10


			class SimpleHalTestBase():

			def setup_test(self, dut, cert):
			self.dut_hal = PyHal(dut)
			self.cert_hal = PyHal(cert)

			self.dut_hal.reset()
			self.cert_hal.reset()

			def teardown_test(self):
			self.dut_hal.close()
			self.cert_hal.close()

			def test_stream_events(self):
			self.dut_hal.send_hci_command(
			hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))

			assertThat(self.dut_hal.get_hci_event_stream()).emits(
			HciMatchers.Exactly(hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)))

			def test_loopback_hci_command(self):
			self.dut_hal.send_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))

			command = hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,
			'0C:05:04:03:02:01')
			self.dut_hal.send_hci_command(command)

			assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LoopbackOf(command))

			def test_inquiry_from_dut(self):
			self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))

			lap = hci_packets.Lap()
			lap.lap = 0x33
			self.dut_hal.send_hci_command(hci_packets.InquiryBuilder(lap, 0x30, 0xff))

			assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'\x02\x0f' in packet.payload
			# Expecting an HCI Event (code 0x02, length 0x0f)
			)

			def test_le_ad_scan_cert_advertises(self):
			self.dut_hal.set_random_le_address('0D:05:04:03:02:01')

			self.dut_hal.set_scan_parameters()
			self.dut_hal.start_scanning()

			advertisement = self.cert_hal.create_advertisement(
			0,
			'0C:05:04:03:02:01',
			min_interval=512,
			max_interval=768,
			peer_address='A6:A5:A4:A3:A2:A1',
			tx_power=0x7f,
			sid=1)
			advertisement.set_data(b'Im_A_Cert')
			advertisement.start()

			assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'Im_A_Cert' in packet.payload)

			advertisement.stop()

			self.dut_hal.stop_scanning()

			def test_le_connection_dut_advertises(self):
			self.cert_hal.set_random_le_address('0C:05:04:03:02:01')
			self.cert_hal.initiate_le_connection('0D:05:04:03:02:01')

			# DUT Advertises
			advertisement = self.dut_hal.create_advertisement(0, '0D:05:04:03:02:01')
			advertisement.set_data(b'Im_The_DUT')
			advertisement.set_scan_response(b'Im_The_D')
			advertisement.start()

			cert_acl = self.cert_hal.complete_le_connection()
			dut_acl = self.dut_hal.complete_le_connection()

			dut_acl.send_first(b'Just SomeAclData')
			cert_acl.send_first(b'Just SomeMoreAclData')

			assertThat(self.cert_hal.get_acl_stream()).emits(lambda packet: b'SomeAclData' in packet.payload)
			assertThat(self.dut_hal.get_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.payload)

			def test_le_connect_list_connection_cert_advertises(self):
			self.dut_hal.set_random_le_address('0D:05:04:03:02:01')
			self.dut_hal.add_to_connect_list('0C:05:04:03:02:01')
			self.dut_hal.initiate_le_connection_by_connect_list('BA:D5:A4:A3:A2:A1')

			advertisement = self.cert_hal.create_advertisement(
			1,
			'0C:05:04:03:02:01',
			min_interval=512,
			max_interval=768,
			peer_address='A6:A5:A4:A3:A2:A1',
			tx_power=0x7F,
			sid=0)
			advertisement.set_data(b'Im_A_Cert')
			advertisement.start()

			assertThat(self.cert_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())
			assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())

system/gd/hci/cert/controller_test.py

+5 −20

Original line number	Original line	Diff line number	Diff line
	@@ -14,32 +14,17 @@
	# See the License for the specific language governing permissions and		# See the License for the specific language governing permissions and
	# limitations under the License.		# limitations under the License.

	import time

	from cert.gd_base_test import GdBaseTestClass		from cert.gd_base_test import GdBaseTestClass
	from cert.truth import assertThat		from hci.cert.controller_test_lib import ControllerTestBase
	from google.protobuf import empty_pb2 as empty_proto
	from facade import rootservice_pb2 as facade_rootservice
	from hci.facade import controller_facade_pb2 as controller_facade


	class ControllerTest(GdBaseTestClass):		class ControllerTest(GdBaseTestClass, ControllerTestBase):

	def setup_class(self):		def setup_class(self):
	super().setup_class(dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')		GdBaseTestClass.setup_class(self, dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')

	def test_get_addresses(self):		def test_get_addresses(self):
	cert_address = self.cert.hci_controller.GetMacAddressSimple()		ControllerTestBase.test_get_addresses(self, self.dut, self.cert)
	dut_address = self.dut.hci_controller.GetMacAddressSimple()

	assertThat(cert_address).isNotEqualTo(dut_address)
	time.sleep(1) # This shouldn't be needed b/149120542

	def test_write_local_name(self):		def test_write_local_name(self):
	self.dut.hci_controller.WriteLocalName(controller_facade.NameMsg(name=b'ImTheDUT'))		ControllerTestBase.test_write_local_name(self, self.dut, self.cert)
	self.cert.hci_controller.WriteLocalName(controller_facade.NameMsg(name=b'ImTheCert'))
	cert_name = self.cert.hci_controller.GetLocalNameSimple()
	dut_name = self.dut.hci_controller.GetLocalNameSimple()

	assertThat(dut_name).isEqualTo(b'ImTheDUT')
	assertThat(cert_name).isEqualTo(b'ImTheCert')