Migrate gd cert real device tests to Mobly (9fffdb1e) · Commits · e / os / android_packages_modules_Bluetooth

system/blueberry/tests/gd/cert/adb.py

0 → 100644

+156 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		#
		# Copyright 2016 - 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.

		import encodings
		import logging
		import re
		import shlex
		import shutil

		from mobly.controllers.android_device_lib.adb import AdbProxy

		ROOT_USER_ID = '0'
		SHELL_USER_ID = '2000'
		UTF_8 = encodings.utf_8.getregentry().name


		class BlueberryAdbProxy(AdbProxy):
		"""Proxy class for ADB.

		For syntactic reasons, the '-' in adb commands need to be replaced with
		'_'. Can directly execute adb commands on an object:
		>> adb = BlueberryAdbProxy(<serial>)
		>> adb.start_server()
		>> adb.devices() # will return the console output of "adb devices".
		"""

		def __init__(self, serial="", ssh_connection=None):
		"""Construct an instance of AdbProxy.

		Args:
		serial: str serial number of Android device from `adb devices`
		ssh_connection: SshConnection instance if the Android device is
		connected to a remote host that we can reach via SSH.
		"""
		super().__init__(serial)
		self._server_local_port = None
		adb_path = shutil.which('adb')
		adb_cmd = [shlex.quote(adb_path)]
		if serial:
		adb_cmd.append("-s %s" % serial)
		if ssh_connection is not None:
		# Kill all existing adb processes on the remote host (if any)
		# Note that if there are none, then pkill exits with non-zero status
		ssh_connection.run("pkill adb", ignore_status=True)
		# Copy over the adb binary to a temp dir
		temp_dir = ssh_connection.run("mktemp -d").stdout.strip()
		ssh_connection.send_file(adb_path, temp_dir)
		# Start up a new adb server running as root from the copied binary.
		remote_adb_cmd = "%s/adb %s root" % (temp_dir, "-s %s" % serial if serial else "")
		ssh_connection.run(remote_adb_cmd)
		# Proxy a local port to the adb server port
		local_port = ssh_connection.create_ssh_tunnel(5037)
		self._server_local_port = local_port

		if self._server_local_port:
		adb_cmd.append("-P %d" % local_port)
		self.adb_str = " ".join(adb_cmd)
		self._ssh_connection = ssh_connection

		def get_user_id(self):
		"""Returns the adb user. Either 2000 (shell) or 0 (root)."""
		return self.shell('id -u').decode(UTF_8).rstrip()

		def is_root(self, user_id=None):
		"""Checks if the user is root.

		Args:
		user_id: if supplied, the id to check against.
		Returns:
		True if the user is root. False otherwise.
		"""
		if not user_id:
		user_id = self.get_user_id()
		return user_id == ROOT_USER_ID

		def ensure_root(self):
		"""Ensures the user is root after making this call.

		Note that this will still fail if the device is a user build, as root
		is not accessible from a user build.

		Returns:
		False if the device is a user build. True otherwise.
		"""
		self.ensure_user(ROOT_USER_ID)
		return self.is_root()

		def ensure_user(self, user_id=SHELL_USER_ID):
		"""Ensures the user is set to the given user.

		Args:
		user_id: The id of the user.
		"""
		if self.is_root(user_id):
		self.root()
		else:
		self.unroot()
		self.wait_for_device()
		return self.get_user_id() == user_id

		def tcp_forward(self, host_port, device_port):
		"""Starts tcp forwarding from localhost to this android device.

		Args:
		host_port: Port number to use on localhost
		device_port: Port number to use on the android device.

		Returns:
		Forwarded port on host as int or command output string on error
		"""
		if self._ssh_connection:
		# We have to hop through a remote host first.
		# 1) Find some free port on the remote host's localhost
		# 2) Setup forwarding between that remote port and the requested
		# device port
		remote_port = self._ssh_connection.find_free_port()
		host_port = self._ssh_connection.create_ssh_tunnel(remote_port, local_port=host_port)
		output = self.forward(["tcp:%d" % host_port, "tcp:%d" % device_port])
		# If hinted_port is 0, the output will be the selected port.
		# Otherwise, there will be no output upon successfully
		# forwarding the hinted port.
		if not output:
		return host_port
		try:
		output_int = int(output)
		except ValueError:
		return output
		return output_int

		def remove_tcp_forward(self, host_port):
		"""Stop tcp forwarding a port from localhost to this android device.

		Args:
		host_port: Port number to use on localhost
		"""
		if self._ssh_connection:
		remote_port = self._ssh_connection.close_ssh_tunnel(host_port)
		if remote_port is None:
		logging.warning("Cannot close unknown forwarded tcp port: %d", host_port)
		return
		# The actual port we need to disable via adb is on the remote host.
		host_port = remote_port
		self.forward(["--remove", "tcp:%d" % host_port])

