Python: format all Python files in system/bt using YAPF (36384abd) · Commits · e / os / android_packages_modules_Bluetooth

system/build/toolchain/clang/get_clang_suffix.py

+30 −28

Original line number	Diff line number	Diff line
		@@ -3,6 +3,7 @@ import subprocess
		import re
		import sys


		def which(cmd):
		for p in os.environ["PATH"].split(os.pathsep):
		clang_path = os.path.join(p, cmd)
		@@ -10,14 +11,15 @@ def which(cmd):
		return clang_path
		return None


		CLANG_VERSION_REGEX = ".version\s([0-9]\.[0-9])\.*"
		clang_path = which("clang++")
		clang_version_major = 0
		clang_version_minor = 0

		if clang_path:
		clang_version_out = subprocess.Popen([clang_path, "--version"],
		stdout=subprocess.PIPE).communicate()[0]
		clang_version_out = subprocess.Popen(
		[clang_path, "--version"], stdout=subprocess.PIPE).communicate()[0]
		clang_version_match = re.search(CLANG_VERSION_REGEX, clang_version_out)
		clang_version_str = clang_version_match.group(1)
		clang_version_array = clang_version_str.split('.')

system/gd/cert/bluetooth_packets_python3_setup.py

+22 −21

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


		# Usage:
		# 1. Run envsetup and lunch first in an Android checkout
		# 2. Make target bluetooth_packets_python3 that will generate C++ sources for the
		@@ -25,7 +24,6 @@
		# 4. Install:
		# python3 bluetooth_packets_python3_setup.py install --user


		import os
		import glob
		from setuptools import setup, Extension
		@@ -34,10 +32,13 @@ ANDROID_BUILD_TOP = os.getenv("ANDROID_BUILD_TOP")
		PYBIND11_INCLUDE_DIR = os.path.join(ANDROID_BUILD_TOP,
		"external/python/pybind11/include")
		GD_DIR = os.path.join(ANDROID_BUILD_TOP, "packages/modules/Bluetooth/system/gd")
		BT_PACKETS_GEN_DIR = os.path.join(ANDROID_BUILD_TOP,
		BT_PACKETS_GEN_DIR = os.path.join(
		ANDROID_BUILD_TOP,
		"out/soong/.intermediates/packages/modules/Bluetooth/system/gd/BluetoothGeneratedPackets_h/gen")
		BT_PACKETS_PY3_GEN_DIR = os.path.join(ANDROID_BUILD_TOP,
		"out/soong/.intermediates/packages/modules/Bluetooth/system/gd/BluetoothGeneratedPackets_python3_cc/gen")
		BT_PACKETS_PY3_GEN_DIR = os.path.join(
		ANDROID_BUILD_TOP,
		"out/soong/.intermediates/packages/modules/Bluetooth/system/gd/BluetoothGeneratedPackets_python3_cc/gen"
		)

		BT_PACKETS_BASE_SRCS = [
		os.path.join(GD_DIR, "l2cap/fcs.cc"),
		@@ -57,16 +58,16 @@ BT_PACKETS_PY3_SRCs = \
		+ glob.glob(os.path.join(BT_PACKETS_PY3_GEN_DIR, "l2cap", "*.cc")) \
		+ glob.glob(os.path.join(BT_PACKETS_PY3_GEN_DIR, "security", "*.cc"))

		bluetooth_packets_python3_module = Extension('bluetooth_packets_python3',
		bluetooth_packets_python3_module = Extension(
		'bluetooth_packets_python3',
		sources=BT_PACKETS_BASE_SRCS + BT_PACKETS_PY3_SRCs,
		include_dirs=[GD_DIR,
		BT_PACKETS_GEN_DIR,
		BT_PACKETS_PY3_GEN_DIR,
		PYBIND11_INCLUDE_DIR],
		extra_compile_args=['-std=c++17']
		)
		include_dirs=[
		GD_DIR, BT_PACKETS_GEN_DIR, BT_PACKETS_PY3_GEN_DIR, PYBIND11_INCLUDE_DIR
		],
		extra_compile_args=['-std=c++17'])

		setup(name='bluetooth_packets_python3',
		setup(
		name='bluetooth_packets_python3',
		version='1.0',
		author="Android Open Source Project",
		description="""Bluetooth Packet Library""",

system/gd/cert/event_asserts.py

+41 −35

Original line number	Diff line number	Diff line
		@@ -24,6 +24,7 @@ from google.protobuf import text_format

		from cert.event_callback_stream import EventCallbackStream


		class EventAsserts(object):
		"""
		A class that handles various asserts with respect to a gRPC unary stream
		@@ -62,15 +63,16 @@ class EventAsserts(object):
		logging.debug("assert_none")
		try:
		event = self.event_queue.get(timeout=timeout.seconds)
		asserts.assert_equal(event, None,
		msg=(
		"Expected None, but got %s" % text_format.MessageToString(
		asserts.assert_equal(
		event,
		None,
		msg=("Expected None, but got %s" % text_format.MessageToString(
		event, as_one_line=True)))
		except Empty:
		return

		def assert_none_matching(self, match_fn,
		timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):
		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
		@@ -88,23 +90,24 @@ class EventAsserts(object):
		logging.debug("Waiting for event iteration %d" % iter_count)
		try:
		time_before = datetime.now()
		current_event = self.event_queue.get(
		timeout=timeout_seconds)
		current_event = self.event_queue.get(timeout=timeout_seconds)
		time_elapsed = datetime.now() - time_before
		timeout_seconds -= time_elapsed.seconds
		if match_fn(current_event):
		event = current_event
		except Empty:
		continue
		logging.debug(
		"Done waiting for event, got %s" % text_format.MessageToString(
		event, as_one_line=True))
		asserts.assert_equal(event, None,
		msg=(
		"Expected None matching, but got %s" % text_format.MessageToString(
		event, as_one_line=True)))

		def assert_event_occurs(self, match_fn, at_least_times=1,
		logging.debug("Done waiting for event, got %s" %
		text_format.MessageToString(event, as_one_line=True))
		asserts.assert_equal(
		event,
		None,
		msg=("Expected None matching, but got %s" %
		text_format.MessageToString(event, as_one_line=True)))

		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
		@@ -125,21 +128,24 @@ class EventAsserts(object):
		logging.debug("Waiting for event iteration %d" % iter_count)
		try:
		time_before = datetime.now()
		current_event = self.event_queue.get(
		timeout=timeout_seconds)
		current_event = self.event_queue.get(timeout=timeout_seconds)
		time_elapsed = datetime.now() - time_before
		timeout_seconds -= time_elapsed.seconds
		if match_fn(current_event):
		event.append(current_event)
		except Empty:
		continue
		logging.debug(
		"Done waiting for event, got %s" % text_format.MessageToString(
		event, as_one_line=True))
		asserts.assert_true(len(event) == at_least_times,
		msg=("Expected at least %d events, but got %d" % at_least_times, len(event)))

		def assert_event_occurs_at_most(self, match_fn, at_most_times,
		logging.debug("Done waiting for event, got %s" %
		text_format.MessageToString(event, as_one_line=True))
		asserts.assert_true(
		len(event) == at_least_times,
		msg=("Expected at least %d events, but got %d" % at_least_times,
		len(event)))

		def assert_event_occurs_at_most(
		self,
		match_fn,
		at_most_times,
		timeout=timedelta(seconds=DEFAULT_TIMEOUT_SECONDS)):
		"""
		Assert at most \|at_most_times\| instances of events happen where
		@@ -160,16 +166,16 @@ class EventAsserts(object):
		logging.debug("Waiting for event iteration %d" % iter_count)
		try:
		time_before = datetime.now()
		current_event = self.event_queue.get(
		timeout=timeout_seconds)
		current_event = self.event_queue.get(timeout=timeout_seconds)
		time_elapsed = datetime.now() - time_before
		timeout_seconds -= time_elapsed.seconds
		if match_fn(current_event):
		event.append(current_event)
		except Empty:
		continue
		logging.debug(
		"Done waiting for event, got %s" % text_format.MessageToString(
		event, as_one_line=True))
		asserts.assert_true(len(event) <= at_most_times,
		msg=("Expected at most %d events, but got %d" % at_most_times, len(event)))
		No newline at end of file
		logging.debug("Done waiting for event, got %s" %
		text_format.MessageToString(event, as_one_line=True))
		asserts.assert_true(
		len(event) <= at_most_times,
		msg=("Expected at most %d events, but got %d" % at_most_times,
		len(event)))

system/gd/cert/gd_base_test.py

+13 −10

Original line number	Diff line number	Diff line
		@@ -25,12 +25,17 @@ import subprocess

		ANDROID_BUILD_TOP = os.environ.get('ANDROID_BUILD_TOP')

		sys.path.append(ANDROID_BUILD_TOP + '/out/soong/.intermediates/packages/modules/Bluetooth/system/gd/BluetoothFacadeAndCertGeneratedStub_py/gen')
		sys.path.append(
		ANDROID_BUILD_TOP +
		'/out/soong/.intermediates/packages/modules/Bluetooth/system/gd/BluetoothFacadeAndCertGeneratedStub_py/gen'
		)

		ANDROID_HOST_OUT = os.environ.get('ANDROID_HOST_OUT')
		ROOTCANAL = ANDROID_HOST_OUT + "/nativetest64/root-canal/root-canal"


		class GdBaseTestClass(BaseTestClass):

		def __init__(self, configs):
		BaseTestClass.__init__(self, configs)

		@@ -41,7 +46,8 @@ class GdBaseTestClass(BaseTestClass):
		self.rootcanal_running = False
		if 'rootcanal' in self.controller_configs:
		self.rootcanal_running = True
		rootcanal_logpath = os.path.join(log_path_base, 'rootcanal_logs.txt')
		rootcanal_logpath = os.path.join(log_path_base,
		'rootcanal_logs.txt')
		self.rootcanal_logs = open(rootcanal_logpath, 'w')
		rootcanal_config = self.controller_configs['rootcanal']
		rootcanal_hci_port = str(rootcanal_config.get("hci_port", "6402"))
		@@ -55,19 +61,16 @@ class GdBaseTestClass(BaseTestClass):
		cwd=ANDROID_BUILD_TOP,
		env=os.environ.copy(),
		stdout=self.rootcanal_logs,
		stderr=self.rootcanal_logs
		)
		stderr=self.rootcanal_logs)
		for gd_device in gd_devices:
		gd_device["rootcanal_port"] = rootcanal_hci_port
		for gd_cert_device in gd_cert_devices:
		gd_cert_device["rootcanal_port"] = rootcanal_hci_port

		self.register_controller(
		importlib.import_module('cert.gd_device'),
		builtin=True)
		importlib.import_module('cert.gd_device'), builtin=True)
		self.register_controller(
		importlib.import_module('cert.gd_cert_device'),
		builtin=True)
		importlib.import_module('cert.gd_cert_device'), builtin=True)

		def teardown_class(self):
		if self.rootcanal_running:
		@@ -76,6 +79,6 @@ class GdBaseTestClass(BaseTestClass):
		self.rootcanal_logs.close()
		if rootcanal_return_code != 0 and\
		rootcanal_return_code != -signal.SIGINT:
		logging.error("rootcanal stopped with code: %d" %
		rootcanal_return_code)
		logging.error(
		"rootcanal stopped with code: %d" % rootcanal_return_code)
		return False

