Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit 8198497f authored by Jack He's avatar Jack He
Browse files

Python: format all Python files in system/bt using YAPF 

python3 yapf -p -i $(git ls-tree --name-only -r aosp/master | grep "\.py")

Bug: 146016811
Test: run Python tests
Change-Id: Ic5fe6a21151d1abc8eb013f8c8070ba8238a5249
parent cb11407c

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)