system/blueberry/tests/gd/cert/asserts.py

0 → 100644

+57 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		#
		# Copyright 2018 - 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 mobly.asserts import *


		# Have an instance of unittest.TestCase so we could reuse some logic from
		# python's own unittest.
		# _ProxyTest is required because py2 does not allow instantiating
		# unittest.TestCase directly.
		class _ProxyTest(unittest.TestCase):

		def runTest(self):
		pass


		_pyunit_proxy = _ProxyTest()


		def assert_almost_equal(first, second, places=7, msg=None, delta=None, extras=None):
		"""
		Assert FIRST to be within +/- DELTA to SECOND, otherwise fail the
		test.
		:param first: The first argument, LHS
		:param second: The second argument, RHS
		:param places: For floating points, how many decimal places to look into
		:param msg: Message to display on failure
		:param delta: The +/- first and second could be apart from each other
		:param extras: Extra object passed to test failure handler
		:return:
		"""
		my_msg = None
		try:
		if delta:
		_pyunit_proxy.assertAlmostEqual(first, second, msg=msg, delta=delta)
		else:
		_pyunit_proxy.assertAlmostEqual(first, second, places=places, msg=msg)
		except Exception as e:
		my_msg = str(e)
		if msg:
		my_msg = "%s %s" % (my_msg, msg)
		# This is a hack to remove the stacktrace produced by the above exception.
		if my_msg is not None:
		fail(my_msg, extras=extras)

system/blueberry/tests/gd/cert/gd_base_test.py

+12 −1

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@

		import importlib
		import logging
		import os
		import traceback

		from functools import wraps
		@@ -57,6 +58,10 @@ class GdBaseTestClass(base_test.BaseTestClass):
		generate_coverage_report_for_host(self.cert_coverage_info)
		self.cert_coverage_info = None

		def set_controller_properties_path(self, path):
		GD_DIR = os.path.join(os.getcwd(), os.pardir)
		self.controller_properties_file = os.path.join(GD_DIR, path)

		def setup_test(self):
		append_test_context(test_class_name=self.TAG, test_name=self.current_test_info.name)
		self.log_path_base = get_current_context().get_full_output_path()
		@@ -64,12 +69,18 @@ class GdBaseTestClass(base_test.BaseTestClass):
		for config in self.controller_configs[CONTROLLER_CONFIG_NAME]:
		config['verbose_mode'] = self.verbose_mode

		try:
		controller_properties_file = self.controller_properties_file
		except AttributeError:
		controller_properties_file = ''

		self.info = setup_rootcanal(
		dut_module=self.dut_module,
		cert_module=self.cert_module,
		verbose_mode=self.verbose_mode,
		log_path_base=self.log_path_base,
		controller_configs=self.controller_configs)
		controller_configs=self.controller_configs,
		controller_properties_file=controller_properties_file)
		self.rootcanal_running = self.info['rootcanal_running']
		self.rootcanal_logpath = self.info['rootcanal_logpath']
		self.rootcanal_process = self.info['rootcanal_process']

system/blueberry/tests/gd/cert/gd_device.py

+59 −22

