1#!/usr/bin/env python3 2# 3# Copyright (C) 2020 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16# 17 18import subprocess 19import unittest 20 21def run_cmd(cmd, ignore_error=False): 22 print("Running command:", cmd) 23 p = subprocess.Popen(cmd, shell=True) 24 p.communicate() 25 if not ignore_error and p.returncode: 26 raise subprocess.CalledProcessError(p.returncode, cmd) 27 return p.returncode 28 29class TestFmq(unittest.TestCase): 30 pass 31 32def make_test(client, server, client_filter=None): 33 def test(self): 34 try: 35 run_cmd("adb shell killall %s >/dev/null 2>&1" % client, ignore_error=True) 36 run_cmd("adb shell killall %s >/dev/null 2>&1" % server, ignore_error=True) 37 run_cmd("adb shell \"( %s ) </dev/null >/dev/null 2>&1 &\"" % server) 38 client_cmd = client 39 if client_filter: 40 client_cmd += " --gtest_filter=" + client_filter 41 run_cmd("adb shell %s" % client_cmd) 42 finally: 43 run_cmd("adb shell killall %s >/dev/null 2>&1" % client, ignore_error=True) 44 run_cmd("adb shell killall %s >/dev/null 2>&1" % server, ignore_error=True) 45 return test 46 47def has_bitness(bitness): 48 return 0 == run_cmd("echo '[[ \"$(getprop ro.product.cpu.abilist%s)\" != \"\" ]]' | adb shell sh" % bitness, ignore_error=True) 49 50if __name__ == '__main__': 51 clients = [] 52 servers = [] 53 54 def add_clients_and_servers(clients: list[str], servers: list[str], base: str): 55 clients += [ 56 base + "/fmq_test_client/fmq_test_client", 57 base + "/fmq_rust_test_client/fmq_rust_test_client", 58 ] 59 servers += [base + "/[email protected]/[email protected]"] 60 servers += [base + "/[email protected]/[email protected]"] 61 62 if has_bitness(32): 63 add_clients_and_servers(clients, servers, "/data/nativetest") 64 65 if has_bitness(64): 66 add_clients_and_servers(clients, servers, "/data/nativetest64") 67 68 assert len(clients) > 0 69 assert len(servers) > 0 70 71 def bitness(binary_path: str) -> str: 72 if "64" in binary_path: 73 return "64" 74 return "32" 75 76 def short_name(binary_path: str) -> str: 77 base = "rust" if "rust" in binary_path else "" 78 return base + bitness(binary_path) 79 80 for client in clients: 81 for server in servers: 82 test_name = 'test_%s_to_%s' % (short_name(client), short_name(server)) 83 # Tests in the C++ test client that are fully supported by the Rust test server 84 rust_tests = ":".join([ 85 # Only run AIDL tests 0,1, 3 ,4, not HIDL tests 2 and 5 86 "SynchronizedReadWriteClient/0.*", 87 "SynchronizedReadWriteClient/1.*", 88 "SynchronizedReadWriteClient/3.*", 89 "SynchronizedReadWriteClient/4.*", 90 # Skip blocking tests until the Rust FMQ interface supports them: TODO(b/339999649) 91 "-*Blocking*", 92 ]) 93 # Enable subset of tests if testing C++ client against the rust server 94 gtest_filter = rust_tests if "rust" in server and not "rust" in client else None 95 test = make_test(client, server, gtest_filter) 96 setattr(TestFmq, test_name, test) 97 98 suite = unittest.TestLoader().loadTestsFromTestCase(TestFmq) 99 unittest.TextTestRunner(verbosity=2).run(suite) 100