Start PyAclManager, to simplify interaction with AclManager (76a6c90b) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/hci/cert/acl_manager_test.py

+7 −6

Original line number	Diff line number	Diff line
		@@ -32,6 +32,7 @@ from captures import ReadBdAddrCompleteCapture
		from captures import ConnectionCompleteCapture
		from captures import ConnectionRequestCapture
		from py_hci import PyHci
		from py_acl_manager import PyAclManager


		class AclManagerTest(GdFacadeOnlyBaseTestClass):
		@@ -41,7 +42,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):

		def test_dut_connects(self):
		with PyHci(self.cert) as cert_hci, \
		EventStream(self.dut.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:
		PyAclManager(self.dut) as dut_acl_manager:

		cert_hci.enable_inquiry_and_page_scan()
		cert_address = cert_hci.read_own_address()
		@@ -72,7 +73,7 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):

		assertThat(cert_hci.get_acl_stream()).emits(
		lambda packet: b'SomeMoreAclData' in packet.data)
		assertThat(acl_data_stream).emits(
		assertThat(dut_acl_manager.get_acl_stream()).emits(
		lambda packet: b'SomeAclData' in packet.payload)

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

		def test_recombination_l2cap_packet(self):
		with PyHci(self.cert) as cert_hci, \
		EventStream(self.dut.hci_acl_manager.FetchAclData(empty_proto.Empty())) as acl_data_stream:
		PyAclManager(self.dut) as dut_acl_manager:

		# CERT Enables scans and gets its address
		cert_hci.enable_inquiry_and_page_scan()
		@@ -135,6 +136,6 @@ class AclManagerTest(GdFacadeOnlyBaseTestClass):
		connection_event_stream.assert_event_occurs(connection_complete)
		dut_handle = connection_complete.get().GetConnectionHandle()

		assertThat(acl_data_stream).emits(
		lambda packet: b'Hello!' in packet.payload).then(
		lambda packet: b'Hello' * 200 in packet.payload)
		assertThat(dut_acl_manager.get_acl_stream()).emits(
		lambda packet: b'Hello!' in packet.payload,
		lambda packet: b'Hello' * 200 in packet.payload).inOrder()

system/gd/hci/cert/py_acl_manager.py

0 → 100644

+49 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		#
		# Copyright 2020 - 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 google.protobuf import empty_pb2 as empty_proto
		from cert.event_stream import EventStream
		from captures import ReadBdAddrCompleteCapture
		from captures import ConnectionCompleteCapture
		from captures import ConnectionRequestCapture
		from bluetooth_packets_python3 import hci_packets
		from cert.truth import assertThat
		from hci.facade import facade_pb2 as hci_facade


		class PyAclManager(object):

		def __init__(self, device):
		self.device = device

		self.acl_stream = EventStream(
		self.device.hci_acl_manager.FetchAclData(empty_proto.Empty()))

		def __enter__(self):
		return self

		def __exit__(self, type, value, traceback):
		self.clean_up()
		return traceback is None

		def __del__(self):
		self.clean_up()

		def clean_up(self):
		self.acl_stream.shutdown()

		def get_acl_stream(self):
		return self.acl_stream