Original line number	Diff line number	Diff line
		@@ -45,13 +45,18 @@ from cert.os_utils import is_subprocess_alive
		from cert.os_utils import make_ports_available
		from cert.os_utils import TerminalColor

		from blueberry.tests.gd.cert import asserts
		from blueberry.tests.gd.cert.adb import BlueberryAdbProxy
		from blueberry.tests.gd.cert.adb import UTF_8
		from blueberry.tests.gd.cert.context import get_current_context

		from mobly import asserts
		from mobly import utils
		from mobly.controllers.android_device_lib.adb import AdbProxy
		from mobly.controllers.android_device_lib.adb import AdbError

		ADB_FILE_NOT_EXIST_ERROR = "No such file or directory"
		PORT_FORWARDING_ERROR_MSG_PREFIX = "During port forwarding cleanup: "
		PULL_LOG_FILE_ERROR_MSG_PREFIX = "While trying to pull log files"


		def create(configs):
		create_core(configs)
		@@ -220,7 +225,7 @@ class GdAndroidDevice(GdDeviceBase):
		super().__init__(grpc_port, grpc_root_server_port, signal_port, cmd, label, type_identifier, name, verbose_mode)
		asserts.assert_true(serial_number, "serial_number must not be None nor empty")
		self.serial_number = serial_number
		self.adb = AdbProxy(serial_number)
		self.adb = BlueberryAdbProxy(serial_number)

		def setup(self):
		logging.info("Setting up device %s %s" % (self.label, self.serial_number))
		@@ -249,21 +254,25 @@ class GdAndroidDevice(GdDeviceBase):
		try:
		self.adb.shell("rm /data/misc/bluetooth/logs/btsnoop_hci.log")
		except AdbError as error:
		if ADB_FILE_NOT_EXIST_ERROR not in str(error):
		logging.error("Error during setup: " + str(error))

		try:
		self.adb.shell("rm /data/misc/bluetooth/logs/btsnooz_hci.log")
		except AdbError as error:
		if ADB_FILE_NOT_EXIST_ERROR not in str(error):
		logging.error("Error during setup: " + str(error))

		try:
		self.adb.shell("rm /data/misc/bluedroid/bt_config.conf")
		except AdbError as error:
		if ADB_FILE_NOT_EXIST_ERROR not in str(error):
		logging.error("Error during setup: " + str(error))

		try:
		self.adb.shell("rm /data/misc/bluedroid/bt_config.bak")
		except AdbError as error:
		if ADB_FILE_NOT_EXIST_ERROR not in str(error):
		logging.error("Error during setup: " + str(error))

		self.ensure_no_output(self.adb.shell("svc bluetooth disable"))
		@@ -309,28 +318,55 @@ class GdAndroidDevice(GdDeviceBase):
		logging.error("logcat_process %s_%s stopped with code: %d" % (self.label, self.serial_number, return_code))
		self.logcat_logger.stop()
		self.cleanup_port_forwarding()
		self.adb.pull("/data/misc/bluetooth/logs/btsnoop_hci.log %s" % os.path.join(self.log_path_base,
		"%s_btsnoop_hci.log" % self.label))
		try:
		self.adb.pull("/data/misc/bluetooth/logs/btsnoop_hci.log %s" % os.path.join(
		self.log_path_base, "%s_btsnoop_hci.log" % self.label))
		except AdbError as error:
		# Some tests have no snoop logs, and that's OK
		if ADB_FILE_NOT_EXIST_ERROR not in str(error):
		logging.error(PULL_LOG_FILE_ERROR_MSG_PREFIX + str(error))
		try:
		self.adb.pull("/data/misc/bluedroid/bt_config.conf %s" % os.path.join(self.log_path_base,
		"%s_bt_config.conf" % self.label))
		self.adb.pull(
		"/data/misc/bluedroid/bt_config.bak %s" % os.path.join(self.log_path_base, "%s_bt_config.bak" % self.label))
		except AdbError as error:
		# Some tests have no config file, and that's OK
		if ADB_FILE_NOT_EXIST_ERROR not in str(error):
		logging.error(PULL_LOG_FILE_ERROR_MSG_PREFIX + str(error))
		try:
		self.adb.pull("/data/misc/bluedroid/bt_config.bak %s" % os.path.join(self.log_path_base,
		"%s_bt_config.bak" % self.label))
		except AdbError as error:
		# Some tests have no config.bak file, and that's OK
		if ADB_FILE_NOT_EXIST_ERROR not in str(error):
		logging.error(PULL_LOG_FILE_ERROR_MSG_PREFIX + str(error))

		def cleanup_port_forwarding(self):
		try:
		self.adb.remove_tcp_forward(self.grpc_port)
		except AdbError as error:
		logging.error("Error during port forwarding cleanup: " + str(error))
		msg = PORT_FORWARDING_ERROR_MSG_PREFIX + str(error)
		if "not found" in msg:
		logging.info(msg)
		else:
		logging.error(msg)

		try:
		self.adb.remove_tcp_forward(self.grpc_root_server_port)
		except AdbError as error:
		logging.error("Error during port forwarding cleanup: " + str(error))
		msg = PORT_FORWARDING_ERROR_MSG_PREFIX + str(error)
		if "not found" in msg:
		logging.info(msg)
		else:
		logging.error(msg)

		try:
		self.adb.reverse("--remove tcp:%d" % self.signal_port)
		self.adb.reverse(["--remove", "tcp:%d" % self.signal_port])
		except AdbError as error:
		logging.error("Error during port forwarding cleanup: " + str(error))
		msg = PORT_FORWARDING_ERROR_MSG_PREFIX + str(error)
		if "not found" in msg:
		logging.info(msg)
		else:
		logging.error(msg)

		@staticmethod
		def ensure_no_output(result):
		@@ -343,7 +379,7 @@ class GdAndroidDevice(GdDeviceBase):
		def sync_device_time(self):
		self.adb.shell("settings put global auto_time 0")
		self.adb.shell("settings put global auto_time_zone 0")
		device_tz = self.adb.shell("date +%z")
		device_tz = self.adb.shell("date +%z").decode(UTF_8).rstrip()
		asserts.assert_true(device_tz, "date +%z must return device timezone, "
		"but returned {} instead".format(device_tz))
		host_tz = time.strftime("%z")
		@@ -361,7 +397,8 @@ class GdAndroidDevice(GdDeviceBase):
		self.adb.shell("date %s" % time.strftime("%m%d%H%M%Y.%S"))
		datetime_format = "%Y-%m-%dT%H:%M:%S%z"
		try:
		device_time = datetime.strptime(self.adb.shell("date +'%s'" % datetime_format), datetime_format)
		device_time = datetime.strptime(
		self.adb.shell("date +'%s'" % datetime_format).decode(UTF_8).rstrip(), datetime_format)
		except ValueError:
		asserts.fail("Failed to get time after sync")
		return
		@@ -383,7 +420,7 @@ class GdAndroidDevice(GdDeviceBase):
		dst_file_path: The destination of the file.
		push_timeout: How long to wait for the push to finish in seconds
		"""
		out = self.adb.push('%s %s' % (src_file_path, dst_file_path), timeout=push_timeout)
		out = self.adb.push([src_file_path, dst_file_path], timeout=push_timeout).decode(UTF_8).rstrip()
		if 'error' in out:
		asserts.fail('Unable to push file %s to %s due to %s' % (src_file_path, dst_file_path, out))

		@@ -421,7 +458,7 @@ class GdAndroidDevice(GdDeviceBase):
		:param num_retry: number of times to reboot and retry this before dying
		:return: device port int
		"""
		error_or_port = self.adb.reverse("tcp:%d tcp:%d" % (device_port, host_port))
		error_or_port = self.adb.reverse(["tcp:%d" % device_port, "tcp:%d" % host_port])
		if not error_or_port:
		logging.debug("device port %d was already reversed" % device_port)
		return device_port
		@@ -482,7 +519,7 @@ class GdAndroidDevice(GdDeviceBase):
		break
		minutes_left = timeout_minutes - (time.time() - timeout_start) / 60.0
		self.wait_for_boot_completion(timeout_minutes=minutes_left)
		asserts.assert_true(self.adb.ensure_root(), "device %s cannot run as root after reboot", self.serial_number)
		asserts.assert_true(self.adb.ensure_root(), "device %s cannot run as root after reboot" % self.serial_number)

		def wait_for_boot_completion(self, timeout_minutes=15.0):
		"""

system/blueberry/tests/gd/devices_config.yaml

0 → 100644

+44 −0

Original line number	Diff line number	Diff line
		_description: Bluetooth cert testing
		TestBeds:
		- Name: AndroidDeviceCert
		Controllers:
		GdDevice:
		- grpc_port: '8898'
		grpc_root_server_port: '8896'
		signal_port: '8894'
		label: cert
		serial_number: 'CERT'
		name: Cert Device
		cmd:
		- "adb"
		- "-s"
		- "$(serial_number)"
		- "shell"
		- "ASAN_OPTIONS=detect_container_overflow=0"
		- "/system/bin/bluetooth_stack_with_facade"
		- "--grpc-port=$(grpc_port)"
		- "--root-server-port=$(grpc_root_server_port)"
		- "--btsnoop=/data/misc/bluetooth/logs/btsnoop_hci.log"
		- "--btsnooz=/data/misc/bluetooth/logs/btsnooz_hci.log"
		- "--btconfig=/data/misc/bluedroid/bt_config.conf"
		- "--signal-port=$(signal_port)"
		- grpc_port: '8899'
		grpc_root_server_port: '8897'
		signal_port: '8895'
		label: dut
		serial_number: 'DUT'
		name: DUT Device
		cmd:
		- "adb"
		- "-s"
		- "$(serial_number)"
		- "shell"
		- "ASAN_OPTIONS=detect_container_overflow=0"
		- "/system/bin/bluetooth_stack_with_facade"
		- "--grpc-port=$(grpc_port)"
		- "--root-server-port=$(grpc_root_server_port)"
		- "--btsnoop=/data/misc/bluetooth/logs/btsnoop_hci.log"
		- "--btsnooz=/data/misc/bluetooth/logs/btsnooz_hci.log"
		- "--btconfig=/data/misc/bluedroid/bt_config.conf"
		- "--signal-port=$(signal_port)"
		logpath: "/tmp/logs"
		No newline at end of file