system/gd/cert/gd_cert_device.py

+14 −8

Original line number	Diff line number	Diff line
		@@ -25,6 +25,7 @@ from l2cap.classic.cert import api_pb2_grpc as l2cap_cert_pb2_grpc
		ACTS_CONTROLLER_CONFIG_NAME = "GdCertDevice"
		ACTS_CONTROLLER_REFERENCE_NAME = "gd_cert_devices"


		def create(configs):
		if not configs:
		raise GdDeviceConfigError("Configuration is empty")
		@@ -52,20 +53,25 @@ def get_instances_with_configs(configs):
		resolved_cmd = []
		for entry in config["cmd"]:
		resolved_cmd.append(replace_vars(entry, config))
		devices.append(GdCertDevice(config["grpc_port"],
		config["grpc_root_server_port"],
		config["signal_port"],
		resolved_cmd, config["label"]))
		devices.append(
		GdCertDevice(config["grpc_port"], config["grpc_root_server_port"],
		config["signal_port"], resolved_cmd, config["label"]))
		return devices


		class GdCertDevice(GdDeviceBase):
		def __init__(self, grpc_port, grpc_root_server_port, signal_port, cmd, label):

		def __init__(self, grpc_port, grpc_root_server_port, signal_port, cmd,
		label):
		super().__init__(grpc_port, grpc_root_server_port, signal_port, cmd,
		label, ACTS_CONTROLLER_CONFIG_NAME)

		# Cert stubs
		self.rootservice = cert_rootservice_pb2_grpc.RootCertStub(self.grpc_root_server_channel)
		self.rootservice = cert_rootservice_pb2_grpc.RootCertStub(
		self.grpc_root_server_channel)
		self.hal = hal_cert_pb2_grpc.HciHalCertStub(self.grpc_channel)
		self.controller_read_only_property = cert_rootservice_pb2_grpc.ReadOnlyPropertyStub(self.grpc_channel)
		self.controller_read_only_property = cert_rootservice_pb2_grpc.ReadOnlyPropertyStub(
		self.grpc_channel)
		self.hci = hci_cert_pb2_grpc.AclManagerCertStub(self.grpc_channel)
		self.l2cap = l2cap_cert_pb2_grpc.L2capClassicModuleCertStub(self.grpc_channel)
		self.l2cap = l2cap_cert_pb2_grpc.L2capClassicModuleCertStub(
		self.grpc_